Implement write queue for iomux

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

src/wqueue.c

@@ -0,0 +1,112 @@
+#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;
+    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;
+}

test/console_print.c

@@ -0,0 +1,80 @@
+#include <string.h>
+#include <unistd.h>
+#include <stdio.h>
+#include <errno.h>
+#include <ucore/utils.h>
+#include <ucore/utils.h>
+#include <ucore/iomux.h>
+#include <ucore/wqueue.h>
+
+
+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;
+}
+