diff options
Diffstat (limited to 'portage_with_autodep/pym/_emerge/AsynchronousLock.py')
| -rw-r--r-- | portage_with_autodep/pym/_emerge/AsynchronousLock.py | 90 |
1 files changed, 51 insertions, 39 deletions
diff --git a/portage_with_autodep/pym/_emerge/AsynchronousLock.py b/portage_with_autodep/pym/_emerge/AsynchronousLock.py index 637ba73..587aa46 100644 --- a/portage_with_autodep/pym/_emerge/AsynchronousLock.py +++ b/portage_with_autodep/pym/_emerge/AsynchronousLock.py @@ -1,15 +1,16 @@ -# Copyright 2010-2011 Gentoo Foundation +# Copyright 2010-2012 Gentoo Foundation # Distributed under the terms of the GNU General Public License v2 import dummy_threading import fcntl +import errno import logging import sys try: import threading except ImportError: - import dummy_threading as threading + threading = dummy_threading import portage from portage import os @@ -19,7 +20,6 @@ from portage.locks import lockfile, unlockfile from portage.util import writemsg_level from _emerge.AbstractPollTask import AbstractPollTask from _emerge.AsynchronousTask import AsynchronousTask -from _emerge.PollConstants import PollConstants from _emerge.SpawnProcess import SpawnProcess class AsynchronousLock(AsynchronousTask): @@ -35,7 +35,7 @@ class AsynchronousLock(AsynchronousTask): __slots__ = ('path', 'scheduler',) + \ ('_imp', '_force_async', '_force_dummy', '_force_process', \ - '_force_thread', '_waiting') + '_force_thread') _use_process_by_default = True @@ -66,8 +66,7 @@ class AsynchronousLock(AsynchronousTask): def _imp_exit(self, imp): # call exit listeners - if not self._waiting: - self.wait() + self.wait() def _cancel(self): if isinstance(self._imp, AsynchronousTask): @@ -81,9 +80,7 @@ class AsynchronousLock(AsynchronousTask): def _wait(self): if self.returncode is not None: return self.returncode - self._waiting = True self.returncode = self._imp.wait() - self._waiting = False return self.returncode def unlock(self): @@ -114,13 +111,13 @@ class _LockThread(AbstractPollTask): def _start(self): pr, pw = os.pipe() self._files = {} - self._files['pipe_read'] = os.fdopen(pr, 'rb', 0) - self._files['pipe_write'] = os.fdopen(pw, 'wb', 0) - for k, f in self._files.items(): - fcntl.fcntl(f.fileno(), fcntl.F_SETFL, - fcntl.fcntl(f.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK) - self._reg_id = self.scheduler.register(self._files['pipe_read'].fileno(), - PollConstants.POLLIN, self._output_handler) + self._files['pipe_read'] = pr + self._files['pipe_write'] = pw + for f in self._files.values(): + fcntl.fcntl(f, fcntl.F_SETFL, + fcntl.fcntl(f, fcntl.F_GETFL) | os.O_NONBLOCK) + self._reg_id = self.scheduler.register(self._files['pipe_read'], + self.scheduler.IO_IN, self._output_handler) self._registered = True threading_mod = threading if self._force_dummy: @@ -130,26 +127,27 @@ class _LockThread(AbstractPollTask): def _run_lock(self): self._lock_obj = lockfile(self.path, wantnewlockfile=True) - self._files['pipe_write'].write(b'\0') + os.write(self._files['pipe_write'], b'\0') def _output_handler(self, f, event): - buf = self._read_buf(self._files['pipe_read'], event) + buf = None + if event & self.scheduler.IO_IN: + try: + buf = os.read(self._files['pipe_read'], self._bufsize) + except OSError as e: + if e.errno not in (errno.EAGAIN,): + raise if buf: self._unregister() self.returncode = os.EX_OK self.wait() + return True + def _cancel(self): # There's currently no way to force thread termination. pass - def _wait(self): - if self.returncode is not None: - return self.returncode - if self._registered: - self.scheduler.schedule(self._reg_id) - return self.returncode - def unlock(self): if self._lock_obj is None: raise AssertionError('not locked') @@ -171,7 +169,7 @@ class _LockThread(AbstractPollTask): if self._files is not None: for f in self._files.values(): - f.close() + os.close(f) self._files = None class _LockProcess(AbstractPollTask): @@ -190,12 +188,12 @@ class _LockProcess(AbstractPollTask): in_pr, in_pw = os.pipe() out_pr, out_pw = os.pipe() self._files = {} - self._files['pipe_in'] = os.fdopen(in_pr, 'rb', 0) - self._files['pipe_out'] = os.fdopen(out_pw, 'wb', 0) + 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._reg_id = self.scheduler.register(in_pr, - PollConstants.POLLIN, self._output_handler) + self.scheduler.IO_IN, self._output_handler) self._registered = True self._proc = SpawnProcess( args=[portage._python_interpreter, @@ -209,9 +207,22 @@ class _LockProcess(AbstractPollTask): 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 @@ -242,21 +253,22 @@ class _LockProcess(AbstractPollTask): self._proc.poll() return self.returncode - def _wait(self): - if self.returncode is not None: - return self.returncode - if self._registered: - self.scheduler.schedule(self._reg_id) - return self.returncode - def _output_handler(self, f, event): - buf = self._read_buf(self._files['pipe_in'], event) + buf = None + if event & self.scheduler.IO_IN: + try: + buf = os.read(self._files['pipe_in'], self._bufsize) + except OSError as e: + if e.errno not in (errno.EAGAIN,): + raise if buf: self._acquired = True self._unregister() self.returncode = os.EX_OK self.wait() + return True + def _unregister(self): self._registered = False @@ -270,7 +282,7 @@ class _LockProcess(AbstractPollTask): except KeyError: pass else: - pipe_in.close() + os.close(pipe_in) def unlock(self): if self._proc is None: @@ -281,8 +293,8 @@ class _LockProcess(AbstractPollTask): raise AssertionError("lock process failed with returncode %s" \ % (self.returncode,)) self._unlocked = True - self._files['pipe_out'].write(b'\0') - self._files['pipe_out'].close() + os.write(self._files['pipe_out'], b'\0') + os.close(self._files['pipe_out']) self._files = None self._proc.wait() self._proc = None |