X-Git-Url: http://git.archive.openwrt.org/?p=project%2Fubus.git;a=blobdiff_plain;f=libubus-io.c;h=3280e41a37489ff084624675e76c1966bf25dbab;hp=1f7b0fb6d0076a813eae9a59a6e66a9c32b703cc;hb=b79e808944675040ec2cbc2534592a5d4e62477b;hpb=2b4fc4c9163eacaf95b09773746a3dbb12cfe958 diff --git a/libubus-io.c b/libubus-io.c index 1f7b0fb..3280e41 100644 --- a/libubus-io.c +++ b/libubus-io.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 2011-2012 Felix Fietkau + * Copyright (C) 2011-2014 Felix Fietkau * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU Lesser General Public License version 2.1 @@ -28,6 +28,8 @@ #define STATIC_IOV(_var) { .iov_base = (char *) &(_var), .iov_len = sizeof(_var) } +#define UBUS_MSGBUF_REDUCTION_INTERVAL 16 + static const struct blob_attr_info ubus_policy[UBUS_ATTR_MAX] = { [UBUS_ATTR_STATUS] = { .type = BLOB_ATTR_INT32 }, [UBUS_ATTR_OBJID] = { .type = BLOB_ATTR_INT32 }, @@ -54,12 +56,37 @@ static void wait_data(int fd, bool write) poll(&pfd, 1, 0); } -static int writev_retry(int fd, struct iovec *iov, int iov_len) +static int writev_retry(int fd, struct iovec *iov, int iov_len, int sock_fd) { + static struct { + struct cmsghdr h; + int fd; + } fd_buf = { + .h = { + .cmsg_len = sizeof(fd_buf), + .cmsg_level = SOL_SOCKET, + .cmsg_type = SCM_RIGHTS, + } + }; + struct msghdr msghdr = { + .msg_iov = iov, + .msg_iovlen = iov_len, + .msg_control = &fd_buf, + .msg_controllen = sizeof(fd_buf), + }; int len = 0; do { - int cur_len = writev(fd, iov, iov_len); + int cur_len; + + if (sock_fd < 0) { + msghdr.msg_control = NULL; + msghdr.msg_controllen = 0; + } else { + fd_buf.fd = sock_fd; + } + + cur_len = sendmsg(fd, &msghdr, 0); if (cur_len < 0) { switch(errno) { case EAGAIN: @@ -72,6 +99,10 @@ static int writev_retry(int fd, struct iovec *iov, int iov_len) } continue; } + + if (len > 0) + sock_fd = -1; + len += cur_len; while (cur_len >= iov->iov_len) { cur_len -= iov->iov_len; @@ -80,7 +111,10 @@ static int writev_retry(int fd, struct iovec *iov, int iov_len) if (!iov_len) return len; } + iov->iov_base += cur_len; iov->iov_len -= cur_len; + msghdr.msg_iov = iov; + msghdr.msg_iovlen = iov_len; } while (1); /* Should never reach here */ @@ -88,12 +122,13 @@ static int writev_retry(int fd, struct iovec *iov, int iov_len) } int __hidden ubus_send_msg(struct ubus_context *ctx, uint32_t seq, - struct blob_attr *msg, int cmd, uint32_t peer) + struct blob_attr *msg, int cmd, uint32_t peer, int fd) { struct ubus_msghdr hdr; struct iovec iov[2] = { STATIC_IOV(hdr) }; + int ret; hdr.version = 0; hdr.type = cmd; @@ -108,76 +143,155 @@ int __hidden ubus_send_msg(struct ubus_context *ctx, uint32_t seq, iov[1].iov_base = (char *) msg; iov[1].iov_len = blob_raw_len(msg); - return writev_retry(ctx->sock.fd, iov, ARRAY_SIZE(iov)); + ret = writev_retry(ctx->sock.fd, iov, ARRAY_SIZE(iov), fd); + if (ret < 0) + ctx->sock.eof = true; + + if (fd >= 0) + close(fd); + + return ret; } -static bool recv_retry(int fd, struct iovec *iov, bool wait) +static int recv_retry(int fd, struct iovec *iov, bool wait, int *recv_fd) { - int bytes; + int bytes, total = 0; + static struct { + struct cmsghdr h; + int fd; + } fd_buf = { + .h = { + .cmsg_type = SCM_RIGHTS, + .cmsg_level = SOL_SOCKET, + .cmsg_len = sizeof(fd_buf), + }, + }; + struct msghdr msghdr = { + .msg_iov = iov, + .msg_iovlen = 1, + }; while (iov->iov_len > 0) { if (wait) wait_data(fd, false); - bytes = read(fd, iov->iov_base, iov->iov_len); + if (recv_fd) { + msghdr.msg_control = &fd_buf; + msghdr.msg_controllen = sizeof(fd_buf); + } else { + msghdr.msg_control = NULL; + msghdr.msg_controllen = 0; + } + + fd_buf.fd = -1; + bytes = recvmsg(fd, &msghdr, 0); + if (!bytes) + return -1; + if (bytes < 0) { bytes = 0; if (uloop_cancelled) - return false; + return 0; if (errno == EINTR) continue; if (errno != EAGAIN) - return false; + return -1; } if (!wait && !bytes) - return false; + return 0; + + if (recv_fd) + *recv_fd = fd_buf.fd; + + recv_fd = NULL; wait = true; iov->iov_len -= bytes; iov->iov_base += bytes; + total += bytes; } - return true; + return total; } static bool ubus_validate_hdr(struct ubus_msghdr *hdr) { + struct blob_attr *data = (struct blob_attr *) (hdr + 1); + if (hdr->version != 0) return false; - if (blob_raw_len(hdr->data) < sizeof(*hdr->data)) + if (blob_raw_len(data) < sizeof(*data)) return false; - if (blob_pad_len(hdr->data) > UBUS_MAX_MSGLEN) + if (blob_pad_len(data) > UBUS_MAX_MSGLEN) return false; return true; } -static bool get_next_msg(struct ubus_context *ctx) +static bool get_next_msg(struct ubus_context *ctx, int *recv_fd) { - struct iovec iov = STATIC_IOV(ctx->msgbuf.hdr); + struct { + struct ubus_msghdr hdr; + struct blob_attr data; + } hdrbuf; + struct iovec iov = STATIC_IOV(hdrbuf); + int len; + int r; /* receive header + start attribute */ - iov.iov_len += sizeof(struct blob_attr); - if (!recv_retry(ctx->sock.fd, &iov, false)) + r = recv_retry(ctx->sock.fd, &iov, false, recv_fd); + if (r <= 0) { + if (r < 0) + ctx->sock.eof = true; + + return false; + } + + if (!ubus_validate_hdr(&hdrbuf.hdr)) + return false; + + len = blob_raw_len(&hdrbuf.data); + if (len > ctx->msgbuf_data_len) { + ctx->msgbuf_reduction_counter = UBUS_MSGBUF_REDUCTION_INTERVAL; + } else if (ctx->msgbuf_data_len > UBUS_MSG_CHUNK_SIZE) { + if (ctx->msgbuf_reduction_counter > 0) { + len = -1; + --ctx->msgbuf_reduction_counter; + } else + len = UBUS_MSG_CHUNK_SIZE; + } else + len = -1; + + if (len > -1) { + ctx->msgbuf.data = realloc(ctx->msgbuf.data, len * sizeof(char)); + if (ctx->msgbuf.data) + ctx->msgbuf_data_len = len; + } + if (!ctx->msgbuf.data) return false; - iov.iov_len = blob_len(ctx->msgbuf.hdr.data); - if (iov.iov_len > 0 && !recv_retry(ctx->sock.fd, &iov, true)) + memcpy(&ctx->msgbuf.hdr, &hdrbuf.hdr, sizeof(hdrbuf.hdr)); + memcpy(ctx->msgbuf.data, &hdrbuf.data, sizeof(hdrbuf.data)); + + iov.iov_base = (char *)ctx->msgbuf.data + sizeof(hdrbuf.data); + iov.iov_len = blob_len(ctx->msgbuf.data); + if (iov.iov_len > 0 && !recv_retry(ctx->sock.fd, &iov, true, NULL)) return false; - return ubus_validate_hdr(&ctx->msgbuf.hdr); + return true; } void __hidden ubus_handle_data(struct uloop_fd *u, unsigned int events) { struct ubus_context *ctx = container_of(u, struct ubus_context, sock); struct ubus_msghdr *hdr = &ctx->msgbuf.hdr; + int recv_fd = -1; - while (get_next_msg(ctx)) { - ubus_process_msg(ctx, hdr); + while (get_next_msg(ctx, &recv_fd)) { + ubus_process_msg(ctx, hdr, recv_fd); if (uloop_cancelled) break; } @@ -186,10 +300,23 @@ void __hidden ubus_handle_data(struct uloop_fd *u, unsigned int events) ctx->connection_lost(ctx); } +void __hidden ubus_poll_data(struct ubus_context *ctx, int timeout) +{ + struct pollfd pfd = { + .fd = ctx->sock.fd, + .events = POLLIN | POLLERR, + }; + + poll(&pfd, 1, timeout); + ubus_handle_data(&ctx->sock, ULOOP_READ); +} + static void ubus_refresh_state(struct ubus_context *ctx) { struct ubus_object *obj, *tmp; + struct ubus_object **objs; + int n, i = 0; /* clear all type IDs, they need to be registered again */ avl_for_each_element(&ctx->objects, obj, avl) @@ -197,11 +324,14 @@ ubus_refresh_state(struct ubus_context *ctx) obj->type->id = 0; /* push out all objects again */ - avl_for_each_element_safe(&ctx->objects, obj, avl, tmp) { + objs = alloca(ctx->objects.count * sizeof(*objs)); + avl_remove_all_elements(&ctx->objects, obj, avl, tmp) { + objs[i++] = obj; obj->id = 0; - avl_delete(&ctx->objects, &obj->avl); - ubus_add_object(ctx, obj); } + + for (n = i, i = 0; i < n; i++) + ubus_add_object(ctx, objs[i]); } int ubus_reconnect(struct ubus_context *ctx, const char *path) @@ -249,7 +379,7 @@ int ubus_reconnect(struct ubus_context *ctx, const char *path) goto out_free; ret = UBUS_STATUS_OK; - fcntl(ctx->sock.fd, F_SETFL, fcntl(ctx->sock.fd, F_GETFL) | O_NONBLOCK); + fcntl(ctx->sock.fd, F_SETFL, fcntl(ctx->sock.fd, F_GETFL) | O_NONBLOCK | O_CLOEXEC); ubus_refresh_state(ctx);