session.c: refactor.

artiq/coredevice/comm_generic.py

@@ -122,8 +122,9 @@ class CommGeneric:
 
     def _read_chunk(self, length):
         if self._read_length < length:
-            raise IOError("Read overrun while trying to read {} bytes ({} remaining)".
+            raise IOError("Read overrun while trying to read {} bytes ({} remaining)"
-                          format(length, self._read_length))
+                          " in packet {}".
+                          format(length, self._read_length, self._read_type))
 
         self._read_length -= length
         return self.read(length)
@@ -144,6 +145,12 @@ class CommGeneric:
         (value, ) = struct.unpack(">d", self._read_chunk(8))
         return value
 
+    def _read_bytes(self):
+        return self._read_chunk(self._read_int32())
+
+    def _read_string(self):
+        return self._read_bytes()[:-1].decode('utf-8')
+
     #
     # Writer interface
     #
@@ -170,6 +177,9 @@ class CommGeneric:
         self._write_header(ty)
         self._write_flush()
 
+    def _write_chunk(self, chunk):
+        self._write_buffer.append(chunk)
+
     def _write_int8(self, value):
         self._write_buffer.append(struct.pack("B", value))
 
@@ -182,9 +192,13 @@ class CommGeneric:
     def _write_float64(self, value):
         self._write_buffer.append(struct.pack(">d", value))
 
-    def _write_string(self, value):
+    def _write_bytes(self, value):
+        self._write_int32(len(value))
         self._write_buffer.append(value)
 
+    def _write_string(self, value):
+        self._write_bytes(value.encode("utf-8") + b"\0")
+
     #
     # Exported APIs
     #
@@ -211,7 +225,7 @@ class CommGeneric:
 
     def load(self, kernel_library):
         self._write_header(_H2DMsgType.LOAD_LIBRARY)
-        self._write_string(kernel_library)
+        self._write_chunk(kernel_library)
         self._write_flush()
 
         self._read_empty(_D2HMsgType.LOAD_COMPLETED)
@@ -232,8 +246,7 @@ class CommGeneric:
     def flash_storage_write(self, key, value):
         self._write_header(_H2DMsgType.FLASH_WRITE_REQUEST)
         self._write_string(key)
-        self._write_string(b"\x00")
+        self._write_bytes(value)
-        self._write_string(value)
         self._write_flush()
 
         self._read_header()

artiq/coredevice/comm_serial.py

@@ -28,10 +28,10 @@ class Comm(CommGeneric):
         del self.port
 
     def read(self, length):
-        r = bytes()
+        result = bytes()
-        while len(r) < length:
+        while len(result) < length:
-            r += self.port.read(length - len(r))
+            result += self.port.read(length - len(result))
-        return r
+        return result
 
     def write(self, data):
         remaining = len(data)

artiq/frontend/artiq_coretool.py

@@ -26,7 +26,7 @@ def get_argparser():
     # Configuration Read command
     p_read = subparsers.add_parser("cfg-read",
                                    help="read key from core device config")
-    p_read.add_argument("key", type=to_bytes,
+    p_read.add_argument("key", type=str,
                         help="key to be read from core device config")
 
     # Configuration Write command
@@ -34,11 +34,11 @@ def get_argparser():
                                     help="write key-value records to core "
                                          "device config")
     p_write.add_argument("-s", "--string", nargs=2, action="append",
-                         default=[], metavar=("KEY", "STRING"), type=to_bytes,
+                         default=[], metavar=("KEY", "STRING"), type=str,
                          help="key-value records to be written to core device "
                               "config")
     p_write.add_argument("-f", "--file", nargs=2, action="append",
-                         type=to_bytes, default=[],
+                         type=str, default=[],
                          metavar=("KEY", "FILENAME"),
                          help="key and file whose content to be written to "
                               "core device config")
@@ -47,7 +47,7 @@ def get_argparser():
     p_delete = subparsers.add_parser("cfg-delete",
                                      help="delete key from core device config")
     p_delete.add_argument("key", nargs=argparse.REMAINDER,
-                          default=[], type=to_bytes,
+                          default=[], type=str,
                           help="key to be deleted from core device config")
 
     # Configuration Erase command
@@ -61,6 +61,7 @@ def main():
     dmgr = DeviceManager(FlatFileDB(args.ddb))
     try:
         comm = dmgr.get("comm")
+        comm.check_ident()
 
         if args.action == "log":
           print(comm.get_log())
@@ -72,7 +73,7 @@ def main():
                 print(value)
         elif args.action == "cfg-write":
             for key, value in args.string:
-                comm.flash_storage_write(key, value)
+                comm.flash_storage_write(key, value.encode('utf-8'))
             for key, filename in args.file:
                 with open(filename, "rb") as fi:
                     comm.flash_storage_write(key, fi.read())

soc/runtime/flash_storage.c

@@ -115,7 +115,8 @@ static int is_empty(struct record *record)
     return record->value_len == 0;
 }
 
-static int key_exists(char *buff, char *key, char *end, char accept_empty, struct record *found_record)
+static int key_exists(char *buff, const char *key, char *end, char accept_empty,
+                      struct record *found_record)
 {
     struct iter_state is;
     struct record iter_record;
@@ -170,7 +171,7 @@ static char check_for_empty_records(char *buff)
     return 0;
 }
 
-static unsigned int try_to_flush_duplicates(char *new_key, unsigned int buf_len)
+static unsigned int try_to_flush_duplicates(const char *new_key, unsigned int buf_len)
 {
     unsigned int key_size, new_record_size, ret = 0, can_rollback = 0;
     struct record record, previous_record;
@@ -210,7 +211,8 @@ static unsigned int try_to_flush_duplicates(char *new_key, unsigned int buf_len)
     return ret;
 }
 
-static void write_at_offset(char *key, void *buffer, int buf_len, unsigned int sector_offset)
+static void write_at_offset(const char *key, const void *buffer,
+                            int buf_len, unsigned int sector_offset)
 {
     int key_len = strlen(key) + 1;
     unsigned int record_size = key_len + buf_len + sizeof(record_size);
@@ -223,7 +225,7 @@ static void write_at_offset(char *key, void *buffer, int buf_len, unsigned int s
 }
 
 
-int fs_write(char *key, void *buffer, unsigned int buf_len)
+int fs_write(const char *key, const void *buffer, unsigned int buf_len)
 {
     struct record record;
     unsigned int key_size = strlen(key) + 1;
@@ -269,7 +271,7 @@ void fs_erase(void)
     flush_cpu_dcache();
 }
 
-unsigned int fs_read(char *key, void *buffer, unsigned int buf_len, unsigned int *remain)
+unsigned int fs_read(const char *key, void *buffer, unsigned int buf_len, unsigned int *remain)
 {
     unsigned int read_length = 0;
     struct iter_state is;
@@ -295,7 +297,7 @@ unsigned int fs_read(char *key, void *buffer, unsigned int buf_len, unsigned int
     return read_length;
 }
 
-void fs_remove(char *key)
+void fs_remove(const char *key)
 {
     fs_write(key, NULL, 0);
 }

soc/runtime/flash_storage.h

@@ -5,9 +5,9 @@
 #ifndef __FLASH_STORAGE_H
 #define __FLASH_STORAGE_H
 
-void fs_remove(char *key);
+void fs_remove(const char *key);
 void fs_erase(void);
-int fs_write(char *key, void *buffer, unsigned int buflen);
+int fs_write(const char *key, const void *buffer, unsigned int buflen);
-unsigned int fs_read(char *key, void *buffer, unsigned int buflen, unsigned int *remain);
+unsigned int fs_read(const char *key, void *buffer, unsigned int buflen, unsigned int *remain);
 
 #endif /* __FLASH_STORAGE_H */

soc/runtime/ksupport.h

@@ -5,7 +5,7 @@ long long int now_init(void);
 void now_save(long long int now);
 int watchdog_set(int ms);
 void watchdog_clear(int id);
-int rpc(int rpc_num, ...);
+int rpc(int service, ...);
 void lognonl(const char *fmt, ...);
 void log(const char *fmt, ...);
 

soc/runtime/main.c

@@ -189,12 +189,12 @@ static void serial_service(void)
 
     session_poll((void **)&txdata, &txlen);
     if(txlen > 0) {
-        for(i=0;i<txlen;i++)
+        for(i = 0; i < txlen; i++)
             uart_write(txdata[i]);
-        session_ack_data(txlen);
+        session_ack(txlen);
-        session_ack_mem(txlen);
+    } else if(txlen < 0) {
-    } else if(txlen < 0)
         reset_serial_session(1);
+    }
 }
 
 static void regular_main(void)

soc/runtime/net_server.c

@@ -89,7 +89,7 @@ static err_t net_server_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err
 
 static err_t net_server_sent(void *arg, struct tcp_pcb *pcb, u16_t len)
 {
-    session_ack_mem(len);
+    session_ack(len);
     return ERR_OK;
 }
 
@@ -205,7 +205,6 @@ void net_server_service(void)
             if(len > sndbuf)
                 len = sndbuf;
             tcp_write(active_pcb, data, len, 0);
-            session_ack_data(len);
         }
         if(len < 0)
             net_server_close(active_cs, active_pcb);

soc/runtime/session.c

@@ -18,43 +18,230 @@
 #define BUFFER_IN_SIZE (1024*1024)
 #define BUFFER_OUT_SIZE (1024*1024)
 
-static int buffer_in_index;
+static int process_input();
-/* The 9th byte (right after the header) of buffer_in must be aligned
+static int out_packet_available();
- * to a 32-bit boundary for elf_loader to work.
+
- */
+// ============================= Reader interface =============================
+
+// Align the 9th byte (right after the header) of buffer_in so that
+// the payload can be deserialized directly from the buffer using word reads.
 static struct {
     char padding[3];
-    char data[BUFFER_IN_SIZE];
+    union {
-} __attribute__((packed)) _buffer_in __attribute__((aligned(4)));
+        char data[BUFFER_IN_SIZE];
-#define buffer_in _buffer_in.data
+        struct {
-static int buffer_out_index_data;
+            int32_t sync;
-static int buffer_out_index_mem;
+            int32_t length;
-static char buffer_out[BUFFER_OUT_SIZE];
+            int8_t  type;
+        } __attribute__((packed)) header;
+    };
+} __attribute__((packed, aligned(4))) buffer_in;
 
-static int get_in_packet_len(void)
+static int buffer_in_write_cursor, buffer_in_read_cursor;
-{
-    int r;
 
-    memcpy(&r, &buffer_in[4], 4);
+static void in_packet_reset() {
-    return r;
+    buffer_in_write_cursor = 0;
+    buffer_in_read_cursor  = 0;
 }
 
-static int get_out_packet_len(void)
+static int in_packet_fill(uint8_t *data, int length)
 {
-    int r;
+    int consumed = 0;
+    while(consumed < length) {
+        /* Make sure the output buffer is available for any reply
+         * we might need to send. */
+        if(!out_packet_available())
+            break;
 
-    memcpy(&r, &buffer_out[4], 4);
+        if(buffer_in_write_cursor < 4) {
-    return r;
+            /* Haven't received the synchronization sequence yet. */
+            buffer_in.data[buffer_in_write_cursor++] = data[consumed];
+
+            /* Framing error? */
+            if(data[consumed++] != 0x5a) {
+                buffer_in_write_cursor = 0;
+                continue;
+            }
+        } else if(buffer_in_write_cursor < 8) {
+            /* Haven't received the packet length yet. */
+            buffer_in.data[buffer_in_write_cursor++] = data[consumed++];
+        } else if(buffer_in.header.length == 0) {
+            /* Zero-length packet means session reset. */
+            return -2;
+        } else if(buffer_in.header.length > BUFFER_IN_SIZE) {
+            /* Packet wouldn't fit in the buffer. */
+            return -1;
+        } else if(buffer_in.header.length > buffer_in_write_cursor) {
+            /* Receiving payload. */
+            int remaining = buffer_in.header.length - buffer_in_write_cursor;
+            int amount = length - consumed > remaining ? remaining : length - consumed;
+            memcpy(&buffer_in.data[buffer_in_write_cursor], &data[consumed],
+                   amount);
+            buffer_in_write_cursor += amount;
+            consumed += amount;
+        }
+
+        if(buffer_in.header.length == buffer_in_write_cursor) {
+            /* We have a complete packet. */
+
+            buffer_in_read_cursor = sizeof(buffer_in.header);
+            if(!process_input())
+                return -1;
+
+            if(buffer_in_read_cursor < buffer_in_write_cursor) {
+                log("session.c: read underrun (%d bytes remaining)",
+                    buffer_in_write_cursor - buffer_in_read_cursor);
+            }
+
+            in_packet_reset();
+        }
+    }
+
+    return consumed;
 }
 
-static void submit_output(int len)
+static void in_packet_chunk(void *ptr, int length) {
-{
+    if(buffer_in_read_cursor + length > buffer_in_write_cursor) {
-    memset(&buffer_out[0], 0x5a, 4);
+        log("session.c: read overrun while trying to read %d bytes"
-    memcpy(&buffer_out[4], &len, 4);
+            " (%d remaining)",
-    buffer_out_index_data = 0;
+            length, buffer_in_write_cursor - buffer_in_read_cursor);
-    buffer_out_index_mem = 0;
+    }
+
+    if(ptr != NULL)
+        memcpy(ptr, &buffer_in.data[buffer_in_read_cursor], length);
+    buffer_in_read_cursor += length;
 }
 
+static int8_t in_packet_int8() {
+    int8_t result;
+    in_packet_chunk(&result, sizeof(result));
+    return result;
+}
+
+static int32_t in_packet_int32() {
+    int32_t result;
+    in_packet_chunk(&result, sizeof(result));
+    return result;
+}
+
+static const void *in_packet_bytes(int *length) {
+    *length = in_packet_int32();
+    const void *ptr = &buffer_in.data[buffer_in_read_cursor];
+    in_packet_chunk(NULL, *length);
+    return ptr;
+}
+
+static const char *in_packet_string() {
+    int length;
+    const char *string = in_packet_bytes(&length);
+    if(string[length] != 0) {
+        log("session.c: string is not zero-terminated");
+        return "";
+    }
+    return string;
+}
+
+// ============================= Writer interface =============================
+
+static union {
+    char data[BUFFER_OUT_SIZE];
+    struct {
+        int32_t sync;
+        int32_t length;
+        int8_t  type;
+    } __attribute__((packed)) header;
+} buffer_out;
+
+static int buffer_out_read_cursor, buffer_out_write_cursor;
+
+static void out_packet_reset() {
+    buffer_out_read_cursor  = 0;
+    buffer_out_write_cursor = 0;
+}
+
+static int out_packet_available() {
+    return buffer_out_write_cursor == 0;
+}
+
+static void out_packet_extract(void **data, int *length) {
+    if(buffer_out_write_cursor > 0 &&
+       buffer_out.header.length > 0) {
+        *data   = &buffer_out.data[buffer_out_read_cursor];
+        *length = buffer_out_write_cursor - buffer_out_read_cursor;
+    } else {
+        *length = 0;
+    }
+}
+
+static void out_packet_advance(int length) {
+    if(buffer_out_read_cursor + length > buffer_out_write_cursor) {
+        log("session.c: write underrun while trying to acknowledge %d bytes"
+            " (%d remaining)",
+            length, buffer_out_write_cursor - buffer_out_read_cursor);
+        return;
+    }
+
+    buffer_out_read_cursor += length;
+    if(buffer_out_read_cursor == buffer_out_write_cursor)
+        out_packet_reset();
+}
+
+static int out_packet_chunk(const void *ptr, int length) {
+    if(buffer_out_write_cursor + length > BUFFER_OUT_SIZE) {
+        log("session.c: write overrun while trying to write %d bytes"
+            " (%d remaining)",
+            length, BUFFER_OUT_SIZE - buffer_out_write_cursor);
+        return 0;
+    }
+
+    memcpy(&buffer_out.data[buffer_out_write_cursor], ptr, length);
+    buffer_out_write_cursor += length;
+    return 1;
+}
+
+static void out_packet_start(int type) {
+    buffer_out.header.sync   = 0x5a5a5a5a;
+    buffer_out.header.type   = type;
+    buffer_out.header.length = 0;
+    buffer_out_write_cursor  = sizeof(buffer_out.header);
+}
+
+static void out_packet_finish() {
+    buffer_out.header.length = buffer_out_write_cursor;
+}
+
+static void out_packet_empty(int type) {
+    out_packet_start(type);
+    out_packet_finish();
+}
+
+static int out_packet_int8(int8_t value) {
+    return out_packet_chunk(&value, sizeof(value));
+}
+
+static int out_packet_int32(int32_t value) {
+    return out_packet_chunk(&value, sizeof(value));
+}
+
+static int out_packet_int64(int64_t value) {
+    return out_packet_chunk(&value, sizeof(value));
+}
+
+static int out_packet_float64(double value) {
+    return out_packet_chunk(&value, sizeof(value));
+}
+
+static int out_packet_bytes(const void *ptr, int length) {
+    return out_packet_int32(length) &&
+           out_packet_chunk(ptr, length);
+}
+
+static int out_packet_string(const char *string) {
+    return out_packet_bytes(string, strlen(string) + 1);
+}
+
+// =============================== API handling ===============================
+
 static int user_kernel_state;
 
 enum {
@@ -66,11 +253,12 @@ enum {
 
 void session_start(void)
 {
-    buffer_in_index = 0;
+    in_packet_reset();
-    memset(&buffer_out[4], 0, 4);
+    out_packet_reset();
+
     kloader_stop();
-    user_kernel_state = USER_KERNEL_NONE;
     now = -1;
+    user_kernel_state = USER_KERNEL_NONE;
 }
 
 void session_end(void)
@@ -118,69 +306,108 @@ enum {
     REMOTEMSG_TYPE_FLASH_ERROR_REPLY
 };
 
-static int check_flash_storage_key_len(char *key, unsigned int key_len)
-{
-    if(key_len == get_in_packet_len() - 8) {
-        log("Invalid key: not a null-terminated string");
-        buffer_out[8] = REMOTEMSG_TYPE_FLASH_ERROR_REPLY;
-        submit_output(9);
-        return 0;
-    }
-    return 1;
-}
-
 static int process_input(void)
 {
-    switch(buffer_in[8]) {
+    switch(buffer_in.header.type) {
+        case REMOTEMSG_TYPE_IDENT_REQUEST:
+            out_packet_start(REMOTEMSG_TYPE_IDENT_REPLY);
+            out_packet_chunk("AROR", 4);
+            out_packet_finish();
+            break;
+
+        case REMOTEMSG_TYPE_SWITCH_CLOCK: {
+            int clk = in_packet_int8();
+
+            if(user_kernel_state >= USER_KERNEL_RUNNING) {
+                log("Attempted to switch RTIO clock while kernel running");
+                out_packet_empty(REMOTEMSG_TYPE_CLOCK_SWITCH_FAILED);
+                break;
+            }
+
+            if(rtiocrg_switch_clock(clk))
+                out_packet_empty(REMOTEMSG_TYPE_CLOCK_SWITCH_COMPLETED);
+            else
+                out_packet_empty(REMOTEMSG_TYPE_CLOCK_SWITCH_FAILED);
+            break;
+        }
+
         case REMOTEMSG_TYPE_LOG_REQUEST:
 #if (LOG_BUFFER_SIZE + 9) > BUFFER_OUT_SIZE
 #error Output buffer cannot hold the log buffer
 #endif
-            buffer_out[8] = REMOTEMSG_TYPE_LOG_REPLY;
+            out_packet_start(REMOTEMSG_TYPE_LOG_REPLY);
-            log_get(&buffer_out[9]);
+            log_get(&buffer_out.data[buffer_out_write_cursor]);
-            submit_output(9 + LOG_BUFFER_SIZE);
+            buffer_out_write_cursor += LOG_BUFFER_SIZE;
+            out_packet_finish();
             break;
-        case REMOTEMSG_TYPE_IDENT_REQUEST:
+
-            buffer_out[8] = REMOTEMSG_TYPE_IDENT_REPLY;
+        case REMOTEMSG_TYPE_FLASH_READ_REQUEST: {
-            buffer_out[9] = 'A';
+#if SPIFLASH_SECTOR_SIZE - 4 > BUFFER_OUT_SIZE - 9
-            buffer_out[10] = 'R';
+#error Output buffer cannot hold the flash storage data
-            buffer_out[11] = 'O';
+#endif
-            buffer_out[12] = 'R';
+            const char *key = in_packet_string();
-            submit_output(13);
+            int value_length;
+
+            out_packet_start(REMOTEMSG_TYPE_FLASH_READ_REPLY);
+            value_length = fs_read(key, &buffer_out.data[buffer_out_write_cursor],
+                                   sizeof(buffer_out.data) - buffer_out_write_cursor, NULL);
+            buffer_out_write_cursor += value_length;
+            out_packet_finish();
             break;
-        case REMOTEMSG_TYPE_SWITCH_CLOCK:
+        }
-            if(user_kernel_state >= USER_KERNEL_RUNNING) {
+
-                log("Attempted to switch RTIO clock while kernel running");
+        case REMOTEMSG_TYPE_FLASH_WRITE_REQUEST: {
-                buffer_out[8] = REMOTEMSG_TYPE_CLOCK_SWITCH_FAILED;
+#if SPIFLASH_SECTOR_SIZE - 4 > BUFFER_IN_SIZE - 9
-                submit_output(9);
+#error Input buffer cannot hold the flash storage data
-                break;
+#endif
-            }
+            const char *key, *value;
-            if(rtiocrg_switch_clock(buffer_in[9]))
+            int value_length;
-                buffer_out[8] = REMOTEMSG_TYPE_CLOCK_SWITCH_COMPLETED;
+            key   = in_packet_string();
+            value = in_packet_bytes(&value_length);
+
+            if(fs_write(key, value, value_length))
+                out_packet_empty(REMOTEMSG_TYPE_FLASH_OK_REPLY);
             else
-                buffer_out[8] = REMOTEMSG_TYPE_CLOCK_SWITCH_FAILED;
+                out_packet_empty(REMOTEMSG_TYPE_FLASH_ERROR_REPLY);
-            submit_output(9);
             break;
-        case REMOTEMSG_TYPE_LOAD_LIBRARY:
+        }
+
+        case REMOTEMSG_TYPE_FLASH_ERASE_REQUEST:
+            fs_erase();
+            out_packet_empty(REMOTEMSG_TYPE_FLASH_OK_REPLY);
+            break;
+
+        case REMOTEMSG_TYPE_FLASH_REMOVE_REQUEST: {
+            const char *key = in_packet_string();
+
+            fs_remove(key);
+            out_packet_empty(REMOTEMSG_TYPE_FLASH_OK_REPLY);
+            break;
+        }
+
+        case REMOTEMSG_TYPE_LOAD_LIBRARY: {
+            const void *kernel = &buffer_in.data[buffer_in_read_cursor];
+            buffer_in_read_cursor = buffer_in_write_cursor;
+
             if(user_kernel_state >= USER_KERNEL_RUNNING) {
                 log("Attempted to load new kernel library while already running");
-                buffer_out[8] = REMOTEMSG_TYPE_LOAD_FAILED;
+                out_packet_empty(REMOTEMSG_TYPE_LOAD_FAILED);
-                submit_output(9);
                 break;
             }
-            if(kloader_load_library(&buffer_in[9])) {
+
-                buffer_out[8] = REMOTEMSG_TYPE_LOAD_COMPLETED;
+            if(kloader_load_library(kernel)) {
+                out_packet_empty(REMOTEMSG_TYPE_LOAD_COMPLETED);
                 user_kernel_state = USER_KERNEL_LOADED;
             } else {
-                buffer_out[8] = REMOTEMSG_TYPE_LOAD_FAILED;
+                out_packet_empty(REMOTEMSG_TYPE_LOAD_FAILED);
             }
-            submit_output(9);
             break;
-        case REMOTEMSG_TYPE_RUN_KERNEL: {
+        }
+
+        case REMOTEMSG_TYPE_RUN_KERNEL:
             if(user_kernel_state != USER_KERNEL_LOADED) {
                 log("Attempted to run kernel while not in the LOADED state");
-                buffer_out[8] = REMOTEMSG_TYPE_KERNEL_STARTUP_FAILED;
+                out_packet_empty(REMOTEMSG_TYPE_KERNEL_STARTUP_FAILED);
-                submit_output(9);
                 break;
             }
 
@@ -189,254 +416,100 @@ static int process_input(void)
 
             user_kernel_state = USER_KERNEL_RUNNING;
             break;
-        }
+
         case REMOTEMSG_TYPE_RPC_REPLY: {
             struct msg_rpc_reply reply;
 
             if(user_kernel_state != USER_KERNEL_WAIT_RPC) {
                 log("Unsolicited RPC reply");
-                return 0;
+                return 0; // restart session
             }
 
             reply.type = MESSAGE_TYPE_RPC_REPLY;
             // FIXME memcpy(&reply.eid, &buffer_in[9], 4);
-            memcpy(&reply.retval, &buffer_in[13], 4);
+            // memcpy(&reply.retval, &buffer_in[13], 4);
             mailbox_send_and_wait(&reply);
             user_kernel_state = USER_KERNEL_RUNNING;
             break;
         }
-        case REMOTEMSG_TYPE_FLASH_READ_REQUEST: {
-#if SPIFLASH_SECTOR_SIZE - 4 > BUFFER_OUT_SIZE - 9
-#error Output buffer cannot hold the flash storage data
-#elif SPIFLASH_SECTOR_SIZE - 4 > BUFFER_IN_SIZE - 9
-#error Input buffer cannot hold the flash storage data
-#endif
-            unsigned int ret, in_packet_len;
-            char *key;
 
-            in_packet_len = get_in_packet_len();
-            key = &buffer_in[9];
-            buffer_in[in_packet_len] = '\0';
-
-            buffer_out[8] = REMOTEMSG_TYPE_FLASH_READ_REPLY;
-            ret = fs_read(key, &buffer_out[9], sizeof(buffer_out) - 9, NULL);
-            submit_output(9 + ret);
-            break;
-        }
-        case REMOTEMSG_TYPE_FLASH_WRITE_REQUEST: {
-            char *key, *value;
-            unsigned int key_len, value_len, in_packet_len;
-            int ret;
-
-            in_packet_len = get_in_packet_len();
-            key = &buffer_in[9];
-            key_len = strnlen(key, in_packet_len - 9) + 1;
-            if(!check_flash_storage_key_len(key, key_len))
-                break;
-
-            value_len = in_packet_len - key_len - 9;
-            value = key + key_len;
-            ret = fs_write(key, value, value_len);
-
-            if(ret)
-                buffer_out[8] = REMOTEMSG_TYPE_FLASH_OK_REPLY;
-            else
-                buffer_out[8] = REMOTEMSG_TYPE_FLASH_ERROR_REPLY;
-            submit_output(9);
-            break;
-        }
-        case REMOTEMSG_TYPE_FLASH_ERASE_REQUEST: {
-            fs_erase();
-            buffer_out[8] = REMOTEMSG_TYPE_FLASH_OK_REPLY;
-            submit_output(9);
-            break;
-        }
-        case REMOTEMSG_TYPE_FLASH_REMOVE_REQUEST: {
-            char *key;
-            unsigned int in_packet_len;
-
-            in_packet_len = get_in_packet_len();
-            key = &buffer_in[9];
-            buffer_in[in_packet_len] = '\0';
-
-            fs_remove(key);
-            buffer_out[8] = REMOTEMSG_TYPE_FLASH_OK_REPLY;
-            submit_output(9);
-            break;
-        }
         default:
             return 0;
     }
+
     return 1;
 }
 
-/* Returns -1 in case of irrecoverable error
+static int send_rpc_value(const char **tag, void *value) {
- * (the session must be dropped and session_end called)
+    out_packet_int8(**tag);
- */
-int session_input(void *data, int len)
-{
-    unsigned char *_data = data;
-    int consumed;
 
-    consumed = 0;
+    int size = 0;
-    while(len > 0) {
+    switch(**tag) {
-        /* Make sure the output buffer is available for any reply
+        case 0: // last tag
-         * we might need to send. */
+        case 'n': // None
-        if(get_out_packet_len() != 0)
+            break;
-            return consumed;
 
-        if(buffer_in_index < 4) {
+        case 'b': // bool
-            /* synchronizing */
+            size = 1;
-            if(_data[consumed] == 0x5a)
+            out_packet_chunk(value, size);
-                buffer_in[buffer_in_index++] = 0x5a;
+            break;
-            else
-                buffer_in_index = 0;
-            consumed++; len--;
-        } else if(buffer_in_index < 8) {
-            /* receiving length */
-            buffer_in[buffer_in_index++] = _data[consumed];
-            consumed++; len--;
-            if((buffer_in_index == 8) && (get_in_packet_len() == 0))
-                /* zero-length packet = session reset */
-                return -2;
-        } else {
-            /* receiving payload */
-            int packet_len;
-            int count;
 
-            packet_len = get_in_packet_len();
+        case 'i': // int(width=32)
-            if(packet_len > BUFFER_IN_SIZE)
+            size = 4;
-                return -1;
+            out_packet_chunk(value, size);
-            count = packet_len - buffer_in_index;
+            break;
-            if(count > len)
-                count = len;
-            memcpy(&buffer_in[buffer_in_index], &_data[consumed], count);
-            buffer_in_index += count;
 
-            if(buffer_in_index == packet_len) {
+        case 'I': // int(width=64)
-                if(!process_input())
+        case 'f': // float
+            size = 8;
+            out_packet_chunk(value, size);
+            break;
+
+        case 'F': // Fraction
+            size = 16;
+            out_packet_chunk(value, size);
+            break;
+
+        case 'l': { // list(elt='a)
+            size = sizeof(void*);
+
+            struct { uint32_t length; void *elements; } *list = value;
+            void *element = list->elements;
+
+            const char *tag_copy = *tag + 1;
+            for(int i = 0; i < list->length; i++) {
+                int element_size = send_rpc_value(&tag_copy, element);
+                if(element_size < 0)
                     return -1;
-                buffer_in_index = 0;
+                element = (void*)((intptr_t)element + element_size);
             }
-
+            *tag = tag_copy;
-            consumed += count; len -= count;
+            break;
         }
-    }
-    return consumed;
-}
 
-static int add_base_rpc_value(char base_type, void *value, char *buffer_out, int available_space)
-{
-    switch(base_type) {
-        case 'n':
-            return 0;
-        case 'b':
-            if(available_space < 1)
-                return -1;
-            if(*(char *)value)
-                buffer_out[0] = 1;
-            else
-                buffer_out[0] = 0;
-            return 1;
-        case 'i':
-            if(available_space < 4)
-                return -1;
-            memcpy(buffer_out, value, 4);
-            return 4;
-        case 'I':
-        case 'f':
-            if(available_space < 8)
-                return -1;
-            memcpy(buffer_out, value, 8);
-            return 8;
-        case 'F':
-            if(available_space < 16)
-                return -1;
-            memcpy(buffer_out, value, 16);
-            return 16;
         default:
             return -1;
     }
+
+    (*tag)++;
+    return size;
 }
 
-static int add_rpc_value(int bi, int type_tag, void *value)
+static int send_rpc_request(int service, va_list args)
 {
-    char base_type;
+    out_packet_start(REMOTEMSG_TYPE_RPC_REQUEST);
-    int obi, r;
+    out_packet_int32(service);
 
-    obi = bi;
+    const char *tag = va_arg(args, const char*);
-    base_type = type_tag;
+    while(*tag) {
-
+        void *value = va_arg(args, void*);
-    if((bi + 1) > BUFFER_OUT_SIZE)
+        if(!kloader_validate_kpointer(value))
-        return -1;
+            return 0;
-    buffer_out[bi++] = base_type;
+        if(send_rpc_value(&tag, &value) < 0)
-
-    if(base_type == 'l') {
-        char elt_type;
-        int len;
-        int i, p;
-
-        elt_type = type_tag >> 8;
-        if((bi + 1) > BUFFER_OUT_SIZE)
-            return -1;
-        buffer_out[bi++] = elt_type;
-
-        len = *(int *)value;
-        if((bi + 4) > BUFFER_OUT_SIZE)
-            return -1;
-        memcpy(&buffer_out[bi], &len, 4);
-        bi += 4;
-
-        p = 4;
-        for(i=0;i<len;i++) {
-            r = add_base_rpc_value(elt_type, (char *)value + p,
-                                   &buffer_out[bi], BUFFER_OUT_SIZE - bi);
-            if(r < 0)
-                return r;
-            bi += r;
-            p += r;
-        }
-    } else {
-        r = add_base_rpc_value(base_type, value,
-                               &buffer_out[bi], BUFFER_OUT_SIZE - bi);
-        if(r < 0)
-            return r;
-        bi += r;
-    }
-
-    return bi - obi;
-}
-
-static int send_rpc_request(int rpc_num, va_list args)
-{
-    int r;
-    int bi = 8;
-    int type_tag;
-    void *v;
-
-    buffer_out[bi++] = REMOTEMSG_TYPE_RPC_REQUEST;
-
-    memcpy(&buffer_out[bi], &rpc_num, 4);
-    bi += 4;
-
-    while((type_tag = va_arg(args, int))) {
-        if(type_tag == 'n')
-            v = NULL;
-        else {
-            v = va_arg(args, void *);
-            if(!kloader_validate_kpointer(v))
-                return 0;
-        }
-        r = add_rpc_value(bi, type_tag, v);
-        if(r < 0)
             return 0;
-        bi += r;
     }
-    if((bi + 1) > BUFFER_OUT_SIZE)
-        return 0;
-    buffer_out[bi++] = 0;
 
-    submit_output(bi);
+    out_packet_finish();
     return 1;
 }
 
@@ -454,26 +527,27 @@ static int process_kmsg(struct msg_base *umsg)
 
     switch(umsg->type) {
         case MESSAGE_TYPE_FINISHED:
-            buffer_out[8] = REMOTEMSG_TYPE_KERNEL_FINISHED;
+            out_packet_empty(REMOTEMSG_TYPE_KERNEL_FINISHED);
-            submit_output(9);
 
             kloader_stop();
             user_kernel_state = USER_KERNEL_LOADED;
             mailbox_acknowledge();
             break;
+
         case MESSAGE_TYPE_EXCEPTION: {
             struct msg_exception *msg = (struct msg_exception *)umsg;
 
-            buffer_out[8] = REMOTEMSG_TYPE_KERNEL_EXCEPTION;
+            out_packet_empty(REMOTEMSG_TYPE_KERNEL_EXCEPTION);
             // memcpy(&buffer_out[9], &msg->eid, 4);
             // memcpy(&buffer_out[13], msg->eparams, 3*8);
-            submit_output(9+4+3*8);
+            // submit_output(9+4+3*8);
 
             kloader_stop();
             user_kernel_state = USER_KERNEL_LOADED;
             mailbox_acknowledge();
             break;
         }
+
         case MESSAGE_TYPE_WATCHDOG_SET_REQUEST: {
             struct msg_watchdog_set_request *msg = (struct msg_watchdog_set_request *)umsg;
             struct msg_watchdog_set_reply reply;
@@ -483,6 +557,7 @@ static int process_kmsg(struct msg_base *umsg)
             mailbox_send_and_wait(&reply);
             break;
         }
+
         case MESSAGE_TYPE_WATCHDOG_CLEAR: {
             struct msg_watchdog_clear *msg = (struct msg_watchdog_clear *)umsg;
 
@@ -490,76 +565,72 @@ static int process_kmsg(struct msg_base *umsg)
             mailbox_acknowledge();
             break;
         }
+
         case MESSAGE_TYPE_RPC_REQUEST: {
             struct msg_rpc_request *msg = (struct msg_rpc_request *)umsg;
 
-            if(!send_rpc_request(msg->rpc_num, msg->args))
+            if(!send_rpc_request(msg->rpc_num, msg->args)) {
+                log("Failed to send RPC request");
                 return 0;
+            }
+
             user_kernel_state = USER_KERNEL_WAIT_RPC;
             mailbox_acknowledge();
             break;
         }
-        default: {
+
-            log("Received invalid message type from kernel CPU");
+        default:
+            log("Received invalid message type %d from kernel CPU",
+                umsg->type);
             return 0;
-        }
     }
     return 1;
 }
 
-/* len is set to -1 in case of irrecoverable error
+/* Returns amount of bytes consumed on success.
+ * Returns -1 in case of irrecoverable error
+ * (the session must be dropped and session_end called).
+ * Returns -2 if the host has requested session reset.
+ */
+int session_input(void *data, int length) {
+    return in_packet_fill((uint8_t*)data, length);
+}
+
+/* *length is set to -1 in case of irrecoverable error
  * (the session must be dropped and session_end called)
  */
-void session_poll(void **data, int *len)
+void session_poll(void **data, int *length)
 {
-    int l;
-
     if(user_kernel_state == USER_KERNEL_RUNNING) {
         if(watchdog_expired()) {
             log("Watchdog expired");
-            *len = -1;
+            *length = -1;
             return;
         }
         if(!rtiocrg_check()) {
             log("RTIO clock failure");
-            *len = -1;
+            *length = -1;
             return;
         }
     }
 
-    l = get_out_packet_len();
-
     /* If the output buffer is available,
      * check if the kernel CPU has something to transmit.
      */
-    if(l == 0) {
+    if(out_packet_available()) {
-        struct msg_base *umsg;
+        struct msg_base *umsg = mailbox_receive();
-
-        umsg = mailbox_receive();
         if(umsg) {
             if(!process_kmsg(umsg)) {
-                *len = -1;
+                *length = -1;
                 return;
             }
         }
-        l = get_out_packet_len();
     }
 
-    if(l > 0) {
+    out_packet_extract(data, length);
-        *len = l - buffer_out_index_data;
-        *data = &buffer_out[buffer_out_index_data];
-    } else
-        *len = 0;
 }
 
-void session_ack_data(int len)
+void session_ack(int length)
 {
-    buffer_out_index_data += len;
+    out_packet_advance(length);
-}
-
-void session_ack_mem(int len)
-{
-    buffer_out_index_mem += len;
-    if(buffer_out_index_mem >= get_out_packet_len())
-        memset(&buffer_out[4], 0, 4);
 }

soc/runtime/session.h

@@ -4,11 +4,8 @@
 void session_start(void);
 void session_end(void);
 
-int session_input(void *data, int len);
+int session_input(void *data, int length);
-void session_poll(void **data, int *len);
+void session_poll(void **data, int *length);
-void session_ack_data(int len);
+void session_ack(int length);
-void session_ack_mem(int len);
-
-int rpc(int rpc_num, ...);
 
 #endif /* __SESSION_H */