commit: 916a0733c7201b7a8b22f5262bd5be8cbc8992a6 Author: Zac Medico <zmedico <AT> gentoo <DOT> org> AuthorDate: Sun Apr 2 20:50:09 2017 +0000 Commit: Zac Medico <zmedico <AT> gentoo <DOT> org> CommitDate: Mon Apr 3 20:07:59 2017 +0000 URL: https://gitweb.gentoo.org/proj/portage.git/commit/?id=916a0733
AsynchronousLock: add async_unlock method (bug 614108) Add an async_unlock method, in order to avoid event loop recursion which is incompatible with asyncio. X-Gentoo-bug: 614108 X-Gentoo-bug-url: https://bugs.gentoo.org/show_bug.cgi?id=614108 Acked-by: Brian Dolbec <dolsen <AT> gentoo.org> pym/_emerge/AsynchronousLock.py | 89 +++++++++++++++++++++-- pym/portage/tests/locks/test_asynchronous_lock.py | 15 +++- 2 files changed, 92 insertions(+), 12 deletions(-) diff --git a/pym/_emerge/AsynchronousLock.py b/pym/_emerge/AsynchronousLock.py index c0b9b26dc..6a32d2d40 100644 --- a/pym/_emerge/AsynchronousLock.py +++ b/pym/_emerge/AsynchronousLock.py @@ -35,7 +35,7 @@ class AsynchronousLock(AsynchronousTask): __slots__ = ('path', 'scheduler',) + \ ('_imp', '_force_async', '_force_dummy', '_force_process', \ - '_force_thread') + '_force_thread', '_unlock_future') _use_process_by_default = True @@ -84,6 +84,11 @@ class AsynchronousLock(AsynchronousTask): return self.returncode def unlock(self): + """ + This method is deprecated in favor of async_unlock, since waiting + for the child process to respond can trigger event loop recursion + which is incompatible with asyncio. + """ if self._imp is None: raise AssertionError('not locked') if isinstance(self._imp, (_LockProcess, _LockThread)): @@ -92,6 +97,28 @@ class AsynchronousLock(AsynchronousTask): unlockfile(self._imp) self._imp = None + def async_unlock(self): + """ + Release the lock asynchronously. Release notification is available + via the add_done_callback method of the returned Future instance. + + @returns: Future, result is None + """ + if self._imp is None: + raise AssertionError('not locked') + if self._unlock_future is not None: + raise AssertionError("already unlocked") + if isinstance(self._imp, (_LockProcess, _LockThread)): + unlock_future = self._imp.async_unlock() + else: + unlockfile(self._imp) + unlock_future = self.scheduler.create_future() + self.scheduler.call_soon(unlock_future.set_result, None) + self._imp = None + self._unlock_future = unlock_future + return unlock_future + + class _LockThread(AbstractPollTask): """ This uses the portage.locks module to acquire a lock asynchronously, @@ -105,7 +132,7 @@ class _LockThread(AbstractPollTask): """ __slots__ = ('path',) + \ - ('_force_dummy', '_lock_obj', '_thread',) + ('_force_dummy', '_lock_obj', '_thread', '_unlock_future') def _start(self): self._registered = True @@ -132,13 +159,35 @@ class _LockThread(AbstractPollTask): pass def unlock(self): + """ + This method is deprecated in favor of async_unlock, for compatibility + with _LockProcess. + """ + self._unlock() + self._unlock_future.set_result(None) + + def _unlock(self): if self._lock_obj is None: raise AssertionError('not locked') if self.returncode is None: raise AssertionError('lock not acquired yet') + if self._unlock_future is not None: + raise AssertionError("already unlocked") + self._unlock_future = self.scheduler.create_future() unlockfile(self._lock_obj) self._lock_obj = None + def async_unlock(self): + """ + Release the lock asynchronously. Release notification is available + via the add_done_callback method of the returned Future instance. + + @returns: Future, result is None + """ + self._unlock() + self.scheduler.call_soon(self._unlock_future.set_result, None) + return self._unlock_future + def _unregister(self): self._registered = False @@ -156,7 +205,8 @@ class _LockProcess(AbstractPollTask): """ __slots__ = ('path',) + \ - ('_acquired', '_kill_test', '_proc', '_files', '_reg_id', '_unlocked') + ('_acquired', '_kill_test', '_proc', '_files', + '_reg_id','_unlock_future') def _start(self): in_pr, in_pw = os.pipe() @@ -223,13 +273,16 @@ class _LockProcess(AbstractPollTask): return if not self.cancelled and \ - not self._unlocked: + self._unlock_future is None: # We don't want lost locks going unnoticed, so it's # only safe to ignore if either the cancel() or # unlock() methods have been previously called. raise AssertionError("lock process failed with returncode %s" \ % (proc.returncode,)) + if self._unlock_future is not None: + self._unlock_future.set_result(None) + def _cancel(self): if self._proc is not None: self._proc.cancel() @@ -271,16 +324,36 @@ class _LockProcess(AbstractPollTask): os.close(pipe_in) def unlock(self): + """ + This method is deprecated in favor of async_unlock, since waiting + for the child process to respond can trigger event loop recursion + which is incompatible with asyncio. + """ + self._unlock() + self._proc.wait() + self._proc = None + + def _unlock(self): if self._proc is None: raise AssertionError('not locked') - if self.returncode is None: + if not self._acquired: raise AssertionError('lock not acquired yet') if self.returncode != os.EX_OK: raise AssertionError("lock process failed with returncode %s" \ % (self.returncode,)) - self._unlocked = True + if self._unlock_future is not None: + raise AssertionError("already unlocked") + self._unlock_future = self.scheduler.create_future() os.write(self._files['pipe_out'], b'\0') os.close(self._files['pipe_out']) self._files = None - self._proc.wait() - self._proc = None + + def async_unlock(self): + """ + Release the lock asynchronously. Release notification is available + via the add_done_callback method of the returned Future instance. + + @returns: Future, result is None + """ + self._unlock() + return self._unlock_future diff --git a/pym/portage/tests/locks/test_asynchronous_lock.py b/pym/portage/tests/locks/test_asynchronous_lock.py index 3a2ccfb84..ab67242d5 100644 --- a/pym/portage/tests/locks/test_asynchronous_lock.py +++ b/pym/portage/tests/locks/test_asynchronous_lock.py @@ -1,6 +1,7 @@ # Copyright 2010-2011 Gentoo Foundation # Distributed under the terms of the GNU General Public License v2 +import itertools import signal import tempfile @@ -17,7 +18,8 @@ class AsynchronousLockTestCase(TestCase): tempdir = tempfile.mkdtemp() try: path = os.path.join(tempdir, 'lock_me') - for force_async in (True, False): + for force_async, async_unlock in itertools.product( + (True, False), repeat=2): for force_dummy in (True, False): async_lock = AsynchronousLock(path=path, scheduler=scheduler, _force_async=force_async, @@ -26,7 +28,10 @@ class AsynchronousLockTestCase(TestCase): async_lock.start() self.assertEqual(async_lock.wait(), os.EX_OK) self.assertEqual(async_lock.returncode, os.EX_OK) - async_lock.unlock() + if async_unlock: + scheduler.run_until_complete(async_lock.async_unlock()) + else: + async_lock.unlock() async_lock = AsynchronousLock(path=path, scheduler=scheduler, _force_async=force_async, @@ -34,8 +39,10 @@ class AsynchronousLockTestCase(TestCase): async_lock.start() self.assertEqual(async_lock.wait(), os.EX_OK) self.assertEqual(async_lock.returncode, os.EX_OK) - async_lock.unlock() - + if async_unlock: + scheduler.run_until_complete(async_lock.async_unlock()) + else: + async_lock.unlock() finally: shutil.rmtree(tempdir)
