second-transfer 0.5.5.1 → 0.6.0.0
raw patch · 28 files changed
+1871/−1072 lines, 28 filesdep +SafeSemaphoredep +clockdep +deepseqdep ~http2
Dependencies added: SafeSemaphore, clock, deepseq, pqueue, stm, unordered-containers
Dependency ranges changed: http2
Files
- README.md +10/−14
- cbits/tlsinc.c +111/−46
- changelog.md +6/−2
- hs-src/SecondTransfer.hs +47/−42
- hs-src/SecondTransfer/Exception.hs +23/−23
- hs-src/SecondTransfer/Http1.hs +1/−1
- hs-src/SecondTransfer/Http1/Parse.hs +99/−99
- hs-src/SecondTransfer/Http1/Session.hs +69/−52
- hs-src/SecondTransfer/Http2/Framer.hs +367/−162
- hs-src/SecondTransfer/Http2/MakeAttendant.hs +11/−16
- hs-src/SecondTransfer/Http2/Session.hs +391/−135
- hs-src/SecondTransfer/MainLoop.hs +3/−2
- hs-src/SecondTransfer/MainLoop/CoherentWorker.hs +169/−50
- hs-src/SecondTransfer/MainLoop/Framer.hs +39/−39
- hs-src/SecondTransfer/MainLoop/Logging.hs +59/−16
- hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs +157/−133
- hs-src/SecondTransfer/MainLoop/PushPullType.hs +48/−22
- hs-src/SecondTransfer/Sessions.hs +1/−1
- hs-src/SecondTransfer/Sessions/Config.hs +48/−31
- hs-src/SecondTransfer/Sessions/Internal.hs +24/−24
- hs-src/SecondTransfer/Types.hs +2/−2
- hs-src/SecondTransfer/Utils.hs +33/−33
- hs-src/SecondTransfer/Utils/DevNull.hs +15/−9
- hs-src/SecondTransfer/Utils/HTTPHeaders.hs +46/−46
- macros/Logging.cpphs +3/−1
- second-transfer.cabal +15/−4
- tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs +61/−55
- tests/tests-hs-src/compiling_ok.hs +13/−12
README.md view
@@ -1,4 +1,4 @@- + Developer README ================ @@ -6,18 +6,18 @@ ------------ This is an early-stage and very experimental library to create HTTP/2 servers-using Haskell. +using Haskell. -To see the package docs, please check the Hackage page or +To see the package docs, please check the Hackage page or the file hs-src/SecondTransfer.hs. Building and installing ----------------------- -You need Haskell GHC compiler installed (version 7.8.3 at least). You also +You need Haskell GHC compiler installed (version 7.8.3 at least). You also need OpenSSL 1.0.2, since the ALPN feature and some very recent cypher-suites-are needed by HTTP/2. this source distribution will try to find them at -directory `/opt/openssl-1.0.2`, but you should be able to +are needed by HTTP/2. this source distribution will try to find them at+directory `/opt/openssl-1.0.2`, but you should be able to alter the options using `cabal configure`. This package uses Haskell's FFI to interface with OpenSSL. Provided that you have all the dependencies, you should be able to just do:@@ -33,7 +33,7 @@ Debugging complicated scenarios ------------------------------- -To access full debugging capabilities, for example from the test suite, use the +To access full debugging capabilities, for example from the test suite, use the following command from the project's directory: $ cabal exec -- ghci -ihs-src -itests/tests-hs-src -itests/support -XCPP -Imacros/ dist/build/cbits/tlsinc.o@@ -41,25 +41,21 @@ Example ------- -There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`. +There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`. Roadmap ------- Done: -- Version 0.1: Having something that can run. No unit-testing, nothing - fancy. +- Version 0.1: Having something that can run. No unit-testing, nothing+ fancy. - Version 0.2: Absolutely minimal amount of unit tests. - Version 0.3: More sensible logging. Pending:--- Better examples.--- Epoll I/O management - Benchmarking.
cbits/tlsinc.c view
@@ -19,7 +19,9 @@ #include <bits/sigthread.h> #include <sys/time.h> #include <sys/types.h>+#include <netinet/tcp.h> + #include <openssl/rand.h> #include <openssl/ssl.h> #include <openssl/err.h>@@ -44,10 +46,10 @@ // This is the public API of this server... // No arguments, all are wired here somewhere... for now connection_t* make_connection();-// Call when you are done +// Call when you are done void close_connection(connection_t* conn); // Wait for the next one...-#define ALL_OK 0 +#define ALL_OK 0 #define BAD_HAPPENED 1 #define TIMEOUT_REACHED 3 // This is also a failed IO with SSL, but this one may be quite@@ -56,6 +58,7 @@ int wait_for_connection(connection_t* conn, int microseconds, wired_session_t** wired_session); int send_data(wired_session_t* ws, char* buffer, int buffer_size); int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);+int recv_data_best_effort(wired_session_t* ws, char* inbuffer, int buffer_size, int can_wait, int* data_recvd); int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; } void dispose_wired_session(wired_session_t* ws); static int thread_setup(void);@@ -66,7 +69,7 @@ static int threads_are_up = 0; -// Leaky implementation now +// Leaky implementation now void close_connection(connection_t* conn) { if (conn->socket)@@ -102,6 +105,7 @@ handle = socket (AF_INET, SOCK_STREAM, 0); int one = 1; setsockopt(handle, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));+ if (handle == -1) { perror ("Socket could not be created");@@ -110,29 +114,46 @@ } 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); - error = bind (handle, (struct sockaddr *) &server,- sizeof (struct sockaddr));- if (error == -1)+ int flag = 1;+ int result = setsockopt(handle, /* socket affected */+ IPPROTO_TCP, /* set option at TCP level */+ TCP_NODELAY, /* name of option */+ (char *) &flag, /* the cast is historical+ cruft */+ sizeof(int)); /* length of option value */+ if (result < 0) {- perror ("Bind");+ perror("Couldn't set TCP_NODELAY"); handle = 0; *errorh = BAD_HAPPENED;- }- else + } else {- error = listen(handle, 5);- if ( error != 0 )+ 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("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;@@ -253,7 +274,7 @@ static DH *tmp_dh_callback(SSL *s, int is_export, int keylength); -// Setup dh params +// Setup dh params static void setup_dh_parms(SSL_CTX *sslContext) { /* Set up ephemeral DH stuff */@@ -336,13 +357,13 @@ incursor += (1+sublen); } - // I think this is what should be returned + // I think this is what should be returned return -1; } int lookup_protocol(- char* selected, int selected_len, + char* selected, int selected_len, char* myprotocol_list, int mpl_len ) {@@ -376,7 +397,7 @@ stored_cursor += (1+sublen2); } - // I think this is what should be returned + // I think this is what should be returned return -1; } @@ -414,7 +435,7 @@ int result; connection_t *c; - // Ok, here it is in the open, for everybody + // Ok, here it is in the open, for everybody // to see: signal(SIGPIPE, SIG_IGN); @@ -422,7 +443,7 @@ { threads_are_up = 1; thread_setup();- } + } c = malloc (sizeof (connection_t)); c->sslContext = NULL;@@ -455,9 +476,9 @@ // Now I set a few options.... /*SSL_CTX_set_verify(c->sslContext, NULL );*/- const long flags = + const long flags = SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;- // const long flags = + // 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 );@@ -468,7 +489,7 @@ // The only cipher supported by HTTP/2 ... sort of.- result = SSL_CTX_set_cipher_list(c->sslContext, + result = SSL_CTX_set_cipher_list(c->sslContext, "ECDHE-RSA-AES128-GCM-SHA256" // -- ibidem ":ECDHE-RSA-AES256-GCM-SHA384" // <-- Good for Chrome and firefox // ---- These suites will fail HTTP/2@@ -497,7 +518,7 @@ // Load a certificate in .pem format. SSL_CTX_use_certificate_chain_file(c->sslContext, certificate_filename); SSL_CTX_use_PrivateKey_file(c->sslContext, privkey_filename, SSL_FILETYPE_PEM);- // Check the private key + // Check the private key result = SSL_CTX_check_private_key(c->sslContext); if ( result != 1) {@@ -525,19 +546,19 @@ { const int len = strlen(hostname); if ( len > 0 && hostname[len-1] == '\n' )- { + { printf("HELLO THERE. I found some weird characters\n"); *( (int*) 0) = 42; // <-- This is the answer } return sslStart(- certificate_filename, privkey_filename, hostname, - portno, my_protocol_list, protocol_list_length); + certificate_filename, privkey_filename, hostname,+ portno, my_protocol_list, protocol_list_length); } int wait_for_connection(- connection_t* c, + connection_t* c, int microseconds, wired_session_t** wired_session) {@@ -549,7 +570,7 @@ struct sockaddr_in cli_addr; clilen = sizeof(cli_addr); - // Use select to get "interruptible" accepts + // Use select to get "interruptible" accepts fd_set rfds; int can_go = 0;@@ -568,10 +589,10 @@ if ( retval == -1 ) {- if ( errno == EINTR ) + if ( errno == EINTR ) { // printf(".");- // I get a lot of these signals due to the fact of the + // I get a lot of these signals due to the fact of the // Haskell runtime having its own use for signals. return TIMEOUT_REACHED; } else {@@ -583,7 +604,7 @@ // We got data, just let it go... // printf("letitgo\n"); can_go = 1;- } else + } else { // We didn't get data, finish and terminate // printf("timeo\n");@@ -608,6 +629,19 @@ return BAD_HAPPENED; } + int flag = 1;+ int nresult = setsockopt(newsockfd, /* socket affected */+ IPPROTO_TCP, /* set option at TCP level */+ TCP_NODELAY, /* name of option */+ (char *) &flag, /* the cast is historical+ cruft */+ sizeof(int)); /* length of option value */+ if ( nresult < 0)+ {+ perror("Couldn't set TCP_NODELAY");+ return BAD_HAPPENED;+ }+ // Create an SSL struct for the connection result->socket = newsockfd;@@ -640,8 +674,8 @@ SSL_get0_alpn_selected(result->sslHandle, &out_protocol, &pr_len); // Wonder who is in charge of releasing that buffer...- result -> protocol_index = lookup_protocol( - (char*)out_protocol, pr_len, + result -> protocol_index = lookup_protocol(+ (char*)out_protocol, pr_len, c->protocol_list, c->protocol_list_length); *wired_session = result;@@ -681,35 +715,67 @@ } else { return BAD_HAPPENED; }- } else + } else { return BAD_HAPPENED; } } +// Now this function does a very good effort to get as much data as requested, and it+// doesn't return until then. That works well for HTTP/2 sessions which are almost+// never closed naturally. It is however more tricky for EOF cases.... int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd) { int received=0, count = 0;- char buffer[1024]; // printf("Recvd entered\n"); - if (ws)+ while ( count < buffer_size ) {- received = SSL_read (ws->sslHandle, inbuffer, buffer_size);+ if (ws)+ {+ received = SSL_read (ws->sslHandle, inbuffer+count, buffer_size - count );+ }+ if ( received <= 0 )+ {+ ERR_print_errors_fp (stderr);+ return BAD_HAPPENED;+ } else+ {+ // Continue reading+ count += received ;+ } }++ // printf("Recvd exited\n");+ *data_recvd=count ;+ return ALL_OK;+}+++int recv_data_best_effort(wired_session_t* ws, char* inbuffer, int buffer_size, int can_wait, int* data_recvd)+{+ int received=0, count = 0;++ if (ws && ( can_wait || SSL_pending(ws->sslHandle) ) )+ {+ received = SSL_read (ws->sslHandle, inbuffer+count, buffer_size - count );+ } if ( received <= 0 ) { ERR_print_errors_fp (stderr); return BAD_HAPPENED;+ } else+ {+ // Continue reading+ count += received ; }- *data_recvd = received ; - // printf("Recvd exited\n");-+ *data_recvd=count ; return ALL_OK; } + #define MUTEX_TYPE pthread_mutex_t #define MUTEX_SETUP(x) pthread_mutex_init(&(x), NULL) #define MUTEX_CLEANUP(x) pthread_mutex_destroy(&(x))@@ -721,10 +787,10 @@ 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); */ + /* exit(-1); */ } -/* This array will store all of the mutexes available to OpenSSL. */ +/* This array will store all of the mutexes available to OpenSSL. */ static MUTEX_TYPE *mutex_buf= NULL; @@ -789,4 +855,3 @@ mutex_buf = NULL; return 1; }-
changelog.md view
@@ -1,5 +1,9 @@-- 0.5.5.1 :- * Restricted HTTP2 package version.+- 0.6.0.0 :+ * Changed the interface to provide more information to and from workers+ * Implemented HTTP/2 push+ * Upgraded dependencies to 1.0. for http2 package+ * Extended channel interface in such a way that latency is reduced, as a result,+ the test suite is completely broken. Please wait for 0.6.0.1 release. - 0.5.5.0 : * Made HeaderEditor a Monoid instance
hs-src/SecondTransfer.hs view
@@ -6,32 +6,32 @@ Stability : experimental Portability : POSIX -This library implements enough of the HTTP/2 to build -compliant HTTP/2 servers. It also implements enough of +This library implements enough of the HTTP/2 to build+compliant HTTP/2 servers. It also implements enough of HTTP/1.1 so you can actually use it to build polyglot web-servers. -For HTTP/2, frame encoding and decoding is done with +For HTTP/2, frame encoding and decoding is done with Kazu Yamamoto's <http://hackage.haskell.org/package/http2 http2> package. This library just takes care of making sense of sent and received frames. -You can find more detailed information about this library at the page +You can find more detailed information about this library at the page <https://www.httptwo.com/second-transfer/>. The library - * Is concurrent, meaning that you can use amazing Haskell lightweight threads to - process the requests. + * Is concurrent, meaning that you can use amazing Haskell lightweight threads to+ process the requests. * Obeys HTTP/2 flow control aspects, when talking HTTP/2. - * And gives you freedom to (ab)use the HTTP/2 protocol in all the ways envisioned - by the standard. In particular you should be able to process streaming requests + * And gives you freedom to (ab)use the HTTP/2 protocol in all the ways envisioned+ by the standard. In particular you should be able to process streaming requests (long uploads in POST or PUT requests) and to deliver streaming responses. You- should even be able to do both simultaneously. + should even be able to do both simultaneously. Setting up TLS for HTTP/2 correctly is a shore, so I have bundled here the-TLS setup logic. Before you read any further, ATTENTION: enable always the threaded +TLS setup logic. Before you read any further, ATTENTION: enable always the threaded ghc runtime in your final programs if you want TLS to work. @@ -40,12 +40,13 @@ @ {-# LANGUAGE OverloadedStrings #-} import SecondTransfer(- CoherentWorker+ AwareWorker , Footers , DataAndConclusion , tlsServeWithALPN , http2Attendant , http11Attendant+ , coherentToAwareWorker ) import SecondTransfer.Sessions( makeSessionsContext@@ -56,11 +57,11 @@ saysHello :: DataAndConclusion-saysHello = do - -- The data in each yield will be automatically split across multiple +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, + -- 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!"@@ -68,8 +69,8 @@ return [] -helloWorldWorker :: CoherentWorker-helloWorldWorker (_request_headers, _maybe_post_data) = do +helloWorldWorker :: AwareWorker+helloWorldWorker (_request_headers, _maybe_post_data) = coherentToAwareWorker $ do dropIncomingData _maybe_post_data return ( [@@ -80,11 +81,11 @@ ) --- For this program to work, it should be run from the top of +-- For this program to work, it should be run from the top of -- the developement directory.-main = do +main = do sessions_context <- makeSessionsContext defaultSessionsConfig- let + let http2_attendant = http2Attendant sessions_context helloWorldWorker http11_attendant = http11Attendant sessions_context helloWorldWorker tlsServeWithALPN@@ -92,29 +93,31 @@ "tests\/support\/privkey.pem" -- Certificate private key "127.0.0.1" -- On which interface to bind [- ("no-protocol", http11_attendant), -- The first protocol in the list is used when + ("no-protocol", http11_attendant), -- The first protocol in the list is used when -- when no ALPN negotiation happens, and the -- name is really a filler. ("h2-14", http2_attendant), -- Protocols present in the ALPN negotiation- ("h2", http2_attendant), -- they may be slightly different, but for this + ("h2", http2_attendant), -- they may be slightly different, but for this -- test it doesn't matter. ("http/1.1", http11_attendant) -- Let's talk HTTP/1.1 if everything else fails. ]- 8000 + 8000 @ -`CoherentWorker` is the type of the basic callback function that you need to implement. -The callback is used to handle all requests to the server on a given negotiated ALPN +`AwareWorker` is the type of the basic callback function that you need to implement, but+most times you can do with a simplified version called `CoherentWorker`. The function+`coherentToAwareWorker` does the conversion.+The callback is used to handle all requests to the server on a given negotiated ALPN protocol. If you need routing functionality (and you most certainly will need it), you need-to build that functionality inside the callback. +to build that functionality inside the callback. The above program uses a test certificate by a fake certificate authority. The certificate is valid for the server name ("authority", in HTTP\/2 lingo) www.httpdos.com. So, in order-for the above program to run, you probably need to add an alias to your \/etc\/hosts file. +for the above program to run, you probably need to add an alias to your \/etc\/hosts file. You also need very up-to-date versions of OpenSSL (I'm using OpenSSL 1.0.2) to be compliant with the cipher suites demanded by HTTP\/2. The easiest way to test the above program is using-a fairly recent version of <http://curl.haxx.se/ curl>. If everything is allright, +a fairly recent version of <http://curl.haxx.se/ curl>. If everything is allright, you should be able to do: @@@ -127,13 +130,13 @@ -- * Types related to coherent workers --- -- | A coherent worker is an abstraction that can dance at the + -- | A coherent worker is an abstraction that can dance at the -- tune of HTTP/2 and HTTP/1.1. That is, it should be able to take- -- headers from a request first, then a source of data coming in the - -- request (for example, POST data). Even before exhausting the source, + -- headers from a request first, then a source of data coming in the+ -- request (for example, POST data). Even before exhausting the source, -- the coherent worker can post the response headers and -- its source for the response data. A coherent worker can also present- -- streams to push to the client. + -- streams to push to the client. Headers , HeaderName , HeaderValue@@ -141,22 +144,24 @@ , Request , Footers , CoherentWorker+ , AwareWorker , PrincipalStream , PushedStreams , PushedStream , DataAndConclusion , InputDataStream , FinalizationHeaders- + -- ** How to couple bidirectional data channels to sessions ,Attendant ,http11Attendant ,http2Attendant+ ,coherentToAwareWorker #ifndef DISABLE_OPENSSL_TLS- -- * High level OpenSSL functions. - -- - -- | Use these functions to create your TLS-compliant + -- * High level OpenSSL functions.+ --+ -- | Use these functions to create your TLS-compliant -- HTTP/2 server in a snap. ,tlsServeWithALPN ,tlsServeWithALPNAndFinishOnRequest@@ -166,15 +171,15 @@ #endif -- DISABLE_OPENSSL_TLS ,dropIncomingData- -- * Logging + -- * 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` + -- | 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 + ) where import SecondTransfer.Http1 (http11Attendant) import SecondTransfer.Http2.MakeAttendant (http2Attendant)
hs-src/SecondTransfer/Exception.hs view
@@ -22,21 +22,21 @@ -- * Internal exceptions ,HTTP2ProtocolException(..)- ) where + ) where import Control.Exception import Data.Typeable --- | Abstract exception. All HTTP/2 exceptions derive from here +-- | Abstract exception. All HTTP/2 exceptions derive from here data HTTP2SessionException = forall e . Exception e => HTTP2SessionException e deriving Typeable instance Show HTTP2SessionException where show (HTTP2SessionException e) = show e -instance Exception HTTP2SessionException +instance Exception HTTP2SessionException convertHTTP2SessionExceptionToException :: Exception e => e -> SomeException convertHTTP2SessionExceptionToException = toException . HTTP2SessionException@@ -76,7 +76,7 @@ cast a --- | Thrown when the HTTP/2 connection prefix doesn't +-- | Thrown when the HTTP/2 connection prefix doesn't -- match the expected prefix. data BadPrefaceException = BadPrefaceException deriving (Typeable, Show)@@ -106,14 +106,14 @@ cast a -- | Abstract exception. It is an error if an exception of this type bubbles--- to this library, but we will do our best to handle it gracefully. +-- to this library, but we will do our best to handle it gracefully. -- All internal error precursors at the workers can thus inherit from here -- to have a fallback option in case they forget to handle the error. -- This exception inherits from HTTP11Exception data HTTP500PrecursorException = forall e . Exception e => HTTP500PrecursorException e deriving Typeable -instance Show HTTP500PrecursorException where +instance Show HTTP500PrecursorException where show (HTTP500PrecursorException e) = show e -- | Use the traditional idiom if you need to derive from 'HTTP500PrecursorException',@@ -128,57 +128,57 @@ HTTP500PrecursorException a <- fromException x cast a --- Here we say how we go with these exceptions.... +-- Here we say how we go with these exceptions.... instance Exception HTTP500PrecursorException where toException = convertHTTP11ExceptionToException fromException = getHTTP11ExceptionFromException -- | Thrown with HTTP/1.1 over HTTP/1.1 sessions when the response body -- or the request body doesn't include a Content-Length header field,--- given that should have included it -data ContentLengthMissingException = ContentLengthMissingException +-- given that should have included it+data ContentLengthMissingException = ContentLengthMissingException deriving (Typeable, Show) -instance Exception ContentLengthMissingException where +instance Exception ContentLengthMissingException where toException = convertHTTP11ExceptionToException fromException = getHTTP11ExceptionFromException -data HTTP11SyntaxException = HTTP11SyntaxException String +data HTTP11SyntaxException = HTTP11SyntaxException String deriving (Typeable, Show) -instance Exception HTTP11SyntaxException where +instance Exception HTTP11SyntaxException where toException = convertHTTP11ExceptionToException fromException = getHTTP11ExceptionFromException -- | Throw exceptions derived from this (e.g, `GenericIOProblem` below)--- to have the HTTP/2 session to terminate gracefully. -data IOProblem = forall e . Exception e => IOProblem e +-- to have the HTTP/2 session to terminate gracefully.+data IOProblem = forall e . Exception e => IOProblem e deriving Typeable instance Show IOProblem where- show (IOProblem e) = show e + show (IOProblem e) = show e -instance Exception IOProblem +instance Exception IOProblem -- | A concrete case of the above exception. Throw one of this--- if you don't want to implement your own type. Use +-- if you don't want to implement your own type. Use -- `IOProblem` in catch signatures. data GenericIOProblem = GenericIOProblem deriving (Show, Typeable) -instance Exception GenericIOProblem where +instance Exception GenericIOProblem where toException = toException . IOProblem- fromException x = do - IOProblem a <- fromException x + fromException x = do+ IOProblem a <- fromException x cast a --- | This exception will be raised inside a `CoherentWorker` when the underlying +-- | This exception will be raised inside a `CoherentWorker` when the underlying -- stream is cancelled (STREAM_RESET in HTTP\/2). Do any necessary cleanup -- in a handler, or simply use the fact that the exception is asynchronously--- delivered --- to your CoherentWorker Haskell thread, giving you an opportunity to +-- delivered+-- to your CoherentWorker Haskell thread, giving you an opportunity to -- interrupt any blocked operations. data StreamCancelledException = StreamCancelledException deriving (Show, Typeable)
hs-src/SecondTransfer/Http1.hs view
@@ -2,6 +2,6 @@ module SecondTransfer.Http1( http11Attendant- ) where + ) where import SecondTransfer.Http1.Session (http11Attendant)
hs-src/SecondTransfer/Http1/Parse.hs view
@@ -15,7 +15,7 @@ ,IncrementalHttp1Parser ,Http1ParserCompletion(..) ,BodyStopCondition(..)- ) where + ) where @@ -60,7 +60,7 @@ -- L.makeLenses ''IncrementalHttp1Parser -instance Show IncrementalHttp1Parser where +instance Show IncrementalHttp1Parser where show (IncrementalHttp1Parser ft _sp ) = show $ Bu.toLazyByteString ft @@ -72,7 +72,7 @@ -- | Was the parser complete?-data Http1ParserCompletion = +data Http1ParserCompletion = -- | No, not even headers are done. Use the returned -- value to continue MustContinue_H1PC IncrementalHttp1Parser@@ -81,61 +81,61 @@ -- argument is a left-overs string. |OnlyHeaders_H1PC Headers B.ByteString -- | For requests with a body. The second argument is a condition- -- to stop receiving the body, the third is leftovers from + -- to stop receiving the body, the third is leftovers from -- parsing the headers. |HeadersAndBody_H1PC Headers BodyStopCondition B.ByteString- -- | Some requests are mal-formed. We can check those cases + -- | Some requests are mal-formed. We can check those cases -- here.- |RequestIsMalformed_H1PC + |RequestIsMalformed_H1PC deriving Show --- | Stop condition when parsing the body. Right now only length --- is supported, given with Content-Length. +-- | Stop condition when parsing the body. Right now only length+-- is supported, given with Content-Length. ----- TODO: Support "chunked" transfer encoding for classical +-- TODO: Support "chunked" transfer encoding for classical -- HTTP/1.1, uploads will need it.-data BodyStopCondition = - UseBodyLength_BSC Int +data BodyStopCondition =+ UseBodyLength_BSC Int deriving (Show, Eq) -data RequestOrResponseLine = +data RequestOrResponseLine = -- First argument is the URI, second the method Request_RoRL B.ByteString B.ByteString -- First argument is the status code |Response_RoRL Int deriving (Show, Eq) - + addBytes :: IncrementalHttp1Parser -> B.ByteString -> Http1ParserCompletion addBytes (IncrementalHttp1Parser full_text header_parse_closure) new_bytes = let -- Just feed the bytes (positions, length_so_far, last_char ) = header_parse_closure new_bytes new_full_text = full_text `mappend` (Bu.byteString new_bytes)- could_finish = twoCRLFsAreConsecutive positions - in - case could_finish of + could_finish = twoCRLFsAreConsecutive positions+ in+ case could_finish of Just at_position -> elaborateHeaders new_full_text positions at_position - Nothing -> MustContinue_H1PC - $ IncrementalHttp1Parser - new_full_text + Nothing -> MustContinue_H1PC+ $ IncrementalHttp1Parser+ new_full_text (locateCRLFs length_so_far positions last_char) elaborateHeaders :: Bu.Builder -> [Int] -> Int -> Http1ParserCompletion-elaborateHeaders full_text crlf_positions last_headers_position = - let +elaborateHeaders full_text crlf_positions last_headers_position =+ let -- Start by getting a full byte-string representation of the headers, -- no need to be silly with chunks. full_headers_text = Lb.toStrict $ Bu.toLazyByteString full_text -- Filter out CRLF pairs corresponding to multiline headers.- no_cont_positions_reverse = filter - (\ pos -> if pos >= last_headers_position then True else + no_cont_positions_reverse = filter+ (\ pos -> if pos >= last_headers_position then True else not . isWsCh8 $- (Ch8.index + (Ch8.index full_headers_text (pos + 2) )@@ -145,14 +145,14 @@ no_cont_positions = reverse . tail $ no_cont_positions_reverse -- Now get the headers as slices from the original string.- headers_pre = map - (\ (start, stop) -> + headers_pre = map+ (\ (start, stop) -> subByteString start stop full_headers_text ) (zip ((:) 0- (map + (map ( + 2 ) no_cont_positions )@@ -169,23 +169,23 @@ headers_1 = headers_0 - (headers_2, has_body) = case request_or_response of + (headers_2, has_body) = case request_or_response of - Request_RoRL uri method -> + Request_RoRL uri method -> let -- No lowercase, methods are case sensitive -- lc_method = bsToLower method- has_body' = case method of - "POST" -> True - "PUT" -> True + has_body' = case method of+ "POST" -> True+ "PUT" -> True _ -> False- in + in ( (":path", uri):(":method",method):headers_1, has_body' ) - Response_RoRL status -> - let - status_str = pack . show $ status - excludes_body = + Response_RoRL status ->+ let+ status_str = pack . show $ status+ excludes_body = ( (Ch8.head status_str) == '1') || ( status == 204 || status == 304)@@ -197,54 +197,54 @@ ( (stripBs . bsToLower $ hn), stripBs hv ) | (hn, hv) <- headers_2 ] - content_length :: Int - content_length = + content_length :: Int+ content_length = let cnt_length_header = find (\ x -> (fst x) == "content-length" ) headers_3- in case cnt_length_header of - Just (_, hv) -> read . unpack $ hv + in case cnt_length_header of+ Just (_, hv) -> read . unpack $ hv Nothing -> throw ContentLengthMissingException leftovers = B.drop (last_headers_position + 4) full_headers_text- in - if has_body - then + in+ if has_body+ then HeadersAndBody_H1PC headers_3 (UseBodyLength_BSC content_length) leftovers- else + else OnlyHeaders_H1PC headers_3 leftovers splitByColon :: B.ByteString -> (B.ByteString, B.ByteString)-splitByColon = L.over L._2 (B.tail) . Ch8.break (== ':') +splitByColon = L.over L._2 (B.tail) . Ch8.break (== ':') parseFirstLine :: B.ByteString -> RequestOrResponseLine-parseFirstLine s = - let - either_error_or_rrl = Ap.parseOnly httpFirstLine s +parseFirstLine s =+ let+ either_error_or_rrl = Ap.parseOnly httpFirstLine s exc = HTTP11SyntaxException "BadMessageFirstLine"- in - case either_error_or_rrl of - Left _ -> throw exc + in+ case either_error_or_rrl of+ Left _ -> throw exc Right rrl -> rrl bsToLower :: B.ByteString -> B.ByteString-bsToLower = Ch8.map toLower +bsToLower = Ch8.map toLower -- This ought to be slow! stripBs :: B.ByteString -> B.ByteString-stripBs s = - fst +stripBs s =+ fst . last $- takeWhile + takeWhile ( \ (_, ch) -> isWsCh8 ch ) $- iterate - ( \ (bs, _) -> - case Ch8.unsnoc bs of + iterate+ ( \ (bs, _) ->+ case Ch8.unsnoc bs of Just (newbs, w8) -> (newbs, w8) Nothing -> ("", 'n') )@@ -253,13 +253,13 @@ locateCRLFs :: Int -> [Int] -> Word8 -> B.ByteString -> ([Int], Int, Word8) locateCRLFs initial_offset other_positions prev_last_char next_chunk =- let + let (last_char, positions_list, strlen) =- B.foldl - (\ (prev_char, lst, i) w8 -> - let + B.foldl+ (\ (prev_char, lst, i) w8 ->+ let j = i + 1- in case (prev_char, w8) of + in case (prev_char, w8) of (13,10) -> (w8, (i-1):lst, j) _ -> (w8, lst, j) )@@ -268,7 +268,7 @@ in (positions_list, strlen, last_char) -twoCRLFsAreConsecutive :: [Int] -> Maybe Int +twoCRLFsAreConsecutive :: [Int] -> Maybe Int twoCRLFsAreConsecutive (p2:p1:_) | p2 - p1 == 2 = Just p1 twoCRLFsAreConsecutive _ = Nothing @@ -289,7 +289,7 @@ http1Token = Ap.string "HTTP/1.1" <|> Ap.string "HTTP/1.0" http1Method :: Ap.Parser B.ByteString-http1Method = +http1Method = Ap.string "GET" <|> Ap.string "POST" <|> Ap.string "HEAD"@@ -306,13 +306,13 @@ space = Ap.word8 32 requestLine :: Ap.Parser RequestOrResponseLine-requestLine = +requestLine = flip Request_RoRL- <$> - http1Method + <$>+ http1Method <* space <*>- unspacedUri + unspacedUri <* space <* http1Token @@ -320,7 +320,7 @@ digit = Ap.satisfy (Ap.inClass "0-9") responseLine :: Ap.Parser RequestOrResponseLine-responseLine = +responseLine = (pure Response_RoRL) <* http1Token@@ -329,7 +329,7 @@ <*> ( read . map (toEnum . fromIntegral ) <$> Ap.count 3 digit ) <*- space + space <* Ap.takeByteString @@ -338,9 +338,9 @@ headerListToHTTP11Text :: Headers -> Bu.Builder-headerListToHTTP11Text headers = - case headers of - -- According to the specs, :status can be only +headerListToHTTP11Text headers =+ case headers of+ -- According to the specs, :status can be only -- the first header (hn,hv): rest | hn == ":status" -> (@@ -349,16 +349,16 @@ (go rest) ) - rest -> + rest -> ( (first_line 200) `mappend` (go rest) )- where + where go [] = mempty- go ((hn,hv):rest) = - (Bu.byteString hn) `mappend` ":" `mappend` " " `mappend` (Bu.byteString hv) + go ((hn,hv):rest) =+ (Bu.byteString hn) `mappend` ":" `mappend` " " `mappend` (Bu.byteString hv) `mappend` "\r\n" `mappend` (go rest) first_line :: Int -> Bu.Builder@@ -372,43 +372,43 @@ serializeHTTPResponse :: Headers -> [B.ByteString] -> Lb.ByteString-serializeHTTPResponse response_headers fragments = +serializeHTTPResponse response_headers fragments = let- -- So got some data in an answer. Now there are three ways to go about + -- So got some data in an answer. Now there are three ways to go about -- the returned data: to force a chunked transfer-encoding, to read all- -- the data and add/set the Content-Length header, or to let the user + -- the data and add/set the Content-Length header, or to let the user -- decide which one she prefers. -- -- Right now I'm going for the second one, until somebody complains- -- This is equivalent to a lazy byte-string...but I just need the - -- length + -- This is equivalent to a lazy byte-string...but I just need the+ -- length -- I promised to minimize the number of interventions of the library,- -- so it could be a good idea to remove this one further down the - -- road. + -- so it could be a good idea to remove this one further down the+ -- road. h2 = E.lowercaseHeaders response_headers data_size = foldl' (\ n bs -> n + B.length bs) 0 fragments headers_editor = E.fromList h2 content_length_header_lens = E.headerLens "content-length"- he2 = L.set - content_length_header_lens - (Just (pack . show $ data_size)) - headers_editor + he2 = L.set+ content_length_header_lens+ (Just (pack . show $ data_size))+ headers_editor h3 = E.toList he2 -- Next, I must serialize the headers.... headers_text_as_builder = headerListToHTTP11Text h3 - -- We dump the headers first... unfortunately when talking - -- HTTP/1.1 the most efficient way to write those bytes is + -- We dump the headers first... unfortunately when talking+ -- HTTP/1.1 the most efficient way to write those bytes is -- to create a big buffer and pass it on to OpenSSL.- -- However the Builder generating the headers above says + -- However the Builder generating the headers above says -- it generates fragments between 4k and 32 kb, I checked it -- and it is true, so we can use it - -- Now we need to insert an extra \r\n, even it the response is + -- Now we need to insert an extra \r\n, even it the response is -- empty -- And then we use the builder to re-format the fragments returned- -- by the coherent worker + -- by the coherent worker -- TODO: This could be a good place to introduce chunked responses. body_builder = mconcat $ map Bu.byteString fragments @@ -467,10 +467,10 @@ ----------------------------------------------------------------------------------------- -- assertEqual :: Eq a => String -> a -> a -> IO ()--- assertEqual label v1 v2 = do +-- assertEqual label v1 v2 = do -- putStrLn label--- if v1 == v2 --- then +-- if v1 == v2+-- then -- putStrLn "Ok"--- else +-- else -- putStrLn "NoOk"
hs-src/SecondTransfer/Http1/Session.hs view
@@ -2,9 +2,9 @@ {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.Http1.Session( http11Attendant- ) where + ) where - + import Control.Lens import Control.Exception (catch) import Control.Concurrent (forkIO)@@ -17,84 +17,101 @@ import Data.Conduit.List (consume) -- import Data.Monoid (mconcat, mappend) -import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker,Headers)-import SecondTransfer.MainLoop.PushPullType (Attendant)-import SecondTransfer.Sessions.Internal (SessionsContext, acquireNewSessionTag)+import SecondTransfer.MainLoop.CoherentWorker+import SecondTransfer.MainLoop.PushPullType+import SecondTransfer.Sessions.Internal (SessionsContext, acquireNewSessionTag, sessionsConfig) -- Logging utilities-import System.Log.Logger +import System.Log.Logger+-- And we need the time+import System.Clock import SecondTransfer.Http1.Parse import SecondTransfer.Exception (IOProblem) import SecondTransfer.Sessions.Config-import SecondTransfer.Sessions.Internal (sessionsConfig) import qualified SecondTransfer.Utils.HTTPHeaders as He -- import Debug.Trace (traceShow) -- | Session attendant that speaks HTTP/1.1--- -http11Attendant :: SessionsContext -> CoherentWorker -> Attendant-http11Attendant sessions_context coherent_worker - push_action pull_action close_action - = do +--+http11Attendant :: SessionsContext -> AwareWorker -> Attendant+http11Attendant sessions_context coherent_worker attendant_callbacks+ =+ do new_session_tag <- acquireNewSessionTag sessions_context infoM "Session.Session_HTTP11" $ "Starting new session with tag: " ++(show new_session_tag)- forkIO $ go new_session_tag (Just "")+ forkIO $ go new_session_tag (Just "") 1 return ()- where - go :: Int -> Maybe B.ByteString -> IO ()- go session_tag (Just leftovers) = do - infoM "Session.Session_HTTP11" $ "(Re)Using session with tag: " ++(show session_tag)- maybe_leftovers <- add_data newIncrementalHttp1Parser leftovers session_tag- go session_tag maybe_leftovers+ where+ push_action = attendant_callbacks ^. pushAction_AtC+ -- pull_action = attendant_callbacks ^. pullAction_AtC+ close_action = attendant_callbacks ^. closeAction_AtC+ best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_AtC - go _ Nothing = + go :: Int -> Maybe B.ByteString -> Int -> IO ()+ go session_tag (Just leftovers) reuse_no = do+ infoM "Session.Session_HTTP11" $ "(Re)Using session with tag: " ++ (show session_tag)+ maybe_leftovers <- add_data newIncrementalHttp1Parser leftovers session_tag reuse_no+ go session_tag maybe_leftovers (reuse_no + 1)++ go _ Nothing _ = return () - add_data :: IncrementalHttp1Parser -> B.ByteString -> Int -> IO (Maybe B.ByteString)- add_data parser bytes session_tag = do - let - completion = addBytes parser bytes + add_data :: IncrementalHttp1Parser -> B.ByteString -> Int -> Int -> IO (Maybe B.ByteString)+ add_data parser bytes session_tag reuse_no = do+ let+ completion = addBytes parser bytes -- completion = addBytes parser $ traceShow ("At session " ++ (show session_tag) ++ " Received: " ++ (unpack bytes) ) bytes- case completion of + case completion of - MustContinue_H1PC new_parser -> do + MustContinue_H1PC new_parser -> -- print "MustContinue_H1PC"- catch + catch (do- new_bytes <- pull_action- r <- add_data new_parser new_bytes session_tag- return r+ -- Try to get at least 16 bytes. For HTTP/1 requests, that may not be always+ -- possible+ new_bytes <- best_effort_pull_action True+ add_data new_parser new_bytes session_tag reuse_no ) ( (\ _e -> do- -- This is a pretty harmless condition that happens + -- This is a pretty harmless condition that happens -- often when the remote peer closes the connection debugM "Session.HTTP1" "Could not receive data" close_action return Nothing ) :: IOProblem -> IO (Maybe B.ByteString) )- - OnlyHeaders_H1PC headers leftovers -> do ++ OnlyHeaders_H1PC headers leftovers -> do -- print "OnlyHeaders_H1PC" -- Ready for action... -- ATTENTION: Not use for pushed streams here.... -- We must decide what to do if the user return those -- anyway.- let + let modified_headers = addExtraHeaders sessions_context headers- (response_headers, _, data_and_conclusion) <- coherent_worker (- modified_headers, - Nothing- )- (_, fragments) <- runConduit $ fuseBoth data_and_conclusion consume - let + started_time <- getTime Monotonic+ --(response_headers, _, data_and_conclusion)+ principal_stream <- coherent_worker Request {+ _headers_RQ = modified_headers,+ _inputData_RQ = Nothing,+ _perception_RQ = Perception {+ _startedTime_Pr = started_time,+ _streamId_Pr = reuse_no,+ _sessionId_Pr = session_tag+ }+ }+ let+ data_and_conclusion = principal_stream ^. dataAndConclusion_PS+ response_headers = principal_stream ^. headers_PS+ (_, fragments) <- runConduit $ fuseBoth data_and_conclusion consume+ let response_text = serializeHTTPResponse response_headers fragments - catch + catch (do push_action response_text return $ Just leftovers@@ -107,30 +124,30 @@ HeadersAndBody_H1PC _headers _stopcondition _recv_leftovers -> do -- print "HeadersAndBody_H1PC"- -- Let's see if I can go through the basic movements first, then through + -- Let's see if I can go through the basic movements first, then through -- more complicated things. -- TODO: Implement posts and other requests with bodies.... close_action error "NotImplemented requests with bodies" -addExtraHeaders :: SessionsContext -> Headers -> Headers -addExtraHeaders sessions_context headers = - let +addExtraHeaders :: SessionsContext -> Headers -> Headers+addExtraHeaders sessions_context headers =+ let enriched_lens :: Lens' SessionsContext SessionsEnrichedHeaders enriched_lens = (sessionsConfig . sessionsEnrichedHeaders) -- Haskell laziness here!- headers_editor = He.fromList headers - -- TODO: Figure out which is the best way to put this contact in the + headers_editor = He.fromList headers+ -- TODO: Figure out which is the best way to put this contact in the -- source code protocol_lens = He.headerLens "second-transfer-eh--used-protocol" add_used_protocol = sessions_context ^. (enriched_lens . addUsedProtocol )- he1 = if add_used_protocol + he1 = if add_used_protocol then set protocol_lens (Just "HTTP/1.1") headers_editor else headers_editor- result = He.toList he1 + result = He.toList he1 - in if add_used_protocol + in if add_used_protocol -- Nothing will be computed here if the headers are not modified.- then result - else headers + then result+ else headers
hs-src/SecondTransfer/Http2/Framer.hs view
@@ -15,38 +15,54 @@ -import Control.Concurrent+import Control.Concurrent hiding (yield)+import qualified Control.Concurrent as CC(yield)+import qualified Control.Concurrent.MSem as MS+import Control.Concurrent.STM.TMVar+import qualified Control.Concurrent.STM as STM import Control.Exception import qualified Control.Exception as E-import Control.Lens (view)+import Control.Lens (view, (^.) ) import qualified Control.Lens as L-import Control.Monad (unless, when)+import Control.Monad (unless, when, replicateM_) import Control.Monad.IO.Class (liftIO) import qualified Control.Monad.Catch as C import Control.Monad.Trans.Class (lift)+import Control.DeepSeq (($!!)) import Control.Monad.Trans.Reader import Data.Binary (decode) import qualified Data.ByteString as B+import Data.ByteString.Char8 (pack) import qualified Data.ByteString.Lazy as LB+import qualified Data.ByteString.Builder as Bu import Data.Conduit import Data.Foldable (find)+import qualified Data.PQueue.Min as PQ+import Data.Maybe (fromMaybe) import qualified Network.HTTP2 as NH2 -- Logging utilities import System.Log.Logger import qualified Data.HashTable.IO as H+import System.Clock (Clock(..),getTime) -import SecondTransfer.Sessions.Internal (sessionExceptionHandler, nextSessionId, SessionsContext)+import SecondTransfer.Sessions.Internal (+ sessionExceptionHandler,+ nextSessionId,+ sessionsConfig,+ SessionsContext)+import SecondTransfer.Sessions.Config import SecondTransfer.Http2.Session-import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker)+import SecondTransfer.MainLoop.CoherentWorker (AwareWorker, fragmentDeliveryCallback_Ef, priorityEffect_Ef) import qualified SecondTransfer.MainLoop.Framer as F-import SecondTransfer.MainLoop.PushPullType (Attendant, CloseAction,- PullAction, PushAction)+import SecondTransfer.MainLoop.PushPullType import SecondTransfer.Utils (Word24, word24ToInt) import SecondTransfer.Exception-import SecondTransfer.MainLoop.Logging (logWithExclusivity)+import SecondTransfer.MainLoop.Logging (logWithExclusivity, logit) +import Debug.Trace (trace)+ #include "Logging.cpphs" @@ -63,9 +79,10 @@ data FlowControlCommand = AddBytes_FCM Int+ |Finish_FCM -- A hashtable from stream id to channel of availabiliy increases-type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand)+type Stream2AvailSpace = HashTable GlobalStreamId (MVar FlowControlCommand) data CanOutput = CanOutput@@ -74,9 +91,38 @@ data NoHeadersInChannel = NoHeadersInChannel +-- Maximum number of packets in the priority queue waiting for+-- delivery. More than this, and I will simply block...+maxPacketsInQueue :: Int+maxPacketsInQueue = 32+++-----In this implementation, this v- and this v- member should not change during the lieftime+----- of the stream.+-----------------------------------v--priority, v-- stream_id ----|------v-ordinal+newtype PrioPacket = PrioPacket ( (Int , Int, Int ), LB.ByteString)+ deriving Show++instance Eq PrioPacket where+ (==) (PrioPacket (a,_)) (PrioPacket (b,_)) = a == b++instance Ord PrioPacket where+ compare (PrioPacket (a,_)) (PrioPacket (b,_)) = compare a b+++-- Simple thread to prioritize frames in the session+data PrioritySendState = PrioritySendState {+ -- We need a semaphore so that this doesn't get stagnated waiting on writes+ _semToSend :: MS.MSem Int+ ,_prioQ :: TMVar (PQ.MinQueue PrioPacket)+ }++ data FramerSessionData = FramerSessionData {- _stream2flow :: Stream2AvailSpace- , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString)+ _stream2flow :: MVar Stream2AvailSpace+ -- Two members below: the place where you put data which is going to be flow-controlled,+ -- and the place with the ordinal+ , _stream2outputBytes :: MVar ( HashTable GlobalStreamId (MVar LB.ByteString, MVar Int) ) , _defaultStreamWindow :: MVar Int -- Wait variable to output bytes to the channel@@ -89,15 +135,17 @@ , _closeAction :: CloseAction -- Global id of the session, used for e.g. error reporting.- , _sessionId :: Int+ , _sessionIdAtFramer :: Int -- Sessions context, used for thing like e.g. error reporting , _sessionsContext :: SessionsContext -- For GoAway frames , _lastStream :: MVar Int- } + -- For sending data orderly+ , _prioritySendState :: PrioritySendState+ } L.makeLenses ''FramerSessionData @@ -105,44 +153,59 @@ type FramerSession = ReaderT FramerSessionData IO -wrapSession :: CoherentWorker -> SessionsContext -> Attendant-wrapSession coherent_worker sessions_context push_action pull_action close_action = do+wrapSession :: AwareWorker -> SessionsContext -> Attendant+wrapSession aware_worker sessions_context attendant_callbacks = do let session_id_mvar = view nextSessionId sessions_context+ push_action = attendant_callbacks ^. pushAction_AtC+ pull_action = attendant_callbacks ^. pullAction_AtC+ close_action = attendant_callbacks ^. closeAction_AtC+ best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_AtC + new_session_id <- modifyMVarMasked session_id_mvar $ \ session_id -> return (session_id+1, session_id) (session_input, session_output) <- http2Session- coherent_worker+ aware_worker new_session_id sessions_context -- TODO : Add type annotations.... s2f <- H.new+ stream2flow_mvar <- newMVar s2f s2o <- H.new+ stream2output_bytes_mvar <- newMVar s2o 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 + prio_mvar <- STM.atomically $ newTMVar PQ.empty+ sem_to_send <- MS.new maxPacketsInQueue ++ -- We need some shared state let framer_session_data = FramerSessionData {- _stream2flow = s2f- ,_stream2outputBytes = s2o+ _stream2flow = stream2flow_mvar+ ,_stream2outputBytes = stream2output_bytes_mvar ,_defaultStreamWindow = default_stream_size_mvar ,_canOutput = can_output ,_noHeadersInChannel = no_headers_in_channel ,_pushAction = push_action ,_closeAction = close_action- ,_sessionId = new_session_id+ ,_sessionIdAtFramer = new_session_id ,_sessionsContext = sessions_context ,_lastStream = last_stream_id ,_outputIsForbidden = output_is_forbidden+ ,_prioritySendState = PrioritySendState {+ _semToSend = sem_to_send,+ _prioQ = prio_mvar+ } } @@ -180,6 +243,10 @@ forkIO $ close_on_error new_session_id sessions_context $ runReaderT (outputGatherer session_output ) framer_session_data+ -- Actual data is reordered before being sent+ forkIO+ $ close_on_error new_session_id sessions_context+ $ runReaderT sendReordering framer_session_data return () @@ -203,16 +270,70 @@ return True addCapacity stream_id delta_cap = do- table <- view stream2flow- val <- liftIO $ H.lookup table stream_id+ table_mvar <- view stream2flow+ val <- liftIO $ withMVar table_mvar $ \ table ->+ H.lookup table stream_id+ last_stream_mvar <- view lastStream+ last_stream <- liftIO . readMVar $ last_stream_mvar case val of- Nothing -> return False+ Nothing | stream_id > last_stream ->+ return False+ | otherwise -> -- If the stream was seen already, interpret this as a+ -- rogue WINDOW_UPDATE and do nothing+ return True Just command_chan -> do- liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+ liftIO $ putMVar command_chan $ AddBytes_FCM delta_cap return True +finishFlowControlForStream :: GlobalStreamId -> FramerSession ()+finishFlowControlForStream stream_id =+ do+ table_mvar <- view stream2flow+ liftIO . withMVar table_mvar $ \ table -> do+ val <- H.lookup table stream_id+ case val of+ -- Weird+ Nothing -> return ()++ Just command_chan -> do+ liftIO $+ H.delete table stream_id+ return ()+ table2_mvar <- view stream2outputBytes+ liftIO . withMVar table2_mvar $ \ table -> do+ val <- H.lookup table stream_id+ case val of+ Nothing -> return ()++ Just _ ->+ liftIO $ H.delete table stream_id+++readNextFrame :: Monad m =>+ (Int -> m B.ByteString) -- ^ Generator action+ -> Source m (Maybe NH2.Frame) -- ^ Packet and leftovers, if we could get them+readNextFrame pull_action = do+ -- First get 9 bytes with the frame header+ frame_header_bs <- lift $ pull_action 9+ -- decode it+ let+ (frame_type_id, frame_header) = NH2.decodeFrameHeader frame_header_bs+ NH2.FrameHeader payload_length _ _ = frame_header+ -- Get as many bytes as the payload length identifies+ payload_bs <- lift $ pull_action payload_length+ -- Return the entire frame, or raise an exception...+ let+ either_frame = NH2.decodeFramePayload frame_type_id frame_header payload_bs+ case either_frame of+ Right frame_payload -> do+ yield . Just $ NH2.Frame frame_header frame_payload+ readNextFrame pull_action+ Left _ ->+ yield Nothing++ -- 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.@@ -222,19 +343,19 @@ 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+ prefix <- liftIO $ pull_action http2PrefixLength if prefix /= NH2.connectionPreface then do sendGoAwayFrame NH2.ProtocolError liftIO $- -- We just the the GoAway frame, although this is awfully early+ -- We just use the GoAway frame, although this is awfully early -- and probably wrong throwIO BadPrefaceException else INSTRUMENTATION( debugM "HTTP2.Framer" "Prologue validated" ) let- source::Source FramerSession B.ByteString- source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action+ source::Source FramerSession (Maybe NH2.Frame)+ source = transPipe liftIO $ readNextFrame pull_action source $$ consume True where @@ -247,7 +368,7 @@ else sendMiddleFrameToSession session_input frame - abortSession :: Sink B.ByteString FramerSession ()+ abortSession :: Sink a FramerSession () abortSession = lift $ do sendGoAwayFrame NH2.ProtocolError@@ -258,78 +379,65 @@ consume_continue = consume False - consume :: Bool -> Sink B.ByteString FramerSession ()+ consume :: Bool -> Sink (Maybe NH2.Frame) FramerSession () consume starting = do- maybe_bytes <- await-- 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+ maybe_maybe_frame <- await - case error_or_frame of+ case maybe_maybe_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.- INSTRUMENTATION( errorM "HTTP2.Framer" "CouldNotDecodeFrame" )- -- Send frames like GoAway and such...- abortSession+ Just Nothing ->+ abortSession - Right right_frame -> do- case right_frame of+ Just (Just 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)- succeeded <- lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit)- unless succeeded abortSession+ (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)+ succeeded <- lift $ addCapacity stream_id (fromIntegral credit)+ unless succeeded abortSession - frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do- -- Increase all the stuff....- case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of+ 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) ->- when (k /=0 ) $ writeChan v (AddBytes_FCM general_delta)- )- stream_to_flow+ 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 . withMVar stream_to_flow $ \ stream_to_flow' ->+ H.mapM_ (\ (k,v) ->+ when (k /=0 ) $ putMVar v (AddBytes_FCM $! general_delta)+ )+ stream_to_flow' - -- And set a new value- liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size+ -- And set a new value+ liftIO $ putMVar old_default_stream_size_mvar $! new_default_stream_size - Nothing ->- -- This is a silenced internal error- return ()+ 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 $ sendToSession starting frame+ -- And send the frame down to the session, so that session specific settings+ -- can be applied.+ liftIO $ sendToSession starting $! frame - a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do- -- Update the keep of last stream- lift . startStreamOutputQueueIfNotExists $ NH2.fromStreamIdentifier stream_id- lift . updateLastStream $ NH2.fromStreamIdentifier stream_id+ a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do+ -- Update the keep of last stream+ -- lift . startStreamOutputQueueIfNotExists (NH2.fromStreamIdentifier stream_id) priority+ lift . updateLastStream $ stream_id - -- Send frame to the session- liftIO $ sendToSession starting a_frame+ -- Send frame to the session+ liftIO $ sendToSession starting a_frame - -- tail recursion: go again...- consume_continue+ -- tail recursion: go again...+ consume_continue Nothing -> -- We may as well exit this thread@@ -339,104 +447,146 @@ -- All the output frames come this way first outputGatherer :: SessionOutput -> FramerSession () outputGatherer session_output = do+ frame_sent_report_callback <- view $+ sessionsContext .+ sessionsConfig .+ sessionsCallbacks .+ dataDeliveryCallback_SC - -- We start by sending a settings frame- pushFrame- (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)- (NH2.SettingsFrame [])+ session_id <- view sessionIdAtFramer - loopPart+ let - where+ dataForFrame p1 p2 =+ LB.fromStrict $ NH2.encodeFrame p1 p2 + cont = loopPart session_id frame_sent_report_callback - dataForFrame p1 p2 =- LB.fromStrict $ NH2.encodeFrame p1 p2+ loopPart :: Int -> Maybe DataFrameDeliveryCallback -> FramerSession ()+ loopPart session_id frame_sent_report_callback = do+ command_or_frame <- liftIO $ getFrameFromSession session_output+ case command_or_frame of - 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+ INSTRUMENTATION( debugM "HTTP2.Framer" "CancelSession_SOC processed")+ releaseFramer - Left CancelSession_SOC -> do- -- The session wants to cancel things- INSTRUMENTATION( debugM "HTTP2.Framer" "CancelSession_SOC processed")- releaseFramer+ Left (FinishStream_SOC stream_id ) ->+ -- Session knows that we are done with the given stream, and that we can release+ -- the flow control structures+ -- NOTICE: Unfortunately, this doesn't work, so ignore the message for now+ -- finishFlowControlForStream stream_id.+ cont - 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+ Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _), ef ) -> do+ -- This frame is flow-controlled... I may be unable to send this frame in+ -- some circumstances...+ let+ stream_id = stream_idii+ priority = fromMaybe stream_id $ ef ^. priorityEffect_Ef - Nothing -> do- (bc, _) <- startStreamOutputQueue stream_id- return bc+ startStreamOutputQueueIfNotExists stream_id $ priority - Just bytes_chan -> return bytes_chan+ stream2output_mvar <- view stream2outputBytes+ lookup_result <- liftIO $ withMVar stream2output_mvar $ \ s2o -> H.lookup s2o stream_id+ (stream_bytes_chan, frame_ordinal_mvar) <- case lookup_result of+ Nothing ->+ error "It is the end of the world at Framer.hs"+ Just x -> return x - liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2- loopPart+ -- All the dance below is to avoid a system call if there is no need+ case (frame_sent_report_callback, ef ^. fragmentDeliveryCallback_Ef ) of+ (Just c1, Just c2) -> liftIO $ do+ -- Here we invoke the client's callback.+ when_delivered <- getTime Monotonic+ ordinal <- modifyMVar frame_ordinal_mvar $ \ o -> return (o+1, o)+ c1 session_id stream_id ordinal when_delivered+ c2 ordinal when_delivered+ (Nothing, Just c2) -> liftIO $ do+ when_delivered <- getTime Monotonic+ ordinal <- modifyMVar frame_ordinal_mvar $ \ o -> return (o+1, o)+ c2 ordinal when_delivered+ (Just c1, Nothing) -> liftIO $ do+ when_delivered <- getTime Monotonic+ ordinal <- modifyMVar frame_ordinal_mvar $ \ o -> return (o+1, o)+ c1 session_id stream_id ordinal when_delivered+ (Nothing, Nothing) -> return () - Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do- handleHeadersOfStream p1 p2- loopPart+ liftIO $ putMVar stream_bytes_chan $! dataForFrame p1 p2+ cont - 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+ Right (p1, p2@(NH2.PushPromiseFrame _ _), _effect ) -> do+ handleHeadersOfStream p1 p2+ cont + Right (p1, p2@(NH2.HeadersFrame _ _), _effect ) -> do+ handleHeadersOfStream p1 p2+ cont + Right (p1, p2@(NH2.ContinuationFrame _), _effect ) -> do+ handleHeadersOfStream p1 p2+ cont++ Right (p1, p2, _effect) -> 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+ cont++ -- We start by sending a settings frame+ pushFrame+ (NH2.EncodeInfo NH2.defaultFlags 0 Nothing)+ (NH2.SettingsFrame [])++ -- And then we continue...+ loopPart session_id frame_sent_report_callback++ 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) --startStreamOutputQueueIfNotExists :: GlobalStreamId -> FramerSession ()-startStreamOutputQueueIfNotExists stream_id = do- table <- view stream2flow- val <- liftIO $ H.lookup table stream_id+startStreamOutputQueueIfNotExists :: GlobalStreamId -> Int -> FramerSession ()+startStreamOutputQueueIfNotExists stream_id priority = do+ table_mvar <- view stream2flow+ val <- liftIO . withMVar table_mvar $ \ table -> H.lookup table stream_id case val of Nothing | stream_id /= 0 -> do- startStreamOutputQueue stream_id+ startStreamOutputQueue stream_id priority return () _ -> return () -startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand)-startStreamOutputQueue stream_id = do+-- Handles only Data frames.+startStreamOutputQueue :: Int -> Int -> FramerSession (MVar LB.ByteString, MVar FlowControlCommand)+startStreamOutputQueue stream_id priority = do -- New thread for handling outputs of this stream is needed- bytes_chan <- liftIO newChan- command_chan <- liftIO newChan+ bytes_chan <- liftIO newEmptyMVar+ ordinal_num <- liftIO $ newMVar 0+ command_chan <- liftIO newEmptyMVar - s2o <- view stream2outputBytes+ s2o_mvar <- view stream2outputBytes - liftIO $ H.insert s2o stream_id bytes_chan+ liftIO . withMVar s2o_mvar $ \ s2o -> H.insert s2o stream_id (bytes_chan, ordinal_num) - s2c <- view stream2flow+ stream2flow_mvar <- view stream2flow - liftIO $ H.insert s2c stream_id command_chan+ liftIO . withMVar stream2flow_mvar $ \ s2c -> 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+ session_id' <- view sessionIdAtFramer output_is_forbidden_mvar <- view outputIsForbidden -- And don't forget the thread itself@@ -458,7 +608,7 @@ read_state <- ask liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT- (flowControlOutput stream_id initial_cap "" command_chan bytes_chan)+ (flowControlOutput stream_id priority initial_cap 0 "" command_chan bytes_chan) read_state return (bytes_chan , command_chan)@@ -467,7 +617,8 @@ -- 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...+-- There are more synchronization mechanisms in Http2.Session that ensure we+-- only get here frames from one and the same stream. handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession () handleHeadersOfStream p1@(NH2.EncodeInfo {}) frame_payload | frameIsHeadersAndOpensStream frame_payload && not (frameEndsHeaders p1 frame_payload) = do@@ -475,7 +626,7 @@ no_headers <- view noHeadersInChannel liftIO $ takeMVar no_headers pushFrame p1 frame_payload- -- DONT PUT THE MvAR HERE+ -- Don't put the MvAR HERE | frameIsHeadersAndOpensStream frame_payload && frameEndsHeaders p1 frame_payload = do no_headers <- view noHeadersInChannel@@ -496,16 +647,17 @@ pushFrame p1 frame_payload +-- Used to know when we need to switch the channel to "exclusive" mode. frameIsHeadersAndOpensStream :: NH2.FramePayload -> Bool-frameIsHeadersAndOpensStream (NH2.HeadersFrame _ _ )- = True-frameIsHeadersAndOpensStream _- = False+frameIsHeadersAndOpensStream (NH2.HeadersFrame _ _ ) = True+frameIsHeadersAndOpensStream (NH2.PushPromiseFrame _ _) = True+frameIsHeadersAndOpensStream _ = 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 (NH2.EncodeInfo flags _ _) (NH2.PushPromiseFrame _ _) = NH2.testEndHeader flags frameEndsHeaders _ _ = False @@ -522,13 +674,16 @@ 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 "")+ pushFrame+ (NH2.EncodeInfo NH2.defaultFlags 0 Nothing)+ (NH2.GoAwayFrame last_stream_id error_code "") +-- From this point on data is really serialized, sendBytes :: LB.ByteString -> FramerSession () sendBytes bs = do push_action <- view pushAction+ -- I don't think I need to lock here... can_output <- view canOutput liftIO $ bs `seq`@@ -541,13 +696,16 @@ -- 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 =+--+-- There is one of these for each stream+--+flowControlOutput :: Int -> Int -> Int -> Int -> LB.ByteString -> MVar FlowControlCommand -> MVar LB.ByteString -> FramerSession ()+flowControlOutput stream_id priority capacity ordinal leftovers commands_chan bytes_chan = 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+ bytes_to_send <- liftIO $ takeMVar bytes_chan+ flowControlOutput stream_id priority capacity ordinal bytes_to_send commands_chan bytes_chan else do -- Length? let amount = fromIntegral (LB.length leftovers - 9)@@ -555,20 +713,16 @@ 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+ -- liftIO . logit $ "set-priority (stream_id, prio, ordinal) " `mappend` (pack . show) (__stream_id, priority, ordinal)+ withPrioritySend priority stream_id ordinal leftovers+ flowControlOutput stream_id priority (capacity - amount) (ordinal+1) "" 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+ command <- liftIO $ takeMVar commands_chan case command of AddBytes_FCM delta_cap -> -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id)- flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan+ flowControlOutput stream_id priority (capacity + delta_cap) ordinal leftovers commands_chan bytes_chan releaseFramer :: FramerSession ()@@ -576,3 +730,54 @@ -- Release any resources pending... return ()+++-- This prioritizes DATA packets, in some rudimentary way.+-- This code will be replaced in due time for something compliant.+withPrioritySend :: Int -> Int -> Int -> LB.ByteString -> FramerSession ()+withPrioritySend priority stream_id packet_ordinal datum = do+ PrioritySendState {_semToSend = s, _prioQ = pqm } <- view prioritySendState+ let+ new_record = PrioPacket ( (priority, stream_id, packet_ordinal), datum)+ -- Now, for this to work, I need to have enough capacity to add something to the queue+ liftIO $ do+ -- We are using a semaphore to avoid overflowing this place. Notice that the flow+ -- control output is till using an unbounded queue!!!+ MS.wait s+ -- And add it to the queue....+ STM.atomically $ do+ pq <- takeTMVar pqm+ putTMVar pqm $ PQ.insert new_record pq+++-- In charge of actually sending the data frames, in a special thread (create in the caller)+sendReordering :: FramerSession ()+sendReordering = do+ PrioritySendState {_semToSend = s, _prioQ = pqm } <- view prioritySendState+ no_headers <- view noHeadersInChannel+ -- Get a packet, if possible, or wait for it+ PrioPacket ( (priority_, stream_id, packed_ordinal_), datum) <- liftIO . STM.atomically $ do+ pq <- takeTMVar pqm+ if PQ.null pq+ then STM.retry+ else do+ let+ (record, thinner) = PQ.deleteFindMin pq+ putTMVar pqm thinner+ return record+ -- liftIO . logit $ "prio-send (prio,stream_id, ordinal) " `mappend` (pack . show) (priority, stream_id, packed_ordinal_)+ -- Since I got something to send, I can let one of the guys to put more stuff+ liftIO $ MS.signal s+ -- Now we do the tries and locks for headers+ C.bracket+ (liftIO $ takeMVar no_headers)+ (\ _ -> liftIO $ putMVar no_headers NoHeadersInChannel)+ (\ _ -> sendBytes datum )++ -- Sleep for a little bit, we dont' want too many frames+ -- here too fast. BIG TODO: We need a better way to handle this+ liftIO $ replicateM_ 10 CC.yield+ --liftIO $ threadDelay 100++ -- And tail-recurse+ sendReordering
hs-src/SecondTransfer/Http2/MakeAttendant.hs view
@@ -7,26 +7,21 @@ import SecondTransfer.Http2.Framer (wrapSession) import SecondTransfer.Sessions.Internal (SessionsContext) import SecondTransfer.MainLoop.CoherentWorker-import SecondTransfer.MainLoop.PushPullType (- --CloseAction,- --PullAction, - --PushAction,- Attendant- )+import SecondTransfer.MainLoop.PushPullType (Attendant) -- | The type of this function is equivalent to:--- --- @ --- http2Attendant :: CoherentWorker -> PushAction -> PullAction -> CloseAction -> IO ()+-- -- @--- +-- http2Attendant :: CoherentWorker -> AttendantCallbacks -> IO ()+-- @+-- -- Given a `CoherentWorker`, this function wraps it with flow control, multiplexing,--- and state maintenance needed to run an HTTP/2 session. +-- and state maintenance needed to run an HTTP/2 session. -- -- Notice that this function is using HTTP/2 over TLS. We haven't implemented yet--- a session handling mechanism for HTTP/1.1 . -http2Attendant :: SessionsContext -> CoherentWorker -> Attendant-http2Attendant sessions_context coherent_worker push_action pull_action close_action = do - let +-- a session handling mechanism for HTTP/1.1 .+http2Attendant :: SessionsContext -> AwareWorker -> Attendant+http2Attendant sessions_context coherent_worker attendant_callbacks = do+ let attendant = wrapSession coherent_worker sessions_context- attendant push_action pull_action close_action + attendant attendant_callbacks
hs-src/SecondTransfer/Http2/Session.hs view
@@ -29,17 +29,19 @@ #include "Logging.cpphs" -- System grade utilities-import Control.Concurrent (ThreadId, forkIO)+import Control.Concurrent (ThreadId, forkIO, threadDelay) import Control.Concurrent.Chan import Control.Exception (throwTo) import qualified Control.Exception as E-import Control.Monad (forever)+import Control.Monad (forever,unless, when, mapM_, forM, forM_) import Control.Monad.IO.Class (liftIO)+import Control.DeepSeq ( ($!!), deepseq ) import Control.Monad.Trans.Reader -- import Control.Monad.Catch (throwM) import Control.Concurrent.MVar import qualified Data.ByteString as B+import Data.ByteString.Char8 (pack,unpack) import qualified Data.ByteString.Builder as Bu import qualified Data.ByteString.Lazy as Bl import Data.Conduit@@ -59,34 +61,53 @@ -- Logging utilities import System.Log.Logger +import System.Clock (getTime, TimeSpec, Clock(..))+ -- Imports from other parts of the program import SecondTransfer.MainLoop.CoherentWorker import SecondTransfer.MainLoop.Tokens import SecondTransfer.Sessions.Config-import SecondTransfer.Sessions.Internal (sessionExceptionHandler, SessionsContext, sessionsConfig)+import SecondTransfer.Sessions.Internal (sessionExceptionHandler,+ SessionsContext,+ sessionsConfig) import SecondTransfer.Utils (unfoldChannelAndSource) import SecondTransfer.Exception import qualified SecondTransfer.Utils.HTTPHeaders as He-import SecondTransfer.MainLoop.Logging (logWithExclusivity)+import SecondTransfer.MainLoop.Logging (logWithExclusivity, logit) -- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to -- use :-(-type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload)+type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload, Effect) type InputFrame = NH2.Frame -useChunkLength :: Int-useChunkLength = 16384+--useChunkLength :: Int+-- useChunkLength = 2048 -- 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)+-- end of data. Middle value is delay in microseconds+type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString, Effect) +data HeaderOutputMessage =+ NormalResponse_HM (GlobalStreamId, MVar HeadersSent, Headers, Effect)+ --+ |PushPromise_HM (GlobalStreamId, GlobalStreamId, Headers, Effect)+++-- Settings imposed by the peer+data SessionSettings = SessionSettings {+ _pushEnabled :: Bool+ }++makeLenses ''SessionSettings+++ -- Whatever a worker thread is going to need comes here.... -- this is to make refactoring easier, but not strictly needed. data WorkerThreadEnvironment = WorkerThreadEnvironment {@@ -96,14 +117,17 @@ -- 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)+ , _headersOutput :: Chan HeaderOutputMessage -- And regular contents can come this way and thus be properly mixed -- with everything else.... for now...- ,_dataOutput :: Chan DataOutputToConveyor+ ,_dataOutput :: MVar DataOutputToConveyor ,_streamsCancelled_WTE :: MVar NS.IntSet + ,_sessionSettings_WTE :: MVar SessionSettings++ ,_nextPushStream_WTE :: MVar Int } makeLenses ''WorkerThreadEnvironment@@ -125,9 +149,13 @@ sendCommandToSession :: SessionInput -> SessionInputCommand -> IO () sendCommandToSession (SessionInput chan) command = writeChan chan command +type SessionOutputPacket = Either SessionOutputCommand OutputFrame++type SessionOutputChannelAbstraction = Chan SessionOutputPacket+ -- From outside, one can only read from this one-newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) )-getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame)+newtype SessionOutput = SessionOutput SessionOutputChannelAbstraction+getFrameFromSession :: SessionOutput -> IO SessionOutputPacket getFrameFromSession (SessionOutput chan) = readChan chan @@ -153,6 +181,7 @@ -- temporary data SessionOutputCommand = CancelSession_SOC+ |FinishStream_SOC GlobalStreamId deriving Show @@ -161,17 +190,9 @@ -- Here is how we make a session wrapping a CoherentWorker-type CoherentSession = CoherentWorker -> SessionMaker---data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream)---- Settings imposed by the peer-data SessionSettings = SessionSettings {- _pushEnabled :: Bool- }+type CoherentSession = AwareWorker -> SessionMaker -makeLenses ''SessionSettings+data PostInputMechanism = PostInputMechanism (MVar (Maybe B.ByteString), InputDataStream) -- NH2.Frame != Frame@@ -182,8 +203,8 @@ ,_sessionInput :: Chan SessionInputCommand -- We need to lock this channel occassionally so that we can order multiple- -- header frames properly....- ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame))+ -- header frames properly....that's the reason for the outer MVar+ ,_sessionOutput :: MVar SessionOutputChannelAbstraction -- Use to encode ,_toEncodeHeaders :: MVar HP.DynamicTable@@ -207,7 +228,7 @@ -- I make copies of it in different contexts, and as needed. ,_forWorkerThread :: WorkerThreadEnvironment - ,_coherentWorker :: CoherentWorker+ ,_awareWorker :: AwareWorker -- Some streams may be cancelled ,_streamsCancelled :: MVar NS.IntSet@@ -226,6 +247,9 @@ -- And used to keep peer session settings ,_sessionSettings :: MVar SessionSettings++ -- What is the next stream available for push?+ ,_nextPushStream :: MVar Int } @@ -233,8 +257,8 @@ -- v- {headers table size comes here!!}-http2Session :: CoherentWorker -> Int -> SessionsContext -> IO Session-http2Session coherent_worker session_id sessions_context = do+http2Session :: AwareWorker -> Int -> SessionsContext -> IO Session+http2Session aware_worker session_id sessions_context = do session_input <- newChan session_output <- newChan session_output_mvar <- newMVar session_output@@ -252,8 +276,8 @@ -- 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)+ headers_output <- newChan :: IO (Chan HeaderOutputMessage)+ data_output <- newEmptyMVar :: IO (MVar DataOutputToConveyor) stream2postinputmechanism <- H.new stream2workerthread <- H.new@@ -261,6 +285,7 @@ receiving_headers <- newMVar Nothing session_settings <- newMVar $ SessionSettings { _pushEnabled = True }+ next_push_stream <- newMVar 8 -- What about stream cancellation? cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet)@@ -270,6 +295,8 @@ ,_headersOutput = headers_output ,_dataOutput = data_output ,_streamsCancelled_WTE = cancelled_streams_mvar+ ,_sessionSettings_WTE = session_settings+ ,_nextPushStream_WTE = next_push_stream } let session_data = SessionData {@@ -280,7 +307,7 @@ ,_toEncodeHeaders = encode_headers_table_mvar ,_stream2HeaderBlockFragment = stream_request_headers ,_forWorkerThread = for_worker_thread- ,_coherentWorker = coherent_worker+ ,_awareWorker = aware_worker ,_streamsCancelled = cancelled_streams_mvar ,_stream2PostInputMechanism = stream2postinputmechanism ,_stream2WorkerThread = stream2workerthread@@ -288,6 +315,7 @@ ,_receivingHeaders = receiving_headers ,_sessionSettings = session_settings ,_lastGoodStream = last_good_stream_mvar+ ,_nextPushStream = next_push_stream } let@@ -297,10 +325,12 @@ exc_guard component action = E.catch action (\e -> do- INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" )+ -- INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" ) exc_handler component e ) + use_chunk_length = sessions_context ^. sessionsConfig . dataFrameSize+ -- Create an input thread that decodes frames... forkIO $ exc_guard SessionInputThread_HTTP2SessionComponent $ runReaderT sessionInputThread session_data@@ -309,9 +339,10 @@ forkIO $ exc_guard SessionHeadersOutputThread_HTTP2SessionComponent $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data - -- Create a thread that captures data and sends it down the tube+ -- Create a thread that captures data and sends it down the tube. This is waiting for+ -- stuff coming from all the workers. This function is also in charge of closing streams forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent- $ dataOutputThread data_output session_output_mvar+ $ dataOutputThread use_chunk_length 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@@ -335,19 +366,20 @@ decode_headers_table_mvar <- view toDecodeHeaders stream_request_headers <- view stream2HeaderBlockFragment cancelled_streams_mvar <- view streamsCancelled- coherent_worker <- view coherentWorker+ coherent_worker <- view awareWorker for_worker_thread_uns <- view forWorkerThread stream2workerthread <- view stream2WorkerThread receiving_headers_mvar <- view receivingHeaders last_good_stream_mvar <- view lastGoodStream+ current_session_id <- view sessionIdAtSession input <- liftIO $ readChan session_input case input of FirstFrame_SIC (NH2.Frame- (NH2.FrameHeader _ 1 null_stream_id ) _ )| NH2.toStreamIdentifier 0 == null_stream_id -> do+ (NH2.FrameHeader _ 1 null_stream_id ) _ )| 0 == null_stream_id -> do -- This is a SETTINGS ACK frame, which is okej to have, -- do nothing here continue@@ -356,7 +388,7 @@ (NH2.Frame (NH2.FrameHeader _ 0 null_stream_id ) (NH2.SettingsFrame settings_list)- ) | NH2.toStreamIdentifier 0 == null_stream_id -> do+ ) | 0 == null_stream_id -> do -- Good, handle handleSettingsFrame settings_list continue@@ -417,8 +449,10 @@ liftIO $ putMVar last_good_stream_mvar (stream_id) else do -- We are not golden- INSTRUMENTATION( errorM "HTTP2.Session" "Protocol error: bad stream id")+ -- INSTRUMENTATION( errorM "HTTP2.Session" "Protocol error: bad stream id") closeConnectionBecauseIsInvalid NH2.ProtocolError+ -- So, now we know, we are ignoring priority frames at the moment+ -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame) else do maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar case maybe_rcv_headers_of of@@ -434,11 +468,19 @@ liftIO $ modifyMVar_ receiving_headers_mvar (\ _ -> return Nothing )+ -- Lets get a time+ headers_arrived_time <- liftIO $ getTime Monotonic -- 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+#if LOGIT_SWITCH_TIMINGS+ let+ (Just path) = He.fetchHeader header_list ":path"+ liftIO $ logit $ (pack . show $ stream_id ) `mappend` " -> " `mappend` path+#endif+ -- /DEBUG -- Good moment to remove the headers from the table.... we don't want a space -- leak here liftIO $ do@@ -472,6 +514,18 @@ else do return Nothing + let+ perception = Perception {+ _startedTime_Pr = headers_arrived_time,+ _streamId_Pr = stream_id,+ _sessionId_Pr = current_session_id+ }+ request = Request {+ _headers_RQ = header_list_after,+ _inputData_RQ = post_data_source,+ _perception_RQ = perception+ }+ -- TODO: Handle the cases where a request tries to send data -- even if the method doesn't allow for data. @@ -484,7 +538,9 @@ liftIO $ do thread_id <- forkIO $ E.catch (runReaderT- (workerThread (header_list_after, post_data_source) coherent_worker)+ (workerThread+ request+ coherent_worker) for_worker_thread ) (@@ -508,9 +564,9 @@ MiddleFrame_SIC 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) )+ -- 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) )+ -- 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@@ -520,7 +576,7 @@ error "InterruptingUnexistentStream" Just thread_id -> do- INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" )+ -- INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" ) throwTo thread_id StreamCancelledException continue@@ -529,7 +585,7 @@ -> unlessReceivingHeaders $ do -- So I got data to process -- TODO: Handle end of stream- let stream_id = NH2.fromStreamIdentifier nh2_stream_id+ let stream_id = nh2_stream_id -- TODO: Handle the cases where the stream_id doesn't match an already existent -- stream. In such cases it is justified to reset the connection with a protocol_error. @@ -552,10 +608,11 @@ (NH2.WindowUpdateFrame (fromIntegral (B.length somebytes)) )+ sendOutFrame (NH2.EncodeInfo NH2.defaultFlags- (NH2.toStreamIdentifier 0)+ 0 Nothing ) (NH2.WindowUpdateFrame@@ -577,15 +634,16 @@ MiddleFrame_SIC (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do -- Deal with pings: NOT an Ack, so answer- INSTRUMENTATION( debugM "HTTP2.Session" "Ping processed" )+ -- INSTRUMENTATION( debugM "HTTP2.Session" "Ping processed" ) sendOutFrame (NH2.EncodeInfo (NH2.setAck NH2.defaultFlags)- (NH2.toStreamIdentifier 0)+ 0 Nothing ) (NH2.PingFrame somebytes) + continue MiddleFrame_SIC (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do@@ -594,16 +652,14 @@ -- TODO: Do something with these settings!! MiddleFrame_SIC (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do- INSTRUMENTATION( debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) )+ -- INSTRUMENTATION( debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) ) -- Just acknowledge the frame.... for now handleSettingsFrame settings_list continue MiddleFrame_SIC somethingelse -> unlessReceivingHeaders $ do -- An undhandled case here....- INSTRUMENTATION( errorM "HTTP2.Session" $ "Received problematic frame: " )- INSTRUMENTATION( errorM "HTTP2.Session" $ ".. " ++ (show somethingelse) )-+ -- liftIO . logit $ "Strange frame:" `mappend` (pack . show $ somethingelse) continue where@@ -611,23 +667,38 @@ -- TODO: Do use the settings!!! handleSettingsFrame :: NH2.SettingsList -> ReaderT SessionData IO ()- handleSettingsFrame _settings_list =+ handleSettingsFrame _settings_list = do+ let+ enable_push = lookup NH2.SettingsEnablePush _settings_list+ session_settings_mvar <- view sessionSettings+ case enable_push of+ Just 1 -> liftIO $ modifyMVar_ session_settings_mvar+ $ \ old_settings -> return ( pushEnabled .~ True $ old_settings)+ Just 0 -> liftIO $ modifyMVar_ session_settings_mvar+ $ \ old_settings -> return ( pushEnabled .~ False $ old_settings)+ Just _ -> closeConnectionBecauseIsInvalid NH2.ProtocolError++ Nothing -> return ()+ -- liftIO $ logit $ "Settings " `mappend` (pack . show $ _settings_list) sendOutFrame (NH2.EncodeInfo (NH2.setAck NH2.defaultFlags)- (NH2.toStreamIdentifier 0)- Nothing )+ 0+ Nothing+ ) (NH2.SettingsFrame []) ---sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()+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 $ writeChan session_output $ Right (+ encode_info,+ payload,+ -- Not sending effects in this frame, since it is not related...+ error "sendOutFrameNotFor") liftIO $ putMVar session_output_mvar session_output @@ -697,11 +768,11 @@ sendOutFrame (NH2.EncodeInfo NH2.defaultFlags- (NH2.toStreamIdentifier 0)+ 0 Nothing ) (NH2.GoAwayFrame- (NH2.toStreamIdentifier last_good_stream)+ last_good_stream error_code "" )@@ -766,7 +837,7 @@ case pim_maybe of Just (PostInputMechanism (chan, _)) ->- liftIO $ writeChan chan Nothing+ liftIO $ putMVar chan Nothing Nothing -> -- TODO: This is a protocol error, handle it properly@@ -792,7 +863,7 @@ sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO () sendBytesToPim (PostInputMechanism (chan, _)) bytes =- liftIO $ writeChan chan (Just bytes)+ liftIO $ putMVar chan (Just bytes) postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream@@ -811,7 +882,7 @@ return $ NS.member stream_id cancelled_streams -sendPrimitive500Error :: IO PrincipalStream+sendPrimitive500Error :: IO TupledPrincipalStream sendPrimitive500Error = return ( [@@ -825,39 +896,78 @@ ) --workerThread :: Request -> CoherentWorker -> WorkerMonad ()-workerThread req coherent_worker =+-- Invokes the Coherent worker. Data is sent through pipes to+-- the output thread here in this session.+workerThread :: Request -> AwareWorker -> WorkerMonad ()+workerThread req aware_worker = do headers_output <- view headersOutput stream_id <- view streamId+ session_settings_mvar <- view sessionSettings_WTE+ session_settings <- liftIO $ readMVar session_settings_mvar+ next_push_stream_mvar <- view nextPushStream_WTE -- TODO: Handle exceptions here: what happens if the coherent worker -- throws an exception signaling that the request is ill-formed -- and should be dropped? That could happen in a couple of occassions, -- but really most cases should be handled here in this file...- (headers, _, data_and_conclussion) <-+#if LOGIT_SWITCH_TIMINGS+ liftIO . logit $ "worker-thread " `mappend` (pack . show $ stream_id)+#endif+ -- (headers, _, data_and_conclussion)+ principal_stream <- liftIO $ E.catch- ( do- (h, x, d) <- coherent_worker req- return $! (h,x,d)+ (+ aware_worker req ) (- (\ _ -> sendPrimitive500Error )- :: HTTP500PrecursorException -> IO (Headers, PushedStreams, DataAndConclusion)+ const $ tupledPrincipalStreamToPrincipalStream <$> sendPrimitive500Error+ :: HTTP500PrecursorException -> IO PrincipalStream ) + -- Pieces of the header+ let+ headers = principal_stream ^. headers_PS+ data_and_conclusion = principal_stream ^. dataAndConclusion_PS+ effects = principal_stream ^. effect_PS+ pushed_streams = principal_stream ^. pushedStreams_PS++ can_push = session_settings ^. pushEnabled++ -- There are several possible moments where the PUSH_PROMISEs can be sent,+ -- but a default safe one is before sending the response HEADERS, so that+ -- LINK headers in the response come after any potential promises.+ data_promises <- if can_push then do+ forM pushed_streams $ \ pushed_stream_comp -> do+ -- Initialize pushed stream+ pushed_stream <- liftIO pushed_stream_comp+ -- Send the request headers properly wrapped in a "Push Promise", do+ -- it before sending the actual response headers of this stream+ let+ request_headers = pushed_stream ^. requestHeaders_Psh+ response_headers = pushed_stream ^. responseHeaders_Psh+ pushed_data_and_conclusion = pushed_stream ^. dataAndConclusion_Psh++ child_stream_id <- liftIO $ modifyMVar next_push_stream_mvar+ $ (\ x -> return (x+2,x) )++ liftIO . writeChan headers_output . PushPromise_HM $+ (stream_id, child_stream_id, request_headers, effects)+ return (child_stream_id, response_headers, pushed_data_and_conclusion, effects)+ else+ return []+++ -- Now I send the headers, if that's possible at all- headers_sent <- liftIO $ newEmptyMVar- liftIO $ writeChan headers_output (stream_id, headers_sent, headers)+ headers_sent <- liftIO newEmptyMVar+ liftIO $ writeChan headers_output $ NormalResponse_HM (stream_id, headers_sent, headers, effects) -- 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... this is important for pushed -- streams is_stream_cancelled <- isStreamCancelled stream_id- if not is_stream_cancelled-- then do+ unless is_stream_cancelled $ 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) )@@ -866,31 +976,91 @@ -- NOTE: Exceptions generated here inheriting from HTTP500PrecursorException -- are let to bubble and managed in this thread fork point... (_maybe_footers, _) <- runConduit $- (transPipe liftIO data_and_conclussion)+ transPipe liftIO data_and_conclusion `fuseBothMaybe`- (sendDataOfStream stream_id headers_sent)+ sendDataOfStream stream_id headers_sent effects -- BIG TODO: Send the footers ... likely stream conclusion semantics -- will need to be changed.++ -- Now, time to fork threads for the pusher streams ....+ forM_ data_promises+ $ \ (child_stream_id, response_headers, pushed_data_and_conclusion, effects) -> do+ environment <- ask+ let+ action = pusherThread+ child_stream_id+ response_headers+ pushed_data_and_conclusion+ effects+ -- And let the action run in its own thread+ liftIO . forkIO $ runReaderT action environment+ return ()- else +++-- Invokes the Coherent worker. Data is sent through pipes to+-- the output thread here in this session.+pusherThread :: GlobalStreamId -> Headers -> DataAndConclusion -> Effect -> WorkerMonad ()+pusherThread child_stream_id response_headers pushed_data_and_conclusion effects =+ do+ headers_output <- view headersOutput+ session_settings_mvar <- view sessionSettings_WTE+ session_settings <- liftIO $ readMVar session_settings_mvar++ -- TODO: Handle exceptions here: what happens if the coherent worker+ -- throws an exception signaling that the request is ill-formed+ -- and should be dropped? That could happen in a couple of occassions,+ -- but really most cases should be handled here in this file...+ -- (headers, _, data_and_conclussion)+++ -- Now I send the headers, if that's possible at all+ headers_sent <- liftIO newEmptyMVar+ liftIO . writeChan headers_output+ $ NormalResponse_HM (child_stream_id, headers_sent, response_headers, effects)++ -- 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... this is important for pushed+ -- streams+ is_stream_cancelled <- isStreamCancelled child_stream_id+ unless is_stream_cancelled $ 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+ -- NOTE: Exceptions generated here inheriting from HTTP500PrecursorException+ -- are let to bubble and managed in this thread fork point...+ (_maybe_footers, _) <- runConduit $+ transPipe liftIO pushed_data_and_conclusion+ `fuseBothMaybe`+ sendDataOfStream child_stream_id headers_sent effects+ -- BIG TODO: Send the footers ... likely stream conclusion semantics+ -- will need to be changed.+ return () ++ -- v-- comp. monad.-sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()-sendDataOfStream stream_id headers_sent = do+sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Effect -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()+sendDataOfStream stream_id headers_sent effect = do data_output <- view dataOutput -- Wait for all headers sent liftIO $ takeMVar headers_sent consumer data_output where+ --delay = effect ^. middlePauseForDelivery_Ef consumer data_output = do maybe_bytes <- await case maybe_bytes of Nothing ->- liftIO $ writeChan data_output (stream_id, Nothing)+ -- This is how we finish sending data+ liftIO $ putMVar data_output (stream_id, Nothing, effect) Just bytes -> do- liftIO $ writeChan data_output (stream_id, Just bytes)+ liftIO $ do+ putMVar data_output (stream_id, Just bytes, effect) consumer data_output @@ -920,73 +1090,144 @@ isAboutHeaders :: InputFrame -> Maybe (GlobalStreamId, B.ByteString) isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) )- = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+ = Just (stream_id, block_fragment) isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )- = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+ = Just (stream_id, block_fragment) isAboutHeaders _ = Nothing +getHeadersPriority :: InputFrame -> Maybe NH2.Priority+getHeadersPriority (NH2.Frame _ ( NH2.HeadersFrame prio _ ) ) = prio+getHeadersPriority _ = 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+streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = 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))+headersOutputThread :: Chan HeaderOutputMessage -- (GlobalStreamId, MVar HeadersSent, Headers, Effect)+ -> MVar SessionOutputChannelAbstraction -> ReaderT SessionData IO () headersOutputThread input_chan session_output_mvar = forever $ do- (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan - -- First encode the headers using the table- encode_dyn_table_mvar <- view toEncodeHeaders+ use_chunk_length <- view $ sessionsContext . sessionsConfig . dataFrameSize - 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+ header_output_request <- liftIO $ readChan input_chan+ case header_output_request of+ NormalResponse_HM (stream_id, headers_ready_mvar, headers, effect) -> do - -- Now split the bytestring in chunks of the needed size....- bs_chunks <- return $! bytestringChunk useChunkLength data_to_send+ -- First encode the headers using the table+ encode_dyn_table_mvar <- view toEncodeHeaders - -- 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- -- INSTRUMENTATION( debugM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) )- putMVar headers_ready_mvar HeadersSent- )+ 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....+ -- Note that the only way we can+ let+ bs_chunks = bytestringChunk use_chunk_length data_to_send++ -- And send the chunks through while locking the output place....+ liftIO $ bs_chunks `deepseq` E.bracket+ (takeMVar session_output_mvar)+ (putMVar session_output_mvar )+ (\ session_output -> do+ writeIndividualHeaderFrames session_output stream_id bs_chunks True effect+ -- And say that the headers for this thread are out+ -- INSTRUMENTATION( debugM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) )+ putMVar headers_ready_mvar HeadersSent+ )++ PushPromise_HM (parent_stream_id, child_stream_id, promise_headers, effect) -> do++ 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 promise_headers+ liftIO $ putMVar encode_dyn_table_mvar new_dyn_table++ -- Now split the bytestring in chunks of the needed size....+ bs_chunks <- return $! bytestringChunk use_chunk_length 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+ writePushPromiseFrames session_output parent_stream_id child_stream_id bs_chunks True effect+ -- No notification here... but since everyhing headers is serialized here and the next+ -- piece of information corresponding to the child stream is a headers sequence, we shouldn't+ -- need any other tricks here+ )++ where writeIndividualHeaderFrames ::- Chan (Either SessionOutputCommand OutputFrame)+ SessionOutputChannelAbstraction -> GlobalStreamId -> [B.ByteString] -> Bool+ -> Effect -> IO ()- writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first =+ writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first effect = writeChan session_output $ Right ( NH2.EncodeInfo { NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags- ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id+ ,NH2.encodeStreamId = stream_id ,NH2.encodePadding = Nothing },- (if is_first then NH2.HeadersFrame Nothing last_fragment else NH2.ContinuationFrame last_fragment)+ (if is_first then NH2.HeadersFrame Nothing last_fragment else NH2.ContinuationFrame last_fragment),+ effect )- writeIndividualHeaderFrames session_output stream_id (fragment:xs) is_first = do+ writeIndividualHeaderFrames session_output stream_id (fragment:xs) is_first effect = do writeChan session_output $ Right ( NH2.EncodeInfo { NH2.encodeFlags = NH2.defaultFlags- ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id+ ,NH2.encodeStreamId = stream_id ,NH2.encodePadding = Nothing },- (if is_first then NH2.HeadersFrame Nothing fragment else NH2.ContinuationFrame fragment)+ (if is_first then NH2.HeadersFrame Nothing fragment else NH2.ContinuationFrame fragment),+ effect )- writeIndividualHeaderFrames session_output stream_id xs False+ writeIndividualHeaderFrames session_output stream_id xs False effect + writePushPromiseFrames ::+ SessionOutputChannelAbstraction+ -> GlobalStreamId+ -> GlobalStreamId+ -> [B.ByteString]+ -> Bool+ -> Effect+ -> IO ()+ writePushPromiseFrames session_output parent_stream_id child_stream_id (last_fragment:[]) is_first effect =+ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags+ ,NH2.encodeStreamId = parent_stream_id+ ,NH2.encodePadding = Nothing },+ (if is_first+ then NH2.PushPromiseFrame child_stream_id last_fragment+ else NH2.ContinuationFrame last_fragment+ ),+ effect+ )+ writePushPromiseFrames session_output parent_stream_id child_stream_id (fragment:xs) is_first effect = do+ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.defaultFlags+ ,NH2.encodeStreamId = parent_stream_id+ ,NH2.encodePadding = Nothing },+ (if is_first+ then NH2.PushPromiseFrame child_stream_id fragment+ else NH2.ContinuationFrame fragment+ ),+ effect+ )+ writePushPromiseFrames session_output parent_stream_id child_stream_id xs False effect + bytestringChunk :: Int -> B.ByteString -> [B.ByteString] bytestringChunk len s | (B.length s) < len = [ s ] bytestringChunk len s = h:(bytestringChunk len xs)@@ -994,7 +1235,8 @@ (h, xs) = B.splitAt len s --- TODO: find a clean way to finish this thread (maybe with negative stream ids?)+-- This thread is for the entire session, not for each stream. The thread should+-- die while waiting for input in a garbage-collected mvar. -- 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....@@ -1003,28 +1245,35 @@ -- 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))+dataOutputThread :: Int+ -> MVar DataOutputToConveyor+ -> MVar SessionOutputChannelAbstraction -> IO ()-dataOutputThread input_chan session_output_mvar = forever $ do- (stream_id, maybe_contents) <- readChan input_chan+dataOutputThread use_chunk_length input_chan session_output_mvar = forever $ do+ (stream_id, maybe_contents, effect) <- takeMVar input_chan case maybe_contents of Nothing -> do liftIO $ do 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 ""- )+ (\ session_output -> do+ -- Write an empty data-frame with the right flags+ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags+ ,NH2.encodeStreamId = stream_id+ ,NH2.encodePadding = Nothing },+ NH2.DataFrame "",+ effect+ )+ -- And then write an-end-of-stream command+ writeChan session_output $ Left . FinishStream_SOC $ stream_id ) Just contents -> do -- And now just simply output it...- let bs_chunks = bytestringChunk useChunkLength $! contents+ let bs_chunks = bytestringChunk use_chunk_length contents -- And send the chunks through while locking the output place....- writeContinuations bs_chunks stream_id+ bs_chunks `deepseq` writeContinuations bs_chunks stream_id effect+ return () where @@ -1032,11 +1281,18 @@ (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+ writeContinuations :: [B.ByteString] -> GlobalStreamId -> Effect -> IO ()+ writeContinuations fragments stream_id effect =+ mapM_ (\ fragment ->+ withLockedSessionOutput+ (\ session_output -> writeChan session_output $ Right (+ NH2.EncodeInfo {+ NH2.encodeFlags = NH2.defaultFlags+ ,NH2.encodeStreamId = stream_id+ ,NH2.encodePadding = Nothing+ },+ NH2.DataFrame fragment,+ effect )+ )+ )+ fragments
hs-src/SecondTransfer/MainLoop.hs view
@@ -13,7 +13,8 @@ , #endif - enableConsoleLogging+ enableConsoleLogging,+ logit #ifndef DISABLE_OPENSSL_TLS ,TLSLayerGenericProblem(..)@@ -28,4 +29,4 @@ #ifndef DISABLE_OPENSSL_TLS import SecondTransfer.MainLoop.OpenSSL_TLS #endif-import SecondTransfer.MainLoop.Logging (enableConsoleLogging)+import SecondTransfer.MainLoop.Logging (enableConsoleLogging,logit)
hs-src/SecondTransfer/MainLoop/CoherentWorker.hs view
@@ -1,6 +1,6 @@ {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FunctionalDependencies, FlexibleInstances, DeriveDataTypeable #-} +{-# LANGUAGE OverloadedStrings, TemplateHaskell #-}+{-# LANGUAGE FunctionalDependencies, FlexibleInstances, DeriveDataTypeable #-} -- | A CoherentWorker is one that doesn't need to compute everything at once... -- This one is simpler than the SPDY one, because it enforces certain order.... @@ -15,21 +15,49 @@ , Header , Headers , FinalizationHeaders- , Request+ , Request(..) , Footers- , CoherentWorker- , PrincipalStream+ , Perception(..)+ , Effect(..)+ , AwareWorker+ , PrincipalStream(..) , PushedStreams- , PushedStream+ , PushedStream(..) , DataAndConclusion+ , CoherentWorker , InputDataStream+ , TupledPrincipalStream+ , FragmentDeliveryCallback - ) where + , headers_RQ+ , inputData_RQ+ , perception_RQ+ , headers_PS+ , pushedStreams_PS+ , dataAndConclusion_PS+ , dataAndConclusion_Psh+ , requestHeaders_Psh+ , responseHeaders_Psh+ , effect_PS+ , startedTime_Pr+ , streamId_Pr+ , sessionId_Pr+ , fragmentDeliveryCallback_Ef+ , priorityEffect_Ef + , defaultEffects+ , coherentToAwareWorker + , tupledPrincipalStreamToPrincipalStream+ , requestToTupledRequest+ ) where+++import Control.Lens import qualified Data.ByteString as B import Data.Conduit import Data.Foldable (find)+import System.Clock (TimeSpec) -- | The name part of a header@@ -41,90 +69,181 @@ -- | The complete header type Header = (HeaderName, HeaderValue) --- |List of headers. The first part of each tuple is the header name +-- |List of headers. The first part of each tuple is the header name -- (be sure to conform to the HTTP/2 convention of using lowercase) -- and the second part is the headers contents. This list needs to include--- the special :method, :scheme, :authority and :path pseudo-headers for +-- the special :method, :scheme, :authority and :path pseudo-headers for -- requests; and :status (with a plain numeric value represented in ascii digits) -- for responses. type Headers = [Header] -- |This is a Source conduit (see Haskell Data.Conduit library from Michael Snoyman)--- that you can use to retrieve the data sent by the client piece-wise. +-- that you can use to retrieve the data sent by the client piece-wise. type InputDataStream = Source IO B.ByteString ++-- | Data related to the request+data Perception = Perception {+ -- Monotonic time close to when the request was first seen in+ -- the processing pipeline.+ _startedTime_Pr :: TimeSpec,+ -- The HTTP/2 stream id. Or the serial number of the request in an+ -- HTTP/1.1 session.+ _streamId_Pr :: Int,+ -- You know better than to use this for normal web request+ -- processing. But otherwise a number uniquely identifying the session. + _sessionId_Pr :: Int+ }++makeLenses ''Perception++ -- | A request is a set of headers and a request body....--- which will normally be empty, except for POST and PUT requests. But --- this library enforces none of that. -type Request = (Headers, Maybe InputDataStream)+-- which will normally be empty, except for POST and PUT requests. But+-- this library enforces none of that.+data Request = Request {+ _headers_RQ :: ! Headers,+ _inputData_RQ :: Maybe InputDataStream,+ _perception_RQ :: ! Perception+ } --- | Finalization headers. If you don't know what they are, chances are --- that you don't need to worry about them for now. The support in this --- library for those are at best sketchy. +makeLenses ''Request++-- | Finalization headers. If you don't know what they are, chances are+-- that you don't need to worry about them for now. The support in this+-- library for those are at best sketchy. type FinalizationHeaders = Headers --- | Finalization headers +-- | Finalization headers type Footers = FinalizationHeaders --- | You use this type to answer a request. The `Headers` are thus response +-- | A list of pushed streams.+-- Notice that a list of IO computations is required here. These computations+-- only happen when and if the streams are pushed to the client.+-- The lazy nature of Haskell helps to avoid unneeded computations if the+-- streams are not going to be sent to the client.+type PushedStreams = [ IO PushedStream ]++-- | A source-like conduit with the data returned in the response. The+-- return value of the conduit is a list of footers. For now that list can+-- be anything (even bottom), I'm not handling it just yet.+type DataAndConclusion = ConduitM () B.ByteString IO Footers+++-- | A pushed stream, represented by a list of request headers,+-- a list of response headers, and the usual response body (which+-- may include final footers (not implemented yet)).+data PushedStream = PushedStream {+ _requestHeaders_Psh :: Headers,+ _responseHeaders_Psh :: Headers,+ _dataAndConclusion_Psh :: DataAndConclusion+ }++makeLenses ''PushedStream+++-- | First argument is the ordinal of this data frame, second an approximation of when+-- the frame was delivered, according to the monotonic clock. Do not linger in this call,+-- it may delay some important thread+type FragmentDeliveryCallback = Int -> TimeSpec -> IO ()+++-- | Sometimes a response needs to be handled a bit specially,+-- for example by reporting delivery details back to the worker+data Effect = Effect {+ _fragmentDeliveryCallback_Ef :: Maybe FragmentDeliveryCallback++ -- In certain circunstances a stream can use an internal priority,+ -- not given by the browser and the protocol. Lowest values here are+ -- given more priority. Default (when Nothing) is given zero. Cases+ -- with negative numbers also work.+ ,_priorityEffect_Ef :: Maybe Int+ }++makeLenses ''Effect++defaultEffects :: Effect+defaultEffects = Effect {+ _fragmentDeliveryCallback_Ef = Nothing,+ _priorityEffect_Ef = Nothing+ }+++-- | You use this type to answer a request. The `Headers` are thus response -- headers and they should contain the :status pseudo-header. The `PushedStreams` -- is a list of pushed streams...(I don't thaink that I'm handling those yet)-type PrincipalStream = (Headers, PushedStreams, DataAndConclusion)+data PrincipalStream = PrincipalStream {+ _headers_PS :: Headers,+ _pushedStreams_PS :: PushedStreams,+ _dataAndConclusion_PS :: DataAndConclusion,+ _effect_PS :: Effect+ } +makeLenses ''PrincipalStream --- | A source-like conduit with the data returned in the response. The --- return value of the conduit is a list of footers. For now that list can --- be anything (even bottom), I'm not handling it just yet. -type DataAndConclusion = ConduitM () B.ByteString IO Footers -- | Main type of this library. You implement one of these for your server.--- Basically this is a callback that the library calls as soon as it has+-- This is a callback that the library calls as soon as it has -- all the headers of a request. For GET requests that's the entire request--- basically, but for POST and PUT requests this is just before the data --- starts arriving to the server. +-- basically, but for POST and PUT requests this is just before the data+-- starts arriving to the server. ----- It is important that you consume the data in the cases where there is an +-- It is important that you consume the data in the cases where there is an -- input stream, otherwise the memory is lost for the duration of the request, -- and a malicious client can use that. -- -- Also, notice that when handling requests your worker can be interrupted with -- an asynchronous exception of type 'StreamCancelledException', if the peer -- cancels the stream-type CoherentWorker = Request -> IO PrincipalStream+type AwareWorker = Request -> IO PrincipalStream +-- | A CoherentWorker is a less fuzzy worker, but less aware.+type CoherentWorker = (Headers, Maybe InputDataStream) -> IO (Headers, PushedStreams, DataAndConclusion) --- | A list of pushed streams. --- Notice that a list of IO computations is required here. These computations--- only happen when and if the streams are pushed to the client. --- The lazy nature of Haskell helps to avoid unneeded computations if the --- streams are not going to be sent to the client.-type PushedStreams = [ IO PushedStream ]+-- | Not exactly equivalent of the prinicipal stream+type TupledPrincipalStream = (Headers, PushedStreams, DataAndConclusion) --- | A pushed stream, represented by a list of request headers, --- a list of response headers, and the usual response body (which --- may include final footers (not implemented yet)).-type PushedStream = (Headers, Headers, DataAndConclusion)+type TupledRequest = (Headers, Maybe InputDataStream) ++tupledPrincipalStreamToPrincipalStream :: TupledPrincipalStream -> PrincipalStream+tupledPrincipalStreamToPrincipalStream (headers, pushed_streams, data_and_conclusion) = PrincipalStream+ {+ _headers_PS = headers,+ _pushedStreams_PS = pushed_streams,+ _dataAndConclusion_PS = data_and_conclusion,+ _effect_PS = defaultEffects+ }++requestToTupledRequest :: Request -> TupledRequest+requestToTupledRequest req =+ (req ^. headers_RQ,+ req ^. inputData_RQ )++coherentToAwareWorker :: CoherentWorker -> AwareWorker+coherentToAwareWorker w r =+ fmap tupledPrincipalStreamToPrincipalStream $ w . requestToTupledRequest $ r+ -- | Gets a single header from the list getHeaderFromFlatList :: Headers -> B.ByteString -> Maybe B.ByteString-getHeaderFromFlatList unvl bs = +getHeaderFromFlatList unvl bs = case find (\ (x,_) -> x==bs ) unvl of- Just (_, found_value) -> Just found_value + Just (_, found_value) -> Just found_value - Nothing -> Nothing + Nothing -> Nothing --- | If you want to skip the footers, i.e., they are empty, use this +-- | If you want to skip the footers, i.e., they are empty, use this -- function to convert an ordinary Source to a DataAndConclusion. nullFooter :: Source IO B.ByteString -> DataAndConclusion-nullFooter s = s =$= go - where - go = do - i <- await - case i of - Nothing -> +nullFooter s = s =$= go+ where+ go = do+ i <- await+ case i of+ Nothing -> return [] Just ii -> do- yield ii - go + yield ii+ go
hs-src/SecondTransfer/MainLoop/Framer.hs view
@@ -2,7 +2,7 @@ module SecondTransfer.MainLoop.Framer( readNextChunk ,readNextChunkAndContinue- ,readLength+ ,readLengthFromUntamed ,Framer ,LengthCallback@@ -20,7 +20,7 @@ -- import Debug.Trace (trace) #ifndef IMPLICIT_MONOID-import Data.Monoid +import Data.Monoid #endif @@ -41,76 +41,76 @@ LengthCallback -- ^ How to know if we can split somewhere -> B.ByteString -- ^ Input left-overs -> m B.ByteString -- ^ Generator action- -> Source m B.ByteString -- ^ Packet and leftovers, if we could get them -readNextChunk length_callback input_leftovers gen = do - let + -> Source m B.ByteString -- ^ Packet and leftovers, if we could get them+readNextChunk length_callback input_leftovers gen = do+ let maybe_length = length_callback input_leftovers - case maybe_length of - Just the_length -> do - -- Just need to read the rest .... + case maybe_length of+ Just the_length -> do+ -- Just need to read the rest .... (package_bytes, newnewleftovers) <- lift $ readUpTo gen input_leftovers the_length- yield package_bytes - readNextChunk length_callback newnewleftovers gen + yield package_bytes+ readNextChunk length_callback newnewleftovers gen - Nothing -> do - -- Read a bit more - new_fragment <- lift gen + Nothing -> do+ -- Read a bit more+ new_fragment <- lift gen let new_leftovers = input_leftovers `mappend` new_fragment readNextChunk length_callback new_leftovers gen --- +-- readNextChunkAndContinue :: Monad m => LengthCallback -- ^ How to know if we can split somewhere -> B.ByteString -- ^ Input left-overs -> m B.ByteString -- ^ Generator action -> m (B.ByteString, B.ByteString) -- ^ Packet bytes and left-overs.-readNextChunkAndContinue length_callback input_leftovers gen = do - let +readNextChunkAndContinue length_callback input_leftovers gen = do+ let maybe_length = length_callback input_leftovers - case maybe_length of + case maybe_length of - Just the_length -> do - -- Just need to read the rest .... + Just the_length -> do+ -- Just need to read the rest .... (package_bytes, newnewleftovers) <- readUpTo gen input_leftovers the_length return (package_bytes, newnewleftovers) - Nothing -> do - -- Read a bit more - new_fragment <- gen + Nothing -> do+ -- Read a bit more+ new_fragment <- gen let new_leftovers = input_leftovers `mappend` new_fragment readNextChunkAndContinue length_callback new_leftovers gen readUpTo :: Monad m => m B.ByteString -> B.ByteString -> Int -> m (B.ByteString, B.ByteString)-readUpTo gen input_leftovers the_length = - let +readUpTo gen input_leftovers the_length =+ let initial_length = B.length input_leftovers bu = Bu.byteString input_leftovers- go lo readsofar_length - | readsofar_length >= the_length = + go lo readsofar_length+ | readsofar_length >= the_length = return $ B.splitAt the_length $ LB.toStrict . Bu.toLazyByteString $ lo | otherwise = do- frag <- gen - go (lo `mappend` (Bu.byteString frag)) (readsofar_length + (B.length frag))- in + frag <- gen+ go (lo `mappend` Bu.byteString frag) (readsofar_length + B.length frag)+ in go bu initial_length -- Some protocols, e.g., http/2, have the client transmit a fixed-length -- prefix. This function reads both that prefix and returns whatever get's--- trapped up there.... -readLength :: MonadIO m => Int -> m B.ByteString -> m (B.ByteString, B.ByteString)-readLength the_length gen = - readUpTo_ mempty - where - readUpTo_ lo - | (B.length lo) >= the_length = do+-- trapped up there....+readLengthFromUntamed :: MonadIO m => Int -> m B.ByteString -> m (B.ByteString, B.ByteString)+readLengthFromUntamed the_length gen =+ readUpTo_ mempty+ where+ readUpTo_ lo+ | B.length lo >= the_length = -- liftIO $ putStrLn "Full read" return $ B.splitAt the_length lo- | otherwise = do - -- liftIO $ putStrLn $ "fragment read " ++ (show lo) - frag <- gen + | otherwise = do+ -- liftIO $ putStrLn $ "fragment read " ++ (show lo)+ frag <- gen readUpTo_ (lo `mappend` frag)
hs-src/SecondTransfer/MainLoop/Logging.hs view
@@ -1,12 +1,16 @@ {-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE BangPatterns, OverloadedStrings #-} module SecondTransfer.MainLoop.Logging ( -- | Simple, no fuss enable logging enableConsoleLogging ,logWithExclusivity+ ,logit ) where -import System.IO (stderr)-+import System.IO (stderr,openFile)+import qualified System.IO as SIO+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as Bch -- Logging utilities import System.Log.Formatter (simpleLogFormatter)@@ -15,8 +19,11 @@ -- import System.Log.Handler.Syslog (Facility (..), Option (..), openlog) import System.Log.Logger import System.IO.Unsafe (unsafePerformIO)+import System.Clock as Cl -import Control.Concurrent.MVar +import Control.Concurrent.MVar+import Control.Concurrent.Chan+import Control.Concurrent -- | Activates logging to terminal@@ -26,7 +33,7 @@ -- | Protect logging with a mutex... that is to say, -- this is a horrible hack and you should try to log--- as little as possible or nothing at all. This just +-- as little as possible or nothing at all. This just -- works for instrumentation locks... globallyLogWell :: MVar () {-# NOINLINE globallyLogWell #-}@@ -34,40 +41,76 @@ -- | Used internally to avoid garbled logs logWithExclusivity :: IO () -> IO ()-logWithExclusivity a = withMVar globallyLogWell (\_ -> a )+logWithExclusivity a = withMVar globallyLogWell (const a ) configureLoggingToConsole :: IO ()-configureLoggingToConsole = do - s <- streamHandler stderr DEBUG >>= +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 >>= +-- 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 :: (LogHandler s) => s -> IO () setLoggerLevels s = do updateGlobalLogger rootLoggerName removeHandler updateGlobalLogger "Session" (- setHandlers [s] . - setLevel INFO + setHandlers [s] .+ setLevel INFO ) updateGlobalLogger "OpenSSL" (- setHandlers [s] . - setLevel DEBUG + setHandlers [s] .+ setLevel DEBUG ) updateGlobalLogger "HTTP1" (- setHandlers [s] . + setHandlers [s] . setLevel DEBUG ) updateGlobalLogger "HTTP2" (- setHandlers [s] . + setHandlers [s] . setLevel DEBUG ) +++data Logit = Logit Cl.TimeSpec B.ByteString+++loggerChan :: Chan Logit+{-# NOINLINE loggerChan #-}+loggerChan = unsafePerformIO $ do+ chan <- newChan+ log_file <- openFile "LOGIT" SIO.WriteMode+ SIO.hSetBuffering log_file SIO.LineBuffering+ start_of_time <- Cl.getTime Cl.Monotonic+ forkIO $ readLoggerChan chan log_file start_of_time+ return chan++readLoggerChan :: Chan Logit -> SIO.Handle -> Cl.TimeSpec -> IO ()+readLoggerChan chan_logit file_handle origin_time = do+ Logit timespec bs <- readChan chan_logit+ let+ Cl.TimeSpec sec' nsec' = timespec - origin_time+ SIO.hPutStr file_handle (show sec')+ SIO.hPutStr file_handle "|"+ SIO.hPutStr file_handle (show nsec')+ SIO.hPutStr file_handle "|"+ Bch.hPutStrLn file_handle bs+ SIO.hFlush file_handle+ readLoggerChan chan_logit file_handle origin_time++-- Simple logging function. It logs everything to a file named+-- "logit" in the current directory, adding a time-stamp+logit :: B.ByteString -> IO ()+logit !msg = do+ time <- Cl.getTime Cl.Monotonic+ let+ lg = Logit time msg+ writeChan loggerChan lg
hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs view
@@ -7,16 +7,16 @@ ,TLSLayerGenericProblem(..) ,FinishRequest(..)- ) where + ) where import Control.Monad-import Control.Concurrent.MVar -import Control.Exception +import Control.Concurrent.MVar+import Control.Exception import qualified Control.Exception as E #ifndef IMPLICIT_APPLICATIVE_FOLDABLE import Data.Foldable (foldMap) #endif-import Data.Typeable +import Data.Typeable import Data.Monoid () import Foreign import Foreign.C@@ -30,31 +30,32 @@ -- import System.Log.Logger import SecondTransfer.MainLoop.PushPullType+import SecondTransfer.MainLoop.Logging (logit) import SecondTransfer.Exception #include "Logging.cpphs" --- | Exceptions 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. +-- | Exceptions 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 +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 +data InterruptibleEither a b =+ Left_I a+ |Right_I b |Interrupted --- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, +-- | 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@@ -62,18 +63,17 @@ -- These names are absolutely improper.... -- Session creator-data Connection_t +data Connection_t -- Session data Wired_t -type Connection_Ptr = Ptr Connection_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, +-- 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 :: +foreign import ccall "make_connection" makeConnection :: CString -- cert filename -> CString -- privkey_filename -> CString -- hostname@@ -82,24 +82,27 @@ -> CInt -- protocol list length -> IO Connection_Ptr -allOk :: CInt -allOk = 0 +allOk :: CInt+allOk = 0 -badHappened :: CInt -badHappened = 1 +badHappened :: CInt+badHappened = 1 -timeoutReached :: CInt +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 +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 +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 recv_data_best_effort(wired_session_t* ws, char* inbuffer, int buffer_size, int can_wait, int* data_recvd);+foreign import ccall "recv_data_best_effort" recvDataBestEffort :: Wired_Ptr -> Ptr CChar -> CInt -> 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 @@ -117,89 +120,86 @@ protocolsToWire :: Protocols -> B.ByteString-protocolsToWire protocols = - LB.toStrict . BB.toLazyByteString $ - foldMap (\ protocol +protocolsToWire protocols =+ LB.toStrict . BB.toLazyByteString $+ foldMap (\ protocol -> (BB.lazyByteString . LB.fromChunks) [ B.singleton $ fromIntegral $ B.length protocol,- protocol + protocol ] ) protocols- --- | Simple function to open ++-- | 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 + -- Bear in mind that multiple domains can be served from the same HTTP/2 -- TLS socket, so please create the HTTP/2 certificate accordingly.- -- Also, currently this function only accepts paths to certificates - -- or certificate chains in .pem format. - -> FilePath -- ^ Path to the key of your certificate. + -- Also, currently this function only accepts paths to certificates+ -- or certificate chains in .pem format.+ -> 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 + -> [(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, + -- 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. + -> Int -- ^ Port to open to listen for connections. -> IO ()-tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do +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 + 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 + makeConnection c_certfn c_keyfn c_iname (fromIntegral interface_port)- pchar + pchar (fromIntegral len) - if connection_ptr == nullPtr - then do+ when (connection_ptr == nullPtr) $ throwIO $ TLSLayerGenericProblem "Could not create listening end"- else do- return () - forever $ do - either_wired_ptr <- alloca $ \ wired_ptr_ptr -> - let - tryOnce = do + 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 + let+ r = case result_code of+ re | re == allOk -> do p <- peek wired_ptr_ptr return $ Right p- | re == timeoutReached -> tryOnce + | re == timeoutReached -> tryOnce | re == badHappened -> return $ Left ("A wait for connection failed" :: String)- r + r in tryOnce - case either_wired_ptr of + case either_wired_ptr of -- Disable a warning- Left _msg -> do + Left _msg -> return () - Right wired_ptr -> do + Right wired_ptr -> do - (push_action, pull_action, close_action) <- provideActions wired_ptr + attendant_callbacks <- provideActions wired_ptr use_protocol <- getSelectedProtocol wired_ptr - let - maybe_session_attendant = case fromIntegral use_protocol of - n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + 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 + case maybe_session_attendant of - Just session_attendant -> - E.catch - (session_attendant push_action pull_action close_action)- ((\ e -> do + Just session_attendant ->+ E.catch+ (session_attendant attendant_callbacks)+ ((\ e -> throwIO e )::TLSLayerGenericProblem -> IO () ) @@ -208,143 +208,167 @@ 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` +-- | 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 +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 + 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 + makeConnection c_certfn c_keyfn c_iname (fromIntegral interface_port)- pchar + pchar (fromIntegral len) -- Create a computation that accepts a connection, runs a session on it and recurses- let - recursion = do + let+ recursion = do -- Get a SSL session- either_wired_ptr <- alloca $ \ wired_ptr_ptr -> - let - tryOnce = do + 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 + let+ r = case result_code of+ re | re == allOk -> do p <- peek wired_ptr_ptr return $ Right_I p- | re == timeoutReached -> do + | re == timeoutReached -> do got_finish_request <- tryTakeMVar finish_request- case got_finish_request of + case got_finish_request of Nothing -> tryOnce Just _ ->- return Interrupted + return Interrupted - | re == badHappened -> return $ Left_I "A wait for connection failed"- r + | re == badHappened -> return $ Left_I ("A wait for connection failed" :: String)+ r in tryOnce -- With the potentially obtained SSL session do...- case either_wired_ptr of + case either_wired_ptr of - Left_I _msg -> do + Left_I _msg -> -- // .. // recursion - Right_I wired_ptr -> do - (push_action, pull_action, close_action) <- provideActions wired_ptr + Right_I wired_ptr -> do+ attendant_callbacks <- provideActions wired_ptr use_protocol <- getSelectedProtocol wired_ptr - let - maybe_session_attendant = case fromIntegral use_protocol of - n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + 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 + case maybe_session_attendant of - Just session_attendant -> - session_attendant push_action pull_action close_action+ Just session_attendant ->+ session_attendant attendant_callbacks Nothing -> return () -- // .. //- recursion + recursion - Interrupted -> do+ Interrupted -> closeConnection connection_ptr -- Start the loop defined above...- recursion + recursion --- When we are using the eternal version of this function, wake up --- each second .... +-- When we are using the eternal version of this function, wake up+-- each second .... defaultWaitTime :: CInt defaultWaitTime = 200000--- Okej, more responsiviness needed -smallWaitTime :: CInt +-- Okej, more responsiviness needed+smallWaitTime :: CInt smallWaitTime = 50000 -provideActions :: Wired_Ptr -> IO (LB.ByteString -> IO (), IO B.ByteString, IO ())+-- provideActions :: Wired_Ptr -> IO (LB.ByteString -> IO (), Int -> IO B.ByteString, IO ())+provideActions :: Wired_Ptr -> IO AttendantCallbacks provideActions wired_ptr = do already_closed_mvar <- newMVar False let pushAction :: LB.ByteString -> IO ()- pushAction datum = do + pushAction datum = do already_closed <- readMVar already_closed_mvar- if already_closed - then do+ if already_closed+ then throwIO $ TLSLayerGenericProblem "Tried to send data on closed handle"- else do- BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do + else+ BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do result <- sendData wired_ptr pchar (fromIntegral len)- case result of - r | r == allOk -> do + case result of+ r | r == allOk -> return ()- | r == badHappened -> do + | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data" - pullAction :: IO B.ByteString- pullAction = do+ pullAction :: Int -> IO B.ByteString+ pullAction bytes_to_get = do already_closed <- readMVar already_closed_mvar- if already_closed - then do + if already_closed+ then throwIO $ TLSLayerGenericProblem "Tried to receive on closed handle"- else - allocaBytes useBufferSize $ \ pcharbuffer -> - alloca $ \ data_recvd_ptr -> do - result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr- recvd_bytes <- case result of + else+ allocaBytes bytes_to_get $ \ pcharbuffer ->+ alloca $ \ data_recvd_ptr -> do+ result <- recvData wired_ptr pcharbuffer (fromIntegral bytes_to_get) data_recvd_ptr+ recvd_bytes <- case result of r | r == allOk -> peek data_recvd_ptr- | r == badHappened -> do + | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data" B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes) + bestEffortPullAction :: Bool -> IO B.ByteString+ bestEffortPullAction can_wait = do+ already_closed <- readMVar already_closed_mvar+ if already_closed+ then+ throwIO $ TLSLayerGenericProblem "Tried to receive on closed handle"+ else+ allocaBytes useBufferSize $ \ pcharbuffer ->+ alloca $ \ data_recvd_ptr -> do+ result <- recvDataBestEffort+ wired_ptr+ pcharbuffer+ (fromIntegral useBufferSize)+ (fromIntegral . fromEnum $ can_wait)+ 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 :: IO () -- Ensure that the socket and the struct are only closed once closeAction = do b <- readMVar already_closed_mvar- if not b - then do+ unless b $ do modifyMVar_ already_closed_mvar (\ _ -> return True) disposeWiredSession wired_ptr- -- debugM "OpenSSL" "dispose clalled"- else - return ()- return (pushAction, pullAction, closeAction)+ return AttendantCallbacks {+ _pushAction_AtC = pushAction,+ _pullAction_AtC = pullAction,+ _closeAction_AtC = closeAction,+ _bestEffortPullAction_AtC = bestEffortPullAction+ }
hs-src/SecondTransfer/MainLoop/PushPullType.hs view
@@ -1,51 +1,77 @@-{-# LANGUAGE DeriveDataTypeable, ExistentialQuantification #-}+{-# LANGUAGE ExistentialQuantification, TemplateHaskell #-} {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.MainLoop.PushPullType ( PushAction ,PullAction ,Attendant ,CloseAction- ) where + ,AttendantCallbacks(..) + ,pushAction_AtC+ ,pullAction_AtC+ ,closeAction_AtC+ ,bestEffortPullAction_AtC+ ) where ++import Control.Lens+ import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as LB --- | Callback type to push data to a channel. Part of this --- interface is the abstract exception type IOProblem. Throw an --- instance of it from here to notify the session that the connection has --- been broken. There is no way to signal "normal termination", since --- HTTP/2's normal termination can be observed at a higher level when a --- GO_AWAY frame is seen. +-- | Callback type to push data to a channel. Part of this+-- interface is the abstract exception type IOProblem. Throw an+-- instance of it from here to notify the session that the connection has+-- been broken. There is no way to signal "normal termination", since+-- HTTP/2's normal termination can be observed at a higher level when a+-- GO_AWAY frame is seen. type PushAction = LB.ByteString -> IO () +-- | Callback type to pull data from a channel in a best-effort basis.+-- When the first argument is True, the data-providing backend can+-- block if the input buffers are empty and await for new data.+-- Otherwise, it will return immediately with an empty ByteString+type BestEffortPullAction = Bool -> IO B.ByteString+ -- | Callback type to pull data from a channel. The same--- as to PushAction applies to exceptions thrown from --- there. -type PullAction = IO B.ByteString+-- as to PushAction applies to exceptions thrown from+-- there. The first argument is the number of bytes to+-- pull from the medium. Barring exceptions, we always+-- know how many bytes we are expecting with HTTP/2.+type PullAction = Int -> IO B.ByteString --- | Callback that the session calls to realease resources +-- | Callback that the session calls to realease resources -- associated with the channels. Take into account that your -- callback should be able to deal with non-clean shutdowns -- also, for example, if the connection to the remote peer -- is severed suddenly. type CloseAction = IO () --- | A function which takes three arguments: the first one says --- how to send data (on a socket or similar transport), and the second one how --- to receive data on the transport. The third argument encapsulates --- the sequence of steps needed for a clean shutdown. ++data AttendantCallbacks = AttendantCallbacks {+ _pushAction_AtC :: PushAction,+ _pullAction_AtC :: PullAction,+ _bestEffortPullAction_AtC :: BestEffortPullAction,+ _closeAction_AtC :: CloseAction+ }++makeLenses ''AttendantCallbacks++-- | A function which takes three arguments: the first one says+-- how to send data (on a socket or similar transport), and the second one how+-- to receive data on the transport. The third argument encapsulates+-- the sequence of steps needed for a clean shutdown. ----- You can implement one of these to let somebody else supply the --- push, pull and close callbacks. For example, 'tlsServeWithALPN' will --- supply these arguments to an 'Attendant'. +-- You can implement one of these to let somebody else supply the+-- push, pull and close callbacks. For example, 'tlsServeWithALPN' will+-- supply these arguments to an 'Attendant'. -- -- Attendants encapsulate all the session book-keeping functionality,--- which for HTTP/2 is quite complicated. You use the functions +-- which for HTTP/2 is quite complicated. You use the functions -- http**Attendant -- to create one of these from a 'SecondTransfer.Types.CoherentWorker'. -- -- This library supplies two of such Attendant factories,--- 'SecondTransfer.Http1.http11Attendant' for +-- 'SecondTransfer.Http1.http11Attendant' for -- HTTP 1.1 sessions, and 'SecondTransfer.Http2.http2Attendant' for HTTP/2 sessions.-type Attendant = PushAction -> PullAction -> CloseAction -> IO () +type Attendant = AttendantCallbacks -> IO ()
hs-src/SecondTransfer/Sessions.hs view
@@ -2,7 +2,7 @@ makeSessionsContext ,module SecondTransfer.Sessions.Config ,SessionsContext- ) where + ) where import SecondTransfer.Sessions.Config import SecondTransfer.Sessions.Internal
hs-src/SecondTransfer/Sessions/Config.hs view
@@ -1,7 +1,7 @@ {-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}-{- | Configuration and settings for the server. All constructor names are - exported, but notice that they start with an underscore. - They also have an equivalent lens without the +{- | Configuration and settings for the server. All constructor names are+ exported, but notice that they start with an underscore.+ They also have an equivalent lens without the underscore. Please prefer to use the lens interface. -} module SecondTransfer.Sessions.Config(@@ -10,7 +10,9 @@ ,defaultSessionsEnrichedHeaders ,sessionsCallbacks ,sessionsEnrichedHeaders- ,reportErrorCallback+ ,reportErrorCallback_SC+ ,dataDeliveryCallback_SC+ ,dataFrameSize ,addUsedProtocol @@ -21,38 +23,40 @@ -- ,UsedProtocol(..) ,SessionsConfig(..) ,ErrorCallback- ) where + ,DataFrameDeliveryCallback+ ) where -- import Control.Concurrent.MVar (MVar) import Control.Exception (SomeException) import Control.Lens (makeLenses)+import System.Clock (TimeSpec) --- | Information used to identify a particular session. +-- | Information used to identify a particular session. newtype SessionCoordinates = SessionCoordinates Int deriving Show -instance Eq SessionCoordinates where +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 +-- | 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)+sessionId f (SessionCoordinates session_id) =+ fmap SessionCoordinates (f session_id) -- | Components at an individual session. Used to report--- where in the session an error was produced. This interface is likely +-- 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_HTTP2SessionComponent +data SessionComponent =+ SessionInputThread_HTTP2SessionComponent |SessionHeadersOutputThread_HTTP2SessionComponent |SessionDataOutputThread_HTTP2SessionComponent |Framer_HTTP2SessionComponent@@ -61,28 +65,38 @@ -- Which protocol a session is using... no need for this right now--- data UsedProtocol = +-- data UsedProtocol = -- HTTP11_UsP -- |HTTP2_UsP -- | Used by this session engine to report an error at some component, in a particular--- session. +-- session. type ErrorCallback = (SessionComponent, SessionCoordinates, SomeException) -> IO () --- | Callbacks that you can provide your sessions to notify you --- of interesting things happening in the server. +-- | Used by the session engine to report delivery of each data frame. Keep this callback very+-- light, it runs in the main sending thread. It is called as+-- f session_id stream_id ordinal when_delivered+type DataFrameDeliveryCallback = Int -> Int -> Int -> TimeSpec -> IO ()++-- | Callbacks that you can provide your sessions to notify you+-- of interesting things happening in the server. data SessionsCallbacks = SessionsCallbacks { -- Callback used to report errors during this session- _reportErrorCallback :: Maybe ErrorCallback+ _reportErrorCallback_SC :: Maybe ErrorCallback,+ -- Callback used to report delivery of individual data frames+ _dataDeliveryCallback_SC :: Maybe DataFrameDeliveryCallback } makeLenses ''SessionsCallbacks --- | This is a temporal interface, but an useful one nonetheless. +-- | This is a temporal interface, but an useful one nonetheless. -- By setting some values here to True, second-transfer will add--- some headers to inbound requests, and some headers to outbound --- requests. +-- some headers to inbound requests, and some headers to outbound+-- requests.+--+-- This interface is deprecated in favor of the AwareWorker+-- functionality.... data SessionsEnrichedHeaders = SessionsEnrichedHeaders { -- | Adds a second-transfer-eh--used-protocol header -- to inbound requests. Default: False@@ -91,7 +105,7 @@ makeLenses ''SessionsEnrichedHeaders --- | Don't insert any extra-headers by default. +-- | Don't insert any extra-headers by default. defaultSessionsEnrichedHeaders :: SessionsEnrichedHeaders defaultSessionsEnrichedHeaders = SessionsEnrichedHeaders { _addUsedProtocol = False@@ -101,8 +115,10 @@ -- | Configuration information you can provide to the session maker. data SessionsConfig = SessionsConfig { -- | Session callbacks- _sessionsCallbacks :: SessionsCallbacks- ,_sessionsEnrichedHeaders :: SessionsEnrichedHeaders+ _sessionsCallbacks :: SessionsCallbacks+ ,_sessionsEnrichedHeaders :: SessionsEnrichedHeaders+ -- | Size to use when splitting data in data frames+ ,_dataFrameSize :: Int } makeLenses ''SessionsConfig@@ -111,17 +127,18 @@ -- sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks -- sessionsCallbacks f ( -- SessionsConfig {--- _sessionsCallbacks= s +-- _sessionsCallbacks= s -- }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s) --- | Creates a default sessions context. Modify as needed using +-- | Creates a default sessions context. Modify as needed using -- the lenses interfaces defaultSessionsConfig :: SessionsConfig defaultSessionsConfig = SessionsConfig { _sessionsCallbacks = SessionsCallbacks {- _reportErrorCallback = Nothing+ _reportErrorCallback_SC = Nothing,+ _dataDeliveryCallback_SC = Nothing },- _sessionsEnrichedHeaders = defaultSessionsEnrichedHeaders+ _sessionsEnrichedHeaders = defaultSessionsEnrichedHeaders,+ _dataFrameSize = 2048 }-
hs-src/SecondTransfer/Sessions/Internal.hs view
@@ -1,5 +1,5 @@ {-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}-module SecondTransfer.Sessions.Internal where +module SecondTransfer.Sessions.Internal where import SecondTransfer.Sessions.Config @@ -14,10 +14,10 @@ --- | Contains information that applies to all --- sessions created in the program. Use the lenses --- interface to access members of this struct. --- +-- | 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@@ -27,41 +27,41 @@ makeLenses ''SessionsContext --- Session tags are simple session identifiers -acquireNewSessionTag :: SessionsContext -> IO Int -acquireNewSessionTag sessions_context = - modifyMVar +-- Session tags are simple session identifiers+acquireNewSessionTag :: SessionsContext -> IO Int+acquireNewSessionTag sessions_context =+ modifyMVar (sessions_context ^. nextSessionId )- (\ next_id -> return ((next_id+1), next_id))+ (\ next_id -> return (next_id+1, next_id)) --- Adds runtime data to a context, and let it work.... +-- 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 {+makeSessionsContext sessions_config = do+ next_session_id_mvar <- newMVar 1+ return SessionsContext { _sessionsConfig = sessions_config, _nextSessionId = next_session_id_mvar } -sessionExceptionHandler :: +sessionExceptionHandler :: E.Exception e => SessionComponent -> Int -> SessionsContext -> e -> IO ()-sessionExceptionHandler session_component session_id sessions_context e = do +sessionExceptionHandler session_component session_id sessions_context e = let - getit = ( sessionsConfig . sessionsCallbacks . reportErrorCallback ) - maybe_error_callback = sessions_context ^. getit - component_tag = "Session." ++ (show session_component)+ getit = ( sessionsConfig . sessionsCallbacks . reportErrorCallback_SC )+ maybe_error_callback = sessions_context ^. getit+ component_tag = "Session." ++ show session_component error_tuple = ( session_component,- SessionCoordinates session_id, + SessionCoordinates session_id, E.toException e )- case maybe_error_callback of - Nothing -> - errorM component_tag (show (e))+ in case maybe_error_callback of+ Nothing ->+ errorM component_tag (show e) - Just callback -> + Just callback -> callback error_tuple
hs-src/SecondTransfer/Types.hs view
@@ -1,7 +1,7 @@ module SecondTransfer.Types( module SecondTransfer.MainLoop.PushPullType ,module SecondTransfer.MainLoop.CoherentWorker- ) where + ) where -import SecondTransfer.MainLoop.PushPullType +import SecondTransfer.MainLoop.PushPullType import SecondTransfer.MainLoop.CoherentWorker
hs-src/SecondTransfer/Utils.hs view
@@ -11,10 +11,10 @@ ,stripString ,domainFromUrl ,subByteString- ) where + ) where -import Control.Concurrent.Chan+import Control.Concurrent.MVar import Control.Monad.Trans.Class (lift) import Data.Binary (Binary, get, put, putWord8) import Data.Binary.Get (Get, getWord16be, getWord8)@@ -28,7 +28,7 @@ import qualified Network.URI as U -strToInt::String -> Int +strToInt::String -> Int strToInt = fromIntegral . toInteger . (read::String->Integer) @@ -36,77 +36,77 @@ deriving (Show) -word24ToInt :: Word24 -> Int +word24ToInt :: Word24 -> Int word24ToInt (Word24 w24) = w24 instance Binary Word24 where - put (Word24 w24) = - do - let - high_stuff = w24 `shiftR` 24 - low_stuff = w24 `mod` (1 `shiftL` 24) + put (Word24 w24) =+ do+ let+ high_stuff = w24 `shiftR` 24+ low_stuff = w24 `mod` (1 `shiftL` 24) putWord8 $ fromIntegral high_stuff- putWord16be $ fromIntegral low_stuff + putWord16be $ fromIntegral low_stuff get = do- high_stuff <- getWord8 + high_stuff <- getWord8 low_stuff <- getWord16be- let - value = (fromIntegral low_stuff) + ( (fromIntegral high_stuff) `shiftL` 24 ) + let+ value = (fromIntegral low_stuff) + ( (fromIntegral high_stuff) `shiftL` 24 ) return $ Word24 value getWord24be :: Get Int-getWord24be = do +getWord24be = do w24 <- get return $ word24ToInt w24 -putWord24be :: Int -> Put +putWord24be :: Int -> Put putWord24be x = put (Word24 x) lowercaseText :: B.ByteString -> B.ByteString-lowercaseText bs0 = - encodeUtf8 ts1 - where - ts1 = T.toLower ts0 +lowercaseText bs0 =+ encodeUtf8 ts1+ where+ ts1 = T.toLower ts0 ts0 = decodeUtf8 bs0 -unfoldChannelAndSource :: IO (Chan (Maybe a), Source IO a)-unfoldChannelAndSource = do - chan <- newChan - let - source = do - e <- lift $ readChan chan- case e of +unfoldChannelAndSource :: IO (MVar (Maybe a), Source IO a)+unfoldChannelAndSource = do+ chan <- newEmptyMVar+ let+ source = do+ e <- lift $ takeMVar chan+ case e of Just ee -> do yield ee- source + source - Nothing -> + Nothing -> return () return (chan, source) -stripString :: String -> String +stripString :: String -> String stripString = filter $ \ ch -> (ch /= '\n') && ( ch /= ' ') domainFromUrl :: B.ByteString -> B.ByteString-domainFromUrl url = let +domainFromUrl url = let Just (U.URI {- scheme -} _ authority _ _ _) = U.parseURI $ unpack url Just (U.URIAuth _ use_host _) = authority- in + in pack use_host -- Returns the sub-bytestring that starts at start_pos and ends -- just before end_pos subByteString :: Int -> Int -> B.ByteString -> B.ByteString-subByteString start_pos end_pos = - B.take (end_pos - start_pos ) . B.drop start_pos +subByteString start_pos end_pos =+ B.take (end_pos - start_pos ) . B.drop start_pos
hs-src/SecondTransfer/Utils/DevNull.hs view
@@ -1,24 +1,30 @@-+{-# LANGUAGE OverloadedStrings #-} module SecondTransfer.Utils.DevNull( dropIncomingData- ) where + ) where import Control.Concurrent (forkIO)-import Data.Conduit +import Control.Monad.IO.Class (liftIO) -import SecondTransfer.MainLoop.CoherentWorker +import Data.Conduit --- TODO: Handling unnecessary data should be done in some other, less +import SecondTransfer.MainLoop.CoherentWorker+import SecondTransfer.MainLoop.Logging(logit)+++-- TODO: Handling unnecessary data should be done in some other, less -- harmfull way... need to think about that. -- | If you are not processing the potential POST input in a request,--- use this consumer to drop the data to oblivion. Otherwise it will --- remain in an internal queue until the client closes the +-- use this consumer to drop the data to oblivion. Otherwise it will+-- remain in an internal queue until the client closes the -- stream, and if the client doesn't want to do so.... dropIncomingData :: Maybe InputDataStream -> IO () dropIncomingData Nothing = return () dropIncomingData (Just data_source) = do- forkIO $ - data_source $$ (awaitForever (\ _ -> return () ) )+ forkIO $+ data_source $$ awaitForever (\ _ -> do+ liftIO $ logit "stream discarded"+ return () ) return ()
hs-src/SecondTransfer/Utils/HTTPHeaders.hs view
@@ -1,14 +1,14 @@ {-# LANGUAGE OverloadedStrings, Rank2Types #-} {-| -Utilities for working with headers. +Utilities for working with headers. -} module SecondTransfer.Utils.HTTPHeaders ( -- * Simple manipulation- -- + -- -- | These transformations are simple enough that don't require- -- going away from the list representation (see type `Headers`) + -- going away from the list representation (see type `Headers`) lowercaseHeaders ,headersAreLowercase ,headersAreLowercaseAtHeaderEditor@@ -16,22 +16,22 @@ -- * Transformations based on maps -- -- | Many operations benefit- -- from transforming the list to a map with custom sorting and + -- from transforming the list to a map with custom sorting and -- doing a set of operations on that representation. -- ,HeaderEditor ,Headers -- ** Introducing and removing the `HeaderEditor`- ,fromList - ,toList + ,fromList+ ,toList -- ** Access to a particular header ,headerLens ,replaceHeaderValue -- ** HTTP utilities ,replaceHostByAuthority ,introduceDateHeader- ) where - + ) where+ import qualified Control.Lens as L import Control.Lens ( (^.) ) @@ -57,86 +57,86 @@ import SecondTransfer.MainLoop.CoherentWorker (Headers) -- Why having a headers library? The one at Network.HTTP.Headers works--- with Strings and is not very friendly to custom headers. +-- with Strings and is not very friendly to custom headers. -- This is a very basic, lightweight normalizer. --- | HTTP headers are case-insensitive, so we can use lowercase versions +-- | HTTP headers are case-insensitive, so we can use lowercase versions -- everywhere lowercaseHeaders :: Headers -> Headers lowercaseHeaders = map (\(h,v) -> (low h, v))- where + where low = encodeUtf8 . toLower . decodeUtf8 -- | Checks that headers are lowercase-headersAreLowercase :: Headers -> Bool -headersAreLowercase headers = +headersAreLowercase :: Headers -> Bool+headersAreLowercase headers = foldl (\ prev (hn, _) -> (flip (&&)) (aTitleIsLowercase hn) $! prev) True headers -headersAreLowercaseAtHeaderEditor :: HeaderEditor -> Bool -headersAreLowercaseAtHeaderEditor header_editor = +headersAreLowercaseAtHeaderEditor :: HeaderEditor -> Bool+headersAreLowercaseAtHeaderEditor header_editor = Ms.foldlWithKey' (\ prev hn _ -> (flip (&&)) (aTitleIsLowercase . toFlatBs $ hn) $! prev) True (innerMap header_editor) -aTitleIsLowercase :: B.ByteString -> Bool +aTitleIsLowercase :: B.ByteString -> Bool aTitleIsLowercase a_title = not . T.any isUpper . decodeUtf8 $ a_title -- | Looks for a given header fetchHeader :: Headers -> B.ByteString -> Maybe B.ByteString-fetchHeader headers header_name = - snd - <$> - find ( \ x -> fst x == header_name ) headers +fetchHeader headers header_name =+ snd+ <$>+ find ( \ x -> fst x == header_name ) headers newtype Autosorted = Autosorted { toFlatBs :: B.ByteString } deriving Eq -colon :: Word8 +colon :: Word8 colon = fromIntegral . fromEnum $ ':' -instance Ord Autosorted where - compare (Autosorted a) (Autosorted b) | (B.head a) == colon, (B.head b) /= colon = LT - compare (Autosorted a) (Autosorted b) | (B.head a) /= colon, (B.head b) == colon = GT +instance Ord Autosorted where+ compare (Autosorted a) (Autosorted b) | (B.head a) == colon, (B.head b) /= colon = LT+ compare (Autosorted a) (Autosorted b) | (B.head a) /= colon, (B.head b) == colon = GT compare (Autosorted a) (Autosorted b) = compare a b --- | Abstract data-type. Use `fromList` to get one of these from `Headers`. +-- | Abstract data-type. Use `fromList` to get one of these from `Headers`. -- The underlying representation admits better asymptotics. newtype HeaderEditor = HeaderEditor { innerMap :: Ms.Map Autosorted B.ByteString } -- This is a pretty uninteresting instance-instance Monoid HeaderEditor where +instance Monoid HeaderEditor where mempty = HeaderEditor mempty mappend (HeaderEditor x) (HeaderEditor y) = HeaderEditor (x `mappend` y) --- | /O(n*log n)/ Builds the editor from a list. -fromList :: Headers -> HeaderEditor +-- | /O(n*log n)/ Builds the editor from a list.+fromList :: Headers -> HeaderEditor fromList = HeaderEditor . Ms.fromList . map (\(hn, hv) -> (Autosorted hn, hv)) -- | /O(n)/ Takes the HeaderEditor back to Headers-toList :: HeaderEditor -> Headers +toList :: HeaderEditor -> Headers toList (HeaderEditor m) = [ (toFlatBs x, v) | (x,v) <- Ms.toList m ] --- | replaceHeaderValue headers header_name maybe_header_value looks for --- header_name. If header_name is found and maybe_header_value is nothing, it +-- | replaceHeaderValue headers header_name maybe_header_value looks for+-- header_name. If header_name is found and maybe_header_value is nothing, it -- returns a new headers list with the header deleted. If header_name is found--- and header_value is Just new_value, it returns a new list with the header +-- and header_value is Just new_value, it returns a new list with the header -- containing the new value. If header_name is not in headers and maybe_header_value -- is Nothing, it returns the original headers list. If header_name is not in headers -- and maybe_header_value is Just new_value, it returns a new list where the last element -- is (header_name, new_value) replaceHeaderValue :: HeaderEditor -> B.ByteString -> Maybe B.ByteString -> HeaderEditor-replaceHeaderValue (HeaderEditor m) header_name maybe_header_value = +replaceHeaderValue (HeaderEditor m) header_name maybe_header_value = HeaderEditor $ Ms.alter (const maybe_header_value) (Autosorted header_name) m @@ -144,43 +144,43 @@ -- and you can use it then to add, alter and remove headers. -- It uses the same semantics than `replaceHeaderValue` headerLens :: B.ByteString -> L.Lens' HeaderEditor (Maybe B.ByteString)-headerLens name = - L.lens - (Ms.lookup hname . innerMap ) +headerLens name =+ L.lens+ (Ms.lookup hname . innerMap ) (\(HeaderEditor hs) mhv -> HeaderEditor $ Ms.alter (const mhv) hname hs)- where + where hname = Autosorted name --- | Replaces a \"host\" HTTP\/1.1 header by an ":authority" HTTP\/2 +-- | Replaces a \"host\" HTTP\/1.1 header by an ":authority" HTTP\/2 -- header. -- The list is expected to be already in lowercase, so nothing will happen if there -- the header name portion is \"Host\" instead of \"host\". ----- Notice that having a "Host" header in an HTTP\/2 message is perfectly valid in certain +-- Notice that having a "Host" header in an HTTP\/2 message is perfectly valid in certain -- circumstances, check <https://http2.github.io/http2-spec/#rfc.section.8.1.2.3 Section 8.1.2.3> -- of the spec for details. replaceHostByAuthority :: HeaderEditor -> HeaderEditor-replaceHostByAuthority headers = +replaceHostByAuthority headers = let host_lens :: L.Lens' HeaderEditor (Maybe B.ByteString) host_lens = headerLens "host" authority_lens = headerLens ":authority" maybe_host_header = headers ^. host_lens no_hosts = L.set host_lens Nothing headers- in - case maybe_host_header of - Nothing -> headers + in+ case maybe_host_header of+ Nothing -> headers Just host -> L.set authority_lens (Just host) no_hosts --- | Given a header editor, introduces a "Date" header. This function has +-- | Given a header editor, introduces a "Date" header. This function has -- a side-effect: to get the current time introduceDateHeader :: HeaderEditor -> IO HeaderEditor introduceDateHeader header_editor = do current_time <- getCurrentTime- let + let date_header_lens = headerLens "date" formatted_date = Just . pack $ formatTime defaultTimeLocale "%a, %d %b %Y %H:%M:%S %Z" current_time- new_editor = L.set date_header_lens formatted_date header_editor + new_editor = L.set date_header_lens formatted_date header_editor return new_editor
macros/Logging.cpphs view
@@ -3,9 +3,11 @@ #ifdef ENABLE_DEBUG #define INSTRUMENTATION(x) (liftIO $ logWithExclusivity (x))+#define LOGIT_SWITCH_TIMINGS 1 #else #define INSTRUMENTATION(x) (return ())+#define LOGIT_SWITCH_TIMINGS 0 -#endif +#endif
second-transfer.cabal view
@@ -7,7 +7,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version : 0.5.5.1+version : 0.6.0.0 synopsis : Second Transfer HTTP/2 web server @@ -53,7 +53,7 @@ source-repository this type: git location: git@github.com:alcidesv/second-transfer.git- tag: 0.5.5.1+ tag: 0.6.0.0 library @@ -121,10 +121,15 @@ network-uri >= 2.6 && < 2.7, hashtables >= 1.2 && < 1.3, lens >= 4.7 ,- http2 >= 0.7 && <= 0.9.1,+ http2 >= 1.0.2, hslogger >= 1.2.6, hashable >= 1.2, attoparsec >= 0.12,+ clock >= 0.4,+ SafeSemaphore >= 0.10,+ pqueue >= 1.3.0,+ stm >= 2.3,+ deepseq >= 1.4.1, time >= 1.5.0 && < 1.8 -- Directories containing source files.@@ -181,7 +186,7 @@ ,lens >= 4.7 ,HUnit >= 1.2 && < 1.5 ,bytestring >= 0.10.4.0- ,http2 == 0.9.1+ ,http2 ,exceptions >= 0.8 && < 0.9 ,base16-bytestring >= 0.1.1 ,network >= 2.6 && < 2.7@@ -191,10 +196,16 @@ ,transformers >=0.3 && <= 0.5 ,network-uri >= 2.6 && < 2.7 ,hashtables >= 1.2 && < 1.3+ ,unordered-containers ,hslogger >= 1.2.6 ,hashable >= 1.2 ,attoparsec >= 0.12 ,time >= 1.5.0 && < 1.8+ ,SafeSemaphore >= 0.10+ ,pqueue >= 1.3.0+ ,clock >= 0.4+ ,stm >= 2.3+ ,deepseq >= 1.4.1 build-tools : cpphs default-extensions : CPP include-dirs : macros/
tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs view
@@ -1,6 +1,6 @@ {-# LANGUAGE TemplateHaskell, OverloadedStrings #-} -module SecondTransfer.Test.DecoySession where +module SecondTransfer.Test.DecoySession where @@ -14,7 +14,7 @@ import qualified Network.HTTP2 as NH2 import qualified Network.HPACK as HP -import qualified Data.ByteString as B +import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as LB import SecondTransfer.Http2@@ -29,7 +29,7 @@ ,_remainingOutputBit:: MVar B.ByteString ,_nh2Settings :: NH2.Settings ,_sessionThrowed :: MVar Bool- ,_waiting :: MVar ThreadId + ,_waiting :: MVar ThreadId ,_encodeHeadersHere :: MVar HP.DynamicTable ,_decodeHeadersHere :: MVar HP.DynamicTable }@@ -41,16 +41,22 @@ -- Supposed to be an HTTP/2 attendant createDecoySession :: Attendant -> IO DecoySession createDecoySession attendant = do- input_data_channel <- newChan + input_data_channel <- newChan output_data_channel <- newChan- let + let push_action :: PushAction push_action = writeChan output_data_channel- pull_action :: PullAction + pull_action :: PullAction pull_action = readChan input_data_channel- close_action :: CloseAction - close_action = do- return ()+ close_action :: CloseAction+ best_effort_pull_action _ = error "NotImplemented"+ close_action = return ()+ attendant_callbacks = AttendantCallbacks {+ _pushAction_AtC = push_action,+ _pullAction_AtC = pull_action,+ _closeAction_AtC = close_action,+ _bestEffortPullAction_AtC = best_effort_pull_action+ } dtable_for_encoding <- HP.newDynamicTableForEncoding 4096 dtable_for_encoding_mvar <- newMVar dtable_for_encoding@@ -58,15 +64,15 @@ dtable_for_decoding_mvar <- newMVar dtable_for_decoding waiting_mvar <- newEmptyMVar- - thread_id <- forkIO $ catch - (do ++ thread_id <- forkIO $ catch+ (do attendant push_action pull_action close_action )- ((\ e -> do + ((\ e -> do putStrLn $ "Exception: " ++ (show e) maybe_waiting <- tryTakeMVar waiting_mvar- case maybe_waiting of + case maybe_waiting of Just thread_id -> throwTo thread_id e Nothing -> return () ):: SomeException -> IO ())@@ -90,25 +96,25 @@ sessionIsUndone session = error "NotImplemented" --- Send a frame to a session +-- Send a frame to a session sendFrameToSession :: DecoySession -> OutputFrame -> IO ()-sendFrameToSession session (encode_info, frame_payload) = do - let +sendFrameToSession session (encode_info, frame_payload) = do+ let bs_list = NH2.encodeFrameChunks encode_info frame_payload input_data_channel = session ^. inputDataChannel - mapM_ (\ x -> writeChan input_data_channel x ) bs_list+ writeChan2Chan input_data_channel bs_list --- Read a frame from a session... if possible. It will block --- until the frame comes out, but should fail if there is +-- Read a frame from a session... if possible. It will block+-- until the frame comes out, but should fail if there is -- an exception and/or the session is closed. recvFrameFromSession :: DecoySession -> IO (Maybe NH2.Frame) recvFrameFromSession decoy_session = do- let + let output_data_channel = decoy_session ^. outputDataChannel remaining_output_bit_mvar = decoy_session ^. remainingOutputBit- pull_action = fmap LB.toStrict $ readChan output_data_channel + pull_action = fmap LB.toStrict $ readChan output_data_channel settings = decoy_session ^. nh2Settings waiting_for_read = decoy_session ^. waiting my_thread_id <- myThreadId@@ -117,46 +123,46 @@ (packet, rest) <- readNextChunkAndContinue http2FrameLength remaining_output_bit pull_action putMVar remaining_output_bit_mvar rest takeMVar waiting_for_read- let + let error_or_frame = NH2.decodeFrame settings packet- case error_or_frame of - Left _ -> return Nothing + case error_or_frame of+ Left _ -> return Nothing Right frame -> return $ Just frame --- Encode headers to send to the session +-- Encode headers to send to the session -- TODO: There is an important bug here... we are using the default encoding -- strategy everywhere encodeHeadersForSession :: DecoySession -> Headers -> IO B.ByteString-encodeHeadersForSession decoy_session headers = - do - let - mv = (decoy_session ^. encodeHeadersHere )- dtable <- takeMVar mv- (dtable', bs ) <- HP.encodeHeader HP.defaultEncodeStrategy dtable headers - putMVar mv dtable'- return bs+encodeHeadersForSession decoy_session headers =+ do+ let+ mv = decoy_session ^. encodeHeadersHere+ dtable <- takeMVar mv+ (dtable', bs ) <- HP.encodeHeader HP.defaultEncodeStrategy dtable headers+ putMVar mv dtable'+ return bs --- Decode headers to receive them from the session -decodeHeadersForSession :: DecoySession -> B.ByteString -> IO Headers -decodeHeadersForSession decoy_session bs = - do - let - mv = (decoy_session ^. decodeHeadersHere )- dtable <- takeMVar mv - (dtable', headers) <- HP.decodeHeader dtable bs- putMVar mv dtable'- return headers+-- Decode headers to receive them from the session+decodeHeadersForSession :: DecoySession -> B.ByteString -> IO Headers+decodeHeadersForSession decoy_session bs =+ do+ let+ mv = decoy_session ^. decodeHeadersHere+ dtable <- takeMVar mv+ (dtable', headers) <- HP.decodeHeader dtable bs+ putMVar mv dtable'+ return headers --- Send raw data to a session +-- Send raw data to a session sendRawDataToSession :: DecoySession -> B.ByteString -> IO () sendRawDataToSession decoy_session data_to_send = do- let + let input_data_channel = decoy_session ^. inputDataChannel waiting_for_write = decoy_session ^. waiting thread_id <- myThreadId- putMVar waiting_for_write thread_id + putMVar waiting_for_write thread_id writeChan input_data_channel data_to_send takeMVar waiting_for_write return ()@@ -164,22 +170,22 @@ -- Send a headers frame to the session that ends the stream... performRequestSimple :: DecoySession -> Int -> Headers -> IO ()-performRequestSimple decoy_session stream_id headers = do - headers_data <- encodeHeadersForSession decoy_session headers - let - frame = NH2.HeadersFrame +performRequestSimple decoy_session stream_id headers = do+ headers_data <- encodeHeadersForSession decoy_session headers+ let+ frame = NH2.HeadersFrame Nothing -- No priority- headers_data + headers_data frame_data_ei = NH2.EncodeInfo { NH2.encodeFlags = NH2.setEndStream . NH2.setEndHeader $ NH2.defaultFlags,- NH2.encodeStreamId = NH2.toStreamIdentifier stream_id,+ NH2.encodeStreamId = stream_id, NH2.encodePadding = Nothing }- + sendFrameToSession decoy_session (frame_data_ei,frame) -- Read raw data from a session. Normally blocks until data be available.--- It returns Nothing when the session is to be considered finished. +-- It returns Nothing when the session is to be considered finished. recvRawDataFromSession :: DecoySession -> IO (Maybe B.ByteString) recvRawDataFromSession decoy_session = error "NotImplemented"
tests/tests-hs-src/compiling_ok.hs view
@@ -1,12 +1,13 @@ {-# LANGUAGE OverloadedStrings #-} import SecondTransfer(- CoherentWorker+ AwareWorker , Footers , DataAndConclusion , tlsServeWithALPNAndFinishOnRequest , http2Attendant , http11Attendant , dropIncomingData+ , coherentToAwareWorker , FinishRequest(..) ) import SecondTransfer.Sessions(@@ -16,17 +17,17 @@ import Data.Conduit import Control.Concurrent (threadDelay, forkIO)-import Control.Concurrent.MVar +import Control.Concurrent.MVar saysHello :: DataAndConclusion-saysHello = do +saysHello = do yield "Hello world!" -- No footers return [] -helloWorldWorker :: CoherentWorker-helloWorldWorker (_request_headers, _maybe_post_data) = do +helloWorldWorker :: AwareWorker+helloWorldWorker = coherentToAwareWorker $ \ (_request_headers, _maybe_post_data) -> do dropIncomingData _maybe_post_data return ( [@@ -37,17 +38,17 @@ ) --- For this program to work, it should be run from the top of +-- For this program to work, it should be run from the top of -- the developement directory.-main = do +main = do sessions_context <- makeSessionsContext defaultSessionsConfig finish <- newEmptyMVar- -- Make the server work only for small amount of time, so that + -- Make the server work only for small amount of time, so that -- continue running other tests- forkIO $ do + forkIO $ do threadDelay 1000000- putMVar finish FinishRequest - let + putMVar finish FinishRequest+ let http2_attendant = http2Attendant sessions_context helloWorldWorker http11_attendant = http11Attendant sessions_context helloWorldWorker tlsServeWithALPNAndFinishOnRequest@@ -56,7 +57,7 @@ "127.0.0.1" -- On which interface to bind [ ("h2-14", http2_attendant), -- Protocols present in the ALPN negotiation- ("h2", http2_attendant), -- they may be slightly different, but for this + ("h2", http2_attendant), -- they may be slightly different, but for this -- test it doesn't matter. ("http/1.1", http11_attendant) ]