Implement a runtime test that checks whether it's possible to use poll() to

read data from a pty device. The test only runs the first time that the
can_poll_pty() function is called (currently happens when the first
PollScheduler instance is created). The test is known to fail on Darwin and
we're not sure how many other kernels might have a similar problem, this
runtime test allows us to avoid having to maintain special cases for various
kernels.

svn path=/main/trunk/; revision=11025

diff --git a/pym/_emerge/__init__.py b/pym/_emerge/__init__.py
index a282278c960eda15c1c3050a002a8ffa25b35000..bc317156bd0a37594a358d2e826585b776798336 100644 (file)
--- a/pym/_emerge/__init__.py
+++ b/pym/_emerge/__init__.py
@@ -7869,12 +7869,72 @@ class SequentialTaskQueue(SlotObject):
        def __len__(self):
                return len(self._task_queue) + len(self.running_tasks)
 
+_can_poll_pty = None
+
+def can_poll_pty():
+       """
+       Test if it's possible to use poll() on a pty device. This
+       is known to fail on Darwin.
+       @rtype: bool
+       @returns: True if poll() on a pty device succeeds, False otherwise.
+       """
+
+       global _can_poll_pty
+       if _can_poll_pty is not None:
+               return _can_poll_pty
+
+       if not hasattr(select, "poll"):
+               _can_poll_pty = False
+               return _can_poll_pty
+
+       got_pty, master_fd, slave_fd = \
+               portage._create_pty_or_pipe(copy_term_size=sys.stdout.fileno())
+       if not got_pty:
+               _can_poll_pty = False
+               return _can_poll_pty
+
+       test_string = 2 * "blah blah blah\n"
+
+       master_file = os.fdopen(master_fd, 'r')
+
+       task_scheduler = TaskScheduler(max_jobs=2, poll=select.poll())
+       scheduler = task_scheduler.sched_iface
+
+       producer = SpawnProcess(
+               args=["bash", "-c", "echo -n '%s'" % test_string],
+               fd_pipes={1:slave_fd}, scheduler=scheduler)
+
+       consumer = PipeReader(
+               input_files={"producer" : master_file},
+               scheduler=scheduler)
+
+       task_scheduler.add(producer)
+       task_scheduler.add(consumer)
+
+       def producer_start_cb(task):
+               os.close(slave_fd)
+
+       producer.addStartListener(producer_start_cb)
+       task_scheduler.run()
+       _can_poll_pty = test_string == consumer.getvalue()
+       return _can_poll_pty
+
+def create_poll_instance():
+       """
+       Create an instance of select.poll, or an instance of
+       PollSelectAdapter there is no poll() implementation or
+       it is broken somehow.
+       """
+       if can_poll_pty():
+               return select.poll()
+       return PollSelectAdapter()
+
 class PollScheduler(object):
 
        class _sched_iface_class(SlotObject):
                __slots__ = ("register", "schedule", "unregister")
 
-       def __init__(self):
+       def __init__(self, poll=None):
                self._max_jobs = 1
                self._max_load = None
                self._jobs = 0
@@ -7882,10 +7942,9 @@ class PollScheduler(object):
                self._poll_event_handler_ids = {}
                # Increment id for each new handler.
                self._event_handler_id = 0
-               try:
-                       self._poll = select.poll()
-               except AttributeError:
-                       self._poll = PollSelectAdapter()
+               if poll is None:
+                       poll = create_poll_instance()
+               self._poll = poll
 
        def _running_job_count(self):
                return self._jobs
@@ -7974,8 +8033,8 @@ class QueueScheduler(PollScheduler):
        run() method returns when no tasks remain.
        """
 
-       def __init__(self, max_jobs=None, max_load=None):
-               PollScheduler.__init__(self)
+       def __init__(self, max_jobs=None, max_load=None, poll=None):
+               PollScheduler.__init__(self, poll=poll)
 
                if max_jobs is None:
                        max_jobs = 1
@@ -8049,9 +8108,10 @@ class TaskScheduler(object):
        add tasks and call run(). The run() method returns when no tasks remain.
        """
 
-       def __init__(self, max_jobs=None, max_load=None):
+       def __init__(self, max_jobs=None, max_load=None, poll=None):
                self._queue = SequentialTaskQueue(max_jobs=max_jobs)
-               self._scheduler = QueueScheduler(max_jobs=max_jobs, max_load=max_load)
+               self._scheduler = QueueScheduler(
+                       max_jobs=max_jobs, max_load=max_load, poll=poll)
                self.sched_iface = self._scheduler.sched_iface
                self.run = self._scheduler.run
                self._scheduler.add(self._queue)