uasyncio: Auto-unregister poll objects on POLLHUP/POLLERR.

POLLHUP/POLERR may be returned anytime (per POSIX, these flags aren't
even valid in input flags, they just appear in output flags). Subsequent
I/O operation on stream will lead to exception. If an application
doesn't do proper exception handling, the stream won't be closed, and
following calls will return POLLHUP/POLLERR status again (infinitely).
So, proactively unregister such a stream.

This change is questionable, because apps should handle errors properly
and close the stream in such case (or it will be leaked), and closing
will remove the stream from poller too.

But again, if that's not done, it may lead to cascade of adverse effects,
e.g. after eef054d, benchmark/test_http_server_heavy.py regressed and
started and started to throw utimeq queue overflow exceptions. The story
behind it is: Boom benchmarker does an initial probe request to the app
under test which it apparently doen't handle properly, leading to
EPIPE/ECONNRESET on the side of the test app, the app didn't close the
socket, so each invocation to .wait() resulted in that socket being
returned with POLLHUP again and again. Given that after eef054d, .wait()
is called on each even loop iteration, that create positive feedback in
the queue leading to it growing to overflow.

Author: pfalcon

uasyncio/uasyncio/__init__.py

@@ -73,6 +73,13 @@ def wait(self, delay):
         if res:
             for sock, ev in res:
                 cb = self.objmap[id(sock)]
+                if ev & (select.POLLHUP | select.POLLERR):
+                    # These events are returned even if not requested, and
+                    # are sticky, i.e. will be returned again and again.
+                    # If the caller doesn't do proper error handling and
+                    # unregister this sock, we'll busy-loop on it, so we
+                    # as well can unregister it now "just in case".
+                    self.remove_reader(sock)
                 if DEBUG and __debug__:
                     log.debug("Calling IO callback: %r", cb)
                 if isinstance(cb, tuple):