diff --git a/include/ucore/wqueue.h b/include/ucore/wqueue.h new file mode 100644 index 0000000..6dcfad2 --- /dev/null +++ b/include/ucore/wqueue.h @@ -0,0 +1,26 @@ +#ifndef UCORE_WQUEUE_H_ +#define UCORE_WQUEUE_H_ + +#include "mbuf.h" +#include "iomux.h" + +typedef int (*wqueue_write_cb)(struct IOMux *mux, struct IOMuxFD *fd, struct MBuf *mbuf); +typedef int (*wqueue_read_cb)(struct IOMux *mux, struct IOMuxFD *fd); + +struct UCWQueue { + struct IOMuxFD fd; + struct TailQ queue; + struct IOMux *mux; + wqueue_read_cb read_cb; + wqueue_write_cb write_cb; + + unsigned int queue_len; +}; + +void uc_wqueue_init(struct UCWQueue *wqueue, struct IOMux *mux); + +int uc_wqueue_clear(struct UCWQueue *wqueue); + +int uc_wqueue_enqueue(struct UCWQueue *wqueue, struct MBuf *mbuf); + +#endif diff --git a/src/wqueue.c b/src/wqueue.c new file mode 100644 index 0000000..8e7505f --- /dev/null +++ b/src/wqueue.c @@ -0,0 +1,112 @@ +#include +#include +#include "ucore/wqueue.h" +#include "ucore/iomux.h" + +static int uc_wqueue_handle_write(struct UCWQueue *wqueue) +{ + int rc; + 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, fd, 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); + uc_mbuf_free(mbuf); + wqueue->queue_len--; + } + + if (rc >= 0 && uc_tailq_empty(&wqueue->queue) + && (wqueue->fd.what & MUX_EV_WRITE) != 0) { + + int urc; + + urc = iomux_update_events(mux, + fd, + wqueue->fd.what & ~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, fd); + //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; + + uc_tailq_init(&wqueue->queue); +} + +int uc_wqueue_clear(struct UCWQueue *wqueue) +{ + struct MBuf *buf; + int rc; + + while ((buf = uc_mbuf_dequeue(&wqueue->queue)) != NULL) { + uc_mbuf_free(buf); + wqueue->queue_len--; + } + + if ((wqueue->fd.what & MUX_EV_WRITE) != 0) { + rc = iomux_update_events(wqueue->mux, + &wqueue->fd, + wqueue->fd.what & ~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_update_events(wqueue->mux, + &wqueue->fd, + wqueue->fd.what | MUX_EV_WRITE); + assert(rc == 0); + } + + return 0; +} diff --git a/test/console_print.c b/test/console_print.c new file mode 100644 index 0000000..dd3f580 --- /dev/null +++ b/test/console_print.c @@ -0,0 +1,80 @@ +#include +#include +#include +#include +#include +#include +#include +#include + + +int stdout_read(struct IOMux *mux, struct IOMuxFD *fd) +{ + return 1; +} +void stdin_read_cb(struct IOMux *mux, struct IOMuxFD *fd, unsigned int event) +{ + struct MBuf *mbuf = uc_mbuf_new(512); + uint8_t *buf = uc_mbuf_head(mbuf); + + ssize_t len = read(fd->fd, buf, sizeof buf); + + printf("Read %d bytes\n", len); + if (len <= 0) { + uc_mbuf_free(mbuf); + return; + } + uc_mbuf_put(mbuf, len); + + struct UCWQueue *wqueue = fd->cookie_ptr; + uc_wqueue_enqueue(wqueue, mbuf); +} + +struct IOMuxFD stdin_fd = { + .fd = 0, + .what = MUX_EV_READ, + .callback = stdin_read_cb, + +}; + +int stdout_write(struct IOMux *mux, struct IOMuxFD *fd, struct MBuf *mbuf) +{ + uint32_t len; + ssize_t wlen; + uint8_t *data; + + len = uc_mbuf_len(mbuf); + data = uc_mbuf_pull(mbuf, len); + wlen = write(fd->fd, data, len); + + if (wlen < 0) { + perror("write"); + return -1; + } + + if (wlen == len) { + return 1; + } + + return 0; +} + +struct UCWQueue stdout_queue; + +int main(int argc, char *argv[]) +{ + struct IOMux *mux = iomux_create(IOMUX_TYPE_DEFAULT); + uc_wqueue_init(&stdout_queue, mux); + stdin_fd.cookie_ptr = &stdout_queue; + stdout_queue.fd.fd = 1; + stdout_queue.fd.what = MUX_EV_READ; + stdout_queue.write_cb = stdout_write; + stdout_queue.read_cb = stdout_read; + + iomux_register_fd(mux, &stdin_fd); + iomux_register_fd(mux, &stdout_queue.fd); + iomux_run(mux); + + return 0; +} +