add support for parallel requests

client/build.sh

@@ -9,14 +9,16 @@ ES6_FILE="out/libcurl_module.mjs"
 MODULE_FILE="out/emscripten_compiled.js"
 WRAPPER_SOURCE="main.js"
 
-EXPORTED_FUNCS="_load_certs,_perform_request"
+EXPORTED_FUNCS="_init_curl,_start_request,_request_loop"
 RUNTIME_METHODS="addFunction,removeFunction,allocate,ALLOC_NORMAL"
 COMPILER_OPTIONS="-o $MODULE_FILE -lcurl -lssl -lcrypto -lcjson -lz -lbrotlidec -lbrotlicommon -I $INCLUDE_DIR -L $LIB_DIR"
-EMSCRIPTEN_OPTIONS="-lwebsocket.js -sASYNCIFY -sALLOW_TABLE_GROWTH -sEXPORTED_FUNCTIONS=$EXPORTED_FUNCS -sEXPORTED_RUNTIME_METHODS=$RUNTIME_METHODS"
+EMSCRIPTEN_OPTIONS="-lwebsocket.js -sASYNCIFY -sASYNCIFY_ONLY=start_request,request_loop -sALLOW_TABLE_GROWTH -sEXPORTED_FUNCTIONS=$EXPORTED_FUNCS -sEXPORTED_RUNTIME_METHODS=$RUNTIME_METHODS"
 
 if [ "$1" = "release" ]; then
-  COMPILER_OPTIONS="-O3 $COMPILER_OPTIONS"
+  COMPILER_OPTIONS="-O3 -flto $COMPILER_OPTIONS"
   EMSCRIPTEN_OPTIONS="-sSINGLE_FILE $EMSCRIPTEN_OPTIONS"
+else
+  COMPILER_OPTIONS="$COMPILER_OPTIONS --profiling"
 fi
 
 #ensure deps are compiled

client/main.c

@@ -8,12 +8,24 @@
 #include "curl/header.h"
 #include "cjson/cJSON.h"
 #include "cacert.h"
+#include "curl/multi.h"
 
 typedef void(*DataCallback)(char* chunk_ptr, int chunk_size);
 typedef void(*EndCallback)(int error, char* response_json);
+void finish_request(CURLMsg *curl_msg);
 
 #define ERROR_REDIRECT_DISALLOWED -1
 
+CURLM *multi_handle;
+int request_active = 0;
+
+struct RequestInfo {
+  int abort_on_redirect;
+  struct CURLMsg *curl_msg;
+  struct curl_slist* headers_list;
+  EndCallback end_callback;
+};
+
 int write_function(void *data, size_t size, size_t nmemb, DataCallback data_callback) {
   long real_size = size * nmemb;
   char* chunk = malloc(real_size);
@@ -23,14 +35,36 @@ int write_function(void *data, size_t size, size_t nmemb, DataCallback data_call
   return real_size;
 }
 
-void perform_request(const char* url, const char* json_params, DataCallback data_callback, EndCallback end_callback, const char* body, int body_length) {
+void request_loop() {
-  CURL *http_handle;
+  CURLMcode mc;
-  CURLM *multi_handle;
+  struct CURLMsg *curl_msg;
-  int still_running = 1;
+  request_active = 1;
-  int abort_on_redirect = 0;
+  do {
+    mc = curl_multi_perform(multi_handle, &request_active);
  
-  curl_global_init(CURL_GLOBAL_DEFAULT);
+    if(!mc)
-  http_handle = curl_easy_init();
+      mc = curl_multi_poll(multi_handle, NULL, 0, 1000, NULL);
+ 
+    if(mc) {
+      fprintf(stderr, "curl_multi_poll() failed, code %d.\n", (int)mc);
+      break;
+    }
+
+    int msgq = 0;
+    curl_msg = curl_multi_info_read(multi_handle, &msgq);
+    if (curl_msg && curl_msg->msg == CURLMSG_DONE) {
+      finish_request(curl_msg);
+    }
+
+    //ensure we dont block the main thread
+    emscripten_sleep(0);
+ 
+  } while(request_active);
+}
+
+void start_request(const char* url, const char* json_params, DataCallback data_callback, EndCallback end_callback, const char* body, int body_length) {
+  CURL *http_handle = curl_easy_init();  
+  int abort_on_redirect = 0;
  
   curl_easy_setopt(http_handle, CURLOPT_URL, url);
   curl_easy_setopt(http_handle, CURLOPT_CAINFO, "/cacert.pem");
@@ -95,35 +129,30 @@ void perform_request(const char* url, const char* json_params, DataCallback data
     curl_easy_setopt(http_handle, CURLOPT_POSTFIELDSIZE, body_length);
   }
 
-  multi_handle = curl_multi_init();
+  struct RequestInfo *request_info = malloc(sizeof(struct RequestInfo));
+  request_info->abort_on_redirect = abort_on_redirect;
+  request_info->curl_msg = NULL;
+  request_info->headers_list = headers_list;
+  request_info->end_callback = end_callback;
+  curl_easy_setopt(http_handle, CURLOPT_PRIVATE, request_info);
+  
   curl_multi_add_handle(multi_handle, http_handle);
-  
+  if (!request_active) {
-  CURLMcode mc;
+    request_loop();
-  struct CURLMsg *m;
-  do {
-    mc = curl_multi_perform(multi_handle, &still_running);
- 
-    if(!mc)
-      mc = curl_multi_poll(multi_handle, NULL, 0, 1000, NULL);
- 
-    if(mc) {
-      fprintf(stderr, "curl_multi_poll() failed, code %d.\n", (int)mc);
-      break;
   }
+}
 
-    int msgq = 0;
+void finish_request(CURLMsg *curl_msg) {
-    m = curl_multi_info_read(multi_handle, &msgq);
+  //get initial request info from the http handle
+  struct RequestInfo *request_info;
+  CURL *http_handle = curl_msg->easy_handle;
+  curl_easy_getinfo(http_handle, CURLINFO_PRIVATE, &request_info);
 
-    //ensure we dont block the main thread
+  int error = (int) curl_msg->data.result;
-    emscripten_sleep(0);
- 
-  } while(still_running);
-  
-  int error = (int) m->data.result;
   long response_code;
   curl_easy_getinfo(http_handle, CURLINFO_RESPONSE_CODE, &response_code);
 
-  if (abort_on_redirect && response_code / 100 == 3) {
+  if (request_info->abort_on_redirect && response_code / 100 == 3) {
     error = ERROR_REDIRECT_DISALLOWED;
   }
 
@@ -157,13 +186,11 @@ void perform_request(const char* url, const char* json_params, DataCallback data
   cJSON_Delete(response_json);
   
   //clean up curl
-  curl_slist_free_all(headers_list);
+  curl_slist_free_all(request_info->headers_list);
   curl_multi_remove_handle(multi_handle, http_handle);
   curl_easy_cleanup(http_handle);
-  curl_multi_cleanup(multi_handle);
+  (*request_info->end_callback)(error, response_json_str);
-  curl_global_cleanup();
+  free(request_info);
-
-  (*end_callback)(error, response_json_str);
 }
 
 char* copy_bytes(const char* ptr, const int size) {
@@ -172,7 +199,10 @@ char* copy_bytes(const char* ptr, const int size) {
   return new_ptr;
 }
 
-void load_certs() {
+void init_curl() {
+  curl_global_init(CURL_GLOBAL_DEFAULT);
+  multi_handle = curl_multi_init();
+  
   FILE *file = fopen("/cacert.pem", "wb");
   fwrite(_cacert_pem, 1, _cacert_pem_len, file);
   fclose(file);

client/main.js

@@ -149,7 +149,7 @@ function perform_request(url, params, js_data_callback, js_end_callback, body=nu
 
   end_callback_ptr = Module.addFunction(end_callback, "vii");
   data_callback_ptr = Module.addFunction(data_callback, "vii");
-  _perform_request(url_ptr, params_ptr, data_callback_ptr, end_callback_ptr, body_ptr, body_length);
+  _start_request(url_ptr, params_ptr, data_callback_ptr, end_callback_ptr, body_ptr, body_length);
   _free(params_ptr);
 }
 
@@ -231,7 +231,7 @@ function set_websocket_url(url) {
 
 function main() {
   console.log("emscripten module loaded");
-  _load_certs();
+  _init_curl();
   set_websocket_url(websocket_url);
 
   let load_event = new Event("libcurl_load");

server/main.py

@@ -2,7 +2,7 @@ import asyncio
 from websockets.server import serve
 from websockets.exceptions import ConnectionClosed
 
-buffer_size = 64*1024
+buffer_size = 1024*1024
 
 class Connection:
   def __init__(self, ws, path):
@@ -48,7 +48,7 @@ async def connection_handler(websocket, path):
   await asyncio.gather(ws_handler, tcp_handler)
 
 async def main():
-  async with serve(connection_handler, "127.0.0.1", 6001):
+  async with serve(connection_handler, "127.0.0.1", 6001, subprotocols=["binary"]):
     await asyncio.Future()
 
 if __name__ == "__main__":