From f552a3f8f5c98fb23b5a15c74a2f8a3ee070056b Mon Sep 17 00:00:00 2001
From: Guillermo Rodriguez <guillermo@telos.net>
Date: Thu, 13 Mar 2025 22:43:02 -0300
Subject: [PATCH] Handle cancelation on EventFD.read
---
tests/test_ringbuf.py | 7 ++-----
tractor/ipc/_linux.py | 6 ++++--
tractor/ipc/_ringbuf.py | 6 +++---
3 files changed, 9 insertions(+), 10 deletions(-)
diff --git a/tests/test_ringbuf.py b/tests/test_ringbuf.py
index 1c4e88f9..9e457b2a 100644
--- a/tests/test_ringbuf.py
+++ b/tests/test_ringbuf.py
@@ -4,7 +4,6 @@ import trio
import pytest
import tractor
from tractor.ipc import (
- EFD_NONBLOCK,
open_eventfd,
RingBuffSender,
RingBuffReceiver
@@ -95,7 +94,7 @@ async def child_write_shm(
'large_payloads_large_buffer',
]
)
-def test_ring_buff(
+def test_ringbuf(
msg_amount: int,
rand_min: int,
rand_max: int,
@@ -171,8 +170,7 @@ async def child_blocked_receiver(
def test_ring_reader_cancel():
- flags = EFD_NONBLOCK
- write_eventfd = open_eventfd(flags=flags)
+ write_eventfd = open_eventfd()
wrap_eventfd = open_eventfd()
proc_kwargs = {
@@ -201,7 +199,6 @@ def test_ring_reader_cancel():
write_eventfd=write_eventfd,
wrap_eventfd=wrap_eventfd,
shm_key=shm_key,
- flags=flags
) as (sctx, _sent),
):
await trio.sleep(1)
diff --git a/tractor/ipc/_linux.py b/tractor/ipc/_linux.py
index 12b00260..88d80d1c 100644
--- a/tractor/ipc/_linux.py
+++ b/tractor/ipc/_linux.py
@@ -133,8 +133,10 @@ class EventFD:
return write_eventfd(self._fd, value)
async def read(self) -> int:
- #TODO: how to handle signals?
- return await trio.to_thread.run_sync(read_eventfd, self._fd)
+ return await trio.to_thread.run_sync(
+ read_eventfd, self._fd,
+ abandon_on_cancel=True
+ )
def open(self):
self._fobj = os.fdopen(self._fd, self._omode)
diff --git a/tractor/ipc/_ringbuf.py b/tractor/ipc/_ringbuf.py
index 50a9eff1..77a30ab8 100644
--- a/tractor/ipc/_ringbuf.py
+++ b/tractor/ipc/_ringbuf.py
@@ -45,7 +45,7 @@ class RingBuffSender(trio.abc.SendStream):
wrap_eventfd: int,
start_ptr: int = 0,
buf_size: int = 10 * 1024,
- clean_shm_on_exit: bool = True
+ unlink_on_exit: bool = True
):
self._shm = SharedMemory(
name=shm_key,
@@ -55,7 +55,7 @@ class RingBuffSender(trio.abc.SendStream):
self._write_event = EventFD(write_eventfd, 'w')
self._wrap_event = EventFD(wrap_eventfd, 'r')
self._ptr = start_ptr
- self.clean_shm_on_exit = clean_shm_on_exit
+ self.unlink_on_exit = unlink_on_exit
@property
def key(self) -> str:
@@ -104,7 +104,7 @@ class RingBuffSender(trio.abc.SendStream):
async def aclose(self):
self._write_event.close()
self._wrap_event.close()
- if self.clean_shm_on_exit:
+ if self.unlink_on_exit:
self._shm.unlink()
else: