diff --git a/changelogs/fragments/docker-py.yml b/changelogs/fragments/docker-py.yml
new file mode 100644
index 00000000..b092c27a
--- /dev/null
+++ b/changelogs/fragments/docker-py.yml
@@ -0,0 +1,4 @@
+bugfixes:
+  - "vendored Docker SDK for Python code - fix for errors on pipe close in Windows (https://github.com/ansible-collections/community.docker/pull/619)."
+  - "vendored Docker SDK for Python code - use ``poll()`` instead of ``select()`` except on Windows (https://github.com/ansible-collections/community.docker/pull/619)."
+  - "vendored Docker SDK for Python code - respect timeouts on Windows named pipes (https://github.com/ansible-collections/community.docker/pull/619)."
diff --git a/plugins/module_utils/_api/transport/npipesocket.py b/plugins/module_utils/_api/transport/npipesocket.py
index f9d3ed66..5e5a90da 100644
--- a/plugins/module_utils/_api/transport/npipesocket.py
+++ b/plugins/module_utils/_api/transport/npipesocket.py
@@ -21,6 +21,9 @@ PYWIN32_IMPORT_ERROR = None
 try:
     import win32file
     import win32pipe
+    import pywintypes
+    import win32event
+    import win32api
 except ImportError:
     PYWIN32_IMPORT_ERROR = traceback.format_exc()
 
@@ -74,7 +77,9 @@ class NpipeSocket(object):
                 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:
@@ -154,11 +159,22 @@ class NpipeSocket(object):
         if not isinstance(buf, memoryview):
             readbuf = memoryview(buf)
 
-        err, data = win32file.ReadFile(
-            self._handle,
-            readbuf[:nbytes] if nbytes else readbuf
-        )
-        return len(data)
+        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)
 
     def _recv_into_py2(self, buf, nbytes):
         err, data = win32file.ReadFile(self._handle, nbytes or len(buf))
@@ -168,8 +184,18 @@ class NpipeSocket(object):
 
     @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):
@@ -188,15 +214,12 @@ class NpipeSocket(object):
     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
diff --git a/plugins/module_utils/_api/utils/socket.py b/plugins/module_utils/_api/utils/socket.py
index 81c0c4f8..9193ce30 100644
--- a/plugins/module_utils/_api/utils/socket.py
+++ b/plugins/module_utils/_api/utils/socket.py
@@ -15,6 +15,7 @@ import os
 import select
 import socket as pysocket
 import struct
+import sys
 
 from ansible.module_utils.six import PY3, binary_type
 
@@ -42,7 +43,13 @@ def read(socket, n=4096):
     recoverable_errors = (errno.EINTR, errno.EDEADLK, errno.EWOULDBLOCK)
 
     if PY3 and not isinstance(socket, NpipeSocket):
-        select.select([socket], [], [])
+        if sys.platform == 'win32':
+            # Limited to 1024
+            select.select([socket], [], [])
+        else:
+            poll = select.poll()
+            poll.register(socket, select.POLLIN | select.POLLPRI)
+            poll.poll()
 
     try:
         if hasattr(socket, 'recv'):
@@ -60,7 +67,7 @@ def read(socket, n=4096):
         if is_pipe_ended:
             # npipes don't support duplex sockets, so we interpret
             # a PIPE_ENDED error as a close operation (0-length read).
-            return 0
+            return ''
         raise