[gentoo-commits] proj/portage:master commit in: pym/portage/dbapi/, pym/_emerge/

[Zac Medico]

commit:     610831146c27d46256df45f127474d509c9a7d31
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Wed Feb  8 00:36:32 2012 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Wed Feb  8 01:15:47 2012 +0000
URL:        http://git.overlays.gentoo.org/gitweb/?p=proj/portage.git;a=commit;h=61083114

PollScheduler: glib.io_add_watch() compatibility

---
 pym/_emerge/AsynchronousLock.py    |    4 +++
 pym/_emerge/EbuildIpcDaemon.py     |    2 +
 pym/_emerge/EbuildMetadataPhase.py |    2 +
 pym/_emerge/PipeReader.py          |    4 +++
 pym/_emerge/PollScheduler.py       |   50 ++++++++++++++++++++++++-----------
 pym/_emerge/Scheduler.py           |    1 +
 pym/_emerge/SpawnProcess.py        |    4 +++
 pym/portage/dbapi/_MergeProcess.py |    2 +
 8 files changed, 53 insertions(+), 16 deletions(-)

diff --git a/pym/_emerge/AsynchronousLock.py b/pym/_emerge/AsynchronousLock.py
index 3593834..e166df3 100644
--- a/pym/_emerge/AsynchronousLock.py
+++ b/pym/_emerge/AsynchronousLock.py
@@ -143,6 +143,8 @@ class _LockThread(AbstractPollTask):
 			self.returncode = os.EX_OK
 			self.wait()
 
+		return True
+
 	def _cancel(self):
 		# There's currently no way to force thread termination.
 		pass
@@ -280,6 +282,8 @@ class _LockProcess(AbstractPollTask):
 			self.returncode = os.EX_OK
 			self.wait()
 
+		return True
+
 	def _unregister(self):
 		self._registered = False
 

diff --git a/pym/_emerge/EbuildIpcDaemon.py b/pym/_emerge/EbuildIpcDaemon.py
index 5dabe34..6a320cb 100644
--- a/pym/_emerge/EbuildIpcDaemon.py
+++ b/pym/_emerge/EbuildIpcDaemon.py
@@ -84,6 +84,8 @@ class EbuildIpcDaemon(FifoIpcDaemon):
 				if reply_hook is not None:
 					reply_hook()
 
+		return True
+
 	def _send_reply(self, reply):
 		# File streams are in unbuffered mode since we do atomic
 		# read and write of whole pickles. Use non-blocking mode so

diff --git a/pym/_emerge/EbuildMetadataPhase.py b/pym/_emerge/EbuildMetadataPhase.py
index d4f5bc0..f8da866 100644
--- a/pym/_emerge/EbuildMetadataPhase.py
+++ b/pym/_emerge/EbuildMetadataPhase.py
@@ -128,6 +128,8 @@ class EbuildMetadataPhase(SubProcess):
 
 		self._unregister_if_appropriate(event)
 
+		return True
+
 	def _set_returncode(self, wait_retval):
 		SubProcess._set_returncode(self, wait_retval)
 		# self._raw_metadata is None when _start returns

diff --git a/pym/_emerge/PipeReader.py b/pym/_emerge/PipeReader.py
index 0f784cf..a85d794 100644
--- a/pym/_emerge/PipeReader.py
+++ b/pym/_emerge/PipeReader.py
@@ -74,6 +74,8 @@ class PipeReader(AbstractPollTask):
 
 		self._unregister_if_appropriate(event)
 
+		return True
+
 	def _array_output_handler(self, fd, event):
 
 		for f in self.input_files.values():
@@ -93,6 +95,8 @@ class PipeReader(AbstractPollTask):
 
 		self._unregister_if_appropriate(event)
 
+		return True
+
 	def _unregister(self):
 		"""
 		Unregister from the scheduler and close open files.

diff --git a/pym/_emerge/PollScheduler.py b/pym/_emerge/PollScheduler.py
index 519a370..7cf8189 100644
--- a/pym/_emerge/PollScheduler.py
+++ b/pym/_emerge/PollScheduler.py
@@ -24,9 +24,12 @@ from _emerge.PollSelectAdapter import PollSelectAdapter
 class PollScheduler(object):
 
 	class _sched_iface_class(SlotObject):
-		__slots__ = ("output", "register", "schedule",
+		__slots__ = ("io_add_watch", "output", "register", "schedule",
 			"source_remove", "timeout_add", "unregister")
 
+	class _io_handler_class(SlotObject):
+		__slots__ = ("args", "callback", "fd", "source_id")
+
 	class _timeout_handler_class(SlotObject):
 		__slots__ = ("args", "function", "interval", "source_id",
 			"timestamp")
@@ -49,6 +52,7 @@ class PollScheduler(object):
 		self._scheduling = False
 		self._background = False
 		self.sched_iface = self._sched_iface_class(
+			io_add_watch=self._register,
 			output=self._task_output,
 			register=self._register,
 			schedule=self._schedule_wait,
@@ -248,8 +252,9 @@ class PollScheduler(object):
 		try:
 			while event_handlers:
 				f, event = self._next_poll_event()
-				handler, reg_id = event_handlers[f]
-				handler(f, event)
+				x = event_handlers[f]
+				if not x.callback(f, event, *x.args):
+					self._unregister(x.source_id)
 				event_handled = True
 		except StopIteration:
 			event_handled = True
@@ -277,8 +282,9 @@ class PollScheduler(object):
 		try:
 			while event_handlers and self._poll_event_queue:
 				f, event = self._next_poll_event()
-				handler, reg_id = event_handlers[f]
-				handler(f, event)
+				x = event_handlers[f]
+				if not x.callback(f, event, *x.args):
+					self._unregister(x.source_id)
 				events_handled += 1
 		except StopIteration:
 			events_handled += 1
@@ -332,20 +338,31 @@ class PollScheduler(object):
 
 		return bool(ready_timeouts)
 
-	def _register(self, f, eventmask, handler):
+	def _register(self, f, condition, callback, *args):
 		"""
-		@rtype: Integer
-		@return: A unique registration id, for use in schedule() or
-			unregister() calls.
+		Like glib.io_add_watch(), your function should return False to
+		stop being called, or True to continue being called. Any
+		additional positional arguments given here are passed to your
+		function when it's called.
+
+		@type f: int or object with fileno() method
+		@param f: a file descriptor to monitor
+		@type condition: int
+		@param condition: a condition mask
+		@type callback: callable
+		@param callback: a function to call
+		@rtype: int
+		@return: an integer ID of the event source
 		"""
 		if f in self._poll_event_handlers:
 			raise AssertionError("fd %d is already registered" % f)
 		self._event_handler_id += 1
-		reg_id = self._event_handler_id
-		self._poll_event_handler_ids[reg_id] = f
-		self._poll_event_handlers[f] = (handler, reg_id)
-		self._poll_obj.register(f, eventmask)
-		return reg_id
+		source_id = self._event_handler_id
+		self._poll_event_handler_ids[source_id] = f
+		self._poll_event_handlers[f] = self._io_handler_class(
+			args=args, callback=callback, f=f, source_id=source_id)
+		self._poll_obj.register(f, condition)
+		return source_id
 
 	def _unregister(self, reg_id):
 		"""
@@ -409,8 +426,9 @@ class PollScheduler(object):
 			while (wait_ids is None and event_handlers) or \
 				(wait_ids is not None and wait_ids.intersection(handler_ids)):
 				f, event = self._next_poll_event(timeout=remaining_timeout)
-				handler, reg_id = event_handlers[f]
-				handler(f, event)
+				x = event_handlers[f]
+				if not x.callback(f, event, *x.args):
+					self._unregister(x.source_id)
 				event_handled = True
 				if condition is not None and condition():
 					break

diff --git a/pym/_emerge/Scheduler.py b/pym/_emerge/Scheduler.py
index e6f3e0e..55e327f 100644
--- a/pym/_emerge/Scheduler.py
+++ b/pym/_emerge/Scheduler.py
@@ -218,6 +218,7 @@ class Scheduler(PollScheduler):
 			schedule=self._schedule_fetch)
 		self._sched_iface = self._iface_class(
 			fetch=fetch_iface, output=self._task_output,
+			io_add_watch=self._register,
 			register=self._register,
 			schedule=self._schedule_wait,
 			scheduleSetup=self._schedule_setup,

diff --git a/pym/_emerge/SpawnProcess.py b/pym/_emerge/SpawnProcess.py
index ec5bf7d..411e7c6 100644
--- a/pym/_emerge/SpawnProcess.py
+++ b/pym/_emerge/SpawnProcess.py
@@ -218,6 +218,8 @@ class SpawnProcess(SubProcess):
 
 		self._unregister_if_appropriate(event)
 
+		return True
+
 	def _dummy_handler(self, fd, event):
 		"""
 		This method is mainly interested in detecting EOF, since
@@ -240,6 +242,8 @@ class SpawnProcess(SubProcess):
 
 		self._unregister_if_appropriate(event)
 
+		return True
+
 	def _unregister(self):
 		super(SpawnProcess, self)._unregister()
 		if self._log_file_real is not None:

diff --git a/pym/portage/dbapi/_MergeProcess.py b/pym/portage/dbapi/_MergeProcess.py
index df501be..eed7bd4 100644
--- a/pym/portage/dbapi/_MergeProcess.py
+++ b/pym/portage/dbapi/_MergeProcess.py
@@ -83,6 +83,8 @@ class MergeProcess(SpawnProcess):
 					reporter = getattr(portage.elog.messages, funcname)
 					reporter(msg, phase=phase, key=key, out=out)
 
+		return True
+
 	def _spawn(self, args, fd_pipes, **kwargs):
 		"""
 		Fork a subprocess, apply local settings, and call