1 /* 2 * Copyright (C) 2011-2014 Felix Fietkau <nbd@openwrt.org> 3 * 4 * This program is free software; you can redistribute it and/or modify 5 * it under the terms of the GNU Lesser General Public License version 2.1 6 * as published by the Free Software Foundation 7 * 8 * This program is distributed in the hope that it will be useful, 9 * but WITHOUT ANY WARRANTY; without even the implied warranty of 10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 11 * GNU General Public License for more details. 12 */ 13 14 #include <unistd.h> 15 #include "libubus.h" 16 #include "libubus-internal.h" 17 18 struct ubus_pending_data { 19 struct list_head list; 20 int type; 21 struct blob_attr data[]; 22 }; 23 24 static void req_data_cb(struct ubus_request *req, int type, struct blob_attr *data) 25 { 26 struct blob_attr **attr; 27 28 if (req->raw_data_cb) 29 req->raw_data_cb(req, type, data); 30 31 if (!req->data_cb) 32 return; 33 34 attr = ubus_parse_msg(data, blob_raw_len(data)); 35 if (!attr[UBUS_ATTR_DATA]) 36 return; 37 38 req->data_cb(req, type, attr[UBUS_ATTR_DATA]); 39 } 40 41 static void __ubus_process_req_data(struct ubus_request *req) 42 { 43 struct ubus_pending_data *data, *tmp; 44 45 list_for_each_entry_safe(data, tmp, &req->pending, list) { 46 list_del(&data->list); 47 if (!req->cancelled) 48 req_data_cb(req, data->type, data->data); 49 free(data); 50 } 51 } 52 53 int __hidden __ubus_start_request(struct ubus_context *ctx, struct ubus_request *req, 54 struct blob_attr *msg, int cmd, uint32_t peer) 55 { 56 57 if (msg && blob_pad_len(msg) > UBUS_MAX_MSGLEN) { 58 /* 59 * ubus_send_msg() takes ownership of req->fd and closes it on 60 * every path; bailing out before it runs would otherwise leak 61 * the caller-supplied fd. 62 */ 63 if (req->fd >= 0) 64 close(req->fd); 65 return -1; 66 } 67 68 INIT_LIST_HEAD(&req->list); 69 INIT_LIST_HEAD(&req->pending); 70 req->ctx = ctx; 71 req->peer = peer; 72 req->seq = ++ctx->request_seq; 73 74 return ubus_send_msg(ctx, req->seq, msg, cmd, peer, req->fd); 75 } 76 77 int __hidden ubus_start_request(struct ubus_context *ctx, struct ubus_request *req, 78 struct blob_attr *msg, int cmd, uint32_t peer) 79 { 80 memset(req, 0, sizeof(*req)); 81 82 req->fd = -1; 83 84 return __ubus_start_request(ctx, req, msg, cmd, peer); 85 } 86 87 88 void ubus_abort_request(struct ubus_context *ctx, struct ubus_request *req) 89 { 90 if (list_empty(&req->list)) 91 return; 92 93 req->cancelled = true; 94 __ubus_process_req_data(req); 95 list_del_init(&req->list); 96 } 97 98 void ubus_complete_request_async(struct ubus_context *ctx, struct ubus_request *req) 99 { 100 if (!list_empty(&req->list)) 101 return; 102 103 list_add(&req->list, &ctx->requests); 104 } 105 106 static void 107 ubus_req_complete_cb(struct ubus_request *req) 108 { 109 ubus_complete_handler_t cb = req->complete_cb; 110 111 if (!cb) 112 return; 113 114 req->complete_cb = NULL; 115 cb(req, req->status_code); 116 } 117 118 static void 119 ubus_set_req_status(struct ubus_request *req, int ret) 120 { 121 if (!list_empty(&req->list)) 122 list_del_init(&req->list); 123 124 req->status_msg = true; 125 req->status_code = ret; 126 if (!req->blocked) 127 ubus_req_complete_cb(req); 128 } 129 130 static void ubus_sync_req_cb(struct ubus_request *req, int ret) 131 { 132 req->status_msg = true; 133 req->status_code = ret; 134 req->ctx->cancel_poll = true; 135 } 136 137 static int64_t get_time_msec(void) 138 { 139 struct timespec ts; 140 int64_t val; 141 142 clock_gettime(CLOCK_MONOTONIC, &ts); 143 val = (int64_t) ts.tv_sec * 1000LL; 144 val += ts.tv_nsec / 1000000LL; 145 return val; 146 } 147 148 void ubus_flush_requests(struct ubus_context *ctx) 149 { 150 struct ubus_request *req; 151 152 while (!list_empty(&ctx->requests)) { 153 req = list_first_entry(&ctx->requests, struct ubus_request, list); 154 ubus_set_req_status(req, UBUS_STATUS_TIMEOUT); 155 } 156 } 157 158 int ubus_complete_request(struct ubus_context *ctx, struct ubus_request *req, 159 int req_timeout) 160 { 161 ubus_complete_handler_t complete_cb = req->complete_cb; 162 int status = UBUS_STATUS_NO_DATA; 163 int64_t timeout = 0, time_end = 0; 164 165 if (req_timeout) 166 time_end = get_time_msec() + req_timeout; 167 168 ubus_complete_request_async(ctx, req); 169 req->complete_cb = ubus_sync_req_cb; 170 171 ctx->stack_depth++; 172 while (!req->status_msg) { 173 if (req_timeout) { 174 timeout = time_end - get_time_msec(); 175 if (timeout <= 0) { 176 ubus_set_req_status(req, UBUS_STATUS_TIMEOUT); 177 break; 178 } 179 } 180 181 ubus_poll_data(ctx, (unsigned int) timeout); 182 183 if (ctx->sock.eof) { 184 ubus_set_req_status(req, UBUS_STATUS_CONNECTION_FAILED); 185 ctx->cancel_poll = true; 186 break; 187 } 188 } 189 190 ctx->stack_depth--; 191 if (ctx->stack_depth) 192 ctx->cancel_poll = true; 193 194 if (req->status_msg) 195 status = req->status_code; 196 197 req->complete_cb = complete_cb; 198 if (req->complete_cb) 199 req->complete_cb(req, status); 200 201 if (!ctx->stack_depth && !ctx->sock.registered) 202 ctx->pending_timer.cb(&ctx->pending_timer); 203 204 return status; 205 } 206 207 void ubus_complete_deferred_request(struct ubus_context *ctx, struct ubus_request_data *req, int ret) 208 { 209 blob_buf_init(&b, 0); 210 blob_put_int32(&b, UBUS_ATTR_STATUS, ret); 211 blob_put_int32(&b, UBUS_ATTR_OBJID, req->object); 212 ubus_send_msg(ctx, req->seq, b.head, UBUS_MSG_STATUS, req->peer, req->fd); 213 } 214 215 static void ubus_put_data(struct blob_buf *buf, struct blob_attr *msg) 216 { 217 if (msg) 218 blob_put(buf, UBUS_ATTR_DATA, blob_data(msg), blob_len(msg)); 219 else 220 blob_put(buf, UBUS_ATTR_DATA, NULL, 0); 221 } 222 223 int ubus_send_reply(struct ubus_context *ctx, struct ubus_request_data *req, 224 struct blob_attr *msg) 225 { 226 int ret; 227 228 blob_buf_init(&b, 0); 229 blob_put_int32(&b, UBUS_ATTR_OBJID, req->object); 230 ubus_put_data(&b, msg); 231 ret = ubus_send_msg(ctx, req->seq, b.head, UBUS_MSG_DATA, req->peer, -1); 232 if (ret < 0) 233 return UBUS_STATUS_NO_DATA; 234 235 return 0; 236 } 237 238 int ubus_invoke_async_fd(struct ubus_context *ctx, uint32_t obj, 239 const char *method, struct blob_attr *msg, 240 struct ubus_request *req, int fd) 241 { 242 blob_buf_init(&b, 0); 243 blob_put_int32(&b, UBUS_ATTR_OBJID, obj); 244 blob_put_string(&b, UBUS_ATTR_METHOD, method); 245 ubus_put_data(&b, msg); 246 247 memset(req, 0, sizeof(*req)); 248 req->fd = fd; 249 if (__ubus_start_request(ctx, req, b.head, UBUS_MSG_INVOKE, obj) < 0) 250 return UBUS_STATUS_INVALID_ARGUMENT; 251 return 0; 252 } 253 254 int ubus_invoke_fd(struct ubus_context *ctx, uint32_t obj, const char *method, 255 struct blob_attr *msg, ubus_data_handler_t cb, void *priv, 256 int timeout, int fd) 257 { 258 struct ubus_request req; 259 int rc; 260 261 rc = ubus_invoke_async_fd(ctx, obj, method, msg, &req, fd); 262 if (rc) 263 return rc; 264 265 req.data_cb = cb; 266 req.priv = priv; 267 return ubus_complete_request(ctx, &req, timeout); 268 } 269 270 static void 271 ubus_notify_complete_cb(struct ubus_request *req, int ret) 272 { 273 struct ubus_notify_request *nreq; 274 275 nreq = container_of(req, struct ubus_notify_request, req); 276 if (!nreq->complete_cb) 277 return; 278 279 nreq->complete_cb(nreq, 0, 0); 280 } 281 282 static void 283 ubus_notify_data_cb(struct ubus_request *req, int type, struct blob_attr *msg) 284 { 285 struct ubus_notify_request *nreq; 286 287 nreq = container_of(req, struct ubus_notify_request, req); 288 if (!nreq->data_cb) 289 return; 290 291 nreq->data_cb(nreq, type, msg); 292 } 293 294 static int 295 __ubus_notify_async(struct ubus_context *ctx, struct ubus_object *obj, 296 const char *type, struct blob_attr *msg, 297 struct ubus_notify_request *req, bool reply) 298 { 299 if (ubus_context_is_channel(ctx)) 300 return UBUS_STATUS_INVALID_ARGUMENT; 301 302 memset(req, 0, sizeof(*req)); 303 304 blob_buf_init(&b, 0); 305 blob_put_int32(&b, UBUS_ATTR_OBJID, obj->id); 306 blob_put_string(&b, UBUS_ATTR_METHOD, type); 307 ubus_put_data(&b, msg); 308 309 if (!reply) 310 blob_put_int8(&b, UBUS_ATTR_NO_REPLY, true); 311 312 if (ubus_start_request(ctx, &req->req, b.head, UBUS_MSG_NOTIFY, obj->id) < 0) 313 return UBUS_STATUS_INVALID_ARGUMENT; 314 315 /* wait for status message from ubusd first */ 316 req->req.notify = true; 317 req->pending = 1; 318 req->id[0] = obj->id; 319 req->req.complete_cb = ubus_notify_complete_cb; 320 req->req.data_cb = ubus_notify_data_cb; 321 322 return 0; 323 } 324 325 int ubus_notify_async(struct ubus_context *ctx, struct ubus_object *obj, 326 const char *type, struct blob_attr *msg, 327 struct ubus_notify_request *req) 328 { 329 return __ubus_notify_async(ctx, obj, type, msg, req, true); 330 } 331 332 int ubus_notify(struct ubus_context *ctx, struct ubus_object *obj, 333 const char *type, struct blob_attr *msg, int timeout) 334 { 335 struct ubus_notify_request req; 336 int ret; 337 338 ret = __ubus_notify_async(ctx, obj, type, msg, &req, timeout >= 0); 339 if (ret) 340 return ret; 341 342 if (timeout < 0) { 343 ubus_abort_request(ctx, &req.req); 344 return 0; 345 } 346 347 return ubus_complete_request(ctx, &req.req, timeout); 348 } 349 350 static bool ubus_get_status(struct ubus_msghdr_buf *buf, int *ret) 351 { 352 struct blob_attr **attrbuf = ubus_parse_msg(buf->data, blob_raw_len(buf->data)); 353 354 if (!attrbuf[UBUS_ATTR_STATUS]) 355 return false; 356 357 *ret = blob_get_u32(attrbuf[UBUS_ATTR_STATUS]); 358 return true; 359 } 360 361 static int 362 ubus_process_req_status(struct ubus_request *req, struct ubus_msghdr_buf *buf) 363 { 364 int ret = UBUS_STATUS_INVALID_ARGUMENT; 365 366 ubus_get_status(buf, &ret); 367 req->peer = buf->hdr.peer; 368 ubus_set_req_status(req, ret); 369 370 return ret; 371 } 372 373 static void 374 ubus_process_req_data(struct ubus_request *req, struct ubus_msghdr_buf *buf) 375 { 376 struct ubus_pending_data *data; 377 int len; 378 379 if (!req->blocked) { 380 req->blocked = true; 381 req_data_cb(req, buf->hdr.type, buf->data); 382 __ubus_process_req_data(req); 383 req->blocked = false; 384 385 if (req->status_msg) 386 ubus_req_complete_cb(req); 387 388 return; 389 } 390 391 len = blob_raw_len(buf->data); 392 data = calloc(1, sizeof(*data) + len); 393 if (!data) 394 return; 395 396 data->type = buf->hdr.type; 397 memcpy(data->data, buf->data, len); 398 list_add(&data->list, &req->pending); 399 } 400 401 static int 402 ubus_find_notify_id(struct ubus_notify_request *n, uint32_t objid) 403 { 404 uint32_t pending = n->pending; 405 int i; 406 407 for (i = 0; pending; i++, pending >>= 1) { 408 if (!(pending & 1)) 409 continue; 410 411 if (n->id[i] == objid) 412 return i; 413 } 414 415 return -1; 416 } 417 418 static struct ubus_request * 419 ubus_find_request(struct ubus_context *ctx, uint32_t seq, uint32_t peer, int *id) 420 { 421 struct ubus_request *req; 422 423 list_for_each_entry(req, &ctx->requests, list) { 424 struct ubus_notify_request *nreq; 425 nreq = container_of(req, struct ubus_notify_request, req); 426 427 if (seq != req->seq) 428 continue; 429 430 if (req->notify) { 431 if (!nreq->pending) 432 continue; 433 434 *id = ubus_find_notify_id(nreq, peer); 435 if (*id < 0) 436 continue; 437 } else if (peer != req->peer) 438 continue; 439 440 return req; 441 } 442 return NULL; 443 } 444 445 static void ubus_process_notify_status(struct ubus_request *req, int id, struct ubus_msghdr_buf *buf) 446 { 447 struct ubus_notify_request *nreq; 448 struct blob_attr **tb; 449 struct blob_attr *cur; 450 size_t rem; 451 int idx = 1; 452 int ret = 0; 453 454 nreq = container_of(req, struct ubus_notify_request, req); 455 nreq->pending &= ~(1 << id); 456 457 if (!id) { 458 /* first id: ubusd's status message with a list of ids */ 459 tb = ubus_parse_msg(buf->data, blob_raw_len(buf->data)); 460 if (tb[UBUS_ATTR_SUBSCRIBERS]) { 461 blob_for_each_attr(cur, tb[UBUS_ATTR_SUBSCRIBERS], rem) { 462 if (!blob_check_type(blob_data(cur), blob_len(cur), BLOB_ATTR_INT32)) 463 continue; 464 465 nreq->pending |= (1 << idx); 466 nreq->id[idx] = blob_get_int32(cur); 467 idx++; 468 469 if (idx == UBUS_MAX_NOTIFY_PEERS + 1) 470 break; 471 } 472 } 473 } else { 474 ubus_get_status(buf, &ret); 475 if (nreq->status_cb) 476 nreq->status_cb(nreq, id, ret); 477 } 478 479 if (!nreq->pending) 480 ubus_set_req_status(req, 0); 481 } 482 483 void __hidden ubus_process_req_msg(struct ubus_context *ctx, struct ubus_msghdr_buf *buf, int fd) 484 { 485 struct ubus_msghdr *hdr = &buf->hdr; 486 struct ubus_request *req; 487 int id = -1; 488 489 switch(hdr->type) { 490 case UBUS_MSG_STATUS: 491 req = ubus_find_request(ctx, hdr->seq, hdr->peer, &id); 492 if (!req) 493 break; 494 495 if (fd >= 0) { 496 if (req->fd_cb) { 497 req->fd_cb(req, fd); 498 fd = -1; 499 } 500 } 501 502 if (id >= 0) 503 ubus_process_notify_status(req, id, buf); 504 else 505 ubus_process_req_status(req, buf); 506 break; 507 508 case UBUS_MSG_DATA: 509 req = ubus_find_request(ctx, hdr->seq, hdr->peer, &id); 510 if (req && (req->data_cb || req->raw_data_cb)) 511 ubus_process_req_data(req, buf); 512 break; 513 } 514 515 if (fd >= 0) 516 close(fd); 517 } 518 519 int __ubus_monitor(struct ubus_context *ctx, const char *type) 520 { 521 if (ubus_context_is_channel(ctx)) 522 return UBUS_STATUS_INVALID_ARGUMENT; 523 524 blob_buf_init(&b, 0); 525 return ubus_invoke(ctx, UBUS_SYSTEM_OBJECT_MONITOR, type, b.head, NULL, NULL, 1000); 526 } 527
This page was automatically generated by LXR 0.3.1. • OpenWrt