pym/_emerge/AsynchronousLock.py

# Copyright 2010 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2

import dummy_threading
import fcntl
try:
	import threading
except ImportError:
	import dummy_threading as threading

from portage import os
from portage.exception import TryAgain
from portage.locks import lockfile, unlockfile
from _emerge.AbstractPollTask import AbstractPollTask
from _emerge.PollConstants import PollConstants

class AsynchronousLock(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__ = ('lock_obj', 'path',) + \
		('_files', '_force_thread', '_force_dummy',
		'_thread', '_reg_id',)

	def _start(self):

		if self._force_thread:
			self._start_thread()
			return

		try:
			self.lock_obj = lockfile(self.path,
				wantnewlockfile=True, flags=os.O_NONBLOCK)
		except TryAgain:
			self._start_thread()
		else:
			self.returncode = os.EX_OK
			self.wait()

	def _start_thread(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._registered = True
		threading_mod = threading
		if self._force_dummy:
			threading_mod = dummy_threading
		self._thread = threading_mod.Thread(target=self._run_lock)
		self._thread.start()

	def _run_lock(self):
		self.lock_obj = lockfile(self.path, wantnewlockfile=True)
		self._files['pipe_write'].write(b'\0')

	def _output_handler(self, f, event):
		buf = self._read_buf(self._files['pipe_read'], event)
		if buf:
			self._unregister()
			self.returncode = os.EX_OK
			self.wait()

	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')
		unlockfile(self.lock_obj)
		self.lock_obj = None

	def _unregister(self):
		self._registered = False

		if self._thread is not None:
			self._thread.join()
			self._thread = None

		if self._reg_id is not None:
			self.scheduler.unregister(self._reg_id)
			self._reg_id = None

		if self._files is not None:
			for f in self._files.values():
				f.close()
			self._files = None