[pypy-commit] pypy stm-gc: Add 'transaction.remove_epoll()'.

[arigo]

Author: Armin Rigo <arigo at tunes.org>
Branch: stm-gc
Changeset: r54111:22ab9f05fb91
Date: 2012-03-31 17:58 +0200
http://bitbucket.org/pypy/pypy/changeset/22ab9f05fb91/

Log:	Add 'transaction.remove_epoll()'.

diff --git a/lib_pypy/transaction.py b/lib_pypy/transaction.py
--- a/lib_pypy/transaction.py
+++ b/lib_pypy/transaction.py
@@ -15,6 +15,9 @@
     _pending[r] = (f, args)
 
 def add_epoll(ep, callback):
+    for key, (f, args) in _pending.items():
+        if getattr(f, '_reads_from_epoll_', None) is ep:
+            raise ValueError("add_epoll(ep): ep is already registered")
     def poll_reader():
         # assume only one epoll is added.  If the _pending list is
         # now empty, wait.  If not, then just poll non-blockingly.
@@ -26,8 +29,17 @@
         for fd, events in got:
             add(callback, fd, events)
         add(poll_reader)
+    poll_reader._reads_from_epoll_ = ep
     add(poll_reader)
 
+def remove_epoll(ep):
+    for key, (f, args) in _pending.items():
+        if getattr(f, '_reads_from_epoll_', None) is ep:
+            del _pending[key]
+            break
+    else:
+        raise ValueError("remove_epoll(ep): ep is not registered")
+
 def run():
     pending = _pending
     try:
diff --git a/pypy/module/transaction/__init__.py b/pypy/module/transaction/__init__.py
--- a/pypy/module/transaction/__init__.py
+++ b/pypy/module/transaction/__init__.py
@@ -9,7 +9,8 @@
         'set_num_threads': 'interp_transaction.set_num_threads',
         'add': 'interp_transaction.add',
         'run': 'interp_transaction.run',
-        'add_epoll': 'interp_epoll.add_epoll',   # xxx linux only
+        'add_epoll': 'interp_epoll.add_epoll',        # xxx linux only
+        'remove_epoll': 'interp_epoll.remove_epoll',  # xxx linux only
     }
 
     appleveldefs = {
diff --git a/pypy/module/transaction/interp_epoll.py b/pypy/module/transaction/interp_epoll.py
--- a/pypy/module/transaction/interp_epoll.py
+++ b/pypy/module/transaction/interp_epoll.py
@@ -35,6 +35,7 @@
         self.evs = lltype.malloc(rffi.CArray(epoll_event), self.maxevents,
                                  flavor='raw', add_memory_pressure=True,
                                  track_allocation=False)
+        self.force_quit = False
 
     def __del__(self):
         evs = self.evs
@@ -45,7 +46,7 @@
     def run(self):
         # This code is run non-transactionally.  Careful, no GC available.
         state = interp_transaction.state
-        if state.has_exception():
+        if state.has_exception() or self.force_quit:
             return
         fd = rffi.cast(rffi.INT, self.epoller.epfd)
         maxevents = rffi.cast(rffi.INT, self.maxevents)
@@ -107,4 +108,25 @@
 
 @unwrap_spec(epoller=W_Epoll)
 def add_epoll(space, epoller, w_callback):
-    EPollPending(space, epoller, w_callback).register()
+    state = interp_transaction.state
+    if state.epolls is None:
+        state.epolls = {}
+    elif epoller in state.epolls:
+        raise OperationError(space.w_ValueError,
+            space.wrap("add_epoll(ep): ep is already registered"))
+    pending = EPollPending(space, epoller, w_callback)
+    state.epolls[epoller] = pending
+    pending.register()
+
+ at unwrap_spec(epoller=W_Epoll)
+def remove_epoll(space, epoller):
+    state = interp_transaction.state
+    if state.epolls is None:
+        pending = None
+    else:
+        pending = state.epolls.get(epoller, None)
+    if pending is None:
+        raise OperationError(space.w_ValueError,
+            space.wrap("remove_epoll(ep): ep is not registered"))
+    pending.force_quit = True
+    del state.epolls[epoller]
diff --git a/pypy/module/transaction/interp_transaction.py b/pypy/module/transaction/interp_transaction.py
--- a/pypy/module/transaction/interp_transaction.py
+++ b/pypy/module/transaction/interp_transaction.py
@@ -23,6 +23,7 @@
         self.ll_no_tasks_pending_lock = threadintf.null_ll_lock
         self.ll_unfinished_lock = threadintf.null_ll_lock
         self.threadobjs = {}      # empty during translation
+        self.epolls = None
         self.pending = Fifo()
 
     def _freeze_(self):
@@ -305,6 +306,7 @@
     assert not state.is_locked_no_tasks_pending()
     state.clear_all_values_apart_from_main()
     state.running = False
+    state.epolls = None
     #
     # now re-raise the exception that we got in a transaction
     state.close_exceptions()
diff --git a/pypy/module/transaction/test/test_epoll.py b/pypy/module/transaction/test/test_epoll.py
--- a/pypy/module/transaction/test/test_epoll.py
+++ b/pypy/module/transaction/test/test_epoll.py
@@ -50,6 +50,26 @@
         raises(Done, transaction.run)
         assert steps == ['write_stuff', 'callback']
 
+    def test_remove_closed_epoll(self):
+        import transaction, select, posix as os
+
+        fd_read, fd_write = os.pipe()
+
+        epoller = select.epoll()
+        epoller.register(fd_read)
+
+        # we run it 10 times in order to get both possible orders in
+        # the emulator
+        for i in range(10):
+            transaction.add_epoll(epoller, lambda *args: not_actually_callable)
+            transaction.add(transaction.remove_epoll, epoller)
+            transaction.run()
+            # assert didn't deadlock
+            transaction.add(transaction.remove_epoll, epoller)
+            transaction.add_epoll(epoller, lambda *args: not_actually_callable)
+            transaction.run()
+            # assert didn't deadlock
+
 
 class AppTestEpollEmulator(AppTestEpoll):
     def setup_class(cls):