[kernel][ports] Add basic ports functionality

kernel/port.c

+/*
+ * Copyright (c) 2015 Carlos Pizano-Uribe  cpu@chromium.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining
+ * a copy of this software and associated documentation files
+ * (the "Software"), to deal in the Software without restriction,
+ * including without limitation the rights to use, copy, modify, merge,
+ * publish, distribute, sublicense, and/or sell copies of the Software,
+ * and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+ * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+ * CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+ * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+ * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+/**
+ * @file
+ * @brief  Port object functions
+ * @defgroup event Events
+ *
+ */
+
+#include <kernel/port.h>
+#include <debug.h>
+#include <list.h>
+#include <malloc.h>
+#include <assert.h>
+#include <string.h>
+#include <err.h>
+#include <kernel/thread.h>
+
+#define WRITEPORT_MAGIC 'prtw'
+#define READPORT_MAGIC  'prtr'
+#define PORTGROUP_MAGIC 'prtg'
+
+#define PORT_BUFF_SIZE      8
+#define PORT_BUFF_SIZE_BIG 64
+
+#define RESCHEDULE_POLICY true
+
+typedef struct {
+  uint max;
+  uint head;
+  uint tail;
+  port_packet_t packet[1];
+} port_buf_t;
+
+typedef struct {
+  int magic;
+  struct list_node node;
+  char name[PORT_NAME_LEN];

[travisg, 2015/11/09 23:14:16]

should probably put name last, to make sure thumb code can properly use a small
immediate offset into the struct, plus as a general style thing.

[cpu_(ooo_6.6-7.5), 2015/11/11 00:31:13]

Done.

+  struct list_node rp_list;
+  port_buf_t* buf;
+} write_port_t;
+
+// todo (cpu) add the |mode| bit to the write port and check
+// that unicast ports have only one read port.
+
+typedef struct {
+  int magic;
+  wait_queue_t wait;
+  struct list_node rp_list;
+} port_group_t;
+
+typedef struct {
+  int magic;
+  struct list_node node;
+  wait_queue_t wait;
+  void* ctx;
+  write_port_t* wport;
+  port_group_t* gport;
+  port_buf_t* buf;
+} read_port_t;
+
+
+static struct list_node write_port_list;
+
+
+static port_buf_t* make_buf(uint pk_count) {
+    uint size = sizeof(port_buf_t) + ((pk_count - 1) * sizeof(port_packet_t));
+    port_buf_t* buf = (port_buf_t*) malloc(size);
+    if (!buf)
+    return NULL;

[travisg, 2015/11/09 23:14:16]

weird indent here. Is there a mixture of spaces and tabs in this file? For all
new stuff, use 4 space indents across the file.

[cpu_(ooo_6.6-7.5), 2015/11/11 00:31:12]

Done.

+    buf->max = pk_count;
+    buf->head = buf->tail = 0;
+    return buf;
+}
+
+static int buf_write(const port_packet_t* packets, size_t count, port_buf_t* buf)
+{
+    // todo (cpu). circular buffer write.
+    return 0;
+}
+
+static int buf_read(port_buf_t* buf, port_result_t* pr)
+{
+    // todo (cpu). circular buffer read.  
+    return 0;
+}
+
+// must be called before any use of ports.
+void port_init(void)
+{
+    list_initialize(&write_port_list);
+}
+
+status_t port_create(const char* name, port_mode_t mode, port_t* port)
+{

[travisg, 2015/11/09 23:14:16]

test for !name and !port and return ERR_INVALID_ARGS

[cpu_(ooo_6.6-7.5), 2015/11/11 00:31:13]

Done.

+    // only unicast ports can have a large buffer.
+    if (mode & port_mode_broadcast) {
+        if (mode & port_mode_big_buffer)
+            return ERR_INVALID_ARGS;
+    }
+
+    // lookup for existing port, return that if found.
+    write_port_t* wp = NULL;
+    THREAD_LOCK(state1);
+    list_for_every_entry(&write_port_list, wp, write_port_t, node) {
+        if (strcmp(wp->name, name) == 0) {
+            THREAD_UNLOCK(state1);
+            *port = (void*)wp;
+            return NO_ERROR;
+        }
+    }
+    THREAD_UNLOCK(state1);
+
+    // not found, create the write port and the circular buffer.
+    wp = malloc(sizeof(write_port_t));
+    if (!wp)
+        return ERR_NO_MEMORY;
+
+    memset(wp, 0, sizeof(write_port_t));
+    wp->magic = WRITEPORT_MAGIC;  
+    strlcpy(wp->name, name, sizeof(wp->name));
+    list_initialize(&wp->rp_list);
+
+    uint size = mode & port_mode_big_buffer ?  PORT_BUFF_SIZE_BIG : PORT_BUFF_SIZE;
+    wp->buf = make_buf(size);

[travisg, 2015/11/09 23:14:16]

test and unroll wp->buf == NULL here.

[cpu_(ooo_6.6-7.5), 2015/11/11 00:31:13]

Done.

+    
+    THREAD_LOCK(state2);
+    list_add_tail(&write_port_list, &wp->node);

[travisg, 2015/11/09 23:14:16]

add comment that there's a bit of a race here and another write port could be
created with the same name.

[cpu_(ooo_6.6-7.5), 2015/11/11 00:31:13]

good catch!  I'll do that now and later on I'll fix it.

+    THREAD_UNLOCK(state2);
+
+    *port = (void*)wp;
+    return NO_ERROR;
+}
+
+status_t port_open(const char* name, void* ctx, port_t* port)
+{
+    // assume success; create the read port and buffer now.
+    read_port_t* rp = malloc(sizeof(read_port_t));
+    if (!rp)
+        return ERR_NO_MEMORY;
+
+    memset(rp, 0, sizeof(read_port_t));
+    rp->magic = READPORT_MAGIC;
+    wait_queue_init(&rp->wait);
+    rp->ctx = ctx;
+
+    port_buf_t* buf = make_buf(PORT_BUFF_SIZE);

[travisg, 2015/11/09 23:14:16]

test for !buf

+
+    // find the named write port.
+    THREAD_LOCK(state);
+    write_port_t* wp = NULL; 
+    list_for_every_entry(&write_port_list, wp, write_port_t, node) {
+        if (strcmp(wp->name, name) == 0) {
+            // found; add read port to write port list.
+            list_add_tail(&wp->rp_list, &rp->node);
+            rp->wport = wp;
+            if (wp->buf) {
+              // this is the first read port; transfer the circular buffer.
+              rp->buf = wp->buf;
+              wp->buf = NULL;
+            } else {
+              // not first read port; use the new (small) circular buffer.
+              rp->buf = buf;
+              buf = NULL;
+            }
+            THREAD_UNLOCK(state);
+            *port = (void*)rp;
+            return NO_ERROR;
+        }
+    }
+
+    // named port not found.
+    THREAD_UNLOCK(state);
+    free(buf);
+    free(rp);
+    return ERR_NOT_FOUND;
+}
+
+status_t port_group(port_t* ports, size_t count, port_t* group)
+{
+    // assume success; create port group now.
+    port_group_t* pg = malloc(sizeof(port_group_t));
+    if (!pg)
+        return ERR_NO_MEMORY;
+
+    memset(pg, 0, sizeof(port_group_t));
+    pg->magic = PORTGROUP_MAGIC;
+    wait_queue_init(&pg->wait);
+    list_initialize(&pg->rp_list);
+
+    THREAD_LOCK(state);
+    for(size_t ix = 0; ix != count; ix++) {
+        read_port_t* rp = (read_port_t*)ports[ix];
+        if ((rp->magic != READPORT_MAGIC) || rp->gport) {
+            // wrong type of port, or port already part of a group,
+            // in any case, undo the changes to the previous read ports.
+            for (size_t jx = 0; jx != ix; jx++) {
+                ((read_port_t*)ports[jx])->gport = NULL;
+            }
+            THREAD_UNLOCK(state);
+            free(pg);
+            return ERR_BAD_HANDLE;
+        }
+        // link port group and read port.
+        rp->gport = pg;
+        list_add_tail(&pg->rp_list, &rp->node);
+    }
+    THREAD_UNLOCK(state);
+    *group = (port_t*)pg;
+    return NO_ERROR;
+}
+
+status_t port_write(port_t port, const port_packet_t* pk, size_t count)
+{
+    write_port_t* wp = (write_port_t*)port;
+    THREAD_LOCK(state);
+    if (wp->magic != WRITEPORT_MAGIC) {
+        // wrong port type.
+        THREAD_UNLOCK(state);
+        return ERR_BAD_HANDLE;
+    }
+
+    if (wp->buf) {
+        // there are no read ports, just write to the buffer.
+        buf_write(pk, count, wp->buf);
+    } else {
+        // there are read ports. for each, write and attempt to wake a thread
+        // from the port group or from the read port itself.
+        read_port_t* rp;
+        list_for_every_entry(&wp->rp_list, rp, read_port_t, node) {
+            buf_write(pk, count, rp->buf);
+
+            int count = 0;
+            if (rp->gport) {
+                count = wait_queue_wake_one(&rp->gport->wait, RESCHEDULE_POLICY, NO_ERROR);
+            }
+            if (!count) {
+                wait_queue_wake_one(&rp->wait, RESCHEDULE_POLICY, NO_ERROR);
+            }                
+        }
+    }
+
+    THREAD_UNLOCK(state);
+    return NO_ERROR;
+}
+
+static inline status_t read_no_lock(read_port_t* rp, lk_time_t timeout, port_result_t* result)
+{
+    int read = buf_read(rp->buf, result);
+    if (read > 0) {
+        result->ctx = rp->ctx;
+        return NO_ERROR;
+    } else if (read < 0) {
+        return (status_t)read;
+    }
+    status_t rc = wait_queue_block(&rp->wait, timeout);

[travisg, 2015/11/09 23:14:16]

should probably test for timeout == 0 and not bother calling. since this is
inlined and code below calls it with 0 hard coded, it wont emit the entire call
at all.

[cpu_(ooo_6.6-7.5), 2015/11/11 00:31:13]

Done.

+    if (rc != NO_ERROR)
+        return rc;
+    // recursive tail call is usually optimized away with a goto.
+    return read_no_lock(rp, timeout, result);
+}
+
+status_t port_read(port_t port, lk_time_t timeout, port_result_t* result)
+{
+    status_t rc = ERR_GENERIC;
+    read_port_t* rp = (read_port_t*)port;
+
+    THREAD_LOCK(state);
+    if (rp->magic == READPORT_MAGIC) {
+        // dealing with a single port.
+        rc = read_no_lock(rp, timeout, result);
+    } else if (rp->magic == PORTGROUP_MAGIC) {
+        // dealing with a port group.
+        port_group_t* pg = (port_group_t*)port;
+        do {
+            // read each port with no timeout.
+            list_for_every_entry(&pg->rp_list, rp, read_port_t, node) {
+                rc = read_no_lock(rp, 0, result);
+                if (rc != ERR_TIMED_OUT)
+                    goto read_exit;
+            }
+            // no data, block on the group waitqueue.
+            rc = wait_queue_block(&pg->wait, timeout);
+        } while (rc == NO_ERROR);
+    } else {
+        // wrong port type.
+        rc = ERR_BAD_HANDLE;
+    }
+
+read_exit:
+    THREAD_UNLOCK(state);    
+    return rc;
+}
+
+status_t port_destroy(port_t port)
+{
+    // todo (cpu)
+    return NO_ERROR;
+}
+
+status_t port_close(port_t port)
+{
+    // todo (cpu)
+    return NO_ERROR;
+}
+