Issue #23293, asyncio: Rewrite IocpProactor.connect_pipe() as a coroutine

Use a coroutine with asyncio.sleep() instead of call_later() to ensure that the
schedule call is cancelled.

Add also a unit test cancelling connect_pipe().

Lib/asyncio/windows_events.py

@@ -518,28 +518,25 @@ class IocpProactor:
 
         return self._register(ov, pipe, finish_accept_pipe)
 
-    def _connect_pipe(self, fut, address, delay):
-        # Unfortunately there is no way to do an overlapped connect to a pipe.
-        # Call CreateFile() in a loop until it doesn't fail with
-        # ERROR_PIPE_BUSY
-        try:
-            handle = _overlapped.ConnectPipe(address)
-        except OSError as exc:
-            if exc.winerror == _overlapped.ERROR_PIPE_BUSY:
-                # Polling: retry later
-                delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY)
-                self._loop.call_later(delay,
-                                      self._connect_pipe, fut, address, delay)
-            else:
-                fut.set_exception(exc)
-        else:
-            pipe = windows_utils.PipeHandle(handle)
-            fut.set_result(pipe)
-
+    @coroutine
     def connect_pipe(self, address):
-        fut = futures.Future(loop=self._loop)
-        self._connect_pipe(fut, address, CONNECT_PIPE_INIT_DELAY)
-        return fut
+        delay = CONNECT_PIPE_INIT_DELAY
+        while True:
+            # Unfortunately there is no way to do an overlapped connect to a pipe.
+            # Call CreateFile() in a loop until it doesn't fail with
+            # ERROR_PIPE_BUSY
+            try:
+                handle = _overlapped.ConnectPipe(address)
+                break
+            except OSError as exc:
+                if exc.winerror != _overlapped.ERROR_PIPE_BUSY:
+                    raise
+
+            # ConnectPipe() failed with ERROR_PIPE_BUSY: retry later
+            delay = min(delay * 2, CONNECT_PIPE_MAX_DELAY)
+            yield from tasks.sleep(delay, loop=self._loop)
+
+        return windows_utils.PipeHandle(handle)
 
     def wait_for_handle(self, handle, timeout=None):
         """Wait for a handle.