second-transfer 0.2.0.0 → 0.3.0.2
raw patch · 13 files changed
+2182/−2100 lines, 13 files
Files
- README.md +10/−5
- cbits/tlsinc.c +303/−308
- hs-src/SecondTransfer.hs +15/−0
- hs-src/SecondTransfer/Http2/Framer.cpphs +536/−0
- hs-src/SecondTransfer/Http2/Framer.hs +0/−531
- hs-src/SecondTransfer/Http2/Session.cpphs +869/−0
- hs-src/SecondTransfer/Http2/Session.hs +0/−897
- hs-src/SecondTransfer/MainLoop.hs +3/−0
- hs-src/SecondTransfer/MainLoop/Logging.hs +59/−0
- hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs +357/−0
- hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs +0/−350
- second-transfer.cabal +19/−5
- tests/tests-hs-src/compiling_ok.hs +11/−4
README.md view
@@ -24,6 +24,11 @@ $ cabal install second-transfer +Running the tests+-----------------++ $ cabal test+ Example ------- @@ -39,12 +44,12 @@ - Version 0.2: Absolutely minimal amount of unit tests. +- Version 0.3: More sensible logging.+ Pending: -- Version 0.3: Better examples of usage. +- Better examples. -- Version 0.4: By-the-book stream state management. In particular, ensure - that we are not allowing frames to come off-order from the - other peer. +- Epoll I/O management -- Version 0.5: Guaranties about early resource release...+- Benchmarking.
cbits/tlsinc.c view
@@ -63,7 +63,7 @@ //////////////////////////////////////////////////////////////////////// - + static int threads_are_up = 0; // Leaky implementation now @@ -71,79 +71,71 @@ { if (conn->socket) {- close (conn->socket);- conn->socket = 0;+ close (conn->socket);+ conn->socket = 0; } if (conn->sslContext) {- SSL_CTX_free(conn->sslContext);- conn->sslContext = 0;+ SSL_CTX_free(conn->sslContext);+ conn->sslContext = 0; } - free(conn);+ free(conn); } // For this example, we'll be testing on openssl.org -// Wired-in data that I will need to use? -#define CERTIFICATE_PLACE "/home/alcides/projects/rede/mimic/config/servercert.pem"-#define PRIVKEY_PLACE "/home/alcides/projects/rede/mimic/config/privkey.pem"-// This should bind to local interface-#define SERVER "www.httpdos.com"-// Grr-#define PORT 1060- // Establish a regular tcp connection static int tcpStart (char* hostname, int portno, int* errorh) {- int error, handle;- *errorh = 0;- struct hostent *host;- struct sockaddr_in server;+ int error, handle;+ *errorh = 0;+ struct hostent *host;+ struct sockaddr_in server; - bzero((char *) &server, sizeof(server));- + bzero((char *) &server, sizeof(server)); - host = gethostbyname (hostname);- handle = socket (AF_INET, SOCK_STREAM, 0);- int one = 1;- setsockopt(handle, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));- if (handle == -1)++ host = gethostbyname (hostname);+ handle = socket (AF_INET, SOCK_STREAM, 0);+ int one = 1;+ setsockopt(handle, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));+ if (handle == -1) {- perror ("Socket");- handle = 0;- *errorh = BAD_HAPPENED;+ perror ("Socket could not be created");+ handle = 0;+ *errorh = BAD_HAPPENED; }- else+ else {- server.sin_family = AF_INET;- server.sin_port = htons (portno);- server.sin_addr = *((struct in_addr *) host->h_addr);- bzero (&(server.sin_zero), 8);+ server.sin_family = AF_INET;+ server.sin_port = htons (portno);+ server.sin_addr = *((struct in_addr *) host->h_addr);+ bzero (&(server.sin_zero), 8); - error = bind (handle, (struct sockaddr *) &server,- sizeof (struct sockaddr));- if (error == -1)- {- perror ("Bind");- handle = 0;- *errorh = BAD_HAPPENED;- }- else + error = bind (handle, (struct sockaddr *) &server,+ sizeof (struct sockaddr));+ if (error == -1) {- error = listen(handle, 5);- if ( error != 0 )- {- perror("Listen");+ perror ("Bind"); handle = 0; *errorh = BAD_HAPPENED;- } }+ else + {+ error = listen(handle, 5);+ if ( error != 0 )+ {+ perror("Listen");+ handle = 0;+ *errorh = BAD_HAPPENED;+ }+ } } - return handle;+ return handle; } // Dh_Callback {{{@@ -177,22 +169,22 @@ 0x69,0x49,0x22,0x24,0xB6,0xB8,0x45,0xCA,0x8F,0x35,0x71,0x45, 0x4C,0x65,0x38,0x11,0x21,0xD5,0x39,0x03,0x4D,0x05,0xBB,0x95, 0x43,0x58,0xBB,0xF3,- };+ }; static unsigned char dh2236_g[]={ 0x02,- };+ }; DH *dh; if ((dh=DH_new()) == NULL) return(NULL); dh->p=BN_bin2bn(dh2236_p,sizeof(dh2236_p),NULL); dh->g=BN_bin2bn(dh2236_g,sizeof(dh2236_g),NULL); if ((dh->p == NULL) || (dh->g == NULL))- { DH_free(dh); return(NULL); }+ { DH_free(dh); return(NULL); } return(dh); } DH *get_dh2048() {-static unsigned char dh2048_p[]={+ static unsigned char dh2048_p[]={ 0xCE,0x28,0x14,0x1C,0xEF,0x22,0x1D,0x86,0xEA,0x10,0x00,0x50, 0x24,0x42,0x95,0xC3,0x07,0x5A,0x87,0xED,0x0F,0xC5,0xDC,0x0F, 0x5E,0x7E,0x69,0x25,0x85,0x90,0x39,0x60,0x1E,0x87,0x5C,0x4B,@@ -215,22 +207,22 @@ 0x06,0x59,0x4F,0xE7,0x5B,0xD2,0xFC,0x25,0xC6,0x2C,0xA3,0xE8, 0x05,0xE5,0x8D,0xC2,0x94,0x76,0x90,0x63,0x29,0xB4,0xEE,0x2D, 0xD6,0x50,0x83,0x8B,- };-static unsigned char dh2048_g[]={+ };+ static unsigned char dh2048_g[]={ 0x02,- };-DH *dh;+ };+ DH *dh; -if ((dh=DH_new()) == NULL) return(NULL);-dh->p=BN_bin2bn(dh2048_p,sizeof(dh2048_p),NULL);-dh->g=BN_bin2bn(dh2048_g,sizeof(dh2048_g),NULL);-if ((dh->p == NULL) || (dh->g == NULL))- { DH_free(dh); return(NULL); }-return(dh);+ if ((dh=DH_new()) == NULL) return(NULL);+ dh->p=BN_bin2bn(dh2048_p,sizeof(dh2048_p),NULL);+ dh->g=BN_bin2bn(dh2048_g,sizeof(dh2048_g),NULL);+ if ((dh->p == NULL) || (dh->g == NULL))+ { DH_free(dh); return(NULL); }+ return(dh); } DH *get_dh1024()- {+{ static unsigned char dh1024_p[]={ 0x87,0xF4,0xE5,0x5A,0x1E,0x7F,0x98,0x83,0xFD,0x23,0x6A,0xF3, 0x8C,0xE8,0x2F,0x35,0x64,0x3D,0x13,0x34,0x5B,0x0A,0x52,0xEC,@@ -243,17 +235,17 @@ 0x69,0xF8,0x07,0x4F,0x33,0x17,0x72,0xCC,0xEB,0xBD,0x77,0xD1, 0x9C,0x40,0xA6,0x55,0xE3,0x87,0x76,0x66,0x3B,0xE3,0xE4,0x6A, 0x4B,0x6B,0xF1,0x92,0x1A,0x3A,0x25,0xC3,- };+ }; static unsigned char dh1024_g[]={ 0x02,- };+ }; DH *dh; if ((dh=DH_new()) == NULL) return(NULL); dh->p=BN_bin2bn(dh1024_p,sizeof(dh1024_p),NULL); dh->g=BN_bin2bn(dh1024_g,sizeof(dh1024_g),NULL); if ((dh->p == NULL) || (dh->g == NULL))- { DH_free(dh); return(NULL); }+ { DH_free(dh); return(NULL); } return (dh); } @@ -264,8 +256,8 @@ // Setup dh params static void setup_dh_parms(SSL_CTX *sslContext) {- /* Set up ephemeral DH stuff */- SSL_CTX_set_tmp_dh_callback(sslContext, tmp_dh_callback);+ /* Set up ephemeral DH stuff */+ SSL_CTX_set_tmp_dh_callback(sslContext, tmp_dh_callback); } static DH *tmp_dh_callback(SSL *s, int is_export, int keylength)@@ -286,7 +278,7 @@ break; default: /* Generating a key on the fly is very costly, so use what is there */- printf("Keylength %d \n", keylength);+ printf("UNEXPECTED: Keylength %d \n", keylength); } return(dh_tmp); }@@ -295,97 +287,97 @@ static int protocol_select (- SSL *ssl,- const unsigned char **out,- unsigned char *outlen,- const unsigned char *in,- unsigned int inlen,- void *arg)+ SSL *ssl,+ const unsigned char **out,+ unsigned char *outlen,+ const unsigned char *in,+ unsigned int inlen,+ void *arg) {- // Oh well, C is a verbose beast- connection_t* conn = (connection_t*) arg;- static char output[64];-- char* incursor = (char*) in;-- while (incursor < (char*)in + inlen )- {- // Got a protocol.... can I satisfy it?- char sublen = *incursor;- printf("offered prot %.*s \n", sublen, incursor+1);+ // Oh well, C is a verbose beast+ connection_t* conn = (connection_t*) arg;+ static char output[64]; - char* stored_cursor = conn->protocol_list;- int sto_protocol = 0;+ char* incursor = (char*) in; - while( stored_cursor < conn->protocol_list + conn->protocol_list_length)+ while (incursor < (char*)in + inlen ) {- char sublen2 = *stored_cursor;+ // Got a protocol.... can I satisfy it?+ char sublen = *incursor;+ //printf("offered prot %.*s \n", sublen, incursor+1); - if (sublen != sublen2)- {- - } else {- int cmpresult = strncmp( incursor + 1, stored_cursor + 1, sublen);- if (cmpresult == 0)+ char* stored_cursor = conn->protocol_list;+ int sto_protocol = 0;++ while( stored_cursor < conn->protocol_list + conn->protocol_list_length) {- // They are equal, choose this one...- strncpy( output, stored_cursor+1, sublen);- *outlen = sublen;- *out = output;+ char sublen2 = *stored_cursor; + if (sublen != sublen2)+ { - return SSL_TLSEXT_ERR_OK;+ } else {+ int cmpresult = strncmp( incursor + 1, stored_cursor + 1, sublen);+ if (cmpresult == 0)+ {+ // They are equal, choose this one...+ strncpy( output, stored_cursor+1, sublen);+ *outlen = sublen;+ *out = output;+++ return SSL_TLSEXT_ERR_OK;+ }+ }+ sto_protocol += 1;+ stored_cursor += (1+sublen2); }- }- sto_protocol += 1;- stored_cursor += (1+sublen2);- } - incursor += (1+sublen);- }+ incursor += (1+sublen);+ } - // I think this is what should be returned - return -1;+ // I think this is what should be returned + return -1; } int lookup_protocol(- char* selected, int selected_len, - char* myprotocol_list, int mpl_len- )+ char* selected, int selected_len, + char* myprotocol_list, int mpl_len+ ) {- char* incursor = selected;-- char* stored_cursor = myprotocol_list;- int sto_protocol = 0;+ char* incursor = selected; - if (selected_len == 0)- {- // No protocol was selected- return -2;- }+ char* stored_cursor = myprotocol_list;+ int sto_protocol = 0; - while( stored_cursor < myprotocol_list + mpl_len)- {- char sublen2 = *stored_cursor;+ if (selected_len == 0)+ {+ // No protocol was selected+ return -2;+ } - if (sublen2 != selected_len)+ while( stored_cursor < myprotocol_list + mpl_len) {- - } else {- int cmpresult = strncmp( selected, stored_cursor + 1, sublen2);- if (cmpresult == 0)- {- // They are equal, choose this one...- return sto_protocol;- }+ char sublen2 = *stored_cursor;++ if (sublen2 != selected_len)+ {+ // It is not this cone, continue+ } else {+ int cmpresult = strncmp( selected, stored_cursor + 1, sublen2);+ if (cmpresult == 0)+ {+ // They are equal, choose this one...+ return sto_protocol;+ }+ }+ sto_protocol += 1;+ stored_cursor += (1+sublen2); }- sto_protocol += 1;- stored_cursor += (1+sublen2);- } - // I think this is what should be returned - return -1;+ // I think this is what should be returned + return -1; } static int ssl_servername_cb(SSL *s, int *ad, void *arg)@@ -415,126 +407,129 @@ // Establish a connection using an SSL layer static connection_t *sslStart (- char* certificate_filename, char* privkey_filename, char* hostname, int portno,- char* protocol_list, int protocol_list_length- )+ char* certificate_filename, char* privkey_filename, char* hostname, int portno,+ char* protocol_list, int protocol_list_length+ ) {- int result;- connection_t *c;-- if (! threads_are_up)- {- threads_are_up = 1;- thread_setup();- }-- c = malloc (sizeof (connection_t));- c->sslContext = NULL;- c->protocol_list = (char*) malloc(protocol_list_length);- strncpy( c->protocol_list, protocol_list, protocol_list_length );- c->protocol_list_length = protocol_list_length;- c->socket = tcpStart (hostname, portno, &result);+ int result;+ connection_t *c; - if ( result )- {- return 0;- }+ if (! threads_are_up)+ {+ threads_are_up = 1;+ thread_setup();+ } - if (c->socket)- {- // Register the error strings for libcrypto & libssl- SSL_load_error_strings ();- // Register the available ciphers and digests- SSL_library_init ();+ c = malloc (sizeof (connection_t));+ c->sslContext = NULL;+ c->protocol_list = (char*) malloc(protocol_list_length);+ strncpy( c->protocol_list, protocol_list, protocol_list_length );+ c->protocol_list_length = protocol_list_length;+ c->socket = tcpStart (hostname, portno, &result); - // New context saying we are a server, and using SSL 2 or 3- c->sslContext = SSL_CTX_new( TLSv1_2_server_method() );- if (c->sslContext == NULL)+ if ( result ) {- ERR_print_errors_fp (stderr);- perror("Could not create context");- return 0;+ return 0; } + if (c->socket)+ {+ // Register the error strings for libcrypto & libssl+ SSL_load_error_strings ();+ // Register the available ciphers and digests+ SSL_library_init (); - // Now I set a few options....- /*SSL_CTX_set_verify(c->sslContext, NULL );*/- // const long flags = - // SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_TLSv1 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;- const long flags = - SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;- SSL_CTX_set_options(c->sslContext, flags);- setup_dh_parms( c->sslContext );- SSL_CTX_set_ecdh_auto (c->sslContext, 1);+ // New context saying we are a server, and using SSL 2 or 3+ c->sslContext = SSL_CTX_new( TLSv1_2_server_method() );+ if (c->sslContext == NULL)+ {+ ERR_print_errors_fp (stderr);+ perror("Could not create context");+ return 0;+ } - // Give the impression that we are using SNI- SSL_CTX_set_tlsext_servername_callback(c->sslContext, ssl_servername_cb); + // Now I set a few options....+ /*SSL_CTX_set_verify(c->sslContext, NULL );*/+ // const long flags = + // SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_TLSv1 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;+ const long flags = + SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;+ SSL_CTX_set_options(c->sslContext, flags);+ setup_dh_parms( c->sslContext );+ SSL_CTX_set_ecdh_auto (c->sslContext, 1); + // Give the impression that we are using SNI+ SSL_CTX_set_tlsext_servername_callback(c->sslContext, ssl_servername_cb); - // The only cipher supported by HTTP/2 ... sort of.- result = SSL_CTX_set_cipher_list(c->sslContext, "ECDHE-RSA-AES128-GCM-SHA256");- /*result = SSL_CTX_set_cipher_list(c->sslContext, "DEFAULT");*/ - // Be sure we are able to do ALPN- // void SSL_CTX_set_alpn_select_cb(SSL_CTX *ctx,- // int (*cb) (SSL *ssl,- // const unsigned char **out,- // unsigned char *outlen,- // const unsigned char *in,- // unsigned int inlen,- // void *arg), void *arg);- SSL_CTX_set_alpn_select_cb( c->sslContext, protocol_select, c); - if ( result != 1 )- {- ERR_print_errors_fp (stderr);- perror("Could not set cipher");- return 0;- }- // The certificate (well-wired)- SSL_CTX_use_certificate_file(c->sslContext, certificate_filename, SSL_FILETYPE_PEM);- SSL_CTX_use_PrivateKey_file(c->sslContext, privkey_filename, SSL_FILETYPE_PEM);- // Check the private key - result = SSL_CTX_check_private_key(c->sslContext);- if ( result != 1)- {- perror("Check private key failed");- return 0;- }- // //////////////////////////////////////////+ // The only cipher supported by HTTP/2 ... sort of.+ result = SSL_CTX_set_cipher_list(c->sslContext, "ECDHE-RSA-AES128-GCM-SHA256");+ /*result = SSL_CTX_set_cipher_list(c->sslContext, "DEFAULT");*/ + // Be sure we are able to do ALPN+ // void SSL_CTX_set_alpn_select_cb(SSL_CTX *ctx,+ // int (*cb) (SSL *ssl,+ // const unsigned char **out,+ // unsigned char *outlen,+ // const unsigned char *in,+ // unsigned int inlen,+ // void *arg), void *arg);+ SSL_CTX_set_alpn_select_cb( c->sslContext, protocol_select, c);++ if ( result != 1 )+ {+ ERR_print_errors_fp (stderr);+ perror("Could not set cipher");+ return 0;+ }+ // The certificate (well-wired)+ SSL_CTX_use_certificate_file(c->sslContext, certificate_filename, SSL_FILETYPE_PEM);+ SSL_CTX_use_PrivateKey_file(c->sslContext, privkey_filename, SSL_FILETYPE_PEM);+ // Check the private key + result = SSL_CTX_check_private_key(c->sslContext);+ if ( result != 1)+ {+ char msg[600];+ snprintf(msg, 600, "Check certificate file %s and privkey %s failed", certificate_filename,+ privkey_filename);+ perror(msg);+ return 0;+ }+ // //////////////////////////////////////////+ }- else+ else {- perror ("Connect failed");+ perror ("Connect failed"); } - return c;+ return c; } connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno,- char* my_protocol_list, int protocol_list_length- )+ char* my_protocol_list, int protocol_list_length+ ) {- const int len = strlen(hostname);- if ( len > 0 && hostname[len-1] == '\n' )- { - printf("HELLO THERE. I'M A FLIMSY C FUNCTION AND I CAN NOT REMOVE THE END-OF-LINE IN THE PROVIDED HOST NAME WITHOUT A TON OF LINES. I'M GOING TO SEGFAULT NOW. BYE.\n");- *( (int*) 0) = 42; // <-- This is the answer- }+ const int len = strlen(hostname);+ if ( len > 0 && hostname[len-1] == '\n' )+ { + printf("HELLO THERE. I'M A FLIMSY C FUNCTION AND I CAN NOT REMOVE THE END-OF-LINE IN THE PROVIDED HOST NAME WITHOUT A TON OF LINES. I'M GOING TO SEGFAULT NOW. BYE.\n");+ *( (int*) 0) = 42; // <-- This is the answer+ } - return sslStart(- certificate_filename, privkey_filename, hostname, - portno, my_protocol_list, protocol_list_length); + return sslStart(+ certificate_filename, privkey_filename, hostname, + portno, my_protocol_list, protocol_list_length); } int wait_for_connection(- connection_t* c, - int microseconds,- wired_session_t** wired_session)+ connection_t* c, + int microseconds,+ wired_session_t** wired_session) { int clilen, newsockfd; // Don't return anything if there's a failure...@@ -563,24 +558,24 @@ if ( retval == -1 ) {- if ( errno == EINTR )- {- // printf(".");- can_go = 0;- } else {- perror("select()");- return BAD_HAPPENED;- }+ if ( errno == EINTR )+ {+ // printf(".");+ can_go = 0;+ } else {+ perror("select()");+ return BAD_HAPPENED;+ } } else if (retval > 0) {- // We got data, just let it go...- // printf("letitgo\n");- can_go = 1;+ // We got data, just let it go...+ // printf("letitgo\n");+ can_go = 1; } else {- // We didn't get data, finish and terminate- // printf("timeo\n");- return TIMEOUT_REACHED;+ // We didn't get data, finish and terminate+ // printf("timeo\n");+ return TIMEOUT_REACHED; } }@@ -600,7 +595,7 @@ perror("ERROR on accept"); return BAD_HAPPENED; }- + // Create an SSL struct for the connection result->socket = newsockfd;@@ -631,11 +626,11 @@ // After this one is okej, see which protocol was selected const unsigned char* out_protocol; unsigned pr_len; SSL_get0_alpn_selected(result->sslHandle, &out_protocol,- &pr_len);+ &pr_len); // Wonder who is in charge of releasing that buffer... result -> protocol_index = lookup_protocol( - (char*)out_protocol, pr_len, - c->protocol_list, c->protocol_list_length);+ (char*)out_protocol, pr_len, + c->protocol_list, c->protocol_list_length); *wired_session = result; @@ -645,17 +640,17 @@ // Disconnect & free connection struct /*static void sslDisconnect (connection *c)*/ /*{*/- /*if (c->socket)*/- /*close (c->socket);*/- /*if (c->sslHandle)*/- /*{*/- /*SSL_shutdown (c->sslHandle);*/- /*SSL_free (c->sslHandle);*/- /*}*/- /*if (c->sslContext)*/- /*SSL_CTX_free (c->sslContext);*/+/*if (c->socket)*/+/*close (c->socket);*/+/*if (c->sslHandle)*/+/*{*/+/*SSL_shutdown (c->sslHandle);*/+/*SSL_free (c->sslHandle);*/+/*}*/+/*if (c->sslContext)*/+/*SSL_CTX_free (c->sslContext);*/ - /*free (c);*/+/*free (c);*/ /*}*/ @@ -709,77 +704,77 @@ #define MUTEX_LOCK(x) pthread_mutex_lock(&(x)) #define MUTEX_UNLOCK(x) pthread_mutex_unlock(&(x)) #define THREAD_ID pthread_self( )- - ++ void handle_error(const char *file, int lineno, const char *msg){- fprintf(stderr, "** %s:%d %s\n", file, lineno, msg);- ERR_print_errors_fp(stderr);- /* exit(-1); */ - }- + fprintf(stderr, "** %s:%d %s\n", file, lineno, msg);+ ERR_print_errors_fp(stderr);+ /* exit(-1); */ +}+ /* This array will store all of the mutexes available to OpenSSL. */ static MUTEX_TYPE *mutex_buf= NULL;- - ++ static void locking_function(int mode, int n, const char * file, int line) {- if (mode & CRYPTO_LOCK)- MUTEX_LOCK(mutex_buf[n]);- else- MUTEX_UNLOCK(mutex_buf[n]);+ if (mode & CRYPTO_LOCK)+ MUTEX_LOCK(mutex_buf[n]);+ else+ MUTEX_UNLOCK(mutex_buf[n]); }- + static unsigned long id_function(void) {- return ((unsigned long)THREAD_ID);+ return ((unsigned long)THREAD_ID); } void dispose_wired_session(wired_session_t* ws) {- if (ws == 0)- return ;- if ( ws-> sslHandle )- {- SSL_shutdown( ws->sslHandle );- ws -> sslHandle = 0;- }- if (ws->socket)- {- close(ws->socket);- ws -> socket = 0;- }- free(ws);+ if (ws == 0)+ return ;+ if ( ws-> sslHandle )+ {+ SSL_shutdown( ws->sslHandle );+ ws -> sslHandle = 0;+ }+ if (ws->socket)+ {+ close(ws->socket);+ ws -> socket = 0;+ }+ free(ws); } int thread_setup(void) {- int i;+ int i; - // printf("Threads setup\n");- - mutex_buf = malloc(CRYPTO_num_locks( ) * sizeof(MUTEX_TYPE));- if (!mutex_buf)- return 0;- for (i = 0; i < CRYPTO_num_locks( ); i++)- MUTEX_SETUP(mutex_buf[i]);- CRYPTO_set_id_callback(id_function);- CRYPTO_set_locking_callback(locking_function);- return 1;+ // printf("Threads setup\n");++ mutex_buf = malloc(CRYPTO_num_locks( ) * sizeof(MUTEX_TYPE));+ if (!mutex_buf)+ return 0;+ for (i = 0; i < CRYPTO_num_locks( ); i++)+ MUTEX_SETUP(mutex_buf[i]);+ CRYPTO_set_id_callback(id_function);+ CRYPTO_set_locking_callback(locking_function);+ return 1; }- + int thread_cleanup(void) {- int i;- - if (!mutex_buf)- return 0;- CRYPTO_set_id_callback(NULL);- CRYPTO_set_locking_callback(NULL);- for (i = 0; i < CRYPTO_num_locks( ); i++)- MUTEX_CLEANUP(mutex_buf[i]);- free(mutex_buf);- mutex_buf = NULL;- return 1;+ int i;++ if (!mutex_buf)+ return 0;+ CRYPTO_set_id_callback(NULL);+ CRYPTO_set_locking_callback(NULL);+ for (i = 0; i < CRYPTO_num_locks( ); i++)+ MUTEX_CLEANUP(mutex_buf[i]);+ free(mutex_buf);+ mutex_buf = NULL;+ return 1; }
hs-src/SecondTransfer.hs view
@@ -47,6 +47,12 @@ saysHello :: DataAndConclusion saysHello = do + -- The data in each yield will be automatically split across multiple + -- data frames if needed, so you can yield a large block of contents here+ -- if you wish. + -- If you do multiple yields, no data will be left buffered between them, + -- so that you can for example implement a chat client in a single HTTP/2 stream.+ -- Pity browsers hardly support that. yield "Hello world!" -- No footers return []@@ -141,6 +147,15 @@ ,TLSLayerGenericProblem(..) ,FinishRequest(..)++ -- * Logging + --+ -- | The library uses hslogger for its logging. Since logging is + -- expensive, most of the instrumentation needs to be activated + -- at compile time by activating the "debug" flag. And then you + -- need to configure the loggers. The function `enableConsoleLogging` + -- configures them to output a lot of information to standard output.+ ,enableConsoleLogging ) where import SecondTransfer.MainLoop.CoherentWorker
+ hs-src/SecondTransfer/Http2/Framer.cpphs view
@@ -0,0 +1,536 @@+-- The framer has two functions: to convert bytes to Frames and the other way around,+-- and two keep track of flow-control quotas. +{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, + DeriveDataTypeable, TemplateHaskell #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.Framer (+ BadPrefaceException,++ wrapSession,+ http2FrameLength,++ -- Not needed anywhere, but supress the warning about unneeded symbol+ closeAction+ ) where ++++import Control.Concurrent+import Control.Exception+import qualified Control.Exception as E+import Control.Lens (view)+import qualified Control.Lens as L+import Control.Monad.IO.Class (liftIO)+import qualified Control.Monad.Catch as C+import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.Reader+import Data.Binary (decode)+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as LB+import Data.Conduit+import Data.Foldable (find)++import qualified Network.HTTP2 as NH2+-- Logging utilities+import System.Log.Logger++import qualified Data.HashTable.IO as H++import SecondTransfer.Http2.Session +import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker)+import qualified SecondTransfer.MainLoop.Framer as F+import SecondTransfer.MainLoop.PushPullType (Attendant, CloseAction,+ PullAction, PushAction, IOProblem)+import SecondTransfer.Utils (Word24, word24ToInt)+import SecondTransfer.Exception+++#include "Logging.cpphs"+++http2PrefixLength :: Int+http2PrefixLength = B.length NH2.connectionPreface++-- Let's do flow control here here .... ++type HashTable k v = H.CuckooHashTable k v+++type GlobalStreamId = Int+++data FlowControlCommand = + AddBytes_FCM Int ++-- A hashtable from stream id to channel of availabiliy increases+type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand)+++data CanOutput = CanOutput+++data NoHeadersInChannel = NoHeadersInChannel+++data FramerSessionData = FramerSessionData {+ _stream2flow :: Stream2AvailSpace+ , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString)+ , _defaultStreamWindow :: MVar Int++ -- Wait variable to output bytes to the channel+ , _canOutput :: MVar CanOutput+ -- Flag that says if the session has been unwound... if such, + -- threads are adviced to exit as early as possible+ , _outputIsForbidden :: MVar Bool + , _noHeadersInChannel :: MVar NoHeadersInChannel+ , _pushAction :: PushAction+ , _closeAction :: CloseAction++ -- Global id of the session, used for e.g. error reporting.+ , _sessionId :: Int ++ -- Sessions context, used for thing like e.g. error reporting+ , _sessionsContext :: SessionsContext++ -- For GoAway frames+ , _lastStream :: MVar Int + }+++L.makeLenses ''FramerSessionData+++type FramerSession = ReaderT FramerSessionData IO+++wrapSession :: CoherentWorker -> SessionsContext -> Attendant+wrapSession coherent_worker sessions_context push_action pull_action close_action = do++ let + session_id_mvar = view nextSessionId sessions_context++ new_session_id <- modifyMVarMasked+ session_id_mvar+ (\ session_id -> return (session_id+1, session_id))++ (session_input, session_output) <- (http2Session + coherent_worker new_session_id sessions_context)++ -- TODO : Add type annotations....+ s2f <- H.new + s2o <- H.new + default_stream_size_mvar <- newMVar 65536+ can_output <- newMVar CanOutput+ no_headers_in_channel <- newMVar NoHeadersInChannel+ last_stream_id <- newMVar 0+ output_is_forbidden <- newMVar False+++ -- We need some shared state + let framer_session_data = FramerSessionData {+ _stream2flow = s2f+ ,_stream2outputBytes = s2o + ,_defaultStreamWindow = default_stream_size_mvar+ ,_canOutput = can_output + ,_noHeadersInChannel = no_headers_in_channel+ ,_pushAction = push_action+ ,_closeAction = close_action+ ,_sessionId = new_session_id+ ,_sessionsContext = sessions_context+ ,_lastStream = last_stream_id+ ,_outputIsForbidden = output_is_forbidden+ }+++ let + -- TODO: Dodgy exception handling here...+ close_on_error session_id session_context comp = E.finally (+ E.catch comp (exc_handler session_id session_context)) close_action++ exc_handler :: Int -> SessionsContext -> FramerException -> IO ()+ exc_handler x y e = do+ modifyMVar_ output_is_forbidden (\ _ -> return True) + sessionExceptionHandler Framer_SessionComponent x y e+++ forkIO + $ close_on_error new_session_id sessions_context + $ runReaderT (inputGatherer pull_action session_input ) framer_session_data + forkIO + $ close_on_error new_session_id sessions_context + $ runReaderT (outputGatherer session_output ) framer_session_data ++ return ()+++http2FrameLength :: F.LengthCallback+http2FrameLength bs | (B.length bs) >= 3 = let+ word24 = decode input_as_lbs :: Word24+ input_as_lbs = LB.fromStrict bs+ in + Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses+http2FrameLength _ = Nothing+++addCapacity :: + GlobalStreamId ->+ Int -> + FramerSession ()+addCapacity stream_id delta_cap = do ++ if stream_id == 0 + then + -- TODO: Add session flow control+ return ()+ else do+ table <- view stream2flow + val <- liftIO $ H.lookup table stream_id+ case val of + Nothing -> do+ -- ??+ liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id)+ (_, command_chan) <- startStreamOutputQueue stream_id + -- And try again+ liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+++ Just command_chan -> do+ liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+ ++-- This works by pulling bytes from the input side of the pipeline and converting them to frames.+-- The frames are then put in the SessionInput. In the other end of the SessionInput they can be +-- interpreted according to their HTTP/2 meaning. +-- +-- This function also does part of the flow control: it registers WindowUpdate frames and triggers+-- quota updates on the streams. +inputGatherer :: PullAction -> SessionInput -> FramerSession ()+inputGatherer pull_action session_input = do + -- We can start by reading off the prefix....+ INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Entering InputGatherer" )+ (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action++ if prefix /= NH2.connectionPreface + then do + sendGoAwayFrame NH2.ProtocolError+ liftIO $ do + -- We just the the GoAway frame, although this is awfully early+ -- and probably wrong+ INSTRUMENTATION( errorM "HTTP2.Framer" "Invalid prologue")+ throwIO BadPrefaceException+ else + INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Prologue validated" )+ let + source::Source FramerSession B.ByteString+ source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action+ ( source $$ consume)+ where ++ consume :: Sink B.ByteString FramerSession ()+ consume = do + maybe_bytes <- await + -- Deserialize++ case maybe_bytes of ++ Just bytes -> do+ let + error_or_frame = NH2.decodeFrame some_settings bytes+ -- TODO: See how we can change these....+ some_settings = NH2.defaultSettings++ case error_or_frame of ++ Left _ -> do + -- Got an error from the decoder... meaning that a frame could + -- not be decoded.... in this case we send a cancel session command + -- to the session. + liftIO $ errorM "HTTP2.Framer" "CouldNotDecodeFrame"+ -- Send frames like GoAway and such...+ lift $ sendGoAwayFrame NH2.ProtocolError+ -- Inform the session that it can tear down itself+ liftIO $ sendCommandToSession session_input CancelSession_SIC+ -- Any resources remaining here can be disposed+ lift $ releaseFramer+ -- And end this thread++ Right right_frame -> do+ case right_frame of ++ (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do + -- Bookkeep the increase on bytes on that stream+ -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id)+ lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit)+ return ()+++ frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do + -- Increase all the stuff....+ case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of ++ Just (_, new_default_stream_size) -> do + old_default_stream_size_mvar <- view defaultStreamWindow+ old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar+ let general_delta = new_default_stream_size - old_default_stream_size+ stream_to_flow <- view stream2flow+ -- Add capacity to everybody's windows+ liftIO $ + H.mapM_ (+ \ (k,v) -> if k /=0 + then writeChan v (AddBytes_FCM general_delta) + else return () )+ stream_to_flow+++ -- And set a new value + liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size+++ Nothing -> + -- This is a silenced internal error+ return ()++ -- And send the frame down to the session, so that session specific settings+ -- can be applied. + liftIO $ sendFrameToSession session_input frame+++ a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do + -- Update the keep of last stream + lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id++ -- Send frame to the session+ liftIO $ sendFrameToSession session_input a_frame+ -- tail recursion: go again...+ consume ++ Nothing -> + -- We may as well exit this thread+ return ()+++-- All the output frames come this way first+outputGatherer :: SessionOutput -> FramerSession ()+outputGatherer session_output = do ++ -- We start by sending a settings frame + pushFrame + (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)+ (NH2.SettingsFrame [])++ loopPart++ where +++ dataForFrame p1 p2 = + LB.fromStrict $ NH2.encodeFrame p1 p2++ loopPart :: FramerSession ()+ loopPart = do ++ command_or_frame <- liftIO $ getFrameFromSession session_output++ case command_or_frame of ++ Left CancelSession_SOC -> do + -- The session wants to cancel things+ releaseFramer++ Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _) ) -> do+ -- This frame is flow-controlled... I may be unable to send this frame in+ -- some circumstances... + let stream_id = NH2.fromStreamIdentifier stream_idii+ s2o <- view stream2outputBytes+ lookup_result <- liftIO $ H.lookup s2o stream_id + stream_bytes_chan <- case lookup_result of ++ Nothing -> do + (bc, _) <- startStreamOutputQueue stream_id+ return bc++ Just bytes_chan -> return bytes_chan ++ liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2++ loopPart++ Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do+ handleHeadersOfStream p1 p2+ + loopPart++ Right (p1, p2@(NH2.ContinuationFrame _) ) -> do+ handleHeadersOfStream p1 p2++ loopPart++ Right (p1, p2) -> do + -- Most other frames go right away... as long as no headers are in process...+ no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ pushFrame p1 p2 + liftIO $ putMVar no_headers NoHeadersInChannel+ + loopPart+++updateLastStream :: GlobalStreamId -> FramerSession ()+updateLastStream stream_id = do + last_stream_id_mvar <- view lastStream+ liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id)++++startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand)+startStreamOutputQueue stream_id = do+ -- New thread for handling outputs of this stream is needed+ bytes_chan <- liftIO newChan + command_chan <- liftIO newChan ++ s2o <- view stream2outputBytes++ liftIO $ H.insert s2o stream_id bytes_chan ++ s2c <- view stream2flow+++ liftIO $ H.insert s2c stream_id command_chan++ -- + initial_cap_mvar <- view defaultStreamWindow+ initial_cap <- liftIO $ readMVar initial_cap_mvar+ close_action <- view closeAction+ sessions_context <- view sessionsContext + session_id' <- view SecondTransfer.Http2.Framer.sessionId+ output_is_forbidden_mvar <- view outputIsForbidden ++ -- And don't forget the thread itself+ let + close_on_error session_id session_context comp = E.finally (+ E.catch comp (exc_handler session_id session_context)) close_action++ exc_handler :: Int -> SessionsContext -> IOProblem -> IO ()+ exc_handler x y e = do+ -- Let's also decree that other streams don't even try+ modifyMVar_ output_is_forbidden_mvar ( \ _ -> return True)+ sessionExceptionHandler Framer_SessionComponent x y e+++ read_state <- ask + liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT+ (flowControlOutput stream_id initial_cap "" command_chan bytes_chan)+ read_state++ return (bytes_chan , command_chan)+++-- This works in the output side of the HTTP/2 framing session, and it acts as a +-- semaphore ensuring that headers are output without any interleaved frames. +-- +-- There are more synchronization mechanisms in the session, this does not act +-- alone. +handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession ()+handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload+ | (frameIsHeaderOfStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do+ -- Take it + no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ pushFrame p1 frame_payload + -- DONT PUT THE MvAR HERE ++ | (frameIsHeaderOfStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do+ no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ pushFrame p1 frame_payload + -- Since we finish.... + liftIO $ putMVar no_headers NoHeadersInChannel++ | frameEndsHeaders p1 frame_payload = do + -- I can only get here for a continuation frame after something else that is a headers+ no_headers <- view noHeadersInChannel+ liftIO $ putMVar no_headers NoHeadersInChannel+ return ()+++frameIsHeaderOfStream :: NH2.FramePayload -> Bool+frameIsHeaderOfStream (NH2.HeadersFrame _ _ )+ = True+frameIsHeaderOfStream _ + = False +++frameEndsHeaders :: NH2.EncodeInfo -> NH2.FramePayload -> Bool +frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags+frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags+frameEndsHeaders _ _ = False+++-- Push a frame into the output channel... this waits for the +-- channel to be free to send. +pushFrame :: NH2.EncodeInfo+ -> NH2.FramePayload -> FramerSession ()+pushFrame p1 p2 = do+ let bs = LB.fromStrict $ NH2.encodeFrame p1 p2 + sendBytes bs+++sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession ()+sendGoAwayFrame error_code = do+ last_stream_id_mvar <- view lastStream+ last_stream_id <- liftIO $ readMVar last_stream_id_mvar+ pushFrame (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)+ (NH2.GoAwayFrame (NH2.toStreamIdentifier last_stream_id) error_code "")+++sendBytes :: LB.ByteString -> FramerSession ()+sendBytes bs = do+ push_action <- view pushAction+ can_output <- view canOutput + liftIO $ do + bs `seq` + (C.bracket + (takeMVar can_output)+ (\ _ -> push_action bs)+ (\ c -> putMVar can_output c)+ )+++-- A thread in charge of doing flow control transmission....This sends already+-- formatted frames (ByteStrings), not the frames themselves. And it doesn't +-- mess with the structure of the packets.+flowControlOutput :: Int -> Int -> LB.ByteString -> (Chan FlowControlCommand) -> (Chan LB.ByteString) -> FramerSession ()+flowControlOutput stream_id capacity leftovers commands_chan bytes_chan = do + if leftovers == "" + then do+ -- Get more data (possibly block waiting for it)+ bytes_to_send <- liftIO $ readChan bytes_chan+ flowControlOutput stream_id capacity bytes_to_send commands_chan bytes_chan+ else do+ -- Length?+ let amount = fromIntegral $ ((LB.length leftovers) - 9)+ if amount <= capacity + then do+ -- Is + -- I can send ... if no headers are in process....+ no_headers <- view noHeadersInChannel+ C.bracket+ (liftIO $ takeMVar no_headers)+ (\ _ -> liftIO $ putMVar no_headers NoHeadersInChannel)+ (\ _ -> sendBytes leftovers )+ flowControlOutput stream_id (capacity - amount) "" commands_chan bytes_chan+ else do+ -- I can not send because flow-control is full, wait for a command instead + -- liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id)+ command <- liftIO $ readChan commands_chan+ case command of + AddBytes_FCM delta_cap -> do + -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id)+ flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan+++releaseFramer :: FramerSession ()+releaseFramer = do + -- Release any resources pending...++ return ()
− hs-src/SecondTransfer/Http2/Framer.hs
@@ -1,531 +0,0 @@--- The framer has two functions: to convert bytes to Frames and the other way around,--- and two keep track of flow-control quotas. -{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, - DeriveDataTypeable, TemplateHaskell #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.Http2.Framer (- BadPrefaceException,-- wrapSession,- http2FrameLength,-- -- Not needed anywhere, but supress the warning about unneeded symbol- closeAction- ) where ----import Control.Concurrent-import Control.Exception-import qualified Control.Exception as E-import Control.Lens (view)-import qualified Control.Lens as L-import Control.Monad.IO.Class (liftIO)-import qualified Control.Monad.Catch as C-import Control.Monad.Trans.Class (lift)-import Control.Monad.Trans.Reader-import Data.Binary (decode)-import qualified Data.ByteString as B-import qualified Data.ByteString.Lazy as LB-import Data.Conduit-import Data.Foldable (find)--import qualified Network.HTTP2 as NH2--- Logging utilities-import System.Log.Logger--import qualified Data.HashTable.IO as H--import SecondTransfer.Http2.Session -import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker)-import qualified SecondTransfer.MainLoop.Framer as F-import SecondTransfer.MainLoop.PushPullType (Attendant, CloseAction,- PullAction, PushAction, IOProblem)-import SecondTransfer.Utils (Word24, word24ToInt)-import SecondTransfer.Exception---http2PrefixLength :: Int-http2PrefixLength = B.length NH2.connectionPreface---- Let's do flow control here here .... --type HashTable k v = H.CuckooHashTable k v---type GlobalStreamId = Int---data FlowControlCommand = - AddBytes_FCM Int ---- A hashtable from stream id to channel of availabiliy increases-type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand)---data CanOutput = CanOutput---data NoHeadersInChannel = NoHeadersInChannel---data FramerSessionData = FramerSessionData {- _stream2flow :: Stream2AvailSpace- , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString)- , _defaultStreamWindow :: MVar Int-- -- Wait variable to output bytes to the channel- , _canOutput :: MVar CanOutput- -- Flag that says if the session has been unwound... if such, - -- threads are adviced to exit as early as possible- , _outputIsForbidden :: MVar Bool - , _noHeadersInChannel :: MVar NoHeadersInChannel- , _pushAction :: PushAction- , _closeAction :: CloseAction-- -- Global id of the session, used for e.g. error reporting.- , _sessionId :: Int -- -- Sessions context, used for thing like e.g. error reporting- , _sessionsContext :: SessionsContext-- -- For GoAway frames- , _lastStream :: MVar Int - }---L.makeLenses ''FramerSessionData---type FramerSession = ReaderT FramerSessionData IO---wrapSession :: CoherentWorker -> SessionsContext -> Attendant-wrapSession coherent_worker sessions_context push_action pull_action close_action = do-- let - session_id_mvar = view nextSessionId sessions_context-- new_session_id <- modifyMVarMasked- session_id_mvar- (\ session_id -> return (session_id+1, session_id))-- (session_input, session_output) <- (http2Session - coherent_worker new_session_id sessions_context)-- -- TODO : Add type annotations....- s2f <- H.new - s2o <- H.new - default_stream_size_mvar <- newMVar 65536- can_output <- newMVar CanOutput- no_headers_in_channel <- newMVar NoHeadersInChannel- last_stream_id <- newMVar 0- output_is_forbidden <- newMVar False--- -- We need some shared state - let framer_session_data = FramerSessionData {- _stream2flow = s2f- ,_stream2outputBytes = s2o - ,_defaultStreamWindow = default_stream_size_mvar- ,_canOutput = can_output - ,_noHeadersInChannel = no_headers_in_channel- ,_pushAction = push_action- ,_closeAction = close_action- ,_sessionId = new_session_id- ,_sessionsContext = sessions_context- ,_lastStream = last_stream_id- ,_outputIsForbidden = output_is_forbidden- }--- let - -- TODO: Dodgy exception handling here...- close_on_error session_id session_context comp = E.finally (- E.catch comp (exc_handler session_id session_context)) close_action-- exc_handler :: Int -> SessionsContext -> FramerException -> IO ()- exc_handler x y e = do- modifyMVar_ output_is_forbidden (\ _ -> return True) - sessionExceptionHandler Framer_SessionComponent x y e--- forkIO - $ close_on_error new_session_id sessions_context - $ runReaderT (inputGatherer pull_action session_input ) framer_session_data - forkIO - $ close_on_error new_session_id sessions_context - $ runReaderT (outputGatherer session_output ) framer_session_data -- return ()---http2FrameLength :: F.LengthCallback-http2FrameLength bs | (B.length bs) >= 3 = let- word24 = decode input_as_lbs :: Word24- input_as_lbs = LB.fromStrict bs- in - Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses-http2FrameLength _ = Nothing---addCapacity :: - GlobalStreamId ->- Int -> - FramerSession ()-addCapacity stream_id delta_cap = do -- if stream_id == 0 - then - -- TODO: Add session flow control- return ()- else do- table <- view stream2flow - val <- liftIO $ H.lookup table stream_id- case val of - Nothing -> do- -- ??- liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id)- (_, command_chan) <- startStreamOutputQueue stream_id - -- And try again- liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap--- Just command_chan -> do- liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap- ---- This works by pulling bytes from the input side of the pipeline and converting them to frames.--- The frames are then put in the SessionInput. In the other end of the SessionInput they can be --- interpreted according to their HTTP/2 meaning. --- --- This function also does part of the flow control: it registers WindowUpdate frames and triggers--- quota updates on the streams. -inputGatherer :: PullAction -> SessionInput -> FramerSession ()-inputGatherer pull_action session_input = do - -- We can start by reading off the prefix....- (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action-- if prefix /= NH2.connectionPreface - then do - sendGoAwayFrame NH2.ProtocolError- liftIO $ do - -- We just the the GoAway frame, although this is awfully early- -- and probably wrong- throwIO BadPrefaceException- else - return ()- let - source::Source FramerSession B.ByteString- source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action- ( source $$ consume)- where -- consume :: Sink B.ByteString FramerSession ()- consume = do - maybe_bytes <- await - -- Deserialize-- case maybe_bytes of -- Just bytes -> do- let - error_or_frame = NH2.decodeFrame some_settings bytes- -- TODO: See how we can change these....- some_settings = NH2.defaultSettings-- case error_or_frame of -- Left _ -> do - -- Got an error from the decoder... meaning that a frame could - -- not be decoded.... in this case we send a cancel session command - -- to the session. - liftIO $ errorM "HTTP2.Framer" "CouldNotDecodeFrame"- -- Send frames like GoAway and such...- lift $ sendGoAwayFrame NH2.ProtocolError- -- Inform the session that it can tear down itself- liftIO $ sendCommandToSession session_input CancelSession_SIC- -- Any resources remaining here can be disposed- lift $ releaseFramer- -- And end this thread-- Right right_frame -> do- case right_frame of -- (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do - -- Bookkeep the increase on bytes on that stream- -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id)- lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit)- return ()--- frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do - -- Increase all the stuff....- case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of -- Just (_, new_default_stream_size) -> do - old_default_stream_size_mvar <- view defaultStreamWindow- old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar- let general_delta = new_default_stream_size - old_default_stream_size- stream_to_flow <- view stream2flow- -- Add capacity to everybody's windows- liftIO $ - H.mapM_ (- \ (k,v) -> if k /=0 - then writeChan v (AddBytes_FCM general_delta) - else return () )- stream_to_flow--- -- And set a new value - liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size--- Nothing -> - -- This is a silenced internal error- return ()-- -- And send the frame down to the session, so that session specific settings- -- can be applied. - liftIO $ sendFrameToSession session_input frame--- a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do - -- Update the keep of last stream - lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id-- -- Send frame to the session- liftIO $ sendFrameToSession session_input a_frame- -- tail recursion: go again...- consume -- Nothing -> - -- We may as well exit this thread- return ()----- All the output frames come this way first-outputGatherer :: SessionOutput -> FramerSession ()-outputGatherer session_output = do -- -- We start by sending a settings frame - pushFrame - (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)- (NH2.SettingsFrame [])-- loopPart-- where --- dataForFrame p1 p2 = - LB.fromStrict $ NH2.encodeFrame p1 p2-- loopPart :: FramerSession ()- loopPart = do -- command_or_frame <- liftIO $ getFrameFromSession session_output-- case command_or_frame of -- Left CancelSession_SOC -> do - -- The session wants to cancel things- releaseFramer-- Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _) ) -> do- -- This frame is flow-controlled... I may be unable to send this frame in- -- some circumstances... - let stream_id = NH2.fromStreamIdentifier stream_idii- s2o <- view stream2outputBytes- lookup_result <- liftIO $ H.lookup s2o stream_id - stream_bytes_chan <- case lookup_result of -- Nothing -> do - (bc, _) <- startStreamOutputQueue stream_id- return bc-- Just bytes_chan -> return bytes_chan -- liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2-- loopPart-- Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do- handleHeadersOfStream p1 p2- - loopPart-- Right (p1, p2@(NH2.ContinuationFrame _) ) -> do- handleHeadersOfStream p1 p2-- loopPart-- Right (p1, p2) -> do - -- Most other frames go right away... as long as no headers are in process...- no_headers <- view noHeadersInChannel- liftIO $ takeMVar no_headers- pushFrame p1 p2 - liftIO $ putMVar no_headers NoHeadersInChannel- - loopPart---updateLastStream :: GlobalStreamId -> FramerSession ()-updateLastStream stream_id = do - last_stream_id_mvar <- view lastStream- liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id)----startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand)-startStreamOutputQueue stream_id = do- -- New thread for handling outputs of this stream is needed- bytes_chan <- liftIO newChan - command_chan <- liftIO newChan -- s2o <- view stream2outputBytes-- liftIO $ H.insert s2o stream_id bytes_chan -- s2c <- view stream2flow--- liftIO $ H.insert s2c stream_id command_chan-- -- - initial_cap_mvar <- view defaultStreamWindow- initial_cap <- liftIO $ readMVar initial_cap_mvar- close_action <- view closeAction- sessions_context <- view sessionsContext - session_id' <- view SecondTransfer.Http2.Framer.sessionId- output_is_forbidden_mvar <- view outputIsForbidden -- -- And don't forget the thread itself- let - close_on_error session_id session_context comp = E.finally (- E.catch comp (exc_handler session_id session_context)) close_action-- exc_handler :: Int -> SessionsContext -> IOProblem -> IO ()- exc_handler x y e = do- -- Let's also decree that other streams don't even try- modifyMVar_ output_is_forbidden_mvar ( \ _ -> return True)- sessionExceptionHandler Framer_SessionComponent x y e--- read_state <- ask - liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT- (flowControlOutput stream_id initial_cap "" command_chan bytes_chan)- read_state-- return (bytes_chan , command_chan)----- This works in the output side of the HTTP/2 framing session, and it acts as a --- semaphore ensuring that headers are output without any interleaved frames. --- --- There are more synchronization mechanisms in the session, this does not act --- alone. -handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession ()-handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload- | (frameIsHeaderOfStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do- -- Take it - no_headers <- view noHeadersInChannel- liftIO $ takeMVar no_headers- pushFrame p1 frame_payload - -- DONT PUT THE MvAR HERE -- | (frameIsHeaderOfStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do- no_headers <- view noHeadersInChannel- liftIO $ takeMVar no_headers- pushFrame p1 frame_payload - -- Since we finish.... - liftIO $ putMVar no_headers NoHeadersInChannel-- | frameEndsHeaders p1 frame_payload = do - -- I can only get here for a continuation frame after something else that is a headers- no_headers <- view noHeadersInChannel- liftIO $ putMVar no_headers NoHeadersInChannel- return ()---frameIsHeaderOfStream :: NH2.FramePayload -> Bool-frameIsHeaderOfStream (NH2.HeadersFrame _ _ )- = True-frameIsHeaderOfStream _ - = False ---frameEndsHeaders :: NH2.EncodeInfo -> NH2.FramePayload -> Bool -frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags-frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags-frameEndsHeaders _ _ = False----- Push a frame into the output channel... this waits for the --- channel to be free to send. -pushFrame :: NH2.EncodeInfo- -> NH2.FramePayload -> FramerSession ()-pushFrame p1 p2 = do- let bs = LB.fromStrict $ NH2.encodeFrame p1 p2 - sendBytes bs---sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession ()-sendGoAwayFrame error_code = do- last_stream_id_mvar <- view lastStream- last_stream_id <- liftIO $ readMVar last_stream_id_mvar- pushFrame (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)- (NH2.GoAwayFrame (NH2.toStreamIdentifier last_stream_id) error_code "")---sendBytes :: LB.ByteString -> FramerSession ()-sendBytes bs = do- push_action <- view pushAction- can_output <- view canOutput - liftIO $ do - bs `seq` - (C.bracket - (takeMVar can_output)- (\ _ -> push_action bs)- (\ c -> putMVar can_output c)- )----- A thread in charge of doing flow control transmission....This sends already--- formatted frames (ByteStrings), not the frames themselves. And it doesn't --- mess with the structure of the packets.-flowControlOutput :: Int -> Int -> LB.ByteString -> (Chan FlowControlCommand) -> (Chan LB.ByteString) -> FramerSession ()-flowControlOutput stream_id capacity leftovers commands_chan bytes_chan = do - if leftovers == "" - then do- -- Get more data (possibly block waiting for it)- bytes_to_send <- liftIO $ readChan bytes_chan- flowControlOutput stream_id capacity bytes_to_send commands_chan bytes_chan- else do- -- Length?- let amount = fromIntegral $ ((LB.length leftovers) - 9)- if amount <= capacity - then do- -- Is - -- I can send ... if no headers are in process....- no_headers <- view noHeadersInChannel- C.bracket- (liftIO $ takeMVar no_headers)- (\ _ -> liftIO $ putMVar no_headers NoHeadersInChannel)- (\ _ -> sendBytes leftovers )- flowControlOutput stream_id (capacity - amount) "" commands_chan bytes_chan- else do- -- I can not send because flow-control is full, wait for a command instead - -- liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id)- command <- liftIO $ readChan commands_chan- case command of - AddBytes_FCM delta_cap -> do - -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id)- flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan---releaseFramer :: FramerSession ()-releaseFramer = do - -- Release any resources pending...-- return ()
+ hs-src/SecondTransfer/Http2/Session.cpphs view
@@ -0,0 +1,869 @@+-- Session: links frames to streams, and helps in ordering the header frames+-- so that they don't get mixed with header frames from other streams when +-- resources are being served concurrently.+{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.Session(+ http2Session+ ,getFrameFromSession+ ,sendFrameToSession+ ,sendCommandToSession+ ,defaultSessionsConfig+ ,sessionId+ ,reportErrorCallback+ ,sessionsCallbacks+ ,nextSessionId+ ,makeSessionsContext+ ,sessionsConfig+ ,sessionExceptionHandler++ ,CoherentSession+ ,SessionInput(..)+ ,SessionInputCommand(..)+ ,SessionOutput(..)+ ,SessionOutputCommand(..)+ ,SessionsContext(..)+ ,SessionCoordinates(..)+ ,SessionComponent(..)+ ,SessionsCallbacks+ ,SessionsConfig+ ,ErrorCallback++ -- Internal stuff+ ,OutputFrame+ ,InputFrame+ ) where++#include "Logging.cpphs"++-- System grade utilities+import Control.Concurrent (ThreadId, forkIO)+import Control.Concurrent.Chan+import Control.Exception (SomeException, throwTo)+import qualified Control.Exception as E+import Control.Monad (forever)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Trans.Reader++import Control.Concurrent.MVar+import qualified Data.ByteString as B+import qualified Data.ByteString.Builder as Bu+import qualified Data.ByteString.Lazy as Bl+import Data.Conduit+import qualified Data.HashTable.IO as H+import qualified Data.IntSet as NS+import Data.Monoid as Mo++import Control.Lens++-- No framing layer here... let's use Kazu's Yamamoto library+import qualified Network.HPACK as HP+import qualified Network.HTTP2 as NH2++-- Logging utilities+import System.Log.Logger++-- Imports from other parts of the program+import SecondTransfer.MainLoop.CoherentWorker+import SecondTransfer.MainLoop.Tokens+import SecondTransfer.Utils (unfoldChannelAndSource)+import SecondTransfer.Exception++-- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to +-- use :-( +type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload)+type InputFrame = NH2.Frame+++useChunkLength :: Int +useChunkLength = 16384+++-- Singleton instance used for concurrency+data HeadersSent = HeadersSent ++-- All streams put their data bits here. A "Nothing" value signals+-- end of data. +type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString)+++-- Whatever a worker thread is going to need comes here.... +-- this is to make refactoring easier, but not strictly needed. +data WorkerThreadEnvironment = WorkerThreadEnvironment {+ -- What's the header stream id?+ _streamId :: GlobalStreamId++ -- A full block of headers can come here... the mvar in the middle should+ -- be populate to signal end of headers transmission. A thread will be suspended+ -- waiting for that+ , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers)++ -- And regular contents can come this way and thus be properly mixed+ -- with everything else.... for now... + ,_dataOutput :: Chan DataOutputToConveyor++ ,_streamsCancelled_WTE :: MVar NS.IntSet++ }++makeLenses ''WorkerThreadEnvironment+++-- Basically a couple of channels ... +type Session = (SessionInput, SessionOutput)+++-- From outside, one can only write to this one ... the newtype is to enforce +-- this.+newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) )+sendFrameToSession :: SessionInput -> InputFrame -> IO ()+sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame++sendCommandToSession :: SessionInput -> SessionInputCommand -> IO ()+sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command++-- From outside, one can only read from this one +newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) )+getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) +getFrameFromSession (SessionOutput chan) = readChan chan+++type HashTable k v = H.CuckooHashTable k v+++-- Blaze builder could be more proper here... +type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder+++type WorkerMonad = ReaderT WorkerThreadEnvironment IO +++-- Have to figure out which are these...but I would expect to have things+-- like unexpected aborts here in this type.+data SessionInputCommand = + CancelSession_SIC+ deriving Show +++-- temporary+data SessionOutputCommand = + CancelSession_SOC+ deriving Show+++-- | Information used to identify a particular session. +newtype SessionCoordinates = SessionCoordinates Int+ deriving Show++instance Eq SessionCoordinates where + (SessionCoordinates a) == (SessionCoordinates b) = a == b++-- | Get/set a numeric Id from a `SessionCoordinates`. For example, to +-- get the session id with this, import `Control.Lens.(^.)` and then do +--+-- @+-- session_id = session_coordinates ^. sessionId+-- @+-- +sessionId :: Functor f => (Int -> f Int) -> SessionCoordinates -> f SessionCoordinates+sessionId f (SessionCoordinates session_id) = + fmap (\ s' -> (SessionCoordinates s')) (f session_id)++++-- | Components at an individual session. Used to report+-- where in the session an error was produced. This interface is likely +-- to change in the future, as we add more metadata to exceptions+data SessionComponent = + SessionInputThread_SessionComponent + |SessionHeadersOutputThread_SessionComponent+ |SessionDataOutputThread_SessionComponent+ |Framer_SessionComponent+ deriving Show+++-- | Used by this session engine to report an error at some component, in a particular+-- session. +type ErrorCallback = (SessionComponent, SessionCoordinates, SomeException) -> IO ()++-- | Callbacks that you can provide your sessions to notify you +-- of interesting things happening in the server. +data SessionsCallbacks = SessionsCallbacks {+ _reportErrorCallback :: Maybe ErrorCallback+}++makeLenses ''SessionsCallbacks+++-- | Configuration information you can provide to the session maker.+data SessionsConfig = SessionsConfig {+ _sessionsCallbacks :: SessionsCallbacks+}++-- makeLenses ''SessionsConfig++-- | Lens to access sessionsCallbacks in the `SessionsConfig` object.+sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks+sessionsCallbacks f (+ SessionsConfig {+ _sessionsCallbacks= s + }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s)+++-- | Contains information that applies to all +-- sessions created in the program. Use the lenses +-- interface to access members of this struct. +-- +data SessionsContext = SessionsContext {+ _sessionsConfig :: SessionsConfig+ ,_nextSessionId :: MVar Int+ }+++makeLenses ''SessionsContext++-- Here is how we make a session +type SessionMaker = SessionsContext -> IO Session+++-- Here is how we make a session wrapping a CoherentWorker+type CoherentSession = CoherentWorker -> SessionMaker +++-- | Creates a default sessions context. Modify as needed using +-- the lenses interfaces+defaultSessionsConfig :: SessionsConfig+defaultSessionsConfig = SessionsConfig {+ _sessionsCallbacks = SessionsCallbacks {+ _reportErrorCallback = Nothing+ }+ }+++-- Adds runtime data to a context, and let it work.... +makeSessionsContext :: SessionsConfig -> IO SessionsContext+makeSessionsContext sessions_config = do + next_session_id_mvar <- newMVar 1 + return $ SessionsContext {+ _sessionsConfig = sessions_config,+ _nextSessionId = next_session_id_mvar+ }++data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream)+++-- NH2.Frame != Frame+data SessionData = SessionData {+ _sessionsContext :: SessionsContext ++ ,_sessionInput :: Chan (Either SessionInputCommand InputFrame)++ -- We need to lock this channel occassionally so that we can order multiple + -- header frames properly.... + ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame))++ -- Use to encode + ,_toEncodeHeaders :: MVar HP.DynamicTable+ -- And used to decode+ ,_toDecodeHeaders :: MVar HP.DynamicTable++ -- Used for decoding the headers+ ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment++ -- Used for worker threads... this is actually a pre-filled template+ -- I make copies of it in different contexts, and as needed. + ,_forWorkerThread :: WorkerThreadEnvironment++ ,_coherentWorker :: CoherentWorker++ -- Some streams may be cancelled + ,_streamsCancelled :: MVar NS.IntSet++ -- Data input mechanism corresponding to some threads+ ,_stream2PostInputMechanism :: HashTable Int PostInputMechanism ++ -- Worker thread register. This is a dictionary from stream id to + -- the ThreadId of the thread with the worker thread. I use this to + -- raise asynchronous exceptions in the worker thread if the stream + -- is cancelled by the client. This way we get early finalization. + ,_stream2WorkerThread :: HashTable Int ThreadId++ ,_sessionIdAtSession :: Int+ }+++makeLenses ''SessionData+++-- v- {headers table size comes here!!}+http2Session :: CoherentWorker -> Int -> SessionsContext -> IO Session+http2Session coherent_worker session_id sessions_context = do + session_input <- newChan+ session_output <- newChan+ session_output_mvar <- newMVar session_output+++ -- For incremental construction of headers...+ stream_request_headers <- H.new :: IO Stream2HeaderBlockFragment++ -- Warning: we should find a way of coping with different table sizes.+ decode_headers_table <- HP.newDynamicTableForDecoding 4096+ decode_headers_table_mvar <- newMVar decode_headers_table++ encode_headers_table <- HP.newDynamicTableForEncoding 4096+ encode_headers_table_mvar <- newMVar encode_headers_table++ -- These ones need independent threads taking care of sending stuff+ -- their way... + headers_output <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers))+ data_output <- newChan :: IO (Chan DataOutputToConveyor)++ stream2postinputmechanism <- H.new + stream2workerthread <- H.new++ -- What about stream cancellation?+ cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet)++ let for_worker_thread = WorkerThreadEnvironment {+ _streamId = error "NotInitialized" + ,_headersOutput = headers_output+ ,_dataOutput = data_output+ ,_streamsCancelled_WTE = cancelled_streams_mvar+ }++ let session_data = SessionData {+ _sessionsContext = sessions_context+ ,_sessionInput = session_input + ,_sessionOutput = session_output_mvar+ ,_toDecodeHeaders = decode_headers_table_mvar+ ,_toEncodeHeaders = encode_headers_table_mvar+ ,_stream2HeaderBlockFragment = stream_request_headers+ ,_forWorkerThread = for_worker_thread+ ,_coherentWorker = coherent_worker+ ,_streamsCancelled = cancelled_streams_mvar+ ,_stream2PostInputMechanism = stream2postinputmechanism+ ,_stream2WorkerThread = stream2workerthread+ ,_sessionIdAtSession = session_id+ }++ let + exc_handler :: SessionComponent -> HTTP2SessionException -> IO () + exc_handler component e = sessionExceptionHandler component session_id sessions_context e+ exc_guard :: SessionComponent -> IO () -> IO ()+ exc_guard component action = E.catch action $ exc_handler component++ -- Create an input thread that decodes frames...+ forkIO $ exc_guard SessionInputThread_SessionComponent + $ runReaderT sessionInputThread session_data+ + -- Create a thread that captures headers and sends them down the tube + forkIO $ exc_guard SessionHeadersOutputThread_SessionComponent + $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data++ -- Create a thread that captures data and sends it down the tube+ forkIO $ exc_guard SessionDataOutputThread_SessionComponent + $ dataOutputThread data_output session_output_mvar++ -- The two previous threads fill the session_output argument below (they write to it)+ -- the session machinery in the other end is in charge of sending that data through the + -- socket.++ return ( (SessionInput session_input),+ (SessionOutput session_output) )++++sessionInputThread :: ReaderT SessionData IO ()+sessionInputThread = do + INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread" )++ -- This is an introductory and declarative block... all of this is tail-executed+ -- every time that a packet needs to be processed. It may be a good idea to abstract+ -- these values in a closure... + session_input <- view sessionInput + + decode_headers_table_mvar <- view toDecodeHeaders + stream_request_headers <- view stream2HeaderBlockFragment+ cancelled_streams_mvar <- view streamsCancelled+ coherent_worker <- view coherentWorker++ for_worker_thread_uns <- view forWorkerThread+ stream2workerthread <- view stream2WorkerThread++ input <- liftIO $ readChan session_input++ INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input) )++ case input of ++ Left CancelSession_SIC -> do + -- Good place to tear down worker threads... Let the rest of the finalization+ -- to the framer+ liftIO $ do + H.mapM_+ (\ (_, thread_id) -> do+ throwTo thread_id StreamCancelledException+ infoM "HTTP2.Session" $ "Stream successfully interrupted"+ )+ stream2workerthread++ -- We do not continue here, but instead let it finish+ return ()++ Right frame | Just (stream_id, bytes) <- frameIsHeaderOfStream frame -> do + -- Just append the frames to streamRequestHeaders+ appendHeaderFragmentBlock stream_id bytes++ if frameEndsHeaders frame then + do+ -- Let's decode the headers+ let for_worker_thread = set streamId stream_id for_worker_thread_uns + headers_bytes <- getHeaderBytes stream_id+ dyn_table <- liftIO $ takeMVar decode_headers_table_mvar+ (new_table, header_list ) <- liftIO $ HP.decodeHeader dyn_table headers_bytes+ -- Good moment to remove the headers from the table.... we don't want a space+ -- leak here + liftIO $ H.delete stream_request_headers stream_id+ liftIO $ putMVar decode_headers_table_mvar new_table++ -- If the headers end the request.... + post_data_source <- if not (frameEndsStream frame)+ then do + + mechanism <- createMechanismForStream stream_id + let source = postDataSourceFromMechanism mechanism+ return $ Just source+ else do + return Nothing+++ -- I'm clear to start the worker, in its own thread+ -- !!+ liftIO $ do + thread_id <- forkIO $ runReaderT + (workerThread (header_list, post_data_source) coherent_worker)+ for_worker_thread + H.insert stream2workerthread stream_id thread_id++ return ()+ else + -- Frame doesn't end the headers... it was added before... so+ -- probably do nothing + return ()+ + continue ++ Right frame@(NH2.Frame _ (NH2.RSTStreamFrame error_code_id)) -> do+ let stream_id = streamIdFromFrame frame+ liftIO $ do + INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream reset: " ++ (show error_code_id) )+ cancelled_streams <- takeMVar cancelled_streams_mvar+ INSTRUMENTATION( infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id) )+ putMVar cancelled_streams_mvar $ NS.insert stream_id cancelled_streams+ maybe_thread_id <- H.lookup stream2workerthread stream_id+ case maybe_thread_id of + Nothing -> + -- This is actually more like an internal error+ error "InterruptingUnexistentStream"++ Just thread_id -> do+ throwTo thread_id StreamCancelledException+ INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" )++ continue ++ Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) -> do + -- So I got data to process+ -- TODO: Handle end of stream+ let stream_id = NH2.fromStreamIdentifier nh2_stream_id + streamWorkerSendData stream_id somebytes+ -- After that data has been received and forwarded downstream, we can issue a windows update+ --+ -- TODO: We can use wider frames to avoid congestion...+ -- .... and we can also be more compositional with these short bursts of data....+ sendOutFrame+ (NH2.EncodeInfo+ NH2.defaultFlags+ nh2_stream_id+ Nothing+ )+ (NH2.WindowUpdateFrame+ (fromIntegral (B.length somebytes))+ )+ sendOutFrame+ (NH2.EncodeInfo+ NH2.defaultFlags+ (NH2.toStreamIdentifier 0)+ Nothing+ )+ (NH2.WindowUpdateFrame+ (fromIntegral (B.length somebytes))+ ) ++ if frameEndsStream frame + then do + -- Good place to close the source ... + closePostDataSource stream_id + else + return ()++ continue ++ Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do + -- Deal with pings: this is an Ack, so do nothing+ continue ++ Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do + -- Deal with pings: NOT an Ack, so answer+ INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Ping processed" )+ sendOutFrame+ (NH2.EncodeInfo+ (NH2.setAck NH2.defaultFlags)+ (NH2.toStreamIdentifier 0)+ Nothing + )+ (NH2.PingFrame somebytes)++ continue ++ Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do + -- Frame was received by the peer, do nothing here...+ continue +++ Right (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do + INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) )+ -- Just acknowledge the frame.... for now + sendOutFrame + (NH2.EncodeInfo+ (NH2.setAck NH2.defaultFlags)+ (NH2.toStreamIdentifier 0)+ Nothing )+ (NH2.SettingsFrame [])++ continue +++ Right somethingelse -> do + -- An undhandled case here....+ INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" $ "Received problematic frame: " )+ liftIO $ errorM "HTTP2.Session" $ ".. " ++ (show somethingelse)++ continue ++ where ++ continue = sessionInputThread++ sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()+ sendOutFrame encode_info payload = do + session_output_mvar <- view sessionOutput + session_output <- liftIO $ takeMVar session_output_mvar++ liftIO $ writeChan session_output $ Right (encode_info, payload)++ liftIO $ putMVar session_output_mvar session_output+++frameEndsStream :: InputFrame -> Bool +frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags+++createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism+createMechanismForStream stream_id = do + (chan, source) <- liftIO $ unfoldChannelAndSource+ stream2postinputmechanism <- view stream2PostInputMechanism+ let pim = PostInputMechanism (chan, source)+ liftIO $ H.insert stream2postinputmechanism stream_id pim + return pim+++-- TODO: Can be optimized by factoring out the mechanism lookup+closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO ()+closePostDataSource stream_id = do + stream2postinputmechanism <- view stream2PostInputMechanism++ pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id ++ case pim_maybe of ++ Just (PostInputMechanism (chan, _)) -> + liftIO $ writeChan chan Nothing++ Nothing -> + -- This is an internal error, the mechanism should be + -- created when the stream ends+ error "Internal error/closePostDataSource"+++streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO ()+streamWorkerSendData stream_id bytes = do + s2pim <- view stream2PostInputMechanism+ pim_maybe <- liftIO $ H.lookup s2pim stream_id ++ case pim_maybe of ++ Just pim -> + sendBytesToPim pim bytes++ Nothing -> + -- This is an internal error, the mechanism should be + -- created when the headers end+ error "Internal error"+++sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO ()+sendBytesToPim (PostInputMechanism (chan, _)) bytes = + liftIO $ writeChan chan (Just bytes)+++postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream+postDataSourceFromMechanism (PostInputMechanism (_, source)) = source+++isSettingsAck :: NH2.FrameHeader -> Bool +isSettingsAck (NH2.FrameHeader _ flags _) = + NH2.testAck flags+++isStreamCancelled :: GlobalStreamId -> WorkerMonad Bool +isStreamCancelled stream_id = do + cancelled_streams_mvar <- view streamsCancelled_WTE+ cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar+ return $ NS.member stream_id cancelled_streams+++workerThread :: Request -> CoherentWorker -> WorkerMonad ()+workerThread req coherent_worker =+ do+ headers_output <- view headersOutput+ stream_id <- view streamId+ (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req++ -- liftIO $ putStrLn $ "Num pushed streams: " ++ (show $ length pushed_streams)++ -- Now I send the headers, if that's possible at all+ headers_sent <- liftIO $ newEmptyMVar+ liftIO $ writeChan headers_output (stream_id, headers_sent, headers)++ -- At this moment I should ask if the stream hasn't been cancelled by the browser before+ -- commiting to the work of sending addtitional data+ is_stream_cancelled <- isStreamCancelled stream_id+ if not is_stream_cancelled++ then do+ -- I have a beautiful source that I can de-construct...+ -- TODO: Optionally pulling data out from a Conduit ....+ -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )+ -- + -- This threadlet should block here waiting for the headers to finish going+ (maybe_footers, _) <- runConduit $+ (transPipe liftIO data_and_conclussion) + `fuseBothMaybe` + (sendDataOfStream stream_id headers_sent)+ -- BIG TODO: Send the headers ... likely stream conclusion semantics + -- will need to be changed. + return ()+ else ++ return ()++-- v-- comp. monad.+sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()+sendDataOfStream stream_id headers_sent = do+ data_output <- view dataOutput+ -- Wait for all headers sent+ liftIO $ takeMVar headers_sent+ consumer data_output+ where + consumer data_output = do + maybe_bytes <- await + case maybe_bytes of + Nothing -> + liftIO $ writeChan data_output (stream_id, Nothing)+ Just bytes -> do+ liftIO $ writeChan data_output (stream_id, Just bytes)+ consumer data_output+++-- sendDataOfStream :: Sink +++-- Allow this very important function to be used in the future to process footers++-- difficultFunction :: (Monad m)+-- => ConduitM () a2 m r1 -> ConduitM a2 Void m r2+-- -> m (r2, Maybe r1)+-- difficultFunction l r = liftM (fmap getLast) $ runWriterT (l' $$ r')+-- where+-- l' = transPipe lift l >>= lift . tell . Last . Just+-- r' = transPipe lift r++++appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO ()+appendHeaderFragmentBlock global_stream_id bytes = do + ht <- view stream2HeaderBlockFragment + maybe_old_block <- liftIO $ H.lookup ht global_stream_id+ new_block <- case maybe_old_block of ++ Nothing -> do+ INSTRUMENTATION( liftIO $ infoM "HTTP2.Session" $ "Starting stream " ++ (show global_stream_id) )+ return $ Bu.byteString bytes++ Just something -> + return $ something `mappend` (Bu.byteString bytes) ++ liftIO $ H.insert ht global_stream_id new_block+++getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString+getHeaderBytes global_stream_id = do + ht <- view stream2HeaderBlockFragment + Just bytes <- liftIO $ H.lookup ht global_stream_id+ return $ Bl.toStrict $ Bu.toLazyByteString bytes+++frameIsHeaderOfStream :: InputFrame -> Maybe (GlobalStreamId, B.ByteString)+frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) )+ = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )+ = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+frameIsHeaderOfStream _ + = Nothing +++frameEndsHeaders :: InputFrame -> Bool +frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags+++streamIdFromFrame :: InputFrame -> GlobalStreamId+streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id+++-- TODO: Have different size for the headers..... just now going with a default size of 16 k...+-- TODO: Find a way to kill this thread....+headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers)+ -> MVar (Chan (Either SessionOutputCommand OutputFrame)) + -> ReaderT SessionData IO ()+headersOutputThread input_chan session_output_mvar = forever $ do + (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan+ -- liftIO $ putStrLn $ "Output headers: " ++ (show headers)++ -- First encode the headers using the table+ encode_dyn_table_mvar <- view toEncodeHeaders++ encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar+ (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers+ liftIO $ putMVar encode_dyn_table_mvar new_dyn_table++ -- Now split the bytestring in chunks of the needed size.... + bs_chunks <- return $! bytestringChunk useChunkLength data_to_send++ -- And send the chunks through while locking the output place....+ liftIO $ E.bracket+ (takeMVar session_output_mvar)+ (putMVar session_output_mvar )+ (\ session_output -> do+ writeIndividualHeaderFrames session_output stream_id bs_chunks True+ -- And say that the headers for this thread are out + putMVar headers_ready_mvar HeadersSent+ INSTRUMENTATION( infoM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) )+ ) + where + writeIndividualHeaderFrames :: + Chan (Either SessionOutputCommand OutputFrame)+ -> GlobalStreamId + -> [B.ByteString] + -> Bool + -> IO ()+ writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first = + writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + (if is_first then NH2.HeadersFrame Nothing last_fragment else NH2.ContinuationFrame last_fragment)+ )+ writeIndividualHeaderFrames session_output stream_id (fragment:xs) is_first = do + writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + (if is_first then NH2.HeadersFrame Nothing fragment else NH2.ContinuationFrame fragment)+ )+ writeIndividualHeaderFrames session_output stream_id xs False+++bytestringChunk :: Int -> B.ByteString -> [B.ByteString]+bytestringChunk len s | (B.length s) < len = [ s ]+bytestringChunk len s = h:(bytestringChunk len xs)+ where + (h, xs) = B.splitAt len s +++-- TODO: find a clean way to finish this thread (maybe with negative stream ids?)+-- TODO: This function does non-optimal chunking for the case where responses are+-- actually streamed.... in those cases we need to keep state for frames in +-- some other format.... +-- TODO: Right now, we are transmitting an empty last frame with the end-of-stream+-- flag set. I'm afraid that the only+-- way to avoid that is by holding a frame or by augmenting the end-user interface+-- so that the user can signal which one is the last frame. The first approach+-- restricts responsiviness, the second one clutters things.+dataOutputThread :: Chan DataOutputToConveyor+ -> MVar (Chan (Either SessionOutputCommand OutputFrame)) + -> IO ()+dataOutputThread input_chan session_output_mvar = forever $ do + (stream_id, maybe_contents) <- readChan input_chan+ case maybe_contents of + Nothing -> do+ liftIO $ do+ INSTRUMENTATION( debugM "HTTP2.Session" "End-of-stream flag set " )+ withLockedSessionOutput+ (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags+ ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + NH2.DataFrame ""+ )+ )++ Just contents -> do + -- And now just simply output it...+ let bs_chunks = bytestringChunk useChunkLength $! contents+ -- And send the chunks through while locking the output place....+ writeContinuations bs_chunks stream_id+ + where ++ withLockedSessionOutput = E.bracket + (takeMVar session_output_mvar) + (putMVar session_output_mvar) -- <-- There is an implicit argument there!!++ writeContinuations :: [B.ByteString] -> GlobalStreamId -> IO ()+ writeContinuations fragments stream_id = mapM_ (\ fragment -> + withLockedSessionOutput (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, + NH2.DataFrame fragment ) )+ ) fragments++++sessionExceptionHandler :: E.Exception e => SessionComponent -> Int -> SessionsContext -> e -> IO ()+sessionExceptionHandler session_component session_id sessions_context e = do + let+ getit = ( sessionsConfig . sessionsCallbacks . reportErrorCallback ) + maybe_error_callback = sessions_context ^. getit + error_tuple = (+ session_component,+ SessionCoordinates session_id, + E.toException e+ )+ case maybe_error_callback of + Nothing -> + errorM "HTTP2.Session" (show (e))++ Just callback -> + callback error_tuple
− hs-src/SecondTransfer/Http2/Session.hs
@@ -1,897 +0,0 @@--- Session: links frames to streams, and helps in ordering the header frames--- so that they don't get mixed with header frames from other streams when --- resources are being served concurrently.-{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.Http2.Session(- http2Session- ,getFrameFromSession- ,sendFrameToSession- ,sendCommandToSession- ,defaultSessionsConfig- ,sessionId- ,reportErrorCallback- ,sessionsCallbacks- ,nextSessionId- ,makeSessionsContext- ,sessionsConfig- ,sessionExceptionHandler-- ,CoherentSession- ,SessionInput(..)- ,SessionInputCommand(..)- ,SessionOutput(..)- ,SessionOutputCommand(..)- ,SessionsContext(..)- ,SessionCoordinates(..)- ,SessionComponent(..)- ,SessionsCallbacks- ,SessionsConfig- ,ErrorCallback-- -- Internal stuff- ,OutputFrame- ,InputFrame- ) where----- System grade utilities-import Control.Concurrent (ThreadId, forkIO)-import Control.Concurrent.Chan-import Control.Exception (SomeException, throwTo)-import qualified Control.Exception as E-import Control.Monad (forever)-import Control.Monad.IO.Class (liftIO)-import Control.Monad.Trans.Reader--import Control.Concurrent.MVar-import qualified Data.ByteString as B-import qualified Data.ByteString.Builder as Bu-import qualified Data.ByteString.Lazy as Bl-import Data.Conduit-import Data.Conduit.List (foldMapM)-import qualified Data.HashTable.IO as H-import qualified Data.IntSet as NS-import Data.Monoid as Mo--import Control.Lens---- No framing layer here... let's use Kazu's Yamamoto library-import qualified Network.HPACK as HP-import qualified Network.HTTP2 as NH2---- Logging utilities-import System.Log.Logger---- Imports from other parts of the program-import SecondTransfer.MainLoop.CoherentWorker-import SecondTransfer.MainLoop.Tokens-import SecondTransfer.Utils (unfoldChannelAndSource)-import SecondTransfer.Exception---- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to --- use :-( -type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload)-type InputFrame = NH2.Frame---useChunkLength :: Int -useChunkLength = 16384----- Singleton instance used for concurrency-data HeadersSent = HeadersSent ----- Whatever a worker thread is going to need comes here.... --- this is to make refactoring easier, but not strictly needed. -data WorkerThreadEnvironment = WorkerThreadEnvironment {- -- What's the header stream id?- _streamId :: GlobalStreamId-- -- A full block of headers can come here... the mvar in the middle should- -- be populate to signal end of headers transmission. A thread will be suspended- -- waiting for that- , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers)-- -- And regular contents can come this way and thus be properly mixed- -- with everything else.... for now... - ,_dataOutput :: Chan (GlobalStreamId, B.ByteString)-- ,_streamsCancelled_WTE :: MVar NS.IntSet-- }--makeLenses ''WorkerThreadEnvironment----- Basically a couple of channels ... -type Session = (SessionInput, SessionOutput)----- From outside, one can only write to this one ... the newtype is to enforce --- this.-newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) )-sendFrameToSession :: SessionInput -> InputFrame -> IO ()-sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame--sendCommandToSession :: SessionInput -> SessionInputCommand -> IO ()-sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command---- From outside, one can only read from this one -newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) )-getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) -getFrameFromSession (SessionOutput chan) = readChan chan---type HashTable k v = H.CuckooHashTable k v----- Blaze builder could be more proper here... -type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder---type WorkerMonad = ReaderT WorkerThreadEnvironment IO ----- Have to figure out which are these...but I would expect to have things--- like unexpected aborts here in this type.-data SessionInputCommand = - CancelSession_SIC- deriving Show ----- temporary-data SessionOutputCommand = - CancelSession_SOC- deriving Show----- | Information used to identify a particular session. -newtype SessionCoordinates = SessionCoordinates Int- deriving Show--instance Eq SessionCoordinates where - (SessionCoordinates a) == (SessionCoordinates b) = a == b---- | Get/set a numeric Id from a `SessionCoordinates`. For example, to --- get the session id with this, import `Control.Lens.(^.)` and then do ------ @--- session_id = session_coordinates ^. sessionId--- @--- -sessionId :: Functor f => (Int -> f Int) -> SessionCoordinates -> f SessionCoordinates-sessionId f (SessionCoordinates session_id) = - fmap (\ s' -> (SessionCoordinates s')) (f session_id)------ | Components at an individual session. Used to report--- where in the session an error was produced. This interface is likely --- to change in the future, as we add more metadata to exceptions-data SessionComponent = - SessionInputThread_SessionComponent - |SessionHeadersOutputThread_SessionComponent- |SessionDataOutputThread_SessionComponent- |Framer_SessionComponent- deriving Show----- | Used by this session engine to report an error at some component, in a particular--- session. -type ErrorCallback = (SessionComponent, SessionCoordinates, SomeException) -> IO ()---- | Callbacks that you can provide your sessions to notify you --- of interesting things happening in the server. -data SessionsCallbacks = SessionsCallbacks {- _reportErrorCallback :: Maybe ErrorCallback-}--makeLenses ''SessionsCallbacks----- | Configuration information you can provide to the session maker.-data SessionsConfig = SessionsConfig {- _sessionsCallbacks :: SessionsCallbacks-}---- makeLenses ''SessionsConfig---- | Lens to access sessionsCallbacks in the `SessionsConfig` object.-sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks-sessionsCallbacks f (- SessionsConfig {- _sessionsCallbacks= s - }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s)----- | Contains information that applies to all --- sessions created in the program. Use the lenses --- interface to access members of this struct. --- -data SessionsContext = SessionsContext {- _sessionsConfig :: SessionsConfig- ,_nextSessionId :: MVar Int- }---makeLenses ''SessionsContext---- Here is how we make a session -type SessionMaker = SessionsContext -> IO Session----- Here is how we make a session wrapping a CoherentWorker-type CoherentSession = CoherentWorker -> SessionMaker ----- | Creates a default sessions context. Modify as needed using --- the lenses interfaces-defaultSessionsConfig :: SessionsConfig-defaultSessionsConfig = SessionsConfig {- _sessionsCallbacks = SessionsCallbacks {- _reportErrorCallback = Nothing- }- }----- Adds runtime data to a context, and let it work.... -makeSessionsContext :: SessionsConfig -> IO SessionsContext-makeSessionsContext sessions_config = do - next_session_id_mvar <- newMVar 1 - return $ SessionsContext {- _sessionsConfig = sessions_config,- _nextSessionId = next_session_id_mvar- }--data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream)----- NH2.Frame != Frame-data SessionData = SessionData {- _sessionsContext :: SessionsContext -- ,_sessionInput :: Chan (Either SessionInputCommand InputFrame)-- -- We need to lock this channel occassionally so that we can order multiple - -- header frames properly.... - ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame))-- -- Use to encode - ,_toEncodeHeaders :: MVar HP.DynamicTable- -- And used to decode- ,_toDecodeHeaders :: MVar HP.DynamicTable-- -- Used for decoding the headers- ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment-- -- Used for worker threads... this is actually a pre-filled template- -- I make copies of it in different contexts, and as needed. - ,_forWorkerThread :: WorkerThreadEnvironment-- ,_coherentWorker :: CoherentWorker-- -- Some streams may be cancelled - ,_streamsCancelled :: MVar NS.IntSet-- -- Data input mechanism corresponding to some threads- ,_stream2PostInputMechanism :: HashTable Int PostInputMechanism -- -- Worker thread register. This is a dictionary from stream id to - -- the ThreadId of the thread with the worker thread. I use this to - -- raise asynchronous exceptions in the worker thread if the stream - -- is cancelled by the client. This way we get early finalization. - ,_stream2WorkerThread :: HashTable Int ThreadId-- ,_sessionIdAtSession :: Int- }---makeLenses ''SessionData----- v- {headers table size comes here!!}-http2Session :: CoherentWorker -> Int -> SessionsContext -> IO Session-http2Session coherent_worker session_id sessions_context = do - session_input <- newChan- session_output <- newChan- session_output_mvar <- newMVar session_output--- -- For incremental construction of headers...- stream_request_headers <- H.new :: IO Stream2HeaderBlockFragment-- -- Warning: we should find a way of coping with different table sizes.- decode_headers_table <- HP.newDynamicTableForDecoding 4096- decode_headers_table_mvar <- newMVar decode_headers_table-- encode_headers_table <- HP.newDynamicTableForEncoding 4096- encode_headers_table_mvar <- newMVar encode_headers_table-- -- These ones need independent threads taking care of sending stuff- -- their way... - headers_output <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers))- data_output <- newChan :: IO (Chan (GlobalStreamId,B.ByteString))-- stream2postinputmechanism <- H.new - stream2workerthread <- H.new-- -- What about stream cancellation?- cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet)-- let for_worker_thread = WorkerThreadEnvironment {- _streamId = error "NotInitialized" - ,_headersOutput = headers_output- ,_dataOutput = data_output- ,_streamsCancelled_WTE = cancelled_streams_mvar- }-- let session_data = SessionData {- _sessionsContext = sessions_context- ,_sessionInput = session_input - ,_sessionOutput = session_output_mvar- ,_toDecodeHeaders = decode_headers_table_mvar- ,_toEncodeHeaders = encode_headers_table_mvar- ,_stream2HeaderBlockFragment = stream_request_headers- ,_forWorkerThread = for_worker_thread- ,_coherentWorker = coherent_worker- ,_streamsCancelled = cancelled_streams_mvar- ,_stream2PostInputMechanism = stream2postinputmechanism- ,_stream2WorkerThread = stream2workerthread- ,_sessionIdAtSession = session_id- }-- let - exc_handler :: SessionComponent -> HTTP2SessionException -> IO () - exc_handler component e = sessionExceptionHandler component session_id sessions_context e- exc_guard :: SessionComponent -> IO () -> IO ()- exc_guard component action = E.catch action $ exc_handler component-- -- Create an input thread that decodes frames...- forkIO $ exc_guard SessionInputThread_SessionComponent - $ runReaderT sessionInputThread session_data- - -- Create a thread that captures headers and sends them down the tube - forkIO $ exc_guard SessionHeadersOutputThread_SessionComponent - $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data-- -- Create a thread that captures data and sends it down the tube- forkIO $ exc_guard SessionDataOutputThread_SessionComponent - $ dataOutputThread data_output session_output_mvar-- -- The two previous threads fill the session_output argument below (they write to it)- -- the session machinery in the other end is in charge of sending that data through the - -- socket.-- return ( (SessionInput session_input),- (SessionOutput session_output) )----sessionInputThread :: ReaderT SessionData IO ()-sessionInputThread = do - liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread"-- -- This is an introductory and declarative block... all of this is tail-executed- -- every time that a packet needs to be processed. It may be a good idea to abstract- -- these values in a closure... - session_input <- view sessionInput - - decode_headers_table_mvar <- view toDecodeHeaders - stream_request_headers <- view stream2HeaderBlockFragment- cancelled_streams_mvar <- view streamsCancelled- coherent_worker <- view coherentWorker-- for_worker_thread_uns <- view forWorkerThread- stream2workerthread <- view stream2WorkerThread-- input <- liftIO $ readChan session_input-- liftIO $ debugM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input)-- case input of -- Left CancelSession_SIC -> do - -- Good place to tear down worker threads... Let the rest of the finalization- -- to the framer- liftIO $ do - H.mapM_- (\ (_, thread_id) -> do- throwTo thread_id StreamCancelledException- infoM "HTTP2.Session" $ "Stream successfully interrupted"- )- stream2workerthread-- -- We do not continue here, but instead let it finish- return ()-- Right frame | Just (stream_id, bytes) <- frameIsHeaderOfStream frame -> do - -- Just append the frames to streamRequestHeaders- appendHeaderFragmentBlock stream_id bytes-- if frameEndsHeaders frame then - do- -- Let's decode the headers- let for_worker_thread = set streamId stream_id for_worker_thread_uns - headers_bytes <- getHeaderBytes stream_id- dyn_table <- liftIO $ takeMVar decode_headers_table_mvar- (new_table, header_list ) <- liftIO $ HP.decodeHeader dyn_table headers_bytes- -- Good moment to remove the headers from the table.... we don't want a space- -- leak here - liftIO $ H.delete stream_request_headers stream_id- liftIO $ putMVar decode_headers_table_mvar new_table-- -- If the headers end the request.... - post_data_source <- if not (frameEndsStream frame)- then do - - mechanism <- createMechanismForStream stream_id - let source = postDataSourceFromMechanism mechanism- return $ Just source- else do - return Nothing--- -- I'm clear to start the worker, in its own thread- -- !!- liftIO $ do - thread_id <- forkIO $ runReaderT - (workerThread (header_list, post_data_source) coherent_worker)- for_worker_thread - H.insert stream2workerthread stream_id thread_id-- return ()- else - -- Frame doesn't end the headers... it was added before... so- -- probably do nothing - return ()- - continue -- Right frame@(NH2.Frame _ (NH2.RSTStreamFrame error_code_id)) -> do- let stream_id = streamIdFromFrame frame- liftIO $ do - infoM "HTTP2.Session" $ "Stream reset: " ++ (show error_code_id)- cancelled_streams <- takeMVar cancelled_streams_mvar- infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id)- putMVar cancelled_streams_mvar $ NS.insert stream_id cancelled_streams- maybe_thread_id <- H.lookup stream2workerthread stream_id- case maybe_thread_id of - Nothing -> - -- This is actually more like an internal error- error "InterruptingUnexistentStream"-- Just thread_id -> do- throwTo thread_id StreamCancelledException- infoM "HTTP2.Session" $ "Stream successfully interrupted"-- continue -- Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) -> do - -- So I got data to process- -- TODO: Handle end of stream- let stream_id = NH2.fromStreamIdentifier nh2_stream_id - streamWorkerSendData stream_id somebytes- -- After that data has been received and forwarded downstream, we can issue a windows update- --- -- TODO: We can use wider frames to avoid congestion...- -- .... and we can also be more compositional with these short bursts of data....- sendOutFrame- (NH2.EncodeInfo- NH2.defaultFlags- nh2_stream_id- Nothing- )- (NH2.WindowUpdateFrame- (fromIntegral (B.length somebytes))- )- sendOutFrame- (NH2.EncodeInfo- NH2.defaultFlags- (NH2.toStreamIdentifier 0)- Nothing- )- (NH2.WindowUpdateFrame- (fromIntegral (B.length somebytes))- ) -- if frameEndsStream frame - then do - -- Good place to close the source ... - closePostDataSource stream_id - else - return ()-- continue -- Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do - -- Deal with pings: this is an Ack, so do nothing- continue -- Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do - -- Deal with pings: NOT an Ack, so answer- liftIO $ debugM "HTTP2.Session" "Ping processed"- sendOutFrame- (NH2.EncodeInfo- (NH2.setAck NH2.defaultFlags)- (NH2.toStreamIdentifier 0)- Nothing - )- (NH2.PingFrame somebytes)-- continue -- Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do - -- Frame was received by the peer, do nothing here...- continue --- Right (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do - liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list)- -- Just acknowledge the frame.... for now - sendOutFrame - (NH2.EncodeInfo- (NH2.setAck NH2.defaultFlags)- (NH2.toStreamIdentifier 0)- Nothing )- (NH2.SettingsFrame [])-- continue --- Right somethingelse -> do - -- An undhandled case here....- liftIO $ errorM "HTTP2.Session" $ "Received problematic frame: "- liftIO $ errorM "HTTP2.Session" $ ".. " ++ (show somethingelse)-- continue -- where -- continue = sessionInputThread-- sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()- sendOutFrame encode_info payload = do - session_output_mvar <- view sessionOutput - session_output <- liftIO $ takeMVar session_output_mvar-- liftIO $ writeChan session_output $ Right (encode_info, payload)-- liftIO $ putMVar session_output_mvar session_output---frameEndsStream :: InputFrame -> Bool -frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags---createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism-createMechanismForStream stream_id = do - (chan, source) <- liftIO $ unfoldChannelAndSource- stream2postinputmechanism <- view stream2PostInputMechanism- let pim = PostInputMechanism (chan, source)- liftIO $ H.insert stream2postinputmechanism stream_id pim - return pim----- TODO: Can be optimized by factoring out the mechanism lookup-closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO ()-closePostDataSource stream_id = do - stream2postinputmechanism <- view stream2PostInputMechanism-- pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id -- case pim_maybe of -- Just (PostInputMechanism (chan, _)) -> - liftIO $ writeChan chan Nothing-- Nothing -> - -- This is an internal error, the mechanism should be - -- created when the stream ends- error "Internal error/closePostDataSource"---streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO ()-streamWorkerSendData stream_id bytes = do - s2pim <- view stream2PostInputMechanism- pim_maybe <- liftIO $ H.lookup s2pim stream_id -- case pim_maybe of -- Just pim -> - sendBytesToPim pim bytes-- Nothing -> - -- This is an internal error, the mechanism should be - -- created when the headers end- error "Internal error"---sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO ()-sendBytesToPim (PostInputMechanism (chan, _)) bytes = - liftIO $ writeChan chan (Just bytes)---postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream-postDataSourceFromMechanism (PostInputMechanism (_, source)) = source---isSettingsAck :: NH2.FrameHeader -> Bool -isSettingsAck (NH2.FrameHeader _ flags _) = - NH2.testAck flags---isStreamCancelled :: GlobalStreamId -> WorkerMonad Bool -isStreamCancelled stream_id = do - cancelled_streams_mvar <- view streamsCancelled_WTE- cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar- return $ NS.member stream_id cancelled_streams---workerThread :: Request -> CoherentWorker -> WorkerMonad ()-workerThread req coherent_worker =- do- headers_output <- view headersOutput- stream_id <- view streamId- (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req-- -- liftIO $ putStrLn $ "Num pushed streams: " ++ (show $ length pushed_streams)-- -- Now I send the headers, if that's possible at all- headers_sent <- liftIO $ newEmptyMVar- liftIO $ writeChan headers_output (stream_id, headers_sent, headers)-- -- At this moment I should ask if the stream hasn't been cancelled by the browser before- -- commiting to the work of sending addtitional data- is_stream_cancelled <- isStreamCancelled stream_id- if not is_stream_cancelled-- then do- -- I have a beautiful source that I can de-construct...- -- TODO: Optionally pulling data out from a Conduit ....- -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )- -- - -- This threadlet should block here waiting for the headers to finish going- (maybe_footers, _) <- runConduit $- (transPipe liftIO data_and_conclussion) - `fuseBothMaybe` - (sendDataOfStream stream_id headers_sent)- -- BIG TODO: Send the headers ... likely stream conclusion semantics - -- will need to be changed. - return ()- else -- return ()---- v-- comp. monad.-sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()-sendDataOfStream stream_id headers_sent = do- data_output <- view dataOutput- transPipe liftIO $ do - -- Wait for permission to send the data- liftIO $ takeMVar headers_sent- foldMapM $ \ bytes ->- writeChan data_output (stream_id, bytes)----- sendDataOfStream :: Sink ----- Allow this very important function to be used in the future to process footers---- difficultFunction :: (Monad m)--- => ConduitM () a2 m r1 -> ConduitM a2 Void m r2--- -> m (r2, Maybe r1)--- difficultFunction l r = liftM (fmap getLast) $ runWriterT (l' $$ r')--- where--- l' = transPipe lift l >>= lift . tell . Last . Just--- r' = transPipe lift r----appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO ()-appendHeaderFragmentBlock global_stream_id bytes = do - ht <- view stream2HeaderBlockFragment - maybe_old_block <- liftIO $ H.lookup ht global_stream_id- new_block <- return $ case maybe_old_block of -- Nothing -> Bu.byteString bytes-- Just something -> something `mappend` (Bu.byteString bytes) -- liftIO $ H.insert ht global_stream_id new_block---getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString-getHeaderBytes global_stream_id = do - ht <- view stream2HeaderBlockFragment - Just bytes <- liftIO $ H.lookup ht global_stream_id- return $ Bl.toStrict $ Bu.toLazyByteString bytes---frameIsHeaderOfStream :: InputFrame -> Maybe (GlobalStreamId, B.ByteString)-frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) )- = Just (NH2.fromStreamIdentifier stream_id, block_fragment)-frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )- = Just (NH2.fromStreamIdentifier stream_id, block_fragment)-frameIsHeaderOfStream _ - = Nothing ---frameEndsHeaders :: InputFrame -> Bool -frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags---streamIdFromFrame :: InputFrame -> GlobalStreamId-streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id----- TODO: Have different size for the headers..... just now going with a default size of 16 k...--- TODO: Find a way to kill this thread....-headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers)- -> MVar (Chan (Either SessionOutputCommand OutputFrame)) - -> ReaderT SessionData IO ()-headersOutputThread input_chan session_output_mvar = forever $ do - (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan- -- liftIO $ putStrLn $ "Output headers: " ++ (show headers)-- -- First encode the headers using the table- encode_dyn_table_mvar <- view toEncodeHeaders-- encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar- (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers- liftIO $ putMVar encode_dyn_table_mvar new_dyn_table-- -- Now split the bytestring in chunks of the needed size.... - let bs_chunks = bytestringChunk useChunkLength data_to_send-- -- And send the chunks through while locking the output place....- session_output <- liftIO $ takeMVar session_output_mvar-- -- First frame is just a headers frame....- if (length bs_chunks) == 1 - then- do - let flags = NH2.setEndHeader NH2.defaultFlags-- -- Write the first frame - liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = flags- ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.HeadersFrame Nothing (head bs_chunks)- )- else - do - let flags = NH2.defaultFlags- -- Write the first frame - liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = flags- ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.HeadersFrame Nothing (head bs_chunks)- )- -- And write the other frames- let - writeContinuations :: [B.ByteString] -> ReaderT SessionData IO ()- writeContinuations (last_fragment:[]) = liftIO $- writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.ContinuationFrame last_fragment- )- writeContinuations (fragment:xs) = do - liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.ContinuationFrame fragment- )- writeContinuations xs--- writeContinuations (tail bs_chunks)- -- Restore output capability, so that other pieces waiting can send...- liftIO $ putMVar session_output_mvar session_output- -- And say that the headers for this thread are out - liftIO $ putMVar headers_ready_mvar HeadersSent- -- liftIO $ putStrLn "Headers were output"- --bytestringChunk :: Int -> B.ByteString -> [B.ByteString]-bytestringChunk len s | (B.length s) < len = [ s ]-bytestringChunk len s = h:(bytestringChunk len xs)- where - (h, xs) = B.splitAt len s ----- TODO: find a clean way to finish this thread (maybe with negative stream ids?)--- TODO: This function does non-optimal chunking for the case where responses are--- actually streamed.... in those cases we need to keep state for frames in --- some other format.... -dataOutputThread :: Chan (GlobalStreamId, B.ByteString)- -> MVar (Chan (Either SessionOutputCommand OutputFrame)) - -> IO ()-dataOutputThread input_chan session_output_mvar = forever $ do - (stream_id, contents) <- readChan input_chan-- -- And now just simply output it...- let bs_chunks = bytestringChunk useChunkLength contents- -- putStrLn $ "Chunk lengths: " ++ (show (map B.length bs_chunks))-- -- And send the chunks through while locking the output place....- session_output <- liftIO $ takeMVar session_output_mvar-- -- First frame is the only one:- if (length bs_chunks) == 1 - then- do - let flags = NH2.setEndStream NH2.defaultFlags-- -- Write the first frame - liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = flags- ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.DataFrame (head bs_chunks)- )- else - do - let flags = NH2.defaultFlags- -- Write the first frame - liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = flags- ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.DataFrame (head bs_chunks)- )- -- And write the other frames- let - writeContinuations :: [B.ByteString] -> IO ()- writeContinuations (last_fragment:[]) = liftIO $- writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.DataFrame last_fragment- )- writeContinuations (fragment:xs) = do - liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {- NH2.encodeFlags = NH2.defaultFlags - ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id - ,NH2.encodePadding = Nothing }, -- NH2.DataFrame fragment- )- writeContinuations xs- writeContinuations (tail bs_chunks)- -- Restore output capability, so that other pieces waiting can send...- liftIO $ debugM "HTTP2.Session" $ "Output capability restored"- liftIO $ putMVar session_output_mvar session_output ---sessionExceptionHandler :: E.Exception e => SessionComponent -> Int -> SessionsContext -> e -> IO ()-sessionExceptionHandler session_component session_id sessions_context e = do - let- getit = ( sessionsConfig . sessionsCallbacks . reportErrorCallback ) - maybe_error_callback = sessions_context ^. getit - error_tuple = (- session_component,- SessionCoordinates session_id, - E.toException e- )- case maybe_error_callback of - Nothing -> - errorM "HTTP2.Session" (show (e))-- Just callback -> - callback error_tuple
hs-src/SecondTransfer/MainLoop.hs view
@@ -18,6 +18,8 @@ ,tlsServeWithALPN ,tlsServeWithALPNAndFinishOnRequest + ,enableConsoleLogging+ ,TLSLayerGenericProblem(..) ,FinishRequest(..) ) where @@ -29,3 +31,4 @@ ) import SecondTransfer.MainLoop.OpenSSL_TLS+import SecondTransfer.MainLoop.Logging (enableConsoleLogging)
+ hs-src/SecondTransfer/MainLoop/Logging.hs view
@@ -0,0 +1,59 @@+module SecondTransfer.MainLoop.Logging (+ -- | Simple, no fuss enable logging+ enableConsoleLogging+ ) where++import System.IO (stderr)+++-- Logging utilities+import System.Log.Formatter (simpleLogFormatter)+import System.Log.Handler (setFormatter, LogHandler)+import System.Log.Handler.Simple+-- import System.Log.Handler.Syslog (Facility (..), Option (..), openlog)+import System.Log.Logger+++-- | Activates logging to terminal+enableConsoleLogging :: IO ()+enableConsoleLogging = configureLoggingToConsole+++configureLoggingToConsole :: IO ()+configureLoggingToConsole = do + s <- streamHandler stderr DEBUG >>= + \lh -> return $ setFormatter lh (simpleLogFormatter "[$time : $loggername : $prio] $msg")+ setLoggerLevels s+++-- configureLoggingToSyslog :: IO ()+-- configureLoggingToSyslog = do +-- s <- openlog "RehMimic" [PID] DAEMON INFO >>= +-- \lh -> return $ setFormatter lh (simpleLogFormatter "[$time : $loggername : $prio] $msg")+-- setLoggerLevels s+++setLoggerLevels :: (LogHandler s) => s -> IO () +setLoggerLevels s = do+ updateGlobalLogger rootLoggerName removeHandler+ updateGlobalLogger "HTTP2.Session" (+ setHandlers [s] . + setLevel INFO + )+ updateGlobalLogger "OpenSSL" (+ setHandlers [s] . + setLevel INFO + )+ updateGlobalLogger "HarWorker" (+ setHandlers [s] . + setLevel DEBUG + )+ updateGlobalLogger "ResearchWorker" (+ setHandlers [s] . + setLevel DEBUG + )+ updateGlobalLogger "HTTP2.Framer" (+ setHandlers [s] . + setLevel DEBUG+ )+
+ hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs view
@@ -0,0 +1,357 @@+{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings, DeriveDataTypeable #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.MainLoop.OpenSSL_TLS(+ tlsServeWithALPN+ ,tlsServeWithALPNAndFinishOnRequest+ -- ,tlsServeWithALPNOnce++ ,TLSLayerGenericProblem(..)+ ,FinishRequest(..)+ ) where ++++import Control.Monad+import Control.Concurrent.MVar +import Control.Exception +import qualified Control.Exception as E+import Data.Foldable (foldMap)+import Data.Typeable +import Data.Monoid ()+import Foreign+import Foreign.C++import qualified Data.ByteString as B+import qualified Data.ByteString.Builder as BB+import Data.ByteString.Char8 (pack)+import qualified Data.ByteString.Lazy as LB+import qualified Data.ByteString.Unsafe as BU++import System.Log.Logger++import SecondTransfer.MainLoop.PushPullType++#include "Logging.cpphs"++-- | Exception inheriting from `IOProblem`. This is thrown by the +-- OpenSSL subsystem to signal that the connection was broken or that +-- otherwise there was a problem at the SSL layer. +data TLSLayerGenericProblem = TLSLayerGenericProblem String+ deriving (Show, Typeable)+++instance Exception TLSLayerGenericProblem where + toException = toException . IOProblem + fromException x = do + IOProblem a <- fromException x + cast a+++data InterruptibleEither a b = + Left_I a + |Right_I b + |Interrupted+++-- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, +-- the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish+-- at its earliest convenience and call the `CloseAction` for any open sessions.+data FinishRequest = FinishRequest+++-- These names are absolutely improper....+-- Session creator+data Connection_t +-- Session+data Wired_t++type Connection_Ptr = Ptr Connection_t +type Wired_Ptr = Ptr Wired_t+++-- Actually, this makes a listener for new connections+-- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, +-- char* protocol_list, int protocol_list_len)+foreign import ccall "make_connection" makeConnection :: + CString -- cert filename+ -> CString -- privkey_filename+ -> CString -- hostname+ -> CInt -- port+ -> Ptr CChar -- protocol list+ -> CInt -- protocol list length+ -> IO Connection_Ptr++allOk :: CInt +allOk = 0 ++badHappened :: CInt +badHappened = 1 ++timeoutReached :: CInt +timeoutReached = 3++-- int wait_for_connection(connection_t* conn, wired_session_t** wired_session);+foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt ++-- int send_data(wired_session_t* ws, char* buffer, int buffer_size);+foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt ++-- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);+foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt++-- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; }+foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt++-- void dispose_wired_session(wired_session_t* ws);+foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO ()++foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO ()+++useBufferSize :: Int+useBufferSize = 4096+++type Protocols = [B.ByteString]+++protocolsToWire :: Protocols -> B.ByteString+protocolsToWire protocols = + LB.toStrict . BB.toLazyByteString $ + foldMap (\ protocol + -> (BB.lazyByteString . LB.fromChunks)+ [ B.singleton $ fromIntegral $ B.length protocol,+ protocol + ]+ ) protocols+++-- | Simple function to open +tlsServeWithALPN :: FilePath -- ^ Path to a certificate the server is going to use to identify itself.+ -- Bear in mind that multiple domains can be served from the same HTTP/2 + -- TLS socket, so please create the HTTP/2 certificate accordingly.+ -> FilePath -- ^ Path to the key of your certificate. + -> String -- ^ Name of the network interface where you want to start your server+ -> [(String, Attendant)] -- ^ List of protocol names and the corresponding `Attendant` to use for + -- each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the+ -- same socket. When no ALPN negotiation is present during the negotiation, + -- the first protocol in this list is used.+ -> Int -- ^ Port to open to listen for connections. + -> IO ()+tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do ++ let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants+ INSTRUMENTATION( infoM "OpenSSL" "Entering tlsServeWithALPN" )+ withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do ++ connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->+ makeConnection + c_certfn+ c_keyfn+ c_iname+ (fromIntegral interface_port)+ pchar + (fromIntegral len)++ if connection_ptr == nullPtr + then do+ INSTRUMENTATION( errorM "OpenSSL" "Could not create listening socket" )+ throwIO $ TLSLayerGenericProblem "Could not create listening end"+ else do+ INSTRUMENTATION( infoM "OpenSSL" "Listening soxket created" )+ return ()++ forever $ do + either_wired_ptr <- alloca $ \ wired_ptr_ptr -> + let + tryOnce = do + result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr+ let + r = case result_code of + re | re == allOk -> do + p <- peek wired_ptr_ptr+ INSTRUMENTATION( infoM "OpenSSL" "A connection was accepted" )+ return $ Right p+ | re == timeoutReached -> tryOnce + | re == badHappened -> return $ Left "A wait for connection failed"+ r + in tryOnce++ case either_wired_ptr of ++ Left msg -> do + INSTRUMENTATION( errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg )+ return ()++ Right wired_ptr -> do + already_closed_mvar <- newMVar False+ let + pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do + result <- sendData wired_ptr pchar (fromIntegral len)+ case result of + r | r == allOk -> return ()+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data"+ pullAction = do + allocaBytes useBufferSize $ \ pcharbuffer -> + alloca $ \ data_recvd_ptr -> do + result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr+ INSTRUMENTATION( debugM "OpenSSL" "Received data" )+ recvd_bytes <- case result of + r | r == allOk -> peek data_recvd_ptr+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"++ B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)++ -- Ensure that the socket and the struct are only closed once+ closeAction = do+ b <- readMVar already_closed_mvar+ if not b + then do+ putMVar already_closed_mvar True+ disposeWiredSession wired_ptr+ else + return ()++ use_protocol <- getSelectedProtocol wired_ptr+ INSTRUMENTATION( infoM "OpenSSL" $ "Selected protocol:" ++ (show use_protocol))++ let + maybe_session_attendant = case fromIntegral use_protocol of + n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + -- Or just select the first one+ | otherwise -> Just . snd . head $ attendants++ case maybe_session_attendant of ++ Just session_attendant -> + E.catch + (session_attendant pushAction pullAction closeAction)+ ((\ e -> do + INSTRUMENTATION( errorM "OpenSSL" " ** Session ended by TLSLayerGenericProblem (well handled)")+ throwIO e+ )::TLSLayerGenericProblem -> IO () )+++ Nothing ->+ return ()+++-- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to ask +-- the server to finish: you pass an empty MVar and when you want to finish you +-- just populate it. +tlsServeWithALPNAndFinishOnRequest :: FilePath + -> FilePath -- ^ Same as for `tlsServeWithALPN` + -> String -- ^ Same as for `tlsServeWithALPN`+ -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN`+ -> Int -- ^ Same as for `tlsServeWithALPN`+ -> MVar FinishRequest -- ^ Finish request, write a value here to finish serving+ -> IO ()+tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do ++ let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants+ withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do ++ -- Create an accepting endpoint+ connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->+ makeConnection + c_certfn+ c_keyfn+ c_iname+ (fromIntegral interface_port)+ pchar + (fromIntegral len)++ -- Create a computation that accepts a connection, runs a session on it and recurses+ let + recursion = do + -- Get a SSL session+ either_wired_ptr <- alloca $ \ wired_ptr_ptr -> + let + tryOnce = do + result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr+ let + r = case result_code of + re | re == allOk -> do + p <- peek wired_ptr_ptr+ return $ Right_I p+ | re == timeoutReached -> do + got_finish_request <- tryTakeMVar finish_request+ case got_finish_request of + Nothing ->+ tryOnce+ Just _ ->+ return Interrupted ++ | re == badHappened -> return $ Left_I "A wait for connection failed"+ r + in tryOnce++ -- With the potentially obtained SSL session do...+ case either_wired_ptr of ++ Left_I msg -> do + errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg++ -- // .. //+ recursion++ Right_I wired_ptr -> do + already_closed_mvar <- newMVar False+ let + pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do + result <- sendData wired_ptr pchar (fromIntegral len)+ case result of + r | r == allOk -> return ()+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data"+ pullAction = do + allocaBytes useBufferSize $ \ pcharbuffer -> + alloca $ \ data_recvd_ptr -> do + result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr+ recvd_bytes <- case result of + r | r == allOk -> peek data_recvd_ptr+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"++ B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)+ closeAction = do+ b <- readMVar already_closed_mvar+ if not b + then do+ putMVar already_closed_mvar True+ disposeWiredSession wired_ptr+ else + return ()++ use_protocol <- getSelectedProtocol wired_ptr++ infoM "OpenSSL" $ ".. Using protocol: " ++ (show use_protocol)++ let + maybe_session_attendant = case fromIntegral use_protocol of + n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + | otherwise -> Just . snd . head $ attendants++ case maybe_session_attendant of ++ Just session_attendant -> + session_attendant pushAction pullAction closeAction++ Nothing ->+ return ()++ -- // .. //+ recursion ++ Interrupted -> do+ infoM "OpenSSL" "Connection closed"+ closeConnection connection_ptr++ -- Start the loop defined above...+ recursion ++-- When we are using the eternal version of this function, wake up +-- each second .... +defaultWaitTime :: CInt+defaultWaitTime = 200000+-- Okej, more responsiviness needed +smallWaitTime :: CInt +smallWaitTime = 50000
− hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs
@@ -1,350 +0,0 @@-{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings, DeriveDataTypeable #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.MainLoop.OpenSSL_TLS(- tlsServeWithALPN- ,tlsServeWithALPNAndFinishOnRequest- -- ,tlsServeWithALPNOnce-- ,TLSLayerGenericProblem(..)- ,FinishRequest(..)- ) where ----import Control.Monad-import Control.Concurrent.MVar -import Control.Exception -import qualified Control.Exception as E-import Data.Foldable (foldMap)-import Data.Typeable -import Data.Monoid ()-import Foreign-import Foreign.C--import qualified Data.ByteString as B-import qualified Data.ByteString.Builder as BB-import Data.ByteString.Char8 (pack)-import qualified Data.ByteString.Lazy as LB-import qualified Data.ByteString.Unsafe as BU--import System.Log.Logger--import SecondTransfer.MainLoop.PushPullType- ----- | Exception inheriting from `IOProblem`. This is thrown by the --- OpenSSL subsystem to signal that the connection was broken or that --- otherwise there was a problem at the SSL layer. -data TLSLayerGenericProblem = TLSLayerGenericProblem String- deriving (Show, Typeable)---instance Exception TLSLayerGenericProblem where - toException = toException . IOProblem - fromException x = do - IOProblem a <- fromException x - cast a---data InterruptibleEither a b = - Left_I a - |Right_I b - |Interrupted----- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, --- the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish--- at its earliest convenience and call the `CloseAction` for any open sessions.-data FinishRequest = FinishRequest----- These names are absolutely improper....--- Session creator-data Connection_t --- Session-data Wired_t--type Connection_Ptr = Ptr Connection_t -type Wired_Ptr = Ptr Wired_t----- Actually, this makes a listener for new connections--- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, --- char* protocol_list, int protocol_list_len)-foreign import ccall "make_connection" makeConnection :: - CString -- cert filename- -> CString -- privkey_filename- -> CString -- hostname- -> CInt -- port- -> Ptr CChar -- protocol list- -> CInt -- protocol list length- -> IO Connection_Ptr--allOk :: CInt -allOk = 0 --badHappened :: CInt -badHappened = 1 --timeoutReached :: CInt -timeoutReached = 3---- int wait_for_connection(connection_t* conn, wired_session_t** wired_session);-foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt ---- int send_data(wired_session_t* ws, char* buffer, int buffer_size);-foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt ---- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);-foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt---- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; }-foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt---- void dispose_wired_session(wired_session_t* ws);-foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO ()--foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO ()---useBufferSize :: Int-useBufferSize = 4096---type Protocols = [B.ByteString]---protocolsToWire :: Protocols -> B.ByteString-protocolsToWire protocols = - LB.toStrict . BB.toLazyByteString $ - foldMap (\ protocol - -> (BB.lazyByteString . LB.fromChunks)- [ B.singleton $ fromIntegral $ B.length protocol,- protocol - ]- ) protocols----- | Simple function to open -tlsServeWithALPN :: FilePath -- ^ Path to a certificate the server is going to use to identify itself.- -- Bear in mind that multiple domains can be served from the same HTTP/2 - -- TLS socket, so please create the HTTP/2 certificate accordingly.- -> FilePath -- ^ Path to the key of your certificate. - -> String -- ^ Name of the network interface where you want to start your server- -> [(String, Attendant)] -- ^ List of protocol names and the corresponding `Attendant` to use for - -- each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the- -- same socket.- -> Int -- ^ Port to open to listen for connections. - -> IO ()-tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do -- let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants- withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do -- connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->- makeConnection - c_certfn- c_keyfn- c_iname- (fromIntegral interface_port)- pchar - (fromIntegral len)-- if connection_ptr == nullPtr - then - throwIO $ TLSLayerGenericProblem "Could not create listening end"- else - return ()-- forever $ do - either_wired_ptr <- alloca $ \ wired_ptr_ptr -> - let - tryOnce = do - result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr- let - r = case result_code of - re | re == allOk -> do - p <- peek wired_ptr_ptr- return $ Right p- | re == timeoutReached -> tryOnce - | re == badHappened -> return $ Left "A wait for connection failed"- r - in tryOnce-- case either_wired_ptr of -- Left msg -> do - errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg--- Right wired_ptr -> do - already_closed_mvar <- newMVar False- let - pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do - result <- sendData wired_ptr pchar (fromIntegral len)- case result of - r | r == allOk -> return ()- | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data"- pullAction = do - allocaBytes useBufferSize $ \ pcharbuffer -> - alloca $ \ data_recvd_ptr -> do - result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr- recvd_bytes <- case result of - r | r == allOk -> peek data_recvd_ptr- | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"-- B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)-- -- Ensure that the socket and the struct are only closed once- closeAction = do- b <- readMVar already_closed_mvar- if not b - then do- putMVar already_closed_mvar True- disposeWiredSession wired_ptr- else - return ()-- use_protocol <- getSelectedProtocol wired_ptr-- -- putStrLn $ ".. Using protocol: " ++ (show use_protocol)-- let - maybe_session_attendant = case fromIntegral use_protocol of - n | (use_protocol >= 0) -> Just $ snd $ attendants !! n - | otherwise -> Nothing -- case maybe_session_attendant of -- Just session_attendant -> - E.catch - (session_attendant pushAction pullAction closeAction)- ((\ e -> do - errorM "OpenSSL" " ** Session ended by TLSLayerGenericProblem (well handled)"- throwIO e- )::TLSLayerGenericProblem -> IO () )--- Nothing ->- return ()----- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to notify --- the server of finishing. -tlsServeWithALPNAndFinishOnRequest :: FilePath - -> FilePath -- ^ Same as for `tlsServeWithALPN` - -> String -- ^ Same as for `tlsServeWithALPN`- -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN`- -> Int -- ^ Same as for `tlsServeWithALPN`- -> MVar FinishRequest -- ^ Finish request- -> IO ()-tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do -- let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants- withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do -- -- Create an accepting endpoint- connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->- makeConnection - c_certfn- c_keyfn- c_iname- (fromIntegral interface_port)- pchar - (fromIntegral len)-- -- Create a computation that accepts a connection, runs a session on it and recurses- let - recursion = do - -- Get a SSL session- either_wired_ptr <- alloca $ \ wired_ptr_ptr -> - let - tryOnce = do - result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr- let - r = case result_code of - re | re == allOk -> do - p <- peek wired_ptr_ptr- return $ Right_I p- | re == timeoutReached -> do - got_finish_request <- tryTakeMVar finish_request- case got_finish_request of - Nothing ->- tryOnce- Just _ ->- return Interrupted -- | re == badHappened -> return $ Left_I "A wait for connection failed"- r - in tryOnce-- -- With the potentially obtained SSL session do...- case either_wired_ptr of -- Left_I msg -> do - errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg-- -- // .. //- recursion-- Right_I wired_ptr -> do - already_closed_mvar <- newMVar False- let - pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do - result <- sendData wired_ptr pchar (fromIntegral len)- case result of - r | r == allOk -> return ()- | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data"- pullAction = do - allocaBytes useBufferSize $ \ pcharbuffer -> - alloca $ \ data_recvd_ptr -> do - result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr- recvd_bytes <- case result of - r | r == allOk -> peek data_recvd_ptr- | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"-- B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)- closeAction = do- b <- readMVar already_closed_mvar- if not b - then do- putMVar already_closed_mvar True- disposeWiredSession wired_ptr- else - return ()-- use_protocol <- getSelectedProtocol wired_ptr-- infoM "OpenSSL" $ ".. Using protocol: " ++ (show use_protocol)-- let - maybe_session_attendant = case fromIntegral use_protocol of - n | (use_protocol >= 0) -> Just $ snd $ attendants !! n - | otherwise -> Nothing -- case maybe_session_attendant of -- Just session_attendant -> - session_attendant pushAction pullAction closeAction-- Nothing ->- return ()-- -- // .. //- recursion -- Interrupted -> do- infoM "OpenSSL" "Connection closed"- closeConnection connection_ptr-- -- Start the loop defined above...- recursion ---- When we are using the eternal version of this function, wake up --- each second .... -defaultWaitTime :: CInt-defaultWaitTime = 200000--- Okej, more responsiviness needed -smallWaitTime :: CInt -smallWaitTime = 50000
second-transfer.cabal view
@@ -1,7 +1,4 @@--- Initial spdy-ping.cabal generated by cabal init. For further --- documentation-- see http://haskell.org/cabal/users-guide/ --- The name of the package. name : second-transfer -- The package version. See the Haskell package versioning policy (PVP) @@ -10,7 +7,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version : 0.2.0.0+version : 0.3.0.2 synopsis : Second Transfer HTTP/2 web server @@ -42,6 +39,10 @@ -- Constraint on the version of Cabal needed to build this package. cabal-version: >=1.10 +Flag debug + Description: Enable debug support + Default: False+ source-repository head type: git location: git@github.com:alcidesv/second-transfer.git@@ -49,7 +50,7 @@ source-repository this type: git location: git@github.com:alcidesv/second-transfer.git- tag: 0.2.0.0+ tag: 0.3.0.2 library@@ -60,6 +61,7 @@ , SecondTransfer.Http2 , SecondTransfer.MainLoop.Internal , SecondTransfer.Exception+ , SecondTransfer.MainLoop.Logging other-modules: SecondTransfer.MainLoop.CoherentWorker , SecondTransfer.MainLoop.PushPullType@@ -70,6 +72,16 @@ , SecondTransfer.Http2.MakeAttendant , SecondTransfer.Http2.Session ++ build-tools: cpphs++ if flag(debug)+ CPP-Options: -DENABLE_DEBUG+ if !os(windows)+ CC-Options: "-DDEBUG"+ else+ CC-Options: "-DNDEBUG"+ -- LANGUAGE extensions used by modules in this package. -- other-extensions: @@ -148,6 +160,8 @@ extra-libraries: ssl crypto extra-lib-dirs: /opt/openssl-1.0.2/lib++ include-dirs: macros/
tests/tests-hs-src/compiling_ok.hs view
@@ -3,8 +3,9 @@ CoherentWorker , Footers , DataAndConclusion- , tlsServeWithALPN+ , tlsServeWithALPNAndFinishOnRequest , http2Attendant+ , FinishRequest(..) ) import SecondTransfer.Http2( makeSessionsContext@@ -12,7 +13,8 @@ ) import Data.Conduit-+import Control.Concurrent (threadDelay, forkIO)+import Control.Concurrent.MVar saysHello :: DataAndConclusion saysHello = do @@ -35,9 +37,13 @@ -- the developement directory. main = do sessions_context <- makeSessionsContext defaultSessionsConfig+ finish <- newEmptyMVar+ forkIO $ do + threadDelay 1000000+ putMVar finish FinishRequest let http2_attendant = http2Attendant sessions_context helloWorldWorker- tlsServeWithALPN+ tlsServeWithALPNAndFinishOnRequest "tests/support/servercert.pem" -- Server certificate "tests/support/privkey.pem" -- Certificate private key "127.0.0.1" -- On which interface to bind@@ -46,4 +52,5 @@ ("h2", http2_attendant) -- they may be slightly different, but for this -- test it doesn't matter. ]- 8000 + 8000+ finish