ConflictDetector: kill both tasks

newsfragments/2540.feature.rst

@@ -0,0 +1 @@
+ConflictDetector now kills all tasks involved in a conflict.

trio/_util.py

@@ -180,15 +180,20 @@ class ConflictDetector:
  def __init__(self, msg):
  self._msg = msg
  self._held = False
+ self._conflicted = False
 
  def __enter__(self):
  if self._held:
+ self._conflicted = True
  raise trio.BusyResourceError(self._msg)
  else:
  self._held = True
 
  def __exit__(self, *args):
  self._held = False
+ if self._conflicted:
+ self._conflicted = False
+ raise trio.BusyResourceError(self._msg)
 
 
 def async_wraps(cls, wrapped_cls, attr_name):

trio/testing/_check_streams.py

@@ -113,7 +113,9 @@ async def send_empty_then_y():
  nursery.start_soon(do_send_all, b"x")
  assert await do_receive_some(None) == b"x"
 
- with _assert_raises(_core.BusyResourceError):
+ with _assert_raises(
+ (_core.BusyResourceError, _core._multierror.NonBaseMultiError)
+ ):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(do_receive_some, 1)
  nursery.start_soon(do_receive_some, 1)
@@ -307,7 +309,7 @@ async def receiver():
 
  async with _ForceCloseBoth(await clogged_stream_maker()) as (s, r):
  # simultaneous wait_send_all_might_not_block fails
- with _assert_raises(_core.BusyResourceError):
+ with _assert_raises(_core._multierror.NonBaseMultiError):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.wait_send_all_might_not_block)
  nursery.start_soon(s.wait_send_all_might_not_block)
@@ -316,15 +318,15 @@ async def receiver():
  # this test might destroy the stream b/c we end up cancelling
  # send_all and e.g. SSLStream can't handle that, so we have to
  # recreate afterwards)
- with _assert_raises(_core.BusyResourceError):
+ with _assert_raises(_core._multierror.NonBaseMultiError):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.wait_send_all_might_not_block)
  nursery.start_soon(s.send_all, b"123")
 
  async with _ForceCloseBoth(await clogged_stream_maker()) as (s, r):
  # send_all and send_all blocked simultaneously should also raise
  # (but again this might destroy the stream)
- with _assert_raises(_core.BusyResourceError):
+ with _assert_raises(_core._multierror.NonBaseMultiError):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.send_all, b"123")
  nursery.start_soon(s.send_all, b"123")
@@ -496,7 +498,9 @@ async def expect_x_then_eof(r):
  if clogged_stream_maker is not None:
  async with _ForceCloseBoth(await clogged_stream_maker()) as (s1, s2):
  # send_all and send_eof simultaneously is not ok
- with _assert_raises(_core.BusyResourceError):
+ with _assert_raises(
+ (_core.BusyResourceError, _core._multierror.NonBaseMultiError)
+ ):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s1.send_all, b"x")
  await _core.wait_all_tasks_blocked()
@@ -505,7 +509,9 @@ async def expect_x_then_eof(r):
  async with _ForceCloseBoth(await clogged_stream_maker()) as (s1, s2):
  # wait_send_all_might_not_block and send_eof simultaneously is not
  # ok either
- with _assert_raises(_core.BusyResourceError):
+ with _assert_raises(
+ (_core.BusyResourceError, _core._multierror.NonBaseMultiError)
+ ):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s1.wait_send_all_might_not_block)
  await _core.wait_all_tasks_blocked()

trio/tests/test_signals.py

@@ -65,7 +65,7 @@ async def naughty():
 
 
 async def test_open_signal_receiver_conflict():
- with pytest.raises(trio.BusyResourceError):
+ with pytest.raises(_core._multierror.NonBaseMultiError):
  with open_signal_receiver(signal.SIGILL) as receiver:
  async with trio.open_nursery() as nursery:
  nursery.start_soon(receiver.__anext__)

trio/tests/test_ssl.py

@@ -354,28 +354,28 @@ async def test_PyOpenSSLEchoStream_gives_resource_busy_errors():
  # PyOpenSSLEchoStream will notice and complain.
 
  s = PyOpenSSLEchoStream()
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.send_all, b"x")
  nursery.start_soon(s.send_all, b"x")
  assert "simultaneous" in str(excinfo.value)
 
  s = PyOpenSSLEchoStream()
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.send_all, b"x")
  nursery.start_soon(s.wait_send_all_might_not_block)
  assert "simultaneous" in str(excinfo.value)
 
  s = PyOpenSSLEchoStream()
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.wait_send_all_might_not_block)
  nursery.start_soon(s.wait_send_all_might_not_block)
  assert "simultaneous" in str(excinfo.value)
 
  s = PyOpenSSLEchoStream()
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(s.receive_some, 1)
  nursery.start_soon(s.receive_some, 1)
@@ -732,28 +732,28 @@ async def do_wait_send_all_might_not_block():
  await s.wait_send_all_might_not_block()
 
  s, _ = ssl_lockstep_stream_pair(client_ctx)
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(do_send_all)
  nursery.start_soon(do_send_all)
  assert "another task" in str(excinfo.value)
 
  s, _ = ssl_lockstep_stream_pair(client_ctx)
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(do_receive_some)
  nursery.start_soon(do_receive_some)
  assert "another task" in str(excinfo.value)
 
  s, _ = ssl_lockstep_stream_pair(client_ctx)
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(do_send_all)
  nursery.start_soon(do_wait_send_all_might_not_block)
  assert "another task" in str(excinfo.value)
 
  s, _ = ssl_lockstep_stream_pair(client_ctx)
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(do_wait_send_all_might_not_block)
  nursery.start_soon(do_wait_send_all_might_not_block)

trio/tests/test_testing.py

@@ -288,7 +288,7 @@ async def getter(expect):
  nursery.start_soon(putter, b"xyz")
 
  # Two gets at the same time -> BusyResourceError
- with pytest.raises(_core.BusyResourceError):
+ with pytest.raises(_core._multierror.NonBaseMultiError):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(getter, b"asdf")
  nursery.start_soon(getter, b"asdf")
@@ -359,7 +359,7 @@ async def do_send_all_count_resourcebusy():
  nursery.start_soon(do_send_all_count_resourcebusy)
  nursery.start_soon(do_send_all_count_resourcebusy)
 
- assert resource_busy_count == 1
+ assert resource_busy_count == 2
 
  with assert_checkpoints():
  await mss.aclose()
@@ -422,7 +422,7 @@ async def do_receive_some(max_bytes):
  mrs.put_data(b"abc")
  assert await do_receive_some(None) == b"abc"
 
- with pytest.raises(_core.BusyResourceError):
+ with pytest.raises(_core._multierror.NonBaseMultiError):
  async with _core.open_nursery() as nursery:
  nursery.start_soon(do_receive_some, 10)
  nursery.start_soon(do_receive_some, 10)

trio/tests/test_util.py

@@ -53,7 +53,7 @@ async def wait_with_ul1():
  with ul1:
  await wait_all_tasks_blocked()
 
- with pytest.raises(_core.BusyResourceError) as excinfo:
+ with pytest.raises(_core._multierror.NonBaseMultiError) as excinfo:
  async with _core.open_nursery() as nursery:
  nursery.start_soon(wait_with_ul1)
  nursery.start_soon(wait_with_ul1)