• source navigation  • diff markup  • identifier search  • freetext search  • 

Sources/ubus/libubus-req.c

  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