cli: add timeout support for listen command
[project/ubus.git] / libubus-io.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 #define _GNU_SOURCE
15 #include <sys/types.h>
16 #include <sys/uio.h>
17 #include <sys/socket.h>
18
19 #include <unistd.h>
20 #include <fcntl.h>
21 #include <poll.h>
22
23 #include <libubox/usock.h>
24 #include <libubox/blob.h>
25 #include <libubox/blobmsg.h>
26
27 #include "libubus.h"
28 #include "libubus-internal.h"
29
30 #define STATIC_IOV(_var) { .iov_base = (char *) &(_var), .iov_len = sizeof(_var) }
31
32 #define UBUS_MSGBUF_REDUCTION_INTERVAL  16
33
34 static const struct blob_attr_info ubus_policy[UBUS_ATTR_MAX] = {
35         [UBUS_ATTR_STATUS] = { .type = BLOB_ATTR_INT32 },
36         [UBUS_ATTR_OBJID] = { .type = BLOB_ATTR_INT32 },
37         [UBUS_ATTR_OBJPATH] = { .type = BLOB_ATTR_STRING },
38         [UBUS_ATTR_METHOD] = { .type = BLOB_ATTR_STRING },
39         [UBUS_ATTR_ACTIVE] = { .type = BLOB_ATTR_INT8 },
40         [UBUS_ATTR_NO_REPLY] = { .type = BLOB_ATTR_INT8 },
41         [UBUS_ATTR_SUBSCRIBERS] = { .type = BLOB_ATTR_NESTED },
42 };
43
44 static struct blob_attr *attrbuf[UBUS_ATTR_MAX];
45
46 __hidden struct blob_attr **ubus_parse_msg(struct blob_attr *msg)
47 {
48         blob_parse(msg, attrbuf, ubus_policy, UBUS_ATTR_MAX);
49         return attrbuf;
50 }
51
52 static void wait_data(int fd, bool write)
53 {
54         struct pollfd pfd = { .fd = fd };
55
56         pfd.events = write ? POLLOUT : POLLIN;
57         poll(&pfd, 1, -1);
58 }
59
60 static int writev_retry(int fd, struct iovec *iov, int iov_len, int sock_fd)
61 {
62         static struct {
63                 struct cmsghdr h;
64                 int fd;
65         } fd_buf = {
66                 .h = {
67                         .cmsg_len = sizeof(fd_buf),
68                         .cmsg_level = SOL_SOCKET,
69                         .cmsg_type = SCM_RIGHTS,
70                 }
71         };
72         struct msghdr msghdr = {
73                 .msg_iov = iov,
74                 .msg_iovlen = iov_len,
75                 .msg_control = &fd_buf,
76                 .msg_controllen = sizeof(fd_buf),
77         };
78         int len = 0;
79
80         do {
81                 int cur_len;
82
83                 if (sock_fd < 0) {
84                         msghdr.msg_control = NULL;
85                         msghdr.msg_controllen = 0;
86                 } else {
87                         fd_buf.fd = sock_fd;
88                 }
89
90                 cur_len = sendmsg(fd, &msghdr, 0);
91                 if (cur_len < 0) {
92                         switch(errno) {
93                         case EAGAIN:
94                                 wait_data(fd, true);
95                                 break;
96                         case EINTR:
97                                 break;
98                         default:
99                                 return -1;
100                         }
101                         continue;
102                 }
103
104                 if (len > 0)
105                         sock_fd = -1;
106
107                 len += cur_len;
108                 while (cur_len >= iov->iov_len) {
109                         cur_len -= iov->iov_len;
110                         iov_len--;
111                         iov++;
112                         if (!iov_len)
113                                 return len;
114                 }
115                 iov->iov_base += cur_len;
116                 iov->iov_len -= cur_len;
117                 msghdr.msg_iov = iov;
118                 msghdr.msg_iovlen = iov_len;
119         } while (1);
120
121         /* Should never reach here */
122         return -1;
123 }
124
125 int __hidden ubus_send_msg(struct ubus_context *ctx, uint32_t seq,
126                            struct blob_attr *msg, int cmd, uint32_t peer, int fd)
127 {
128         struct ubus_msghdr hdr;
129         struct iovec iov[2] = {
130                 STATIC_IOV(hdr)
131         };
132         int ret;
133
134         hdr.version = 0;
135         hdr.type = cmd;
136         hdr.seq = cpu_to_be16(seq);
137         hdr.peer = cpu_to_be32(peer);
138
139         if (!msg) {
140                 blob_buf_init(&b, 0);
141                 msg = b.head;
142         }
143
144         iov[1].iov_base = (char *) msg;
145         iov[1].iov_len = blob_raw_len(msg);
146
147         ret = writev_retry(ctx->sock.fd, iov, ARRAY_SIZE(iov), fd);
148         if (ret < 0)
149                 ctx->sock.eof = true;
150
151         if (fd >= 0)
152                 close(fd);
153
154         return ret;
155 }
156
157 static int recv_retry(int fd, struct iovec *iov, bool wait, int *recv_fd)
158 {
159         int bytes, total = 0;
160         static struct {
161                 struct cmsghdr h;
162                 int fd;
163         } fd_buf = {
164                 .h = {
165                         .cmsg_type = SCM_RIGHTS,
166                         .cmsg_level = SOL_SOCKET,
167                         .cmsg_len = sizeof(fd_buf),
168                 },
169         };
170         struct msghdr msghdr = {
171                 .msg_iov = iov,
172                 .msg_iovlen = 1,
173         };
174
175         while (iov->iov_len > 0) {
176                 if (wait)
177                         wait_data(fd, false);
178
179                 if (recv_fd) {
180                         msghdr.msg_control = &fd_buf;
181                         msghdr.msg_controllen = sizeof(fd_buf);
182                 } else {
183                         msghdr.msg_control = NULL;
184                         msghdr.msg_controllen = 0;
185                 }
186
187                 fd_buf.fd = -1;
188                 bytes = recvmsg(fd, &msghdr, 0);
189                 if (!bytes)
190                         return -1;
191
192                 if (bytes < 0) {
193                         bytes = 0;
194                         if (uloop_cancelled)
195                                 return 0;
196                         if (errno == EINTR)
197                                 continue;
198
199                         if (errno != EAGAIN)
200                                 return -1;
201                 }
202                 if (!wait && !bytes)
203                         return 0;
204
205                 if (recv_fd)
206                         *recv_fd = fd_buf.fd;
207
208                 recv_fd = NULL;
209
210                 wait = true;
211                 iov->iov_len -= bytes;
212                 iov->iov_base += bytes;
213                 total += bytes;
214         }
215
216         return total;
217 }
218
219 static bool ubus_validate_hdr(struct ubus_msghdr *hdr)
220 {
221         struct blob_attr *data = (struct blob_attr *) (hdr + 1);
222
223         if (hdr->version != 0)
224                 return false;
225
226         if (blob_raw_len(data) < sizeof(*data))
227                 return false;
228
229         if (blob_pad_len(data) > UBUS_MAX_MSGLEN)
230                 return false;
231
232         return true;
233 }
234
235 static bool alloc_msg_buf(struct ubus_context *ctx, int len)
236 {
237         void *ptr;
238         int buf_len = ctx->msgbuf_data_len;
239         int rem;
240
241         if (!ctx->msgbuf.data)
242                 buf_len = 0;
243
244         rem = (len % UBUS_MSG_CHUNK_SIZE);
245         if (rem > 0)
246                 len += UBUS_MSG_CHUNK_SIZE - rem;
247
248         if (len < buf_len &&
249             ++ctx->msgbuf_reduction_counter > UBUS_MSGBUF_REDUCTION_INTERVAL) {
250                 ctx->msgbuf_reduction_counter = 0;
251                 buf_len = 0;
252         }
253
254         if (len <= buf_len)
255                 return true;
256
257         ptr = realloc(ctx->msgbuf.data, len);
258         if (!ptr)
259                 return false;
260
261         ctx->msgbuf.data = ptr;
262         return true;
263 }
264
265 static bool get_next_msg(struct ubus_context *ctx, int *recv_fd)
266 {
267         struct {
268                 struct ubus_msghdr hdr;
269                 struct blob_attr data;
270         } hdrbuf;
271         struct iovec iov = STATIC_IOV(hdrbuf);
272         int len;
273         int r;
274
275         /* receive header + start attribute */
276         r = recv_retry(ctx->sock.fd, &iov, false, recv_fd);
277         if (r <= 0) {
278                 if (r < 0)
279                         ctx->sock.eof = true;
280
281                 return false;
282         }
283
284         hdrbuf.hdr.seq = be16_to_cpu(hdrbuf.hdr.seq);
285         hdrbuf.hdr.peer = be32_to_cpu(hdrbuf.hdr.peer);
286
287         if (!ubus_validate_hdr(&hdrbuf.hdr))
288                 return false;
289
290         len = blob_raw_len(&hdrbuf.data);
291         if (!alloc_msg_buf(ctx, len))
292                 return false;
293
294         memcpy(&ctx->msgbuf.hdr, &hdrbuf.hdr, sizeof(hdrbuf.hdr));
295         memcpy(ctx->msgbuf.data, &hdrbuf.data, sizeof(hdrbuf.data));
296
297         iov.iov_base = (char *)ctx->msgbuf.data + sizeof(hdrbuf.data);
298         iov.iov_len = blob_len(ctx->msgbuf.data);
299         if (iov.iov_len > 0 &&
300             recv_retry(ctx->sock.fd, &iov, true, NULL) <= 0)
301                 return false;
302
303         return true;
304 }
305
306 void __hidden ubus_handle_data(struct uloop_fd *u, unsigned int events)
307 {
308         struct ubus_context *ctx = container_of(u, struct ubus_context, sock);
309         int recv_fd = -1;
310
311         while (get_next_msg(ctx, &recv_fd)) {
312                 ubus_process_msg(ctx, &ctx->msgbuf, recv_fd);
313                 if (uloop_cancelled)
314                         break;
315         }
316
317         if (u->eof)
318                 ctx->connection_lost(ctx);
319 }
320
321 void __hidden ubus_poll_data(struct ubus_context *ctx, int timeout)
322 {
323         struct pollfd pfd = {
324                 .fd = ctx->sock.fd,
325                 .events = POLLIN | POLLERR,
326         };
327
328         poll(&pfd, 1, timeout ? timeout : -1);
329         ubus_handle_data(&ctx->sock, ULOOP_READ);
330 }
331
332 static void
333 ubus_refresh_state(struct ubus_context *ctx)
334 {
335         struct ubus_object *obj, *tmp;
336         struct ubus_object **objs;
337         int n, i = 0;
338
339         /* clear all type IDs, they need to be registered again */
340         avl_for_each_element(&ctx->objects, obj, avl)
341                 if (obj->type)
342                         obj->type->id = 0;
343
344         /* push out all objects again */
345         objs = alloca(ctx->objects.count * sizeof(*objs));
346         avl_remove_all_elements(&ctx->objects, obj, avl, tmp) {
347                 objs[i++] = obj;
348                 obj->id = 0;
349         }
350
351         for (n = i, i = 0; i < n; i++)
352                 ubus_add_object(ctx, objs[i]);
353 }
354
355 int ubus_reconnect(struct ubus_context *ctx, const char *path)
356 {
357         struct {
358                 struct ubus_msghdr hdr;
359                 struct blob_attr data;
360         } hdr;
361         struct blob_attr *buf;
362         int ret = UBUS_STATUS_UNKNOWN_ERROR;
363
364         if (!path)
365                 path = UBUS_UNIX_SOCKET;
366
367         if (ctx->sock.fd >= 0) {
368                 if (ctx->sock.registered)
369                         uloop_fd_delete(&ctx->sock);
370
371                 close(ctx->sock.fd);
372         }
373
374         ctx->sock.fd = usock(USOCK_UNIX, path, NULL);
375         if (ctx->sock.fd < 0)
376                 return UBUS_STATUS_CONNECTION_FAILED;
377
378         if (read(ctx->sock.fd, &hdr, sizeof(hdr)) != sizeof(hdr))
379                 goto out_close;
380
381         if (!ubus_validate_hdr(&hdr.hdr))
382                 goto out_close;
383
384         if (hdr.hdr.type != UBUS_MSG_HELLO)
385                 goto out_close;
386
387         buf = calloc(1, blob_raw_len(&hdr.data));
388         if (!buf)
389                 goto out_close;
390
391         memcpy(buf, &hdr.data, sizeof(hdr.data));
392         if (read(ctx->sock.fd, blob_data(buf), blob_len(buf)) != blob_len(buf))
393                 goto out_free;
394
395         ctx->local_id = hdr.hdr.peer;
396         if (!ctx->local_id)
397                 goto out_free;
398
399         ret = UBUS_STATUS_OK;
400         fcntl(ctx->sock.fd, F_SETFL, fcntl(ctx->sock.fd, F_GETFL) | O_NONBLOCK | O_CLOEXEC);
401
402         ubus_refresh_state(ctx);
403
404 out_free:
405         free(buf);
406 out_close:
407         if (ret)
408                 close(ctx->sock.fd);
409
410         return ret;
411 }