| # Copyright 2010-2020 Gentoo Authors |
| # Distributed under the terms of the GNU General Public License v2 |
| |
| import fcntl |
| import logging |
| import sys |
| |
| try: |
| import dummy_threading |
| except ImportError: |
| dummy_threading = None |
| |
| try: |
| import threading |
| except ImportError: |
| threading = dummy_threading |
| |
| import portage |
| from portage import os |
| from portage.exception import TryAgain |
| from portage.localization import _ |
| from portage.locks import lockfile, unlockfile |
| from portage.util import writemsg_level |
| from _emerge.AbstractPollTask import AbstractPollTask |
| from _emerge.AsynchronousTask import AsynchronousTask |
| from _emerge.SpawnProcess import SpawnProcess |
| |
| |
| class AsynchronousLock(AsynchronousTask): |
| """ |
| This uses the portage.locks module to acquire a lock asynchronously, |
| using either a thread (if available) or a subprocess. |
| |
| The default behavior is to use a process instead of a thread, since |
| there is currently no way to interrupt a thread that is waiting for |
| a lock (notably, SIGINT doesn't work because python delivers all |
| signals to the main thread). |
| """ |
| |
| __slots__ = ("path",) + ( |
| "_imp", |
| "_force_async", |
| "_force_dummy", |
| "_force_process", |
| "_force_thread", |
| "_unlock_future", |
| ) |
| |
| _use_process_by_default = True |
| |
| def _start(self): |
| |
| if not self._force_async: |
| try: |
| self._imp = lockfile( |
| self.path, wantnewlockfile=True, flags=os.O_NONBLOCK |
| ) |
| except TryAgain: |
| pass |
| else: |
| self.returncode = os.EX_OK |
| self._async_wait() |
| return |
| |
| if self._force_process or ( |
| not self._force_thread |
| and (self._use_process_by_default or threading is dummy_threading) |
| ): |
| self._imp = _LockProcess(path=self.path, scheduler=self.scheduler) |
| else: |
| self._imp = _LockThread( |
| path=self.path, scheduler=self.scheduler, _force_dummy=self._force_dummy |
| ) |
| |
| self._imp.addExitListener(self._imp_exit) |
| self._imp.start() |
| |
| def _imp_exit(self, imp): |
| # call exit listeners |
| self.returncode = imp.returncode |
| self._async_wait() |
| |
| def _cancel(self): |
| if isinstance(self._imp, AsynchronousTask): |
| self._imp.cancel() |
| |
| def _poll(self): |
| if isinstance(self._imp, AsynchronousTask): |
| self._imp.poll() |
| return self.returncode |
| |
| 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, |
| using a background thread. After the lock is acquired, the thread |
| writes to a pipe in order to notify a poll loop running in the main |
| thread. |
| |
| If the threading module is unavailable then the dummy_threading |
| module will be used, and the lock will be acquired synchronously |
| (before the start() method returns). |
| """ |
| |
| __slots__ = ("path",) + ("_force_dummy", "_lock_obj", "_thread", "_unlock_future") |
| |
| def _start(self): |
| self._registered = True |
| threading_mod = threading |
| if self._force_dummy: |
| threading_mod = dummy_threading |
| self._thread = threading_mod.Thread(target=self._run_lock) |
| self._thread.daemon = True |
| self._thread.start() |
| |
| def _run_lock(self): |
| self._lock_obj = lockfile(self.path, wantnewlockfile=True) |
| # Thread-safe callback to EventLoop |
| self.scheduler.call_soon_threadsafe(self._run_lock_cb) |
| |
| def _run_lock_cb(self): |
| self._unregister() |
| self.returncode = os.EX_OK |
| self._async_wait() |
| |
| def _cancel(self): |
| # There's currently no way to force thread termination. |
| pass |
| |
| 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 |
| |
| if self._thread is not None: |
| self._thread.join() |
| self._thread = None |
| |
| |
| class _LockProcess(AbstractPollTask): |
| """ |
| This uses the portage.locks module to acquire a lock asynchronously, |
| using a subprocess. After the lock is acquired, the process |
| writes to a pipe in order to notify a poll loop running in the main |
| process. The unlock() method notifies the subprocess to release the |
| lock and exit. |
| """ |
| |
| __slots__ = ("path",) + ( |
| "_acquired", |
| "_kill_test", |
| "_proc", |
| "_files", |
| "_unlock_future", |
| ) |
| |
| def _start(self): |
| in_pr, in_pw = os.pipe() |
| out_pr, out_pw = os.pipe() |
| self._files = {} |
| self._files["pipe_in"] = in_pr |
| self._files["pipe_out"] = out_pw |
| |
| fcntl.fcntl( |
| in_pr, fcntl.F_SETFL, fcntl.fcntl(in_pr, fcntl.F_GETFL) | os.O_NONBLOCK |
| ) |
| |
| self.scheduler.add_reader(in_pr, self._output_handler) |
| self._registered = True |
| self._proc = SpawnProcess( |
| args=[ |
| portage._python_interpreter, |
| os.path.join(portage._bin_path, "lock-helper.py"), |
| self.path, |
| ], |
| env=dict(os.environ, PORTAGE_PYM_PATH=portage._pym_path), |
| fd_pipes={0: out_pr, 1: in_pw, 2: sys.__stderr__.fileno()}, |
| scheduler=self.scheduler, |
| ) |
| self._proc.addExitListener(self._proc_exit) |
| self._proc.start() |
| os.close(out_pr) |
| os.close(in_pw) |
| |
| def _proc_exit(self, proc): |
| |
| if self._files is not None: |
| # Close pipe_out if it's still open, since it's useless |
| # after the process has exited. This helps to avoid |
| # "ResourceWarning: unclosed file" since Python 3.2. |
| try: |
| pipe_out = self._files.pop("pipe_out") |
| except KeyError: |
| pass |
| else: |
| os.close(pipe_out) |
| |
| if proc.returncode != os.EX_OK: |
| # Typically, this will happen due to the |
| # process being killed by a signal. |
| |
| if not self._acquired: |
| # If the lock hasn't been aquired yet, the |
| # caller can check the returncode and handle |
| # this failure appropriately. |
| if not (self.cancelled or self._kill_test): |
| writemsg_level( |
| "_LockProcess: %s\n" |
| % _("failed to acquire lock on '%s'") |
| % (self.path,), |
| level=logging.ERROR, |
| noiselevel=-1, |
| ) |
| self._unregister() |
| self.returncode = proc.returncode |
| self._async_wait() |
| return |
| |
| if not self.cancelled and 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() |
| |
| def _poll(self): |
| if self._proc is not None: |
| self._proc.poll() |
| return self.returncode |
| |
| def _output_handler(self): |
| buf = self._read_buf(self._files["pipe_in"]) |
| if buf: |
| self._acquired = True |
| self._unregister() |
| self.returncode = os.EX_OK |
| self._async_wait() |
| |
| return True |
| |
| def _unregister(self): |
| self._registered = False |
| |
| if self._files is not None: |
| try: |
| pipe_in = self._files.pop("pipe_in") |
| except KeyError: |
| pass |
| else: |
| self.scheduler.remove_reader(pipe_in) |
| os.close(pipe_in) |
| |
| def _unlock(self): |
| if self._proc is None: |
| raise AssertionError("not locked") |
| 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,) |
| ) |
| 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 |
| |
| 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 |
