def waitForNode( self, node ):
     "Wait for a node to finish, and  print its output."
     # Pollers
     nodePoller = poll()
     nodePoller.register( node.stdout )
     bothPoller = poll()
     bothPoller.register( self.stdin, POLLIN )
     bothPoller.register( node.stdout, POLLIN )
     if self.isatty():
         # Buffer by character, so that interactive
         # commands sort of work
         quietRun( 'stty -icanon min 1' )
     while True:
         try:
             bothPoller.poll()
             # XXX BL: this doesn't quite do what we want.
             if False and self.inputFile:
                 key = self.inputFile.read( 1 )
                 if key is not '':
                     node.write(key)
                 else:
                     self.inputFile = None
             if isReadable( self.inPoller ):
                 key = self.stdin.read( 1 )
                 node.write( key )
             if isReadable( nodePoller ):
                 data = node.monitor()
                 output( data )
             if not node.waiting:
                 break
         except KeyboardInterrupt:
             node.sendInt()

        def _poll_mode(self):
            """
            Read and write to device in polling mode.
            """

            pi = select.poll()
            po = select.poll()

            for fd in self.in_files:
                pi.register(fd, select.POLLIN)

            for fd in self.out_files:
                po.register(fd, select.POLLOUT)

            while not self.exit_thread.isSet():
                data = b""
                t_out = self.out_files

                readyf = pi.poll(1.0)
                for i in readyf:
                    data += os.read(i[0], self.cachesize)

                if data != b"":
                    while ((len(t_out) != len(readyf)) and not
                           self.exit_thread.isSet()):
                        readyf = po.poll(1.0)
                    for desc in t_out:
                        os.write(desc, data)

 def test_poll(self):
     import select
     class A(object):
         def __int__(self):
             return 3
     
     select.poll().poll(A()) # assert did not crash

 def __init__(self, nnpy_socket):
     self.send = nnpy_socket.send
     self.recv = nnpy_socket.recv
     self.close = nnpy_socket.close
     self.setsockopt = nnpy_socket.setsockopt
     self.getsockopt = nnpy_socket.getsockopt
     # construct poll function
     try:
         read_p = select.poll()
         read_fd = nnpy_socket.getsockopt(nnpy.SOL_SOCKET, nnpy.RCVFD)
         read_p.register(read_fd, select.POLLIN)
         def check_readable():
             return read_p.poll(0) != []
         self.check_readable = check_readable
     except:
         self.check_readable = None
     try:
         write_p = select.poll()
         write_fd = nnpy_socket.getsockopt(nnpy.SOL_SOCKET, nnpy.SNDFD)
         write_p.register(write_fd, select.POLLOUT)
         def check_writeable():
             return write_p.poll(0) != []
         self.check_writeable = check_writeable
     except:
         self.check_writeable = None

 def waitForNode( self, node ):
     "Wait for a node to finish, and  print its output."
     # Pollers
     nodePoller = poll()
     nodePoller.register( node.stdout )
     bothPoller = poll()
     bothPoller.register( self.stdin )
     bothPoller.register( node.stdout )
     while True:
         try:
             bothPoller.poll()
             # XXX BL: this doesn't quite do what we want.
             if False and self.inputFile:
                 key = self.inputFile.read( 1 )
                 if key is not '':
                     node.write(key)
                 else:
                     self.inputFile = None
             if isReadable( self.inPoller ):
                 key = self.stdin.read( 1 )
                 node.write( key )
             if isReadable( nodePoller ):
                 data = node.monitor()
                 output( data )
             if not node.waiting:
                 break
         except KeyboardInterrupt:
             node.sendInt()

	def startSkulltag(self, deadArg):
		self.skulltag = subprocess.Popen(['/usr/games/skulltag/skulltag-server', '+sv_markchatlines 1']+self.args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=0)
		self.stdinPoll = select.poll()
		self.stdinPoll.register(self.skulltag.stdout, select.POLLIN)
		self.stdoutPoll = select.poll()
		self.stdoutPoll.register(self.skulltag.stdout, select.POLLOUT)
		thread.start_new(self.rwLoop, (None,))

 def __init__(self, fd):
     self.fd = fd
     self._r, self._w = os.pipe()
     self._wp = select.poll()
     self._wp.register(self._w, select.POLLOUT)
     self._rp = select.poll()
     self._rp.register(self.fd, select.POLLIN)
     self._rp.register(self._r, select.POLLIN)

 def __init__(self):
     ImmortalThread.__init__(self, name="Redusa Async Core")
     self._use_poll = 0
     try:
         select.poll()
         self._use_poll = 1
     except:
         msglog.log('broadway', msglog.types.INFO,
                    'Platform does not support poll().')
     return

def main():
    args = aux.parse_args("Echo server with poll-processing model.")
    server_socket = aux.listening_socket(args.port)
    server_fileno = server_socket.fileno()

    fileno_to_data = {}
    fileno_to_socket = {}

    read_poll = select.poll()
    read_poll.register(server_fileno, select.POLLIN)

    while True:
        for fileno, eventmask in read_poll.poll(50):
            if fileno == server_fileno:
                client_socket, _ = server_socket.accept()
                client_fileno = client_socket.fileno()

                fileno_to_data[client_fileno] = ""
                fileno_to_socket[client_fileno] = client_socket

                read_poll.register(client_fileno, select.POLLIN)
            else:
                client_socket = fileno_to_socket[fileno]

                data = client_socket.recv(1024)
                if not data:
                    read_poll.unregister(fileno)
                    del fileno_to_data[fileno]
                    del fileno_to_socket[fileno]
                    client_socket.close()
                else:
                    fileno_to_data[fileno] += data

        check_writability = [f for f, d in fileno_to_data.iteritems() if d]
        if not check_writability:
            continue

        write_poll = select.poll()
        for fileno in check_writability:
            write_poll.register(fileno, select.POLLOUT)

        for fileno, eventmask in write_poll.poll(50):
            if eventmask & (select.POLLERR | select.POLLHUP):
                read_poll.unregister(fileno)
                fileno_to_socket[fileno].close()
                del fileno_to_data[fileno]
                del fileno_to_socket[fileno]
                continue

            client_socket = fileno_to_socket[fileno]
            data = fileno_to_data[fileno]

            n = client_socket.send(data)
            if n > 0:
                fileno_to_data[fileno] = data[n:]

 def test_poll(self):
     import select
     if not hasattr(select, 'poll'):
         skip("no select.poll() on this platform")
     readend, writeend = self.getpair()
     try:
         class A(object):
             def __int__(self):
                 return readend.fileno()
         select.poll().poll(A()) # assert did not crash
     finally:
         readend.close()
         writeend.close()

  def test_select_not_ready_with_poll(self):
    s = self._MockSocket()
    callback = self.mox.CreateMockAnything()
    poll = self.mox.CreateMockAnything()

    select.poll().AndReturn(poll)
    poll.register(s.fileno(), select.POLLIN)
    poll.poll(1000).AndReturn([])

    self.mox.ReplayAll()
    self.select_thread.add_socket(s, callback)
    self.select_thread._select()
    self.mox.VerifyAll()

    def configure(self, config):
        CompositeNode.configure(self, config)
        set_attribute(self, 'ncounters', REQUIRED, config, int)
        set_attribute(self, 'nDIs', REQUIRED, config, int)
        set_attribute(self, 'nrelays', REQUIRED, config, int)
        set_attribute(self, 'ndallas_busses', REQUIRED, config, int)
        set_attribute(self, 'nGPIOs', REQUIRED, config, int)

        # Open the avr devices.
        self.avr = None
        self.avroob = None
        try:
            self.avr = open("/dev/avr", "r+")
            self.avroob = open("/dev/avroob", "r")
            self.p = select.poll()
            self.p.register(self.avroob, select.POLLIN)
            avr_maj_ver = ord(self.invoke_message('\x17\x00\x00')[0])
            if (avr_maj_ver < 2) and (self.nGPIOs > 0):
                self.nGPIOs = 0
                msglog.log('mpx',msglog.types.ERR,'No GPIOs created; AVR version is %s; should be 2.x or greater.' \
                           % self.version())
            # Attach the counters, relays and dallas busses to the AVR.
            config_list = (('mpx.ion.host.avr.counter', 'counter',
                            self.ncounters),
                           ('mpx.ion.host.avr.di', 'DI', self.nDIs),
                           ('mpx.ion.host.avr.relay', 'relay', self.nrelays),
                           ('mpx.ion.host.avr.dallasbus', 'dallas',
                            self.ndallas_busses),
                           ('mpx.ion.host.avr.gpio', 'gpio', self.nGPIOs))
            for module,prefix,count in config_list:
                for i in range(1,count+1):
                    name = prefix + str(i)
                    config = {'name':name, 'id':i, 'avr':self, 'parent':self}
                    ion = mpx.lib.factory(module)
                    ion.configure(config)
        except:
            msglog.log('broadway',msglog.types.ERR,"Failed to open avr device.")
            msglog.exception()
            self.p = select.poll()
            if self.avr:
                self.avr.close()
                self.avr = None
            if self.avroob:
                self.avroob.close()
                self.avroob = None
            pass

        return

def poll2(timeout=0.0, map=None):
    # Use the poll() support added to the select module in Python 2.0
    if map is None:
        map = socket_map
    if timeout is not None:
        # timeout is in milliseconds
        timeout = int(timeout*1000)
    pollster = select.poll()
    if map:
        for fd, obj in map.items():
            flags = 0
            if obj.readable():
                flags |= select.POLLIN | select.POLLPRI
            if obj.writable():
                flags |= select.POLLOUT
            if flags:
                # Only check for exceptions if object was either readable
                # or writable.
                flags |= select.POLLERR | select.POLLHUP | select.POLLNVAL
                pollster.register(fd, flags)
        try:
            r = pollster.poll(timeout)
        except select.error, err:
            if err.args[0] != EINTR:
                raise
            r = []
        for fd, flags in r:
            obj = map.get(fd)
            if obj is None:
                continue
            readwrite(obj, flags)

    def run(self):
        """Start the heartbeat thread."""
        # The first heartbeat happens immediately
        self.log.info('starting heartbeater')
        interval = 0
        self.agent.set_agent_advertise_addr()

        self.reader, self.writer = os.pipe()
        p = select.poll()
        p.register(self.reader, select.POLLIN)
        try:
            while True:
                if p.poll(interval * 1000):
                    if os.read(self.reader, 1) == 'a':
                        break

                self.do_heartbeat()
                interval_multiplier = random.uniform(
                    self.min_jitter_multiplier,
                    self.max_jitter_multiplier)
                interval = self.agent.heartbeat_timeout * interval_multiplier
                log_msg = 'sleeping before next heartbeat, interval: {0}'
                self.log.info(log_msg.format(interval))
        finally:
            os.close(self.reader)
            os.close(self.writer)
            self.reader = None
            self.writer = None

    def startShell( self ):
        if self.shell:
            error( "%s: shell is already running" )
            return
        subprocess.call(["docker stop "+self.name], shell=True, stdout=output)
        subprocess.call(["docker rm -f "+self.name], shell=True, stdout=output)

        cmd = ["docker","run","--privileged","-h",self.name ,"--name="+self.name,"-v", "/vagrant:/home/ubuntu"]
        if self.dargs is not None:
            cmd.extend([self.dargs])
        cmd.extend(["--net='none'",self.image, self.startString])

        self.shell = Popen( cmd, stdin=PIPE, stdout=PIPE, stderr=STDOUT, close_fds=True )
        self.stdin = self.shell.stdin
        self.stdout = self.shell.stdout
        self.pid = self.shell.pid
        self.pollOut = select.poll()
        self.pollOut.register( self.stdout )
        self.outToNode[ self.stdout.fileno() ] = self
        self.inToNode[ self.stdin.fileno() ] = self
        self.execed = False
        self.lastCmd = None
        self.lastPid = None
        self.readbuf = ''
        self.waiting = False
        call("sleep 1", shell=True)
        pid_cmd = ["docker","inspect","--format='{{ .State.Pid }}'",""+self.name]
        pidp = Popen( pid_cmd, stdin=PIPE, stdout=PIPE, stderr=STDOUT, close_fds=False )
        ps_out = pidp.stdout.readlines()
        self.pid = int(ps_out[0])

	def __init__(self, servername, sshcommand="ssh", username=None, ssh_options=[], debug=0, debugopts=dict(), maxopenfiles=10):
		self.debug = DebugMode(debug)
		self.debugopts = debugopts
		self.requests = dict()
		self.collectors = dict()
		self.queue = list()
		self.wantwrite = list()
		self.requestid_try_next = 17
		self.semaphores = {'openfile': maxopenfiles}

		commandline = [sshcommand]
		if ssh_options:
			commandline.extend(ssh_options)
		# those defaults are after the user-supplied ones so they can be overridden.
		# (earlier ones win with ssh).
		commandline.extend(["-oProtocol 2", # "-oLogLevel DEBUG",
		                    "-oForwardX11 no", "-oForwardAgent no",
		                    "-oPermitLocalCommand no",
		                    "-oClearAllForwardings yes"])
		if username:
			commandline.extend(["-l", username])
		commandline.extend(["-s", "--", servername, "sftp"])
		self.connection = subprocess.Popen(commandline,
		                                   close_fds = True,
		                                   stdin = subprocess.PIPE,
		                                   stdout = subprocess.PIPE,
		                                   bufsize = 0)
		self.poll = select.poll()
		self.poll.register(self.connection.stdout, select.POLLIN)
		self.inbuffer = bytes()
		self.send(INIT.bin(self, 3))
		t,b = self.getpacket()
		if t != VERSION.id:
			raise SftpUnexpectedAnswerException(b, "INIT")

    def __init__(self):
        self.poll = select.poll()
        self.pipetrick = os.pipe()
        self.pendingWakeup = False
        self.runningPoll = False
        self.nextHandleID = 1
        self.nextTimerID = 1
        self.handles = []
        self.timers = []
        self.quit = False

        # The event loop can be used from multiple threads at once.
        # Specifically while the main thread is sleeping in poll()
        # waiting for events to occur, another thread may come along
        # and add/update/remove a file handle, or timer. When this
        # happens we need to interrupt the poll() sleep in the other
        # thread, so that it'll see the file handle / timer changes.
        #
        # Using OS level signals for this is very unreliable and
        # hard to implement correctly. Thus we use the real classic
        # "self pipe" trick. A anonymous pipe, with one end registered
        # with the event loop for input events. When we need to force
        # the main thread out of a poll() sleep, we simple write a
        # single byte of data to the other end of the pipe.
        debug("Self pipe watch %d write %d" %(self.pipetrick[0], self.pipetrick[1]))
        self.poll.register(self.pipetrick[0], select.POLLIN)

    def test_threaded_poll(self):
        r, w = os.pipe()
        self.addCleanup(os.close, r)
        self.addCleanup(os.close, w)
        rfds = []
        for i in range(10):
            fd = os.dup(r)
            self.addCleanup(os.close, fd)
            rfds.append(fd)
        pollster = select.poll()
        for fd in rfds:
            pollster.register(fd, select.POLLIN)

        t = threading.Thread(target=pollster.poll)
        t.start()
        try:
            time.sleep(0.5)
            # trigger ufds array reallocation
            for fd in rfds:
                pollster.unregister(fd)
            pollster.register(w, select.POLLOUT)
            self.assertRaises(RuntimeError, pollster.poll)
        finally:
            # and make the call to poll() from the thread return
            os.write(w, b'spam')
            t.join()

    def receive(sdef, slen=SLEN):
        try:
            sdef.setblocking(1)
            poller = select.poll()
            poller.register(sdef, READ_OR_ERROR)
            ready = poller.poll(LTIMEOUT*1000)
            if not ready:
                # logical timeout
                return "*"
            fd, flag = ready[0]
            if (flag & ( select.POLLHUP | select.POLLERR | select.POLLNVAL)):
                # No need to read
                raise RuntimeError("Socket POLLHUP")
            if (flag & (select.POLLIN|select.POLLPRI)):
                data = sdef.recv(slen)
                if not data:
                    # POLLIN and POLLHUP are not exclusive. We can have both.
                    raise RuntimeError("Socket EOF")
                data = int(data)  # receive length
            elif (flag & (select.POLLERR | select.POLLHUP | select.POLLNVAL)):
                raise RuntimeError("Socket error {}".format(flag))
            else:
                raise RuntimeError("Socket Unexpected Error")
            chunks = []
            bytes_recd = 0
            while bytes_recd < data:
                ready = poller.poll(LTIMEOUT*1000)
                if not ready:
                    raise RuntimeError("Socket Timeout2")
                fd, flag = ready[0]
                if (flag & ( select.POLLHUP | select.POLLERR | select.POLLNVAL)):
                    # No need to read
                    raise RuntimeError("Socket POLLHUP2")
                if (flag & (select.POLLIN|select.POLLPRI)):
                    chunk = sdef.recv(min(data - bytes_recd, 2048))
                    if not chunk:
                        raise RuntimeError("Socket EOF2")
                    chunks.append(chunk)
                    bytes_recd = bytes_recd + len(chunk)
                elif (flag & (select.POLLERR | select.POLLHUP | select.POLLNVAL)):
                    raise RuntimeError("Socket Error {}".format(flag))
                else:
                    raise RuntimeError("Socket Unexpected Error")

            poller.unregister(sdef)
            segments = b''.join(chunks).decode("utf-8")
            return json.loads(segments)
        except Exception as e:
            """
            exc_type, exc_obj, exc_tb = sys.exc_info()
            fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
            print(exc_type, fname, exc_tb.tb_lineno)
            """
            # Final cleanup
            try:
                poller.unregister(sdef)
            except Exception as e2:
                pass
                #print ("Exception unregistering: {}".format(e2))
            raise RuntimeError("Connections: {}".format(e))

def register():
    '''
    This function creates a select.poll object that can be used in the same
    manner as signal.pause(). The poll object returns each time a signal was
    received by the process.

    This function has to be called from the main thread.
    '''

    global _signal_poller
    global _signal_read_fd

    if _signal_poller is not None:
        raise RuntimeError('register was already called')

    read_fd, write_fd = os.pipe()

    # Python c-level signal handler requires that the write end will be in
    # non blocking mode
    filecontrol.set_non_blocking(write_fd)

    # Set the read pipe end to non-blocking too, just in case.
    filecontrol.set_non_blocking(read_fd)

    # Prevent subproccesses we execute from inheriting the pipes.
    filecontrol.set_close_on_exec(write_fd)
    filecontrol.set_close_on_exec(read_fd)

    signal.set_wakeup_fd(write_fd)

    poller = select.poll()
    poller.register(read_fd, select.POLLIN)

    _signal_poller = poller
    _signal_read_fd = read_fd