fix message buffering
This commit is contained in:
Felix Fietkau
parent
f6a6b0d492
commit
a84c6cac9a
3 changed files with 64 additions and 46 deletions
74
ubusd.c
74
ubusd.c
|
|
@ -90,31 +90,56 @@ 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] = 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;
|
||||
|
||||
written = ubus_msg_writev(cl->sock.fd, ub, 0);
|
||||
if (written > 0 && written < ub->len + sizeof(ub->hdr)) {
|
||||
cl->buf_head_ofs = written;
|
||||
if (written < 0)
|
||||
written = 0;
|
||||
|
||||
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);
|
||||
|
||||
out:
|
||||
if (free)
|
||||
ubus_msg_free(ub);
|
||||
}
|
||||
|
||||
static struct ubus_msg_buf *ubus_msg_head(struct ubus_client *cl)
|
||||
{
|
||||
return cl->tx_queue[cl->txq_cur];
|
||||
}
|
||||
|
||||
static void ubus_msg_dequeue(struct ubus_client *cl)
|
||||
{
|
||||
struct ubus_msg_buf *ub = ubus_msg_head(cl);
|
||||
|
||||
if (!ub)
|
||||
return;
|
||||
|
||||
ubus_msg_free(ub);
|
||||
return;
|
||||
|
||||
queue:
|
||||
ub = ubus_msg_unshare(ub);
|
||||
ub->next = NULL;
|
||||
*cl->buf_tail = ub;
|
||||
cl->buf_tail = &ub->next;
|
||||
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)
|
||||
|
|
@ -126,6 +151,9 @@ static void handle_client_disconnect(struct ubus_client *cl)
|
|||
ubusd_free_object(obj);
|
||||
}
|
||||
|
||||
while (ubus_msg_head(cl))
|
||||
ubus_msg_dequeue(cl);
|
||||
|
||||
ubus_free_id(&clients, &cl->id);
|
||||
uloop_fd_delete(&cl->sock);
|
||||
close(cl->sock.fd);
|
||||
|
|
@ -138,11 +166,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:
|
||||
|
|
@ -156,19 +183,16 @@ static void client_cb(struct uloop_fd *sock, unsigned int events)
|
|||
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:
|
||||
|
|
@ -220,7 +244,7 @@ retry:
|
|||
}
|
||||
|
||||
out:
|
||||
if (!sock->eof || cl->buf_head)
|
||||
if (!sock->eof || ubus_msg_head(cl))
|
||||
return;
|
||||
|
||||
disconnect:
|
||||
|
|
|
|||
22
ubusd.h
22
ubusd.h
|
|
@ -14,7 +14,6 @@
|
|||
#define UBUS_OBJ_HASH_BITS 4
|
||||
|
||||
struct ubus_msg_buf {
|
||||
struct ubus_msg_buf *next;
|
||||
uint32_t refcount; /* ~0: uses external data buffer */
|
||||
struct ubus_msghdr hdr;
|
||||
struct blob_attr *data;
|
||||
|
|
@ -25,22 +24,17 @@ struct ubus_client {
|
|||
struct ubus_id id;
|
||||
struct uloop_fd sock;
|
||||
|
||||
struct list_head objects;
|
||||
|
||||
struct ubus_msg_buf *tx_queue[UBUSD_CLIENT_BACKLOG];
|
||||
unsigned int txq_cur, txq_tail, txq_ofs;
|
||||
|
||||
struct ubus_msg_buf *pending_msg;
|
||||
int pending_msg_offset;
|
||||
struct {
|
||||
struct ubus_msghdr hdr;
|
||||
struct blob_attr data;
|
||||
} hdrbuf;
|
||||
|
||||
struct list_head objects;
|
||||
|
||||
int pending_msg_offset;
|
||||
struct ubus_msg_buf *pending_msg;
|
||||
|
||||
unsigned int buf_head_ofs;
|
||||
struct ubus_msg_buf *buf_head;
|
||||
struct ubus_msg_buf **buf_tail;
|
||||
|
||||
struct ubus_msg_buf *requests[UBUSD_CLIENT_BACKLOG];
|
||||
unsigned int req_head, req_tail;
|
||||
};
|
||||
|
||||
struct ubus_path {
|
||||
|
|
@ -49,7 +43,7 @@ struct ubus_path {
|
|||
};
|
||||
|
||||
struct ubus_msg_buf *ubus_msg_new(void *data, int len, bool shared);
|
||||
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);
|
||||
struct ubus_msg_buf *ubus_msg_ref(struct ubus_msg_buf *ub);
|
||||
void ubus_msg_free(struct ubus_msg_buf *ub);
|
||||
|
||||
|
|
|
|||
|
|
@ -58,14 +58,14 @@ bool ubusd_send_hello(struct ubus_client *cl)
|
|||
return false;
|
||||
|
||||
ubus_msg_init(ub, UBUS_MSG_HELLO, 0, cl->id.id);
|
||||
ubus_msg_send(cl, ub);
|
||||
ubus_msg_send(cl, ub, true);
|
||||
return true;
|
||||
}
|
||||
|
||||
static int ubusd_send_pong(struct ubus_client *cl, struct ubus_msg_buf *ub, struct blob_attr **attr)
|
||||
{
|
||||
ub->hdr.type = UBUS_MSG_DATA;
|
||||
ubus_msg_send(cl, ubus_msg_ref(ub));
|
||||
ubus_msg_send(cl, ub, false);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -86,7 +86,7 @@ static int ubusd_handle_publish(struct ubus_client *cl, struct ubus_msg_buf *ub,
|
|||
if (!ub)
|
||||
return UBUS_STATUS_NO_DATA;
|
||||
|
||||
ubus_msg_send(cl, ub);
|
||||
ubus_msg_send(cl, ub, true);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -111,7 +111,7 @@ static void ubusd_send_obj(struct ubus_client *cl, struct ubus_msg_buf *ub, stru
|
|||
if (!ub)
|
||||
return;
|
||||
|
||||
ubus_msg_send(cl, ub);
|
||||
ubus_msg_send(cl, ub, true);
|
||||
}
|
||||
|
||||
static int ubusd_handle_lookup(struct ubus_client *cl, struct ubus_msg_buf *ub, struct blob_attr **attr)
|
||||
|
|
@ -195,7 +195,7 @@ static int ubusd_handle_invoke(struct ubus_client *cl, struct ubus_msg_buf *ub,
|
|||
|
||||
ub->hdr.type = UBUS_MSG_INVOKE;
|
||||
ub->hdr.peer = cl->id.id;
|
||||
ubus_msg_send(obj->client, ub);
|
||||
ubus_msg_send(obj->client, ub, true);
|
||||
|
||||
return -1;
|
||||
}
|
||||
|
|
@ -221,7 +221,7 @@ static int ubusd_handle_response(struct ubus_client *cl, struct ubus_msg_buf *ub
|
|||
goto error;
|
||||
|
||||
ub->hdr.peer = blob_get_int32(attr[UBUS_ATTR_OBJID]);
|
||||
ubus_msg_send(cl, ub);
|
||||
ubus_msg_send(cl, ub, true);
|
||||
return -1;
|
||||
|
||||
error:
|
||||
|
|
@ -260,7 +260,7 @@ void ubusd_receive_message(struct ubus_client *cl, struct ubus_msg_buf *ub)
|
|||
ubus_msg_free(ub);
|
||||
|
||||
*retmsg_data = htonl(ret);
|
||||
ubus_msg_send(cl, ubus_msg_ref(retmsg));
|
||||
ubus_msg_send(cl, retmsg, false);
|
||||
}
|
||||
|
||||
static void __init ubusd_proto_init(void)
|
||||
|
|
|
|||
Loading…
Reference in a new issue