[WIP] Extending the capabilities of the HTTP/2.0 server

[dhdavvie]

Setting up this PR now to get some feedback on this. I have extended the functionality of the the H2 server by improving the multithreading aspect mostly. To summarise, when the Http2WebTestRequestHandler receives a frame on a new stream, it creates a thread for that stream specifically and uses a Queue to pass frames from the h2 handler to the individual stream handlers.

In the stream threads, when it receives a RequestReceived frame it creates a H2Request straight away along with a H2Response object and begins handling those in a new thread. This is because at this point there is enough information to begin parsing and checking for invalid requests, mostly giving the ability to check for invalid routes without needing to wait for Data frames (#7693). In the instance of POST requests where Data frames are expected, I set up an os.pipe() pair of file-like objects, and use this to write the data received in the stream handler to the rfile that the request handler will read POST data from. Reason for using os.pipe() is that it ended up being the most appropriate file-like object for the scenario, as other ones like BytesIO didn't provide the blocking and EOF characteristics in a way that allows me to leave the request handler side untouched.

The main reason for this PR is to get some feedback about what the API should look like. In theory (tests to come), someone should currently be able to write a test where a handler can create frames using the H2 Connection object and write them. I want to add something where the handler could create bogus frames (HPACK encoding might make this tricky for headers) and write those. Please let me know what you think this might look like, and any other functionality that should be exposed.

TL;DR:

• Made multithreading more robust, and made it so each stream gets its
  own thread

• Requests start getting parsed immediately and do not wait for the
  entire request to get there. This helps with 404 detection early on and
  other use cases.

• Created H2Request object

• Implemented basic Push Promise functionality

---

This change is 

[jgraham]

Reviewed 3 of 3 files at r1.
Reviewable status: all files reviewed, 8 unresolved discussions (waiting on @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/response.py, line 445 at r1 (raw file):

    def write_push(self, promise_headers, response_headers, status, data=None, stream_id=None):
        with self.h2conn as connection:
            connection.push_stream(self.request.h2_stream_id, stream_id, promise_headers)

Does it make sense that stream_id can be None here?

---

tools/wptserve/wptserve/response.py, line 448 at r1 (raw file):

            self.write(connection)

        has_data = data is not None

Does it make sense to automatically send the actual response here?

---

tools/wptserve/wptserve/server.py, line 15 at r1 (raw file):

from collections import OrderedDict

if sys.version[0] == '2':

You can use six.moves here instead of the explicit conditional.

---

tools/wptserve/wptserve/server.py, line 361 at r1 (raw file):

                for frame in frames:
                    if isinstance(frame, (RequestReceived, DataReceived, StreamEnded, StreamReset)):

Do we actually just want to special case ConnectionTerminated and otherwise take this branch as an else?

---

tools/wptserve/wptserve/server.py, line 364 at r1 (raw file):

                        if frame.stream_id not in self.stream_queues:
                            self.stream_queues[frame.stream_id] = Queue()
                            self.start_stream_thread(frame)

Pass the Queue in here and maybe the stream id.

---

tools/wptserve/wptserve/server.py, line 407 at r1 (raw file):

            if isinstance(frame, RequestReceived):
                # Create a shallow copy of this instance, to avoid issues with multiple threads editing the same object
                self_copy = copy(self)

So I'm not super-happy about this. It might be better to have a special method that returns an object we can pass into this thread with just the relevant data. OR create the Request and Response objects in the parent thread and pass it in

---

tools/wptserve/wptserve/server.py, line 411 at r1 (raw file):

                # If its a POST request, we expect data frames and open the pipe to the request object
                if self_copy.command == 'POST':

So I don't really like the idea of specialising on POST here. Surely we can just handle the data frame when we actually get it rather than making asumptions based on the method.

---

tools/wptserve/wptserve/server.py, line 423 at r1 (raw file):

                # Begin processing the response in another thread,
                # and continue listening for more data here in this one
                t = threading.Thread(

So I'm not clear why this happens in a separate thread. Can't it just be in the same thread as the request handler for this stream?

[dhdavvie]

Reviewable status: all files reviewed, 8 unresolved discussions (waiting on @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/response.py, line 445 at r1 (raw file):

Previously, jgraham wrote…

Does it make sense that stream_id can be None here?

Nope, leftover from a previous iteration, will remove

---

tools/wptserve/wptserve/response.py, line 448 at r1 (raw file):

Previously, jgraham wrote…

Does it make sense to automatically send the actual response here?

So I think there should be a default behaviour, which may be to send the push promise along with the actual content, but I agree that it should also be a possibility to delay and choose when to actually send the data

---

tools/wptserve/wptserve/server.py, line 15 at r1 (raw file):

Previously, jgraham wrote…

You can use six.moves here instead of the explicit conditional.

Good to know, will look into that

---

tools/wptserve/wptserve/server.py, line 361 at r1 (raw file):

Previously, jgraham wrote…

Do we actually just want to special case ConnectionTerminated and otherwise take this branch as an else?

That makes much more sense, it was like this due to previously having multiple branches at this level, now those sit in the stream handler

---

tools/wptserve/wptserve/server.py, line 364 at r1 (raw file):

Previously, jgraham wrote…

Pass the Queue in here and maybe the stream id.

Is there a need to pass the stream id if it is in the frame?

---

tools/wptserve/wptserve/server.py, line 407 at r1 (raw file):

Previously, jgraham wrote…

So I'm not super-happy about this. It might be better to have a special method that returns an object we can pass into this thread with just the relevant data. OR create the Request and Response objects in the parent thread and pass it in

I agree this is super messy and not at all ideal, was just something to get it working quickly and come back to. I like the idea of having a method that creates a pseudo handler to pass along.

---

tools/wptserve/wptserve/server.py, line 411 at r1 (raw file):

Previously, jgraham wrote…

So I don't really like the idea of specialising on POST here. Surely we can just handle the data frame when we actually get it rather than making asumptions based on the method.

I suppose it is unnecessary, I guess I was just trying to avoid creating a pipe unless it was needed, but either way works.

---

tools/wptserve/wptserve/server.py, line 423 at r1 (raw file):

Previously, jgraham wrote…

So I'm not clear why this happens in a separate thread. Can't it just be in the same thread as the request handler for this stream?

Its mainly to avoid having to alter other bits of code further on. If it were to happen in the same thread, you would need to check the queue again for Data frames (in the scenario of a POST request) somewhere down the line. Here, the stream thread is in charge of checking for frames in the queue and forwards them to the request using the pipe, allowing the request to function without needed special checks for H2.

[jgraham]

Reviewed 2 of 7 files at r4, 3 of 9 files at r5, 1 of 2 files at r6.
Reviewable status: 65 of 71 files reviewed, 5 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/handlers.py, line 235 at r5 (raw file):

        return "<%s base_path:%s url_base:%s>" % (self.__class__.__name__, self.base_path, self.url_base)

    def read_file_and_run(self, request, response, func):

I think this doesn't work, because at the time we call the handler we've reset sys.path and sys.modules.

---

tools/wptserve/wptserve/handlers.py, line 266 at r5 (raw file):

    def generate_handler(self, request):

So, I was thinking that we could make this look like

class PythonScripHandler(object):
    def __init__(self, base_path, url_base):
        self.handler_obj = None

   def __call__(self, request, response):
        # This finlizes the response
        if self.handler_obj is None:
            self.handle_setup(request, response)
        with self.handler_env:
            self.handler_obj(request, resp)

   def handle_setup(self, request, resp):
       # Load the .py file, and look for a function or class called main or Handler
      # Store this in self.handler_obj

   def handle_headers(self, request, resp):
       if hasattr(self.handler_obj, "handle_data"):
           with self.handler_env:
               self.handler_obj.handle_headers(request, response)
[…]

Where handler_env is a context manager that sets sys.path and handles the possibility of an IOError (although it occurs to me now that this sys.path manipulation is very non-threadsafe).

What do you think?

---

tools/wptserve/wptserve/server.py, line 427 at r3 (raw file):

Previously, dhdavvie (David H) wrote…

Does it not make sense for the child thread to do this, since otherwise the parent thread would send the connection terminated frame, and might delete the queue before the child thread processes it, unless I make the parent thread sit and wait for the queue to empty/join?

Oh right, so I was also thinking that it would just store the queue as a local variable in the thread rather than accessing it through the dict, so it can't be deleted by the parent.

---

tools/wptserve/wptserve/server.py, line 418 at r5 (raw file):

                req_handler = stream_handler.server.router.get_handler(request)

                if isinstance(req_handler, PythonScriptHandler):

Instead of using isinstance here use if hasattr(req_handler, "generate_handler") so we're defining a protocol that any handler can implement. I'm also not suer-sold on the name and might perfer something like init or create or handle_new_stream or something.

---

tools/wptserve/wptserve/utils.py, line 116 at r3 (raw file):

Previously, dhdavvie (David H) wrote…

I believe I do since the continuation is not actually within a set of brackets. My editor yells at me otherwise

Oh, then add one more set of parens :) Continuation lines are kind of terrible and can usually be avoided.

[dhdavvie]

Reviewable status: 65 of 71 files reviewed, 5 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/handlers.py, line 235 at r5 (raw file):

Previously, jgraham wrote…

I think this doesn't work, because at the time we call the handler we've reset sys.path and sys.modules.

I believe it works, because the func is run whilst the path is in this modified state, and the path is only reset after func is finished.

---

tools/wptserve/wptserve/handlers.py, line 266 at r5 (raw file):

Previously, jgraham wrote…

So, I was thinking that we could make this look like

class PythonScripHandler(object):
    def __init__(self, base_path, url_base):
        self.handler_obj = None

   def __call__(self, request, response):
        # This finlizes the response
        if self.handler_obj is None:
            self.handle_setup(request, response)
        with self.handler_env:
            self.handler_obj(request, resp)

   def handle_setup(self, request, resp):
       # Load the .py file, and look for a function or class called main or Handler
      # Store this in self.handler_obj

   def handle_headers(self, request, resp):
       if hasattr(self.handler_obj, "handle_data"):
           with self.handler_env:
               self.handler_obj.handle_headers(request, response)
[…]

Where handler_env is a context manager that sets sys.path and handles the possibility of an IOError (although it occurs to me now that this sys.path manipulation is very non-threadsafe).

What do you think?

The issue I see with this is that it can cause issues when running multiple tests since PythonScriptHandler is an object shared between all tests and is a generator for FunctionHandler objects. It would feel like bad practice to turn it from a generator to now storing handler attributes for a single handler instance. For example, what does the server do if two different connections request .py resources at the same time? In your proposal, the second .py resource called would overwrite the handle_headers attribute, or see that there is already a handler object and use those.

To implement this, PythonScriptHandler would need to be changed to be a class that you instantiate a new object for every .py resource called.

---

tools/wptserve/wptserve/server.py, line 427 at r3 (raw file):

Previously, jgraham wrote…

Oh right, so I was also thinking that it would just store the queue as a local variable in the thread rather than accessing it through the dict, so it can't be deleted by the parent.

That's a very good point, I'll change it so that the stream thread has its own local variable of the Queue!

---

tools/wptserve/wptserve/server.py, line 418 at r5 (raw file):

Previously, jgraham wrote…

Instead of using isinstance here use if hasattr(req_handler, "generate_handler") so we're defining a protocol that any handler can implement. I'm also not suer-sold on the name and might perfer something like init or create or handle_new_stream or something.

That's a good idea! I believe the name is fairly suitable since PythonScriptHandler is a class for generating handlers, but the name can obviously be changed if need be.

---

tools/wptserve/wptserve/utils.py, line 116 at r3 (raw file):

Previously, jgraham wrote…

Oh, then add one more set of parens :) Continuation lines are kind of terrible and can usually be avoided.

Done.

[jgraham]

Reviewed 1 of 8 files at r3, 3 of 9 files at r5, 1 of 2 files at r6, 4 of 4 files at r7.
Reviewable status: all files reviewed, 4 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/tests/functional/test_response.py, line 339 at r7 (raw file):

    def test_raw_header_frame(self):

Remove all the blank lines between the signature and the method body.

---

tools/wptserve/wptserve/handlers.py, line 277 at r7 (raw file):

    def convert_to_h2_handler(self, request):

Can we rename this something that doesn't specifically mention h2 since it seems at least plasible we'll reuse this infrastructure for some other protcol in the future. frame_handler or something would be fine.

---

tools/wptserve/wptserve/handlers.py, line 294 at r7 (raw file):

                handler.func = environ["main"]
            if "handle_headers" in environ:
                handler.handle_headers = environ["handle_headers"]

So, API-wise this means that there's no way to just return data from one of these functions and have it do something. Maybe that's OK, I'm not sure (don't change it in this PR though).

---

tools/wptserve/wptserve/server.py, line 427 at r3 (raw file):

Previously, dhdavvie (David H) wrote…

That's a very good point, I'll change it so that the stream thread has its own local variable of the Queue!

Are you going to move the removal from stream_queues to the parent?

---

tools/wptserve/wptserve/server.py, line 344 at r7 (raw file):

        self.request.sendall(data)

        self.request_threads = []

This and stream_queues could be local variables, I think. I know we aren't always great at it, but reducing shared state is good!

[dhdavvie]

Reviewable status: all files reviewed, 4 unresolved discussions (waiting on @dhdavvie, @jgraham, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/tests/functional/test_response.py, line 339 at r7 (raw file):

Previously, jgraham wrote…

Remove all the blank lines between the signature and the method body.

Ok

---

tools/wptserve/wptserve/handlers.py, line 277 at r7 (raw file):

Previously, jgraham wrote…

Can we rename this something that doesn't specifically mention h2 since it seems at least plasible we'll reuse this infrastructure for some other protcol in the future. frame_handler or something would be fine.

Sure, making it more general would be a good idea

---

tools/wptserve/wptserve/handlers.py, line 294 at r7 (raw file):

Previously, jgraham wrote…

So, API-wise this means that there's no way to just return data from one of these functions and have it do something. Maybe that's OK, I'm not sure (don't change it in this PR though).

I'm not sure what you mean by this, mind elaborating what you had in mind?

---

tools/wptserve/wptserve/server.py, line 427 at r3 (raw file):

Previously, jgraham wrote…

Are you going to move the removal from stream_queues to the parent?

Yup

---

tools/wptserve/wptserve/server.py, line 344 at r7 (raw file):

Previously, jgraham wrote…

This and stream_queues could be local variables, I think. I know we aren't always great at it, but reducing shared state is good!

Yup, moving the last bit of logic out of the stream thread and then making this local

[jgraham]

Reviewable status: 67 of 71 files reviewed, 4 unresolved discussions (waiting on @jgraham, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/handlers.py, line 294 at r7 (raw file):

Previously, dhdavvie (David H) wrote…

I'm not sure what you mean by this, mind elaborating what you had in mind?

You can't write a function like

def handle_data(request, response):
     return "hello world"

and have the server return hello world as the response body.

[dhdavvie]

Reviewable status: 67 of 71 files reviewed, 4 unresolved discussions (waiting on @jgraham, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/handlers.py, line 294 at r7 (raw file):

Previously, jgraham wrote…

You can't write a function like

def handle_data(request, response):
     return "hello world"

and have the server return hello world as the response body.

Ahh yes, now I see what you mean. I don't see why this shouldn't be possible by using similar code to the __call__ function in FunctionHandler to parse return values from the handler functions. I will look into doing this for the final PR :)

[jgraham]

Reviewed 4 of 4 files at r8.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/handlers.py, line 277 at r7 (raw file):

Previously, dhdavvie (David H) wrote…

Sure, making it more general would be a good idea

Sorry this, is somewhat irrational, but I really dislike convert_to. It sounds like it mutates the object itself, but it doesn't. So I really think we should drop that from the method name.

---

tools/wptserve/wptserve/server.py, line 370 at r8 (raw file):

                        if frame.stream_id not in stream_queues:
                            stream_queues[frame.stream_id] = Queue()
                            request_threads.append(self.start_stream_thread(frame, stream_queues[frame.stream_id]))

This list leaks, right? That's maybe not a huge problem since the number of threads per connection might be limited, but we should perhaps fix it anyway, maybe by having a single data structure like {stream_id: (thread, Queue)}

[dhdavvie]

Reviewable status: all files reviewed, 2 unresolved discussions (waiting on @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/handlers.py, line 277 at r7 (raw file):

Previously, jgraham wrote…

Sorry this, is somewhat irrational, but I really dislike convert_to. It sounds like it mutates the object itself, but it doesn't. So I really think we should drop that from the method name.

Fair enough, the implication is definitely there. should the function just be frame_handler(.....) then?

---

tools/wptserve/wptserve/server.py, line 370 at r8 (raw file):

Previously, jgraham wrote…

This list leaks, right? That's maybe not a huge problem since the number of threads per connection might be limited, but we should perhaps fix it anyway, maybe by having a single data structure like {stream_id: (thread, Queue)}

Yes, I realised that this does leak, but put it on the backburner to fix in the next PR. Want me to fix it in this one instead?

[jgraham]

Reviewed 2 of 2 files at r9.
Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/server.py, line 365 at r9 (raw file):

                        for stream_id, (thread, queue) in stream_queues.items():
                            queue.put(frame)
                            del stream_queues[stream_id]

So, a question. Here we put a message that should cause the threads to terminate, but don't join them.

---

tools/wptserve/wptserve/server.py, line 379 at r9 (raw file):

                self.logger.debug('(%s) ERROR - Closing Connection - \n%s' % (self.uid, str(e)))
                self.close_connection = True
                for stream_id, (thread, queue) in stream_queues:

But here we wait for the threads to exit without sending them any message. That seems wrong?

[dhdavvie]

Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/server.py, line 379 at r9 (raw file):

Previously, jgraham wrote…

But here we wait for the threads to exit without sending them any message. That seems wrong?

So what I could do is move the join code back 2 indents, so that it will run after the while loop, which will happen in the case of ConnectionTerminated and a socket exception. I agree that there should be some communication to the threads that they should finish up. Would creating dummy ConnectionTerminated frames to trigger them to finish make sense?

[jgraham]

Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/server.py, line 379 at r9 (raw file):

Previously, dhdavvie (David H) wrote…

So what I could do is move the join code back 2 indents, so that it will run after the while loop, which will happen in the case of ConnectionTerminated and a socket exception. I agree that there should be some communication to the threads that they should finish up. Would creating dummy ConnectionTerminated frames to trigger them to finish make sense?

So I might just make a message of None act like ConnectionTerminated in the theads, rather than trying to really create a fake message. Then put the join in a finally block outside the while loop.

[dhdavvie]

Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/server.py, line 379 at r9 (raw file):

Previously, jgraham wrote…

So I might just make a message of None act like ConnectionTerminated in the theads, rather than trying to really create a fake message. Then put the join in a finally block outside the while loop.

How does this look? It seems easier to move the join to outside the while loop, instead of putting a try/finally wrap around the while loop, unless I've misunderstood what you meant

+++ tools/wptserve/wptserve/server.py	(date 1534254666000)
@@ -362,7 +362,6 @@
                         # Flood all the streams with connection terminated, this will cause them to stop
                         for stream_id, (thread, queue) in stream_queues.items():
                             queue.put(frame)
-                            del stream_queues[stream_id]
 
                     elif hasattr(frame, 'stream_id'):
                         if frame.stream_id not in stream_queues:
@@ -376,8 +375,11 @@
             except (socket.timeout, socket.error) as e:
                 self.logger.debug('(%s) ERROR - Closing Connection - \n%s' % (self.uid, str(e)))
                 self.close_connection = True
-                for stream_id, (thread, queue) in stream_queues:
-                    thread.join()
+                for stream_id, (thread, queue) in stream_queues.items():
+                    queue.put(None)
+
+        for stream_id, (thread, queue) in stream_queues:
+            thread.join()
 
     def start_stream_thread(self, frame, queue):
         t = threading.Thread(
@@ -433,7 +435,7 @@
 
                 if frame.stream_ended:
                     wfile.close()
-            elif isinstance(frame, (StreamReset, StreamEnded, ConnectionTerminated)):
+            elif isinstance(frame, (StreamReset, StreamEnded, ConnectionTerminated, type(None))):
                 self.logger.debug('(%s - %s) Stream Reset, Thread Closing' % (self.uid, stream_id))
                 break```

[jgraham]

Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/server.py, line 379 at r9 (raw file):

Previously, dhdavvie (David H) wrote…

How does this look? It seems easier to move the join to outside the while loop, instead of putting a try/finally wrap around the while loop, unless I've misunderstood what you meant

+++ tools/wptserve/wptserve/server.py	(date 1534254666000)
@@ -362,7 +362,6 @@
                         # Flood all the streams with connection terminated, this will cause them to stop
                         for stream_id, (thread, queue) in stream_queues.items():
                             queue.put(frame)
-                            del stream_queues[stream_id]
 
                     elif hasattr(frame, 'stream_id'):
                         if frame.stream_id not in stream_queues:
@@ -376,8 +375,11 @@
             except (socket.timeout, socket.error) as e:
                 self.logger.debug('(%s) ERROR - Closing Connection - \n%s' % (self.uid, str(e)))
                 self.close_connection = True
-                for stream_id, (thread, queue) in stream_queues:
-                    thread.join()
+                for stream_id, (thread, queue) in stream_queues.items():
+                    queue.put(None)
+
+        for stream_id, (thread, queue) in stream_queues:
+            thread.join()
 
     def start_stream_thread(self, frame, queue):
         t = threading.Thread(
@@ -433,7 +435,7 @@
 
                 if frame.stream_ended:
                     wfile.close()
-            elif isinstance(frame, (StreamReset, StreamEnded, ConnectionTerminated)):
+            elif isinstance(frame, (StreamReset, StreamEnded, ConnectionTerminated, type(None))):
                 self.logger.debug('(%s - %s) Stream Reset, Thread Closing' % (self.uid, stream_id))
                 break```
</blockquote></details>

I would just make the condition `frame is None or isinstance(…)`. 

The difference here is that other kinds of exceptions aren't handled, which might actually be right, but now I think about it I wonder if we correctly wrap all the user code in exception handlers so that handlers for one frame can't cause us to stop processing other streams.


<!-- Sent from Reviewable.io -->

[dhdavvie]

Reviewable status: all files reviewed, 3 unresolved discussions (waiting on @jgraham, @dhdavvie, @AutomatedTester, @annevk, @gsnedders, and @andreastt)

---

tools/wptserve/wptserve/server.py, line 379 at r9 (raw file):

Previously, jgraham wrote…

I would just make the condition frame is None or isinstance(…).

The difference here is that other kinds of exceptions aren't handled, which might actually be right, but now I think about it I wonder if we correctly wrap all the user code in exception handlers so that handlers for one frame can't cause us to stop processing other streams.

Exceptions raised in on stream thread should not affect the other streams AFAIK

[jgraham]

Thanks for all the fixesd!

Want to rewrite the history to remove [WIP] from the commit messages and document the changes made?

[dhdavvie]

Doing that now