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