test_multiprocessing_spawn.test_processes flaky tests on Windows

[colesbury]

Bug report

Seen on Windows free threading build, but not free threading specific:

https://github.com/python/cpython/actions/runs/13606907022/job/38039459160?pr=130732

This is in Windows specific code. Not yet sure if this is free threading specific.

ERROR: test_async_error_callback (test.test_multiprocessing_spawn.test_processes.WithProcessesTestPoolWorkerErrors.test_async_error_callback)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "D:\a\cpython\cpython\Lib\test\_test_multiprocessing.py", line 3101, in test_async_error_callback
    p.close()
    ~~~~~~~^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\pool.py", line 652, in close
    self._change_notifier.put(None)
    ~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\queues.py", line 394, in put
    self._writer.send_bytes(obj)
    ~~~~~~~~~~~~~~~~~~~~~~~^^^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\connection.py", line 206, in send_bytes
    self._send_bytes(m[offset:offset + size])
    ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\connection.py", line 293, in _send_bytes
    raise ValueError("concurrent send_bytes() calls "
                     "are not supported")
ValueError: concurrent send_bytes() calls are not supported

Also `test_traceback`:

ERROR: test_traceback (test.test_multiprocessing_spawn.test_processes.WithProcessesTestPool.test_traceback)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "D:\a\cpython\cpython\Lib\test\_test_multiprocessing.py", line 2974, in test_traceback
    with self.Pool(1) as p:
         ~~~~~~~~~^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\pool.py", line 739, in __exit__
    self.terminate()
    ~~~~~~~~~~~~~~^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\pool.py", line 657, in terminate
    self._terminate()
    ~~~~~~~~~~~~~~~^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\util.py", line 216, in __call__
    res = self._callback(*self._args, **self._kwargs)
  File "D:\a\cpython\cpython\Lib\multiprocessing\pool.py", line 690, in _terminate_pool
    change_notifier.put(None)
    ~~~~~~~~~~~~~~~~~~~^^^^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\queues.py", line 394, in put
    self._writer.send_bytes(obj)
    ~~~~~~~~~~~~~~~~~~~~~~~^^^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\connection.py", line 206, in send_bytes
    self._send_bytes(m[offset:offset + size])
    ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\a\cpython\cpython\Lib\multiprocessing\connection.py", line 293, in _send_bytes
    raise ValueError("concurrent send_bytes() calls "
                     "are not supported")
ValueError: concurrent send_bytes() calls are not supported

[colesbury]

The Windows implementation of PipeConnection doesn't support concurrent calls:

cpython/Lib/multiprocessing/connection.py

Lines 291 to 294 in 7c3692f

	if self._send_ov is not None:
	# A connection should only be used by a single thread
	raise ValueError("concurrent send_bytes() calls "
	"are not supported")

(Added by @vstinner in #109244)

But multiprocessing's SimpleQueue has always skipped locking because "writes to a message oriented win32 pipe are atomic":

cpython/Lib/multiprocessing/queues.py

Lines 392 to 394 in 7c3692f

	if self._wlock is None:
	# writes to a message oriented win32 pipe are atomic
	self._writer.send_bytes(obj)

[colesbury]

My guess is that we need at least a threading.Lock() for self._wlock in SimpleQueue on Windows. I'm not sure we need an inter-process multiprocessing lock.

[zooba]

If the underlying write is atomic (which I believe is true, but would have to double-check), then a simple lock around the code that sets and clears _send_ov would seem to be all that's needed, yeah.

[vstinner]

The Windows implementation of PipeConnection doesn't support concurrent calls

In #109244, my main need was to be able to cancel an on-going write by calling ov.cancel() in close(). For that, I added a new self._send_ov attribute to share the overlapped operation between _send_bytes() and close(). If you need/want to support multiple "concurrent" writes, you should replace self._send_ov attribute with a list of overlapped operations.