Add multithreaded example for fossa

    PUBLISHED_FROM=680a37157fa24884f27d0255379b54193c21f4e8

Author: cpq

examples/multithreaded_restful_server/Makefile

@@ -0,0 +1,18 @@
+PROG = multithreaded_restful_server
+SOURCES = $(PROG).c ../../fossa.c
+APP_FLAGS = -DNS_ENABLE_THREADS $(CFLAGS_EXTRA)
+
+ifeq ($(OS), Windows_NT)
+APP_FLAGS += advapi32.lib
+endif
+
+all: $(PROG)
+
+$(PROG): $(SOURCES)
+	$(CC) $(SOURCES) -o $@ -W -Wall -I../.. -pthread $(APP_FLAGS)
+
+$(PROG).exe: $(SOURCES)
+	cl $(SOURCES) /I../.. /MD /Fe$@ $(APP_FLAGS)
+
+clean:
+	rm -rf *.gc* *.dSYM *.exe *.obj *.o a.out $(PROG)

examples/multithreaded_restful_server/multithreaded_restful_server.c

@@ -0,0 +1,52 @@
+// Copyright (c) 2015 Cesanta Software Limited
+// All rights reserved
+
+// This example shows how to handle long, blocking requests by
+// handing off computation to different threads. Here, each
+// request spawns a new thread. In a production scenario, a thread
+// pools can be used for efficiency, if required.
+// Long computation is simulated by sleeping for a random interval.
+
+#include "fossa.h"
+
+static const char *s_http_port = "8000";
+
+static void ev_handler(struct ns_connection *c, int ev, void *p) {
+  if (ev == NS_HTTP_REQUEST) {
+    struct http_message *hm = (struct http_message *) p;
+    char reply[100];
+
+    /* Simulate long calculation */
+    sleep(3);
+
+    /* Send the reply */
+    snprintf(reply, sizeof(reply), "{ \"uri\": \"%.*s\" }\n",
+             (int) hm->uri.len, hm->uri.p);
+    ns_printf(c, "HTTP/1.1 200 OK\r\n"
+              "Content-Type: application/json\r\n"
+              "Content-Length: %d\r\n"
+              "\r\n"
+              "%s",
+              (int) strlen(reply), reply);
+  }
+}
+
+int main(void) {
+  struct ns_mgr mgr;
+  struct ns_connection *nc;
+
+  ns_mgr_init(&mgr, NULL);
+  nc = ns_bind(&mgr, s_http_port, ev_handler);
+  ns_set_protocol_http_websocket(nc);
+
+  /* For each new connection, execute ev_handler in a separate thread */
+  ns_enable_multithreading(nc);
+
+  printf("Starting multi-threaded server on port %s\n", s_http_port);
+  for (;;) {
+    ns_mgr_poll(&mgr, 3000);
+  }
+  ns_mgr_free(&mgr);
+
+  return 0;
+}

fossa.c

@@ -72,6 +72,9 @@ NS_INTERNAL struct ns_connection *ns_finish_connect(struct ns_connection *nc,
 NS_INTERNAL int ns_parse_address(const char *str, union socket_address *sa,
                                  int *proto, char *host, size_t host_len);
 NS_INTERNAL void ns_call(struct ns_connection *, int ev, void *ev_data);
+NS_INTERNAL void ns_forward(struct ns_connection *, struct ns_connection *);
+NS_INTERNAL void ns_add_conn(struct ns_mgr *mgr, struct ns_connection *c);
+NS_INTERNAL void ns_remove_conn(struct ns_connection *c);
 
 #ifndef NS_DISABLE_FILESYSTEM
 NS_INTERNAL int find_index_file(char *, size_t, const char *, ns_stat_t *);
@@ -1649,15 +1652,16 @@ static void ns_ev_mgr_free(struct ns_mgr *mgr);
 static void ns_ev_mgr_add_conn(struct ns_connection *nc);
 static void ns_ev_mgr_remove_conn(struct ns_connection *nc);
 
-static void ns_add_conn(struct ns_mgr *mgr, struct ns_connection *c) {
+NS_INTERNAL void ns_add_conn(struct ns_mgr *mgr, struct ns_connection *c) {
+  c->mgr = mgr;
   c->next = mgr->active_connections;
   mgr->active_connections = c;
   c->prev = NULL;
   if (c->next != NULL) c->next->prev = c;
   ns_ev_mgr_add_conn(c);
 }
 
-static void ns_remove_conn(struct ns_connection *conn) {
+NS_INTERNAL void ns_remove_conn(struct ns_connection *conn) {
   if (conn->prev == NULL) conn->mgr->active_connections = conn->next;
   if (conn->prev) conn->prev->next = conn->next;
   if (conn->next) conn->next->prev = conn->prev;
@@ -3063,6 +3067,118 @@ int ns_check_ip_acl(const char *acl, uint32_t remote_ip) {
 
   return allowed == '+';
 }
+
+/* Move data from one connection to another */
+void ns_forward(struct ns_connection *from, struct ns_connection *to) {
+  ns_send(to, from->recv_mbuf.buf, from->recv_mbuf.len);
+  mbuf_remove(&from->recv_mbuf, from->recv_mbuf.len);
+}
+#ifdef NS_MODULE_LINES
+#line 1 "src/multithreading.c"
+/**/
+#endif
+/*
+ * Copyright (c) 2014 Cesanta Software Limited
+ * All rights reserved
+ */
+
+/* Amalgamated: #include "internal.h" */
+
+#ifdef NS_ENABLE_THREADS
+
+static void multithreaded_ev_handler(struct ns_connection *c, int ev, void *p);
+
+/*
+ * This thread function executes user event handler.
+ * It runs an event manager that has only one connection, until that
+ * connection is alive.
+ */
+static void *per_connection_thread_function(void *param) {
+  struct ns_connection *c = (struct ns_connection *) param;
+  struct ns_mgr m;
+
+  ns_mgr_init(&m, NULL);
+  ns_add_conn(&m, c);
+  while (m.active_connections != NULL) {
+    ns_mgr_poll(&m, 1000);
+  }
+  ns_mgr_free(&m);
+
+  return param;
+}
+
+static void link_conns(struct ns_connection *c1, struct ns_connection *c2) {
+  c1->priv_2 = c2;
+  c2->priv_2 = c1;
+}
+
+static void unlink_conns(struct ns_connection *c) {
+  struct ns_connection *peer = (struct ns_connection *) c->priv_2;
+  if (peer != NULL) {
+    peer->flags |= NSF_SEND_AND_CLOSE;
+    peer->priv_2 = NULL;
+  }
+  c->priv_2 = NULL;
+}
+
+static void forwarder_ev_handler(struct ns_connection *c, int ev, void *p) {
+  (void) p;
+  if (ev == NS_RECV && c->priv_2) {
+    ns_forward(c, c->priv_2);
+  } else if (ev == NS_CLOSE) {
+    unlink_conns(c);
+  }
+}
+
+static void spawn_handling_thread(struct ns_connection *nc) {
+  struct ns_mgr dummy = {};
+  sock_t sp[2];
+  struct ns_connection *c[2];
+
+  /*
+   * Create a socket pair, and wrap each socket into the connection with
+   * dummy event manager.
+   * c[0] stays in this thread, c[1] goes to another thread.
+   */
+  ns_socketpair(sp, SOCK_STREAM);
+  c[0] = ns_add_sock(&dummy, sp[0], forwarder_ev_handler);
+  c[1] = ns_add_sock(&dummy, sp[1], nc->listener->priv_1);
+
+  /* Interlink client connection with c[0] */
+  link_conns(c[0], nc);
+
+  /*
+   * Switch c[0] manager from the dummy one to the real one. c[1] manager
+   * will be set in another thread, allocated on stack of that thread.
+   */
+  ns_add_conn(nc->mgr, c[0]);
+
+  /*
+   * Dress c[1] as nc.
+   * TODO(lsm): code in accept_conn() looks similar. Refactor.
+   */
+  c[1]->listener = nc->listener;
+  c[1]->proto_handler = nc->proto_handler;
+  c[1]->proto_data = nc->proto_data;
+  c[1]->user_data = nc->user_data;
+
+  ns_start_thread(per_connection_thread_function, c[1]);
+}
+
+static void multithreaded_ev_handler(struct ns_connection *c, int ev, void *p) {
+  (void) p;
+  if (ev == NS_ACCEPT) {
+    spawn_handling_thread(c);
+    c->handler = forwarder_ev_handler;
+  }
+}
+
+void ns_enable_multithreading(struct ns_connection *nc) {
+  /* Wrap user event handler into our multithreaded_ev_handler */
+  nc->priv_1 = nc->handler;
+  nc->handler = multithreaded_ev_handler;
+}
+#endif
 #ifdef NS_MODULE_LINES
 #line 1 "src/http.c"
 /**/
@@ -3625,12 +3741,6 @@ static void free_http_proto_data(struct ns_connection *nc) {
   }
 }
 
-/* Move data from one connection to another */
-static void ns_forward(struct ns_connection *from, struct ns_connection *to) {
-  ns_send(to, from->recv_mbuf.buf, from->recv_mbuf.len);
-  mbuf_remove(&from->recv_mbuf, from->recv_mbuf.len);
-}
-
 static void transfer_file_data(struct ns_connection *nc) {
   struct proto_data_http *dp = (struct proto_data_http *) nc->proto_data;
   char buf[NS_MAX_HTTP_SEND_IOBUF];
@@ -5833,8 +5943,7 @@ void ns_hexdump_connection(struct ns_connection *nc, const char *path,
   if ((fp = fopen(path, "a")) != NULL) {
     ns_sock_to_str(nc->sock, src, sizeof(src), 3);
     ns_sock_to_str(nc->sock, dst, sizeof(dst), 7);
-    fprintf(fp, "%lu %p %s %s %s %d\n", (unsigned long) time(NULL),
-            nc->user_data, src,
+    fprintf(fp, "%lu %p %s %s %s %d\n", (unsigned long) time(NULL), nc, src,
             ev == NS_RECV ? "<-" : ev == NS_SEND ? "->" : ev == NS_ACCEPT
                                                               ? "<A"
                                                               : ev == NS_CONNECT

fossa.h

@@ -137,6 +137,7 @@
 #else
 #define fseeko(x, y, z) fseek((x), (y), (z))
 #endif
+#define random() rand()
 typedef int socklen_t;
 typedef unsigned char uint8_t;
 typedef unsigned int uint32_t;
@@ -626,6 +627,8 @@ struct ns_connection {
   void *proto_data;                 /* Protocol-specific data */
   ns_event_handler_t handler;       /* Event handler function */
   void *user_data;                  /* User-specific data */
+  void *priv_1;                     /* Used by ns_enable_multithreading() */
+  void *priv_2;                     /* Used by ns_enable_multithreading() */
   void *mgr_data; /* Implementation-specific event manager's data. */
 
   unsigned long flags;
@@ -922,6 +925,15 @@ int ns_resolve(const char *domain_name, char *ip_addr_buf, size_t buf_len);
  */
 int ns_check_ip_acl(const char *acl, uint32_t remote_ip);
 
+/*
+ * Enable multi-threaded handling for the given listening connection `nc`.
+ * For each accepted connection, Mongoose will create a separate thread
+ * and run event handler in that thread. Thus, if an event hanler is doing
+ * a blocking call or some long computation, that will not slow down
+ * other connections.
+ */
+void ns_enable_multithreading(struct ns_connection *nc);
+
 #ifdef __cplusplus
 }
 #endif /* __cplusplus */

src/http.c

@@ -556,12 +556,6 @@ static void free_http_proto_data(struct ns_connection *nc) {
   }
 }
 
-/* Move data from one connection to another */
-static void ns_forward(struct ns_connection *from, struct ns_connection *to) {
-  ns_send(to, from->recv_mbuf.buf, from->recv_mbuf.len);
-  mbuf_remove(&from->recv_mbuf, from->recv_mbuf.len);
-}
-
 static void transfer_file_data(struct ns_connection *nc) {
   struct proto_data_http *dp = (struct proto_data_http *) nc->proto_data;
   char buf[NS_MAX_HTTP_SEND_IOBUF];

src/internal.h

@@ -67,6 +67,9 @@ NS_INTERNAL struct ns_connection *ns_finish_connect(struct ns_connection *nc,
 NS_INTERNAL int ns_parse_address(const char *str, union socket_address *sa,
                                  int *proto, char *host, size_t host_len);
 NS_INTERNAL void ns_call(struct ns_connection *, int ev, void *ev_data);
+NS_INTERNAL void ns_forward(struct ns_connection *, struct ns_connection *);
+NS_INTERNAL void ns_add_conn(struct ns_mgr *mgr, struct ns_connection *c);
+NS_INTERNAL void ns_remove_conn(struct ns_connection *c);
 
 #ifndef NS_DISABLE_FILESYSTEM
 NS_INTERNAL int find_index_file(char *, size_t, const char *, ns_stat_t *);

src/modules.mk

@@ -28,6 +28,7 @@ SOURCES = $(COMMON)/mbuf.c \
           $(COMMON)/dirent.c \
           $(FROZEN)/frozen.c \
           net.c \
+          multithreading.c \
           http.c \
           util.c \
           json-rpc.c \