diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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. diff --git a/cbits/tlsinc.c b/cbits/tlsinc.c --- a/cbits/tlsinc.c +++ b/cbits/tlsinc.c @@ -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; } diff --git a/hs-src/SecondTransfer.hs b/hs-src/SecondTransfer.hs --- a/hs-src/SecondTransfer.hs +++ b/hs-src/SecondTransfer.hs @@ -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 diff --git a/hs-src/SecondTransfer/Http2/Framer.cpphs b/hs-src/SecondTransfer/Http2/Framer.cpphs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/Framer.cpphs @@ -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 () diff --git a/hs-src/SecondTransfer/Http2/Framer.hs b/hs-src/SecondTransfer/Http2/Framer.hs deleted file mode 100644 --- a/hs-src/SecondTransfer/Http2/Framer.hs +++ /dev/null @@ -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 () diff --git a/hs-src/SecondTransfer/Http2/Session.cpphs b/hs-src/SecondTransfer/Http2/Session.cpphs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/Session.cpphs @@ -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 diff --git a/hs-src/SecondTransfer/Http2/Session.hs b/hs-src/SecondTransfer/Http2/Session.hs deleted file mode 100644 --- a/hs-src/SecondTransfer/Http2/Session.hs +++ /dev/null @@ -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 diff --git a/hs-src/SecondTransfer/MainLoop.hs b/hs-src/SecondTransfer/MainLoop.hs --- a/hs-src/SecondTransfer/MainLoop.hs +++ b/hs-src/SecondTransfer/MainLoop.hs @@ -18,6 +18,8 @@ ,tlsServeWithALPN ,tlsServeWithALPNAndFinishOnRequest + ,enableConsoleLogging + ,TLSLayerGenericProblem(..) ,FinishRequest(..) ) where @@ -29,3 +31,4 @@ ) import SecondTransfer.MainLoop.OpenSSL_TLS +import SecondTransfer.MainLoop.Logging (enableConsoleLogging) diff --git a/hs-src/SecondTransfer/MainLoop/Logging.hs b/hs-src/SecondTransfer/MainLoop/Logging.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/Logging.hs @@ -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 + ) + diff --git a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs @@ -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 diff --git a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs deleted file mode 100644 --- a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs +++ /dev/null @@ -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 diff --git a/second-transfer.cabal b/second-transfer.cabal --- a/second-transfer.cabal +++ b/second-transfer.cabal @@ -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/ diff --git a/tests/tests-hs-src/compiling_ok.hs b/tests/tests-hs-src/compiling_ok.hs --- a/tests/tests-hs-src/compiling_ok.hs +++ b/tests/tests-hs-src/compiling_ok.hs @@ -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