runtime/net_server: refactor to support multiple services

m-labs · Dec 17, 2015 · 6ae41e6 · 6ae41e6
1 parent 7f4490a
commit 6ae41e6
Showing 3 changed files with 91 additions and 45 deletions.
diff --git a/artiq/runtime/main.c b/artiq/runtime/main.c
@@ -127,11 +127,21 @@ static void network_init(void)
     netif_set_link_up(&netif);
 }
 
+static struct net_server_instance session_inst = {
+    .port = 1381,
+    .start = session_start,
+    .end = session_end,
+    .input = session_input,
+    .poll = session_poll,
+    .ack_consumed = session_ack_consumed,
+    .ack_sent = session_ack_sent
+};
+
 static void regular_main(void)
 {
     puts("Accepting sessions on Ethernet.");
     network_init();
-    net_server_init();
+    net_server_init(&session_inst);
     moninj_init();
 
     session_end();

diff --git a/artiq/runtime/net_server.c b/artiq/runtime/net_server.c
@@ -13,22 +13,23 @@
 #include <netif/etharp.h>
 #include <liteethif.h>
 
-#include "session.h"
 #include "net_server.h"
 
 struct net_server_connstate {
+    struct net_server_instance *instance;
     int magic_recognized;
     struct pbuf *rp;
     int rp_offset;
 };
 
-static struct net_server_connstate *cs_new(void)
+static struct net_server_connstate *cs_new(struct net_server_instance *instance)
 {
     struct net_server_connstate *cs;
 
     cs = (struct net_server_connstate *)mem_malloc(sizeof(struct net_server_connstate));
     if(!cs)
         return NULL;
+    cs->instance = instance;
     cs->magic_recognized = 0;
     cs->rp = NULL;
     cs->rp_offset = 0;
@@ -49,15 +50,15 @@ static int magic_ok(struct net_server_connstate *cs)
     return cs->magic_recognized >= 14;
 }
 
-static struct net_server_connstate *active_cs;
-static struct tcp_pcb *active_pcb;
-
 static void net_server_close(struct net_server_connstate *cs, struct tcp_pcb *pcb)
 {
-    if(cs == active_cs) {
-        session_end();
-        active_cs = NULL;
-        active_pcb = NULL;
+    struct net_server_instance *instance;
+
+    instance = cs->instance;
+    if(cs == instance->open_session_cs) {
+        instance->end();
+        instance->open_session_cs = NULL;
+        instance->open_session_pcb = NULL;
     }
 
     if(pcb) {
@@ -91,26 +92,32 @@ 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_sent(len);
+    struct net_server_connstate *cs;
+
+    cs = (struct net_server_connstate *)arg;
+    cs->instance->ack_sent(len);
     return ERR_OK;
 }
 
 static void tcp_pcb_service(void *arg, struct tcp_pcb *pcb)
 {
     struct net_server_connstate *cs;
+    struct net_server_instance *instance;
     int remaining_in_pbuf;
     char *rpp;
     struct pbuf *next;
     int r;
 
     cs = (struct net_server_connstate *)arg;
+    instance = cs->instance;
 
+    /* Reader interface */
     while(cs->rp) {
         remaining_in_pbuf = cs->rp->len - cs->rp_offset;
         rpp = (char *)cs->rp->payload;
         while(remaining_in_pbuf > 0) {
-            if(cs == active_cs) {
-                r = session_input(&rpp[cs->rp_offset], remaining_in_pbuf);
+            if(cs == instance->open_session_cs) {
+                r = instance->input(&rpp[cs->rp_offset], remaining_in_pbuf);
                 if(r > 0) {
                     tcp_recved(pcb, r);
                     cs->rp_offset += r;
@@ -123,11 +130,12 @@ static void tcp_pcb_service(void *arg, struct tcp_pcb *pcb)
                 if(rpp[cs->rp_offset] == net_server_magic[cs->magic_recognized]) {
                     cs->magic_recognized++;
                     if(magic_ok(cs)) {
-                        if(active_cs)
-                            net_server_close(active_cs, active_pcb);
-                        session_start();
-                        active_cs = cs;
-                        active_pcb = pcb;
+                        if(instance->open_session_cs)
+                            net_server_close(instance->open_session_cs,
+                                             instance->open_session_pcb);
+                        instance->start();
+                        instance->open_session_cs = cs;
+                        instance->open_session_pcb = pcb;
                         tcp_sent(pcb, net_server_sent);
                     }
                 } else {
@@ -150,6 +158,23 @@ static void tcp_pcb_service(void *arg, struct tcp_pcb *pcb)
             cs->rp = NULL;
         }
     }
+
+    /* Writer interface */
+    if(cs == instance->open_session_cs) {
+        void *data;
+        int len, sndbuf;
+
+        cs->instance->poll(&data, &len);
+        if(len > 0) {
+            sndbuf = tcp_sndbuf(pcb);
+            if(len > sndbuf)
+                len = sndbuf;
+            tcp_write(pcb, data, len, 0);
+            instance->ack_consumed(len);
+        }
+        if(len < 0)
+            net_server_close(cs, pcb);
+    }
 }
 
 static void net_server_err(void *arg, err_t err)
@@ -160,59 +185,51 @@ static void net_server_err(void *arg, err_t err)
     net_server_close(cs, NULL);
 }
 
-static struct tcp_pcb *listen_pcb;
-
 static err_t net_server_accept(void *arg, struct tcp_pcb *newpcb, err_t err)
 {
+    struct net_server_instance *instance;
     struct net_server_connstate *cs;
 
-    cs = cs_new();
+    instance = (struct net_server_instance *)arg;
+    cs = cs_new(instance);
     if(!cs)
         return ERR_MEM;
-    tcp_accepted(listen_pcb);
+    tcp_accepted(instance->listen_pcb);
     tcp_arg(newpcb, cs);
     tcp_recv(newpcb, net_server_recv);
     tcp_err(newpcb, net_server_err);
     return ERR_OK;
 }
 
-void net_server_init(void)
+void net_server_init(struct net_server_instance *instance)
 {
-    listen_pcb = tcp_new();
-    listen_pcb->so_options |= SOF_KEEPALIVE;
-    tcp_bind(listen_pcb, IP_ADDR_ANY, 1381);
-    listen_pcb = tcp_listen(listen_pcb);
-    tcp_accept(listen_pcb, net_server_accept);
+    struct tcp_pcb *bind_pcb;
+
+    bind_pcb = tcp_new();
+    bind_pcb->so_options |= SOF_KEEPALIVE;
+    tcp_bind(bind_pcb, IP_ADDR_ANY, instance->port);
+
+    instance->listen_pcb = tcp_listen(bind_pcb);
+    tcp_arg(instance->listen_pcb, instance);
+    tcp_accept(instance->listen_pcb, net_server_accept);
 }
 
 extern struct tcp_pcb *tcp_active_pcbs;
 
 void net_server_service(void)
 {
     struct tcp_pcb *pcb;
-    void *data;
-    int len, sndbuf;
 
-    /* Assume all active TCP PCBs with a non-NULL arg are our connections. */
+    /*
+     * Assume all active TCP PCBs with a non-NULL arg are our connections.
+     * lwip defines "active PCB" as being able to send/receive data.
+     */
     pcb = tcp_active_pcbs;
     while(pcb) {
         if(pcb->callback_arg)
             tcp_pcb_service(pcb->callback_arg, pcb);
         pcb = pcb->next;
     }
-
-    if(active_cs) {
-        session_poll(&data, &len);
-        if(len > 0) {
-            sndbuf = tcp_sndbuf(active_pcb);
-            if(len > sndbuf)
-                len = sndbuf;
-            tcp_write(active_pcb, data, len, 0);
-            session_ack_consumed(len);
-        }
-        if(len < 0)
-            net_server_close(active_cs, active_pcb);
-    }
 }
 
 #endif /* CSR_ETHMAC_BASE */
diff --git a/artiq/runtime/net_server.h b/artiq/runtime/net_server.h
@@ -1,7 +1,26 @@
 #ifndef __NET_SERVER_H
 #define __NET_SERVER_H
 
-void net_server_init(void);
+struct net_server_connstate;
+struct tcp_pcb;
+
+struct net_server_instance {
+    int port;
+
+    void (*start)(void);
+    void (*end)(void);
+    int (*input)(void *data, int length);
+    void (*poll)(void **data, int *length);
+    void (*ack_consumed)(int length);
+    void (*ack_sent)(int length);
+
+    /* internal use */
+    struct tcp_pcb *listen_pcb;
+    struct net_server_connstate *open_session_cs;
+    struct tcp_pcb *open_session_pcb;
+};
+
+void net_server_init(struct net_server_instance *instance);
 void net_server_service(void);
 
 #endif /* __NET_SERVER_H */