X-Git-Url: http://git.archive.openwrt.org/?p=project%2Fubus.git;a=blobdiff_plain;f=ubusd.c;h=80e5cfa55d986c977bb8053f006b1fc756ba3d57;hp=798761dc7b51508463374ecd12f0b99dfbd41de8;hb=df1af726e211eaaea870f7d2cbc326e5974b9b09;hpb=d80ebf55afcde808265e4702c162d6b40d649260 diff --git a/ubusd.c b/ubusd.c index 798761d..80e5cfa 100644 --- a/ubusd.c +++ b/ubusd.c @@ -12,8 +12,6 @@ #include "ubusd.h" -static struct avl_tree clients; - static struct ubus_msg_buf *ubus_msg_unshare(struct ubus_msg_buf *ub) { ub = realloc(ub, sizeof(*ub) + ub->len); @@ -26,7 +24,7 @@ static struct ubus_msg_buf *ubus_msg_unshare(struct ubus_msg_buf *ub) return ub; } -struct ubus_msg_buf *ubus_msg_ref(struct ubus_msg_buf *ub) +static struct ubus_msg_buf *ubus_msg_ref(struct ubus_msg_buf *ub) { if (ub->refcount == ~0) return ubus_msg_unshare(ub); @@ -90,43 +88,64 @@ static int ubus_msg_writev(int fd, struct ubus_msg_buf *ub, int offset) } } +static void ubus_msg_enqueue(struct ubus_client *cl, struct ubus_msg_buf *ub) +{ + if (cl->tx_queue[cl->txq_tail]) + return; + + cl->tx_queue[cl->txq_tail] = ubus_msg_ref(ub); + cl->txq_tail = (cl->txq_tail + 1) % ARRAY_SIZE(cl->tx_queue); +} + /* takes the msgbuf reference */ -void ubus_msg_send(struct ubus_client *cl, struct ubus_msg_buf *ub) +void ubus_msg_send(struct ubus_client *cl, struct ubus_msg_buf *ub, bool free) { int written; - if (cl->buf_head) - goto queue; + if (!cl->tx_queue[cl->txq_cur]) { + written = ubus_msg_writev(cl->sock.fd, ub, 0); + if (written >= ub->len + sizeof(ub->hdr)) + goto out; + + if (written < 0) + written = 0; - written = ubus_msg_writev(cl->sock.fd, ub, 0); - if (written > 0 && written < ub->len + sizeof(ub->hdr)) { - cl->buf_head_ofs = written; + cl->txq_ofs = written; /* get an event once we can write to the socket again */ uloop_fd_add(&cl->sock, ULOOP_READ | ULOOP_WRITE | ULOOP_EDGE_TRIGGER); - goto queue; } + ubus_msg_enqueue(cl, ub); - ubus_msg_free(ub); - return; +out: + if (free) + ubus_msg_free(ub); +} -queue: - ub = ubus_msg_unshare(ub); - ub->next = NULL; - *cl->buf_tail = ub; - cl->buf_tail = &ub->next; +static struct ubus_msg_buf *ubus_msg_head(struct ubus_client *cl) +{ + return cl->tx_queue[cl->txq_cur]; } -static void handle_client_disconnect(struct ubus_client *cl) +static void ubus_msg_dequeue(struct ubus_client *cl) { - struct ubus_object *obj; + struct ubus_msg_buf *ub = ubus_msg_head(cl); - while (!list_empty(&cl->objects)) { - obj = list_first_entry(&cl->objects, struct ubus_object, list); - ubusd_free_object(obj); - } + if (!ub) + return; + + ubus_msg_free(ub); + cl->txq_ofs = 0; + cl->tx_queue[cl->txq_cur] = NULL; + cl->txq_cur = (cl->txq_cur + 1) % ARRAY_SIZE(cl->tx_queue); +} + +static void handle_client_disconnect(struct ubus_client *cl) +{ + while (ubus_msg_head(cl)) + ubus_msg_dequeue(cl); - ubus_free_id(&clients, &cl->id); + ubusd_proto_free_client(cl); uloop_fd_delete(&cl->sock); close(cl->sock.fd); free(cl); @@ -138,11 +157,10 @@ static void client_cb(struct uloop_fd *sock, unsigned int events) struct ubus_msg_buf *ub; /* first try to tx more pending data */ - while (cl->buf_head) { - struct ubus_msg_buf *ub = cl->buf_head; + while ((ub = ubus_msg_head(cl))) { int written; - written = ubus_msg_writev(sock->fd, ub, cl->buf_head_ofs); + written = ubus_msg_writev(sock->fd, ub, cl->txq_ofs); if (written < 0) { switch(errno) { case EINTR: @@ -153,22 +171,17 @@ static void client_cb(struct uloop_fd *sock, unsigned int events) } break; } - if (written == 0) - break; - cl->buf_head_ofs += written; - if (cl->buf_head_ofs < ub->len + sizeof(ub->hdr)) + cl->txq_ofs += written; + if (cl->txq_ofs < ub->len + sizeof(ub->hdr)) break; - cl->buf_head_ofs = 0; - cl->buf_head = ub->next; - if (!cl->buf_head) - cl->buf_tail = &cl->buf_head; + ubus_msg_dequeue(cl); } /* prevent further ULOOP_WRITE events if we don't have data * to send anymore */ - if (!cl->buf_head && (events & ULOOP_WRITE)) + if (!ubus_msg_head(cl) && (events & ULOOP_WRITE)) uloop_fd_add(sock, ULOOP_READ | ULOOP_EDGE_TRIGGER); retry: @@ -184,7 +197,7 @@ retry: if (cl->pending_msg_offset < sizeof(cl->hdrbuf)) goto out; - if (blob_len(&cl->hdrbuf.data) + sizeof(cl->hdrbuf) > UBUS_MAX_MSGLEN) + if (blob_pad_len(&cl->hdrbuf.data) > UBUS_MAX_MSGLEN) goto disconnect; cl->pending_msg = ubus_msg_new(NULL, blob_raw_len(&cl->hdrbuf.data), false); @@ -215,29 +228,18 @@ retry: /* accept message */ cl->pending_msg_offset = 0; cl->pending_msg = NULL; - ubusd_receive_message(cl, ub); + ubusd_proto_receive_message(cl, ub); goto retry; } out: - if (!sock->eof || cl->buf_head) + if (!sock->eof || ubus_msg_head(cl)) return; disconnect: handle_client_disconnect(cl); } -struct ubus_client *ubusd_get_client_by_id(uint32_t id) -{ - struct ubus_id *clid; - - clid = ubus_find_id(&clients, id); - if (!clid) - return NULL; - - return container_of(clid, struct ubus_client, id); -} - static bool get_next_connection(int fd) { struct ubus_client *cl; @@ -254,25 +256,12 @@ static bool get_next_connection(int fd) } } - cl = calloc(1, sizeof(*cl)); - cl->sock.fd = client_fd; - - INIT_LIST_HEAD(&cl->objects); - if (!ubus_alloc_id(&clients, &cl->id, 0)) - goto error; - - cl->sock.cb = client_cb; - uloop_fd_add(&cl->sock, ULOOP_READ | ULOOP_EDGE_TRIGGER); - if (!ubusd_send_hello(cl)) - goto error_free; - - return true; + cl = ubusd_proto_new_client(client_fd, client_cb); + if (cl) + uloop_fd_add(&cl->sock, ULOOP_READ | ULOOP_EDGE_TRIGGER); + else + close(client_fd); -error_free: - ubus_free_id(&clients, &cl->id); -error: - close(cl->sock.fd); - free(cl); return true; } @@ -289,18 +278,37 @@ static struct uloop_fd server_fd = { .cb = server_cb, }; +static int usage(const char *progname) +{ + fprintf(stderr, "Usage: %s []\n" + "Options: \n" + " -s : Set the unix domain socket to listen on\n" + "\n", progname); + return 1; +} + int main(int argc, char **argv) { + const char *ubus_socket = UBUS_UNIX_SOCKET; int ret = 0; + int ch; signal(SIGPIPE, SIG_IGN); - ubus_init_id_tree(&clients); - uloop_init(); - unlink(UBUS_UNIX_SOCKET); - server_fd.fd = usock(USOCK_UNIX | USOCK_SERVER | USOCK_NONBLOCK, UBUS_UNIX_SOCKET, NULL); + while ((ch = getopt(argc, argv, "s:")) != -1) { + switch (ch) { + case 's': + ubus_socket = optarg; + break; + default: + return usage(argv[0]); + } + } + + unlink(ubus_socket); + server_fd.fd = usock(USOCK_UNIX | USOCK_SERVER | USOCK_NONBLOCK, ubus_socket, NULL); if (server_fd.fd < 0) { perror("usock"); ret = -1; @@ -309,6 +317,7 @@ int main(int argc, char **argv) uloop_fd_add(&server_fd, ULOOP_READ | ULOOP_EDGE_TRIGGER); uloop_run(); + unlink(ubus_socket); out: uloop_done();