Nils O. Selåsdal
124 lines
2.8 KiB
C
124 lines
2.8 KiB
C
#include <assert.h>
|
|
#include <string.h>
|
|
#include "ucore/wqueue.h"
|
|
#include "ucore/iomux.h"
|
|
|
|
static int uc_wqueue_handle_write(struct UCWQueue *wqueue)
|
|
{
|
|
int rc = UC_WQ_DONE;
|
|
struct IOMuxFD *fd = &wqueue->fd;
|
|
struct IOMux *mux = wqueue->mux;
|
|
|
|
while (!uc_tailq_empty(&wqueue->queue)) {
|
|
struct TailQ *q;
|
|
struct MBuf *mbuf;
|
|
q = UC_TAILQ_FIRST(&wqueue->queue);
|
|
|
|
mbuf = UC_TAILQ_CONTAINER(q, struct MBuf, entry);
|
|
rc = wqueue->write_cb(mux, wqueue, mbuf);
|
|
if (rc <= 0) {
|
|
//NOTE for rc < 0, nothing can touch the wqueue or fd
|
|
//any more, as it might been free'd
|
|
break;
|
|
}
|
|
|
|
mbuf = uc_mbuf_dequeue(&wqueue->queue);
|
|
assert(mbuf != NULL);
|
|
wqueue->free_cb(mbuf);
|
|
wqueue->queue_len--;
|
|
}
|
|
|
|
if (rc >= 0 && uc_tailq_empty(&wqueue->queue)
|
|
&& (wqueue->fd.what & MUX_EV_WRITE) != 0) {
|
|
|
|
int urc;
|
|
|
|
urc = iomux_clear_event(mux,
|
|
fd,
|
|
MUX_EV_WRITE);
|
|
assert(urc == 0);
|
|
|
|
}
|
|
|
|
return rc;
|
|
}
|
|
|
|
|
|
|
|
static void uc_wqueue_cb(struct IOMux *mux, struct IOMuxFD *fd, unsigned int what)
|
|
{
|
|
int rc = 0;
|
|
struct UCWQueue *wqueue = UC_CONTAINER_OF(fd, struct UCWQueue, fd);
|
|
|
|
if (what & MUX_EV_WRITE) {
|
|
rc = uc_wqueue_handle_write(wqueue);
|
|
//NOTE - can't touch the wqueue or fd if rc < 0
|
|
//as it might been free'd
|
|
}
|
|
|
|
|
|
if (rc >= 0 && what & MUX_EV_READ) {
|
|
rc = wqueue->read_cb(mux, wqueue);
|
|
//NOTE - can't touch the wqueue or fd after this point,
|
|
//as it might been free'd
|
|
}
|
|
}
|
|
|
|
void uc_wqueue_init(struct UCWQueue *wqueue, struct IOMux *mux)
|
|
{
|
|
assert(mux != NULL);
|
|
|
|
memset(wqueue, 0, sizeof *wqueue);
|
|
wqueue->mux = mux;
|
|
wqueue->fd.callback = uc_wqueue_cb;
|
|
wqueue->free_cb = uc_mbuf_free;
|
|
uc_tailq_init(&wqueue->queue);
|
|
}
|
|
|
|
int uc_wqueue_clear(struct UCWQueue *wqueue)
|
|
{
|
|
struct MBuf *buf;
|
|
int rc = 0;
|
|
|
|
while ((buf = uc_mbuf_dequeue(&wqueue->queue)) != NULL) {
|
|
wqueue->free_cb(buf);
|
|
wqueue->queue_len--;
|
|
}
|
|
|
|
if ((wqueue->fd.what & MUX_EV_WRITE) != 0) {
|
|
rc = iomux_clear_event(wqueue->mux,
|
|
&wqueue->fd,
|
|
MUX_EV_WRITE);
|
|
}
|
|
|
|
assert(rc == 0);
|
|
|
|
return rc;
|
|
}
|
|
|
|
int uc_wqueue_enqueue(struct UCWQueue *wqueue, struct MBuf *mbuf)
|
|
{
|
|
int rc = 0;
|
|
|
|
uc_mbuf_enqueue(mbuf, &wqueue->queue);
|
|
wqueue->queue_len++;
|
|
|
|
if ((wqueue->fd.what & MUX_EV_WRITE) == 0) {
|
|
rc = iomux_add_event(wqueue->mux,
|
|
&wqueue->fd,
|
|
MUX_EV_WRITE);
|
|
assert(rc == 0);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
int uc_wqueue_flush(struct UCWQueue *wqueue)
|
|
{
|
|
if (uc_wqueue_handle_write(wqueue) < 0) {
|
|
return -1;
|
|
}
|
|
|
|
return 0;
|
|
}
|