Rewrite UDP transport from scratch.

The new implementation follows asyncio pretty closely.  We no longer use
uv_udp_* functions from libuv.  Instead, we use uv_poll_* and inherit
asyncio logic from selector_events.py.

When `loop.create_datagram_endpoint()` is called with a `remote_addr`
argument, asyncio attempts to *connect* the socket.  The old uvloop
implementation didn't do that, as there's no 'uv_udp_connect()' function
in libuv (at least yet).

For UDP it doesn't really matter if the socket is connected or not.  The
main difference is that when a UDP socket is connected, we can use
`socket.send(data)` method, instead of `socket.sendto(data, addr)`.

The subtle difference appears in two scenarios:

* using getsockname();

* raising an exception from `create_datagram_endpoint()` when
  `remote_addr` is unreachable.

It doesn't seem practical to wait until we have `uv_udp_connect()`
function in libuv, hence the rewrite.

In the future, when `uv_udp_connect()` is available, it should be easy
to rollback to the current implementation.

Addresses issues #109, #108, #85.

Author: 1st1

tests/test_udp.py

@@ -46,18 +46,19 @@ def datagram_received(self, data, addr):
                 super().datagram_received(data, addr)
                 self.transport.sendto(b'resp:' + data, addr)
 
-        for lc in (('127.0.0.1', 0), None):
-            if lc is None and not isinstance(self.loop, uvloop.Loop):
-                # TODO This looks like a bug in asyncio -- if no local_addr
-                # and no remote_addr are specified, the connection
-                # that asyncio creates is not bound anywhere.
-                return
+        for family, lc_host, lc_port in ((socket.AF_INET, '127.0.0.1', 0),
+                                         (socket.AF_INET6, '::1', 0)):
+
+            lc = (lc_host, lc_port)
 
             with self.subTest(local_addr=lc):
                 coro = self.loop.create_datagram_endpoint(
-                    TestMyDatagramProto, local_addr=lc, family=socket.AF_INET)
+                    TestMyDatagramProto,
+                    local_addr=lc,
+                    family=family)
+
                 s_transport, server = self.loop.run_until_complete(coro)
-                host, port = s_transport.get_extra_info('sockname')
+                host, port, *_ = s_transport.get_extra_info('sockname')
 
                 self.assertIsInstance(server, TestMyDatagramProto)
                 self.assertEqual('INITIALIZED', server.state)
@@ -70,16 +71,16 @@ def datagram_received(self, data, addr):
 
                 coro = self.loop.create_datagram_endpoint(
                     lambda: MyDatagramProto(loop=self.loop),
-                    family=socket.AF_INET,
-                    remote_addr=None if lc is None else (host, port),
+                    family=family,
+                    remote_addr=(host, port),
                     **extra)
                 transport, client = self.loop.run_until_complete(coro)
 
                 self.assertIsInstance(client, MyDatagramProto)
                 self.assertEqual('INITIALIZED', client.state)
                 self.assertIs(client.transport, transport)
 
-                transport.sendto(b'xxx', (host, port) if lc is None else None)
+                transport.sendto(b'xxx')
                 test_utils.run_until(self.loop, lambda: server.nbytes)
                 self.assertEqual(3, server.nbytes)
                 test_utils.run_until(self.loop, lambda: client.nbytes)

uvloop/handles/handle.pxd

@@ -1,10 +1,14 @@
 cdef class UVHandle:
     cdef:
         uv.uv_handle_t *_handle
-        bint _closed
-        bint _inited
         Loop _loop
         readonly _source_traceback
+        bint _closed
+        bint _inited
+
+        # Added to enable current UDPTransport implementation,
+        # which doesn't use libuv handles.
+        bint _has_handle
 
     # All "inline" methods are final
 
@@ -20,7 +24,8 @@ cdef class UVHandle:
 
     cdef _warn_unclosed(self)
 
-    cdef inline _free(self)
+    cdef _dealloc_impl(self)
+    cdef _free(self)
     cdef _close(self)
     cdef _after_close(self)
 
@@ -35,7 +40,7 @@ cdef class UVSocketHandle(UVHandle):
 
     # All "inline" methods are final
 
-    cdef inline _fileno(self)
+    cdef _fileno(self)
 
     cdef _new_socket(self)
     cdef inline _get_socket(self)

uvloop/handles/handle.pyx

@@ -15,6 +15,7 @@ cdef class UVHandle:
     def __cinit__(self):
         self._closed = 0
         self._inited = 0
+        self._has_handle = 1
         self._handle = NULL
         self._loop = None
         self._source_traceback = None
@@ -25,6 +26,9 @@ cdef class UVHandle:
                 self.__class__.__name__))
 
     def __dealloc__(self):
+        self._dealloc_impl()
+
+    cdef _dealloc_impl(self):
         if UVLOOP_DEBUG:
             if self._loop is not None:
                 self._loop._debug_handles_current.subtract([
@@ -36,6 +40,9 @@ cdef class UVHandle:
                     .format(self.__class__.__name__))
 
         if self._handle is NULL:
+            if UVLOOP_DEBUG:
+                if self._has_handle == 0:
+                    self._loop._debug_uv_handles_freed += 1
             return
 
         # -> When we're at this point, something is wrong <-
@@ -67,7 +74,7 @@ cdef class UVHandle:
             self._closed = 1
             self._free()
 
-    cdef inline _free(self):
+    cdef _free(self):
         if self._handle == NULL:
             return
 
@@ -98,6 +105,8 @@ cdef class UVHandle:
         if self._handle is not NULL:
             self._free()
 
+        self._closed = 1
+
         if UVLOOP_DEBUG:
             name = self.__class__.__name__
             if self._inited:
@@ -107,11 +116,10 @@ cdef class UVHandle:
                 raise RuntimeError(
                     '_abort_init: {}._closed is set'.format(name))
 
-        self._closed = 1
-
     cdef inline _finish_init(self):
         self._inited = 1
-        self._handle.data = <void*>self
+        if self._has_handle == 1:
+            self._handle.data = <void*>self
         if self._loop._debug:
             self._source_traceback = extract_stack()
         if UVLOOP_DEBUG:
@@ -134,7 +142,7 @@ cdef class UVHandle:
         cdef bint res
         res = self._closed != 1 and self._inited == 1
         if UVLOOP_DEBUG:
-            if res:
+            if res and self._has_handle == 1:
                 name = self.__class__.__name__
                 if self._handle is NULL:
                     raise RuntimeError(
@@ -228,7 +236,7 @@ cdef class UVSocketHandle(UVHandle):
         self._fileobj = None
         self.__cached_socket = None
 
-    cdef inline _fileno(self):
+    cdef _fileno(self):
         cdef:
             int fd
             int err
@@ -349,6 +357,9 @@ cdef void __uv_close_handle_cb(uv.uv_handle_t* handle) with gil:
         h = <UVHandle>handle.data
         try:
             if UVLOOP_DEBUG:
+                if not h._has_handle:
+                    raise RuntimeError(
+                        'has_handle=0 in __uv_close_handle_cb')
                 h._loop._debug_handles_closed.update([
                     h.__class__.__name__])
             h._free()

uvloop/handles/poll.pxd

@@ -12,6 +12,7 @@ cdef class UVPoll(UVHandle):
 
     cdef int is_active(self)
 
+    cdef is_reading(self)
     cdef start_reading(self, Handle callback)
     cdef start_writing(self, Handle callback)
     cdef stop_reading(self)

uvloop/handles/poll.pyx

@@ -84,6 +84,9 @@ cdef class UVPoll(UVHandle):
                     self.fd,
                     &dummy_event)  # ignore errors
 
+    cdef is_reading(self):
+        return self._is_alive() and self.reading_handle is not None
+
     cdef start_reading(self, Handle callback):
         cdef:
             int mask = 0

uvloop/handles/udp.pxd

@@ -1,25 +1,14 @@
 cdef class UDPTransport(UVBaseTransport):
     cdef:
-        bint __receiving
-        int _family
+        object sock
+        UVPoll poll
+        object address
+        object buffer
 
-        bint _address_set
-        system.sockaddr_storage _address
-        object _cached_py_address
+    cdef _init(self, Loop loop, object sock, object r_addr)
 
-    cdef _init(self, Loop loop, unsigned int family)
+    cdef _on_read_ready(self)
+    cdef _on_write_ready(self)
 
-    cdef _set_remote_address(self, system.sockaddr* addr,
-                             size_t addr_len)
-
-    cdef _bind(self, system.sockaddr* addr, bint reuse_addr)
-    cdef open(self, int family, int sockfd)
-    cdef _set_broadcast(self, bint on)
-
-    cdef inline __receiving_started(self)
-    cdef inline __receiving_stopped(self)
-
-    cdef _send(self, object data, object addr)
-
-    cdef _on_receive(self, bytes data, object exc, object addr)
-    cdef _on_sent(self, object exc)
+    @staticmethod
+    cdef UDPTransport new(Loop loop, object sock, object r_addr)