We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Timeouts don't work on windows
Currently the windows npipe implementation doesn't honour timeouts. Regardless of which api endpoint you use or pretty much anything else this leads to bugs where the docker api waits until the docker daemon finishes instead of timing out properly.

For example, if there is a dockerfile containing at `timeout/`
```
FROM alpine

RUN sleep 1000
```
and you run
```python
from docker import DockerClient

DockerClient.from_env().images.build(path="timeout/", timeout=3)
```
python will hang for the full 1000 seconds instead of raising an error after 3.

Version info:  
docker-py: 6.0.1  
python: 3.11.3  
docker:
Client:
 Cloud integration: v1.0.24
 Version:           20.10.14
 API version:       1.41
 Go version:        go1.16.15
 Git commit:        a224086
 Built:             Thu Mar 24 01:53:11 2022        
 OS/Arch:           windows/amd64
 Context:           default
 Experimental:      true

Server: Docker Desktop 4.8.1 (78998)
 Engine:
  Version:          20.10.14
  API version:      1.41 (minimum version 1.12)     
  Go version:       go1.16.15
  Git commit:       87a90dc
  Built:            Thu Mar 24 01:46:14 2022        
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          1.5.11
  GitCommit:        3df54a852345ae127d1fa3092b95168e4a88e2f8
 runc:
  Version:          1.0.3
  GitCommit:        v1.0.3-0-gf46b6ba
 docker-init:
  Version:          0.19.0
  GitCommit:        de40ad0

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `docker/transport/npipesocket.py`
Content:
```
1 import functools
2 import time
3 import io
4 
5 import win32file
6 import win32pipe
7 
8 cERROR_PIPE_BUSY = 0xe7
9 cSECURITY_SQOS_PRESENT = 0x100000
10 cSECURITY_ANONYMOUS = 0
11 
12 MAXIMUM_RETRY_COUNT = 10
13 
14 
15 def check_closed(f):
16     @functools.wraps(f)
17     def wrapped(self, *args, **kwargs):
18         if self._closed:
19             raise RuntimeError(
20                 'Can not reuse socket after connection was closed.'
21             )
22         return f(self, *args, **kwargs)
23     return wrapped
24 
25 
26 class NpipeSocket:
27     """ Partial implementation of the socket API over windows named pipes.
28         This implementation is only designed to be used as a client socket,
29         and server-specific methods (bind, listen, accept...) are not
30         implemented.
31     """
32 
33     def __init__(self, handle=None):
34         self._timeout = win32pipe.NMPWAIT_USE_DEFAULT_WAIT
35         self._handle = handle
36         self._closed = False
37 
38     def accept(self):
39         raise NotImplementedError()
40 
41     def bind(self, address):
42         raise NotImplementedError()
43 
44     def close(self):
45         self._handle.Close()
46         self._closed = True
47 
48     @check_closed
49     def connect(self, address, retry_count=0):
50         try:
51             handle = win32file.CreateFile(
52                 address,
53                 win32file.GENERIC_READ | win32file.GENERIC_WRITE,
54                 0,
55                 None,
56                 win32file.OPEN_EXISTING,
57                 cSECURITY_ANONYMOUS | cSECURITY_SQOS_PRESENT,
58                 0
59             )
60         except win32pipe.error as e:
61             # See Remarks:
62             # https://msdn.microsoft.com/en-us/library/aa365800.aspx
63             if e.winerror == cERROR_PIPE_BUSY:
64                 # Another program or thread has grabbed our pipe instance
65                 # before we got to it. Wait for availability and attempt to
66                 # connect again.
67                 retry_count = retry_count + 1
68                 if (retry_count < MAXIMUM_RETRY_COUNT):
69                     time.sleep(1)
70                     return self.connect(address, retry_count)
71             raise e
72 
73         self.flags = win32pipe.GetNamedPipeInfo(handle)[0]
74 
75         self._handle = handle
76         self._address = address
77 
78     @check_closed
79     def connect_ex(self, address):
80         return self.connect(address)
81 
82     @check_closed
83     def detach(self):
84         self._closed = True
85         return self._handle
86 
87     @check_closed
88     def dup(self):
89         return NpipeSocket(self._handle)
90 
91     def getpeername(self):
92         return self._address
93 
94     def getsockname(self):
95         return self._address
96 
97     def getsockopt(self, level, optname, buflen=None):
98         raise NotImplementedError()
99 
100     def ioctl(self, control, option):
101         raise NotImplementedError()
102 
103     def listen(self, backlog):
104         raise NotImplementedError()
105 
106     def makefile(self, mode=None, bufsize=None):
107         if mode.strip('b') != 'r':
108             raise NotImplementedError()
109         rawio = NpipeFileIOBase(self)
110         if bufsize is None or bufsize <= 0:
111             bufsize = io.DEFAULT_BUFFER_SIZE
112         return io.BufferedReader(rawio, buffer_size=bufsize)
113 
114     @check_closed
115     def recv(self, bufsize, flags=0):
116         err, data = win32file.ReadFile(self._handle, bufsize)
117         return data
118 
119     @check_closed
120     def recvfrom(self, bufsize, flags=0):
121         data = self.recv(bufsize, flags)
122         return (data, self._address)
123 
124     @check_closed
125     def recvfrom_into(self, buf, nbytes=0, flags=0):
126         return self.recv_into(buf, nbytes, flags), self._address
127 
128     @check_closed
129     def recv_into(self, buf, nbytes=0):
130         readbuf = buf
131         if not isinstance(buf, memoryview):
132             readbuf = memoryview(buf)
133 
134         err, data = win32file.ReadFile(
135             self._handle,
136             readbuf[:nbytes] if nbytes else readbuf
137         )
138         return len(data)
139 
140     def _recv_into_py2(self, buf, nbytes):
141         err, data = win32file.ReadFile(self._handle, nbytes or len(buf))
142         n = len(data)
143         buf[:n] = data
144         return n
145 
146     @check_closed
147     def send(self, string, flags=0):
148         err, nbytes = win32file.WriteFile(self._handle, string)
149         return nbytes
150 
151     @check_closed
152     def sendall(self, string, flags=0):
153         return self.send(string, flags)
154 
155     @check_closed
156     def sendto(self, string, address):
157         self.connect(address)
158         return self.send(string)
159 
160     def setblocking(self, flag):
161         if flag:
162             return self.settimeout(None)
163         return self.settimeout(0)
164 
165     def settimeout(self, value):
166         if value is None:
167             # Blocking mode
168             self._timeout = win32pipe.NMPWAIT_WAIT_FOREVER
169         elif not isinstance(value, (float, int)) or value < 0:
170             raise ValueError('Timeout value out of range')
171         elif value == 0:
172             # Non-blocking mode
173             self._timeout = win32pipe.NMPWAIT_NO_WAIT
174         else:
175             # Timeout mode - Value converted to milliseconds
176             self._timeout = value * 1000
177 
178     def gettimeout(self):
179         return self._timeout
180 
181     def setsockopt(self, level, optname, value):
182         raise NotImplementedError()
183 
184     @check_closed
185     def shutdown(self, how):
186         return self.close()
187 
188 
189 class NpipeFileIOBase(io.RawIOBase):
190     def __init__(self, npipe_socket):
191         self.sock = npipe_socket
192 
193     def close(self):
194         super().close()
195         self.sock = None
196 
197     def fileno(self):
198         return self.sock.fileno()
199 
200     def isatty(self):
201         return False
202 
203     def readable(self):
204         return True
205 
206     def readinto(self, buf):
207         return self.sock.recv_into(buf)
208 
209     def seekable(self):
210         return False
211 
212     def writable(self):
213         return False
214
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/docker/transport/npipesocket.py b/docker/transport/npipesocket.py
--- a/docker/transport/npipesocket.py
+++ b/docker/transport/npipesocket.py
@@ -4,6 +4,9 @@
 
 import win32file
 import win32pipe
+import pywintypes
+import win32event
+import win32api
 
 cERROR_PIPE_BUSY = 0xe7
 cSECURITY_SQOS_PRESENT = 0x100000
@@ -54,7 +57,9 @@
                 0,
                 None,
                 win32file.OPEN_EXISTING,
-                cSECURITY_ANONYMOUS | cSECURITY_SQOS_PRESENT,
+                (cSECURITY_ANONYMOUS
+                    | cSECURITY_SQOS_PRESENT
+                    | win32file.FILE_FLAG_OVERLAPPED),
                 0
             )
         except win32pipe.error as e:
@@ -131,22 +136,37 @@
         if not isinstance(buf, memoryview):
             readbuf = memoryview(buf)
 
-        err, data = win32file.ReadFile(
-            self._handle,
-            readbuf[:nbytes] if nbytes else readbuf
-        )
-        return len(data)
-
-    def _recv_into_py2(self, buf, nbytes):
-        err, data = win32file.ReadFile(self._handle, nbytes or len(buf))
-        n = len(data)
-        buf[:n] = data
-        return n
+        event = win32event.CreateEvent(None, True, True, None)
+        try:
+            overlapped = pywintypes.OVERLAPPED()
+            overlapped.hEvent = event
+            err, data = win32file.ReadFile(
+                self._handle,
+                readbuf[:nbytes] if nbytes else readbuf,
+                overlapped
+            )
+            wait_result = win32event.WaitForSingleObject(event, self._timeout)
+            if wait_result == win32event.WAIT_TIMEOUT:
+                win32file.CancelIo(self._handle)
+                raise TimeoutError
+            return win32file.GetOverlappedResult(self._handle, overlapped, 0)
+        finally:
+            win32api.CloseHandle(event)
 
     @check_closed
     def send(self, string, flags=0):
-        err, nbytes = win32file.WriteFile(self._handle, string)
-        return nbytes
+        event = win32event.CreateEvent(None, True, True, None)
+        try:
+            overlapped = pywintypes.OVERLAPPED()
+            overlapped.hEvent = event
+            win32file.WriteFile(self._handle, string, overlapped)
+            wait_result = win32event.WaitForSingleObject(event, self._timeout)
+            if wait_result == win32event.WAIT_TIMEOUT:
+                win32file.CancelIo(self._handle)
+                raise TimeoutError
+            return win32file.GetOverlappedResult(self._handle, overlapped, 0)
+        finally:
+            win32api.CloseHandle(event)
 
     @check_closed
     def sendall(self, string, flags=0):
@@ -165,15 +185,12 @@
     def settimeout(self, value):
         if value is None:
             # Blocking mode
-            self._timeout = win32pipe.NMPWAIT_WAIT_FOREVER
+            self._timeout = win32event.INFINITE
         elif not isinstance(value, (float, int)) or value < 0:
             raise ValueError('Timeout value out of range')
-        elif value == 0:
-            # Non-blocking mode
-            self._timeout = win32pipe.NMPWAIT_NO_WAIT
         else:
             # Timeout mode - Value converted to milliseconds
-            self._timeout = value * 1000
+            self._timeout = int(value * 1000)
 
     def gettimeout(self):
         return self._timeout

{"golden_diff": "diff --git a/docker/transport/npipesocket.py b/docker/transport/npipesocket.py\n--- a/docker/transport/npipesocket.py\n+++ b/docker/transport/npipesocket.py\n@@ -4,6 +4,9 @@\n \n import win32file\n import win32pipe\n+import pywintypes\n+import win32event\n+import win32api\n \n cERROR_PIPE_BUSY = 0xe7\n cSECURITY_SQOS_PRESENT = 0x100000\n@@ -54,7 +57,9 @@\n                 0,\n                 None,\n                 win32file.OPEN_EXISTING,\n-                cSECURITY_ANONYMOUS | cSECURITY_SQOS_PRESENT,\n+                (cSECURITY_ANONYMOUS\n+                    | cSECURITY_SQOS_PRESENT\n+                    | win32file.FILE_FLAG_OVERLAPPED),\n                 0\n             )\n         except win32pipe.error as e:\n@@ -131,22 +136,37 @@\n         if not isinstance(buf, memoryview):\n             readbuf = memoryview(buf)\n \n-        err, data = win32file.ReadFile(\n-            self._handle,\n-            readbuf[:nbytes] if nbytes else readbuf\n-        )\n-        return len(data)\n-\n-    def _recv_into_py2(self, buf, nbytes):\n-        err, data = win32file.ReadFile(self._handle, nbytes or len(buf))\n-        n = len(data)\n-        buf[:n] = data\n-        return n\n+        event = win32event.CreateEvent(None, True, True, None)\n+        try:\n+            overlapped = pywintypes.OVERLAPPED()\n+            overlapped.hEvent = event\n+            err, data = win32file.ReadFile(\n+                self._handle,\n+                readbuf[:nbytes] if nbytes else readbuf,\n+                overlapped\n+            )\n+            wait_result = win32event.WaitForSingleObject(event, self._timeout)\n+            if wait_result == win32event.WAIT_TIMEOUT:\n+                win32file.CancelIo(self._handle)\n+                raise TimeoutError\n+            return win32file.GetOverlappedResult(self._handle, overlapped, 0)\n+        finally:\n+            win32api.CloseHandle(event)\n \n     @check_closed\n     def send(self, string, flags=0):\n-        err, nbytes = win32file.WriteFile(self._handle, string)\n-        return nbytes\n+        event = win32event.CreateEvent(None, True, True, None)\n+        try:\n+            overlapped = pywintypes.OVERLAPPED()\n+            overlapped.hEvent = event\n+            win32file.WriteFile(self._handle, string, overlapped)\n+            wait_result = win32event.WaitForSingleObject(event, self._timeout)\n+            if wait_result == win32event.WAIT_TIMEOUT:\n+                win32file.CancelIo(self._handle)\n+                raise TimeoutError\n+            return win32file.GetOverlappedResult(self._handle, overlapped, 0)\n+        finally:\n+            win32api.CloseHandle(event)\n \n     @check_closed\n     def sendall(self, string, flags=0):\n@@ -165,15 +185,12 @@\n     def settimeout(self, value):\n         if value is None:\n             # Blocking mode\n-            self._timeout = win32pipe.NMPWAIT_WAIT_FOREVER\n+            self._timeout = win32event.INFINITE\n         elif not isinstance(value, (float, int)) or value < 0:\n             raise ValueError('Timeout value out of range')\n-        elif value == 0:\n-            # Non-blocking mode\n-            self._timeout = win32pipe.NMPWAIT_NO_WAIT\n         else:\n             # Timeout mode - Value converted to milliseconds\n-            self._timeout = value * 1000\n+            self._timeout = int(value * 1000)\n \n     def gettimeout(self):\n         return self._timeout\n", "issue": "Timeouts don't work on windows\nCurrently the windows npipe implementation doesn't honour timeouts. Regardless of which api endpoint you use or pretty much anything else this leads to bugs where the docker api waits until the docker daemon finishes instead of timing out properly.\r\n\r\nFor example, if there is a dockerfile containing at `timeout/`\r\n```\r\nFROM alpine\r\n\r\nRUN sleep 1000\r\n```\r\nand you run\r\n```python\r\nfrom docker import DockerClient\r\n\r\nDockerClient.from_env().images.build(path=\"timeout/\", timeout=3)\r\n```\r\npython will hang for the full 1000 seconds instead of raising an error after 3.\r\n\r\nVersion info:  \r\ndocker-py: 6.0.1  \r\npython: 3.11.3  \r\ndocker:\r\nClient:\r\n Cloud integration: v1.0.24\r\n Version:           20.10.14\r\n API version:       1.41\r\n Go version:        go1.16.15\r\n Git commit:        a224086\r\n Built:             Thu Mar 24 01:53:11 2022        \r\n OS/Arch:           windows/amd64\r\n Context:           default\r\n Experimental:      true\r\n\r\nServer: Docker Desktop 4.8.1 (78998)\r\n Engine:\r\n  Version:          20.10.14\r\n  API version:      1.41 (minimum version 1.12)     \r\n  Go version:       go1.16.15\r\n  Git commit:       87a90dc\r\n  Built:            Thu Mar 24 01:46:14 2022        \r\n  OS/Arch:          linux/amd64\r\n  Experimental:     false\r\n containerd:\r\n  Version:          1.5.11\r\n  GitCommit:        3df54a852345ae127d1fa3092b95168e4a88e2f8\r\n runc:\r\n  Version:          1.0.3\r\n  GitCommit:        v1.0.3-0-gf46b6ba\r\n docker-init:\r\n  Version:          0.19.0\r\n  GitCommit:        de40ad0\n", "before_files": [{"content": "import functools\nimport time\nimport io\n\nimport win32file\nimport win32pipe\n\ncERROR_PIPE_BUSY = 0xe7\ncSECURITY_SQOS_PRESENT = 0x100000\ncSECURITY_ANONYMOUS = 0\n\nMAXIMUM_RETRY_COUNT = 10\n\n\ndef check_closed(f):\n    @functools.wraps(f)\n    def wrapped(self, *args, **kwargs):\n        if self._closed:\n            raise RuntimeError(\n                'Can not reuse socket after connection was closed.'\n            )\n        return f(self, *args, **kwargs)\n    return wrapped\n\n\nclass NpipeSocket:\n    \"\"\" Partial implementation of the socket API over windows named pipes.\n        This implementation is only designed to be used as a client socket,\n        and server-specific methods (bind, listen, accept...) are not\n        implemented.\n    \"\"\"\n\n    def __init__(self, handle=None):\n        self._timeout = win32pipe.NMPWAIT_USE_DEFAULT_WAIT\n        self._handle = handle\n        self._closed = False\n\n    def accept(self):\n        raise NotImplementedError()\n\n    def bind(self, address):\n        raise NotImplementedError()\n\n    def close(self):\n        self._handle.Close()\n        self._closed = True\n\n    @check_closed\n    def connect(self, address, retry_count=0):\n        try:\n            handle = win32file.CreateFile(\n                address,\n                win32file.GENERIC_READ | win32file.GENERIC_WRITE,\n                0,\n                None,\n                win32file.OPEN_EXISTING,\n                cSECURITY_ANONYMOUS | cSECURITY_SQOS_PRESENT,\n                0\n            )\n        except win32pipe.error as e:\n            # See Remarks:\n            # https://msdn.microsoft.com/en-us/library/aa365800.aspx\n            if e.winerror == cERROR_PIPE_BUSY:\n                # Another program or thread has grabbed our pipe instance\n                # before we got to it. Wait for availability and attempt to\n                # connect again.\n                retry_count = retry_count + 1\n                if (retry_count < MAXIMUM_RETRY_COUNT):\n                    time.sleep(1)\n                    return self.connect(address, retry_count)\n            raise e\n\n        self.flags = win32pipe.GetNamedPipeInfo(handle)[0]\n\n        self._handle = handle\n        self._address = address\n\n    @check_closed\n    def connect_ex(self, address):\n        return self.connect(address)\n\n    @check_closed\n    def detach(self):\n        self._closed = True\n        return self._handle\n\n    @check_closed\n    def dup(self):\n        return NpipeSocket(self._handle)\n\n    def getpeername(self):\n        return self._address\n\n    def getsockname(self):\n        return self._address\n\n    def getsockopt(self, level, optname, buflen=None):\n        raise NotImplementedError()\n\n    def ioctl(self, control, option):\n        raise NotImplementedError()\n\n    def listen(self, backlog):\n        raise NotImplementedError()\n\n    def makefile(self, mode=None, bufsize=None):\n        if mode.strip('b') != 'r':\n            raise NotImplementedError()\n        rawio = NpipeFileIOBase(self)\n        if bufsize is None or bufsize <= 0:\n            bufsize = io.DEFAULT_BUFFER_SIZE\n        return io.BufferedReader(rawio, buffer_size=bufsize)\n\n    @check_closed\n    def recv(self, bufsize, flags=0):\n        err, data = win32file.ReadFile(self._handle, bufsize)\n        return data\n\n    @check_closed\n    def recvfrom(self, bufsize, flags=0):\n        data = self.recv(bufsize, flags)\n        return (data, self._address)\n\n    @check_closed\n    def recvfrom_into(self, buf, nbytes=0, flags=0):\n        return self.recv_into(buf, nbytes, flags), self._address\n\n    @check_closed\n    def recv_into(self, buf, nbytes=0):\n        readbuf = buf\n        if not isinstance(buf, memoryview):\n            readbuf = memoryview(buf)\n\n        err, data = win32file.ReadFile(\n            self._handle,\n            readbuf[:nbytes] if nbytes else readbuf\n        )\n        return len(data)\n\n    def _recv_into_py2(self, buf, nbytes):\n        err, data = win32file.ReadFile(self._handle, nbytes or len(buf))\n        n = len(data)\n        buf[:n] = data\n        return n\n\n    @check_closed\n    def send(self, string, flags=0):\n        err, nbytes = win32file.WriteFile(self._handle, string)\n        return nbytes\n\n    @check_closed\n    def sendall(self, string, flags=0):\n        return self.send(string, flags)\n\n    @check_closed\n    def sendto(self, string, address):\n        self.connect(address)\n        return self.send(string)\n\n    def setblocking(self, flag):\n        if flag:\n            return self.settimeout(None)\n        return self.settimeout(0)\n\n    def settimeout(self, value):\n        if value is None:\n            # Blocking mode\n            self._timeout = win32pipe.NMPWAIT_WAIT_FOREVER\n        elif not isinstance(value, (float, int)) or value < 0:\n            raise ValueError('Timeout value out of range')\n        elif value == 0:\n            # Non-blocking mode\n            self._timeout = win32pipe.NMPWAIT_NO_WAIT\n        else:\n            # Timeout mode - Value converted to milliseconds\n            self._timeout = value * 1000\n\n    def gettimeout(self):\n        return self._timeout\n\n    def setsockopt(self, level, optname, value):\n        raise NotImplementedError()\n\n    @check_closed\n    def shutdown(self, how):\n        return self.close()\n\n\nclass NpipeFileIOBase(io.RawIOBase):\n    def __init__(self, npipe_socket):\n        self.sock = npipe_socket\n\n    def close(self):\n        super().close()\n        self.sock = None\n\n    def fileno(self):\n        return self.sock.fileno()\n\n    def isatty(self):\n        return False\n\n    def readable(self):\n        return True\n\n    def readinto(self, buf):\n        return self.sock.recv_into(buf)\n\n    def seekable(self):\n        return False\n\n    def writable(self):\n        return False\n", "path": "docker/transport/npipesocket.py"}], "after_files": [{"content": "import functools\nimport time\nimport io\n\nimport win32file\nimport win32pipe\nimport pywintypes\nimport win32event\nimport win32api\n\ncERROR_PIPE_BUSY = 0xe7\ncSECURITY_SQOS_PRESENT = 0x100000\ncSECURITY_ANONYMOUS = 0\n\nMAXIMUM_RETRY_COUNT = 10\n\n\ndef check_closed(f):\n    @functools.wraps(f)\n    def wrapped(self, *args, **kwargs):\n        if self._closed:\n            raise RuntimeError(\n                'Can not reuse socket after connection was closed.'\n            )\n        return f(self, *args, **kwargs)\n    return wrapped\n\n\nclass NpipeSocket:\n    \"\"\" Partial implementation of the socket API over windows named pipes.\n        This implementation is only designed to be used as a client socket,\n        and server-specific methods (bind, listen, accept...) are not\n        implemented.\n    \"\"\"\n\n    def __init__(self, handle=None):\n        self._timeout = win32pipe.NMPWAIT_USE_DEFAULT_WAIT\n        self._handle = handle\n        self._closed = False\n\n    def accept(self):\n        raise NotImplementedError()\n\n    def bind(self, address):\n        raise NotImplementedError()\n\n    def close(self):\n        self._handle.Close()\n        self._closed = True\n\n    @check_closed\n    def connect(self, address, retry_count=0):\n        try:\n            handle = win32file.CreateFile(\n                address,\n                win32file.GENERIC_READ | win32file.GENERIC_WRITE,\n                0,\n                None,\n                win32file.OPEN_EXISTING,\n                (cSECURITY_ANONYMOUS\n                    | cSECURITY_SQOS_PRESENT\n                    | win32file.FILE_FLAG_OVERLAPPED),\n                0\n            )\n        except win32pipe.error as e:\n            # See Remarks:\n            # https://msdn.microsoft.com/en-us/library/aa365800.aspx\n            if e.winerror == cERROR_PIPE_BUSY:\n                # Another program or thread has grabbed our pipe instance\n                # before we got to it. Wait for availability and attempt to\n                # connect again.\n                retry_count = retry_count + 1\n                if (retry_count < MAXIMUM_RETRY_COUNT):\n                    time.sleep(1)\n                    return self.connect(address, retry_count)\n            raise e\n\n        self.flags = win32pipe.GetNamedPipeInfo(handle)[0]\n\n        self._handle = handle\n        self._address = address\n\n    @check_closed\n    def connect_ex(self, address):\n        return self.connect(address)\n\n    @check_closed\n    def detach(self):\n        self._closed = True\n        return self._handle\n\n    @check_closed\n    def dup(self):\n        return NpipeSocket(self._handle)\n\n    def getpeername(self):\n        return self._address\n\n    def getsockname(self):\n        return self._address\n\n    def getsockopt(self, level, optname, buflen=None):\n        raise NotImplementedError()\n\n    def ioctl(self, control, option):\n        raise NotImplementedError()\n\n    def listen(self, backlog):\n        raise NotImplementedError()\n\n    def makefile(self, mode=None, bufsize=None):\n        if mode.strip('b') != 'r':\n            raise NotImplementedError()\n        rawio = NpipeFileIOBase(self)\n        if bufsize is None or bufsize <= 0:\n            bufsize = io.DEFAULT_BUFFER_SIZE\n        return io.BufferedReader(rawio, buffer_size=bufsize)\n\n    @check_closed\n    def recv(self, bufsize, flags=0):\n        err, data = win32file.ReadFile(self._handle, bufsize)\n        return data\n\n    @check_closed\n    def recvfrom(self, bufsize, flags=0):\n        data = self.recv(bufsize, flags)\n        return (data, self._address)\n\n    @check_closed\n    def recvfrom_into(self, buf, nbytes=0, flags=0):\n        return self.recv_into(buf, nbytes, flags), self._address\n\n    @check_closed\n    def recv_into(self, buf, nbytes=0):\n        readbuf = buf\n        if not isinstance(buf, memoryview):\n            readbuf = memoryview(buf)\n\n        event = win32event.CreateEvent(None, True, True, None)\n        try:\n            overlapped = pywintypes.OVERLAPPED()\n            overlapped.hEvent = event\n            err, data = win32file.ReadFile(\n                self._handle,\n                readbuf[:nbytes] if nbytes else readbuf,\n                overlapped\n            )\n            wait_result = win32event.WaitForSingleObject(event, self._timeout)\n            if wait_result == win32event.WAIT_TIMEOUT:\n                win32file.CancelIo(self._handle)\n                raise TimeoutError\n            return win32file.GetOverlappedResult(self._handle, overlapped, 0)\n        finally:\n            win32api.CloseHandle(event)\n\n    @check_closed\n    def send(self, string, flags=0):\n        event = win32event.CreateEvent(None, True, True, None)\n        try:\n            overlapped = pywintypes.OVERLAPPED()\n            overlapped.hEvent = event\n            win32file.WriteFile(self._handle, string, overlapped)\n            wait_result = win32event.WaitForSingleObject(event, self._timeout)\n            if wait_result == win32event.WAIT_TIMEOUT:\n                win32file.CancelIo(self._handle)\n                raise TimeoutError\n            return win32file.GetOverlappedResult(self._handle, overlapped, 0)\n        finally:\n            win32api.CloseHandle(event)\n\n    @check_closed\n    def sendall(self, string, flags=0):\n        return self.send(string, flags)\n\n    @check_closed\n    def sendto(self, string, address):\n        self.connect(address)\n        return self.send(string)\n\n    def setblocking(self, flag):\n        if flag:\n            return self.settimeout(None)\n        return self.settimeout(0)\n\n    def settimeout(self, value):\n        if value is None:\n            # Blocking mode\n            self._timeout = win32event.INFINITE\n        elif not isinstance(value, (float, int)) or value < 0:\n            raise ValueError('Timeout value out of range')\n        else:\n            # Timeout mode - Value converted to milliseconds\n            self._timeout = int(value * 1000)\n\n    def gettimeout(self):\n        return self._timeout\n\n    def setsockopt(self, level, optname, value):\n        raise NotImplementedError()\n\n    @check_closed\n    def shutdown(self, how):\n        return self.close()\n\n\nclass NpipeFileIOBase(io.RawIOBase):\n    def __init__(self, npipe_socket):\n        self.sock = npipe_socket\n\n    def close(self):\n        super().close()\n        self.sock = None\n\n    def fileno(self):\n        return self.sock.fileno()\n\n    def isatty(self):\n        return False\n\n    def readable(self):\n        return True\n\n    def readinto(self, buf):\n        return self.sock.recv_into(buf)\n\n    def seekable(self):\n        return False\n\n    def writable(self):\n        return False\n", "path": "docker/transport/npipesocket.py"}]}