1075c652fd090b62d9e96a10c36c9866b0f0d513
[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 ctx->msgbuf_data_len = len;
263 return true;
264 }
265
266 static bool get_next_msg(struct ubus_context *ctx, int *recv_fd)
267 {
268 struct {
269 struct ubus_msghdr hdr;
270 struct blob_attr data;
271 } hdrbuf;
272 struct iovec iov = STATIC_IOV(hdrbuf);
273 int len;
274 int r;
275
276 /* receive header + start attribute */
277 r = recv_retry(ctx->sock.fd, &iov, false, recv_fd);
278 if (r <= 0) {
279 if (r < 0)
280 ctx->sock.eof = true;
281
282 return false;
283 }
284
285 hdrbuf.hdr.seq = be16_to_cpu(hdrbuf.hdr.seq);
286 hdrbuf.hdr.peer = be32_to_cpu(hdrbuf.hdr.peer);
287
288 if (!ubus_validate_hdr(&hdrbuf.hdr))
289 return false;
290
291 len = blob_raw_len(&hdrbuf.data);
292 if (!alloc_msg_buf(ctx, len))
293 return false;
294
295 memcpy(&ctx->msgbuf.hdr, &hdrbuf.hdr, sizeof(hdrbuf.hdr));
296 memcpy(ctx->msgbuf.data, &hdrbuf.data, sizeof(hdrbuf.data));
297
298 iov.iov_base = (char *)ctx->msgbuf.data + sizeof(hdrbuf.data);
299 iov.iov_len = blob_len(ctx->msgbuf.data);
300 if (iov.iov_len > 0 &&
301 recv_retry(ctx->sock.fd, &iov, true, NULL) <= 0)
302 return false;
303
304 return true;
305 }
306
307 void __hidden ubus_handle_data(struct uloop_fd *u, unsigned int events)
308 {
309 struct ubus_context *ctx = container_of(u, struct ubus_context, sock);
310 int recv_fd = -1;
311
312 while (get_next_msg(ctx, &recv_fd)) {
313 ubus_process_msg(ctx, &ctx->msgbuf, recv_fd);
314 if (uloop_cancelled)
315 break;
316 }
317
318 if (u->eof)
319 ctx->connection_lost(ctx);
320 }
321
322 void __hidden ubus_poll_data(struct ubus_context *ctx, int timeout)
323 {
324 struct pollfd pfd = {
325 .fd = ctx->sock.fd,
326 .events = POLLIN | POLLERR,
327 };
328
329 poll(&pfd, 1, timeout ? timeout : -1);
330 ubus_handle_data(&ctx->sock, ULOOP_READ);
331 }
332
333 static void
334 ubus_refresh_state(struct ubus_context *ctx)
335 {
336 struct ubus_object *obj, *tmp;
337 struct ubus_object **objs;
338 int n, i = 0;
339
340 /* clear all type IDs, they need to be registered again */
341 avl_for_each_element(&ctx->objects, obj, avl)
342 if (obj->type)
343 obj->type->id = 0;
344
345 /* push out all objects again */
346 objs = alloca(ctx->objects.count * sizeof(*objs));
347 avl_remove_all_elements(&ctx->objects, obj, avl, tmp) {
348 objs[i++] = obj;
349 obj->id = 0;
350 }
351
352 for (n = i, i = 0; i < n; i++)
353 ubus_add_object(ctx, objs[i]);
354 }
355
356 int ubus_reconnect(struct ubus_context *ctx, const char *path)
357 {
358 struct {
359 struct ubus_msghdr hdr;
360 struct blob_attr data;
361 } hdr;
362 struct blob_attr *buf;
363 int ret = UBUS_STATUS_UNKNOWN_ERROR;
364
365 if (!path)
366 path = UBUS_UNIX_SOCKET;
367
368 if (ctx->sock.fd >= 0) {
369 if (ctx->sock.registered)
370 uloop_fd_delete(&ctx->sock);
371
372 close(ctx->sock.fd);
373 }
374
375 ctx->sock.eof = false;
376 ctx->sock.fd = usock(USOCK_UNIX, path, NULL);
377 if (ctx->sock.fd < 0)
378 return UBUS_STATUS_CONNECTION_FAILED;
379
380 if (read(ctx->sock.fd, &hdr, sizeof(hdr)) != sizeof(hdr))
381 goto out_close;
382
383 if (!ubus_validate_hdr(&hdr.hdr))
384 goto out_close;
385
386 if (hdr.hdr.type != UBUS_MSG_HELLO)
387 goto out_close;
388
389 buf = calloc(1, blob_raw_len(&hdr.data));
390 if (!buf)
391 goto out_close;
392
393 memcpy(buf, &hdr.data, sizeof(hdr.data));
394 if (read(ctx->sock.fd, blob_data(buf), blob_len(buf)) != blob_len(buf))
395 goto out_free;
396
397 ctx->local_id = hdr.hdr.peer;
398 if (!ctx->local_id)
399 goto out_free;
400
401 ret = UBUS_STATUS_OK;
402 fcntl(ctx->sock.fd, F_SETFL, fcntl(ctx->sock.fd, F_GETFL) | O_NONBLOCK | O_CLOEXEC);
403
404 ubus_refresh_state(ctx);
405
406 out_free:
407 free(buf);
408 out_close:
409 if (ret)
410 close(ctx->sock.fd);
411
412 return ret;
413 }