Make pyghmi tolerate arbitrary threading models

When calling from various threading code strategies, pyghmi
would cause confusion and eventlet to print debug output.
Create a dedicated IO thread to isolate the shared socket
usage from calling code thread behavior.  Currently, it still
requires that calling code loop on wait_for_rsp to assure session
liveness and/or to do SOL.

Change-Id: I66164adbfd867200af53269553210a70a0619a85

pyghmi/ipmi/console.py

@@ -16,8 +16,6 @@
 #
 # This represents the low layer message framing portion of IPMI
 
-import fcntl
-import os
 import pyghmi.exceptions as exc
 import struct
 
@@ -46,18 +44,7 @@ class Console(object):
                  force=False, kg=None):
         self.connected = False
         self.broken = False
-        if type(iohandler) == tuple:  # two file handles
-            self.console_in = iohandler[0]
-            self.console_out = iohandler[1]
-        elif type(iohandler) == file:  # one full duplex file handle
-            self.console_out = iohandler
-            self.console_in = iohandler
-        elif hasattr(iohandler, '__call__'):
-            self.console_out = None
-            self.console_in = None
-            self.out_handler = iohandler
-        if self.console_in is not None:
-            fcntl.fcntl(self.console_in.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
+        self.out_handler = iohandler
         self.remseq = 0
         self.myseq = 0
         self.lastsize = 0
@@ -139,9 +126,6 @@ class Console(object):
         #ignore data[10:11] for now, the vlan detail, shouldn't matter to this
         #code anyway...
         self.ipmi_session.sol_handler = self._got_sol_payload
-        if self.console_in is not None:
-            self.ipmi_session.register_handle_callback(self.console_in,
-                                                       self._got_cons_input)
         self.connected = True
 
     def _got_cons_input(self, handle):
@@ -218,17 +202,7 @@ class Console(object):
         callback function that this class will use to convey data back to
         caller.
         """
-        if self.console_out is not None:
-            # if we are writing to a dumb stream, format a string ourselves
-            if type(data) == dict:
-                if 'error' in data:
-                    data = 'ERROR: ' + data['error'] + '\n'
-                elif 'info' in data:
-                    data = 'INFO: ' + data['info'] + '\n'
-            self.console_out.write(data)
-            self.console_out.flush()
-        elif self.out_handler:  # callback style..
-            self.out_handler(data)
+        self.out_handler(data)
 
     def _got_sol_payload(self, payload):
         """SOL payload callback

pyghmi/ipmi/private/session.py

@@ -17,13 +17,14 @@
 
 import atexit
 import collections
+import fcntl
 import hashlib
 import os
 import random
 import select
 import socket
 import struct
-import time
+import threading
 
 from Crypto.Cipher import AES
 from Crypto.Hash import HMAC
@@ -36,6 +37,98 @@ from pyghmi.ipmi.private import constants
 initialtimeout = 0.5  # minimum timeout for first packet to retry in any given
                      # session.  This will be randomized to stagger out retries
                      # in case of congestion
+iothread = None  # the thread in which all IO will be performed
+                 # While the model as-is works fine for it's own coroutine
+                 # structure, when combined with threading or something like
+                 # eventlet, it becomes difficult for the calling code to cope
+                 # This thread will tuck away the threading situation such that
+                 # calling code doesn't have to do any gymnastics to cope with
+                 # the nature of things.
+ioqueue = collections.deque([])
+selectbreak = None
+selectdeadline = 0
+running = True
+iosockets = []  # set of iosockets that will be shared amongst Session objects
+
+
+def _ioworker(initialized):
+    global selectbreak
+    global selectdeadline
+    selectbreak = os.pipe()
+    fcntl.fcntl(selectbreak[0], fcntl.F_SETFL, os.O_NONBLOCK)
+    iosockets.append(selectbreak[0])
+    iowaiters = []
+    timeout = 300
+    initialized.set()
+    while running:
+        if timeout < 0:
+            timeout = 0
+        selectdeadline = _monotonic_time() + timeout
+        tmplist, _, _ = select.select(iosockets, (), (), timeout)
+        rdylist = []
+        for handle in tmplist:
+            if handle is selectbreak[0]:
+                try:  # flush all requests to interrupt that may be pending
+                    while True:
+                        os.read(handle, 1)
+                except OSError:
+                    # this means an EWOULDBLOCK occurred, ignore that as that
+                    # was the endgame
+                    pass
+            else:
+                rdylist.append(handle)
+        for w in iowaiters:
+            w[2].append(tuple(rdylist))
+            w[3].set()
+        iowaiters = []
+        timeout = 300
+        while ioqueue:
+            workitem = ioqueue.popleft()
+            # structure is function, args, list to append to ,event to set
+            if isinstance(workitem[1], tuple):  # positional arguments
+                workitem[2].append(workitem[0](*workitem[1]))
+                workitem[3].set()
+            elif isinstance(workitem[1], dict):
+                workitem[2].append(workitem[0](**workitem[1]))
+                workitem[3].set()
+            elif workitem[0] == 'wait':
+                if len(rdylist) > 0:
+                    workitem[2].append(tuple(rdylist))
+                    workitem[3].set()
+                else:
+                    ltimeout = workitem[1] - _monotonic_time()
+                    if ltimeout < timeout:
+                        timeout = ltimeout
+                    iowaiters.append(workitem)
+
+
+def _io_apply(function, args):
+    global selectbreak
+    evt = threading.Event()
+    result = []
+    ioqueue.append((function, args, result, evt))
+    if not (function == 'wait' and selectdeadline < args):
+        os.write(selectbreak[1], '1')
+    evt.wait()
+    return result[0]
+
+
+selectdeadline = 0
+selectwait = None
+
+
+def _io_sendto(mysocket, packet, sockaddr):
+    #Want sendto to act reasonably sane..
+    mysocket.setblocking(1)
+    mysocket.sendto(packet, sockaddr)
+
+
+def _io_recvfrom(mysocket, size):
+    mysocket.setblocking(0)
+    try:
+        return mysocket.recvfrom(size)
+    except socket.error:
+        return None
 
 
 def _monotonic_time():
@@ -47,15 +140,11 @@ def _monotonic_time():
     # Python does not provide one until 3.3, so we make do
     # for most OSes, os.times()[4] works well.
     # for microsoft, GetTickCount64
-    if (os.name == "posix"):
-        return os.times()[4]
-    else:  # last resort, non monotonic time
-        return time.time()
-    #TODO(jbjohnso): Windows variant
+    return os.times()[4]
 
 
 def _poller(readhandles, timeout=0):
-    rdylist, _, _ = select.select(readhandles, (), (), timeout)
+    rdylist = _io_apply('wait', timeout + _monotonic_time())
     return rdylist
 
 
@@ -117,7 +206,6 @@ class Session(object):
     :param port: UDP port to communicate with, pretty much always 623
     :param onlogon: callback to receive notification of login completion
     """
-    _external_handlers = {}
     bmc_handlers = {}
     waiting_sessions = {}
     keepalive_sessions = {}
@@ -128,14 +216,26 @@ class Session(object):
 
     @classmethod
     def _cleanup(cls):
+        global running
         for session in cls.bmc_handlers.itervalues():
             session.cleaningup = True
             session.logout()
+        running = False
 
     @classmethod
     def _createsocket(cls):
+        global iowork
+        global iothread
+        global iosockets
+        if iothread is None:
+            initevt = threading.Event()
+            iothread = threading.Thread(target=_ioworker, args=(initevt,))
+            iothread.daemon = True
+            iothread.start()
+            initevt.wait()
         atexit.register(cls._cleanup)
-        cls.socket = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)  # INET6
+        cls.socket = _io_apply(socket.socket,
+                               (socket.AF_INET6, socket.SOCK_DGRAM))  # INET6
                                     # can do IPv4 if you are nice to it
         try:  # we will try to fixup our receive buffer size if we are smaller
              # than allowed.
@@ -145,15 +245,16 @@ class Session(object):
             curmax = cls.socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
             curmax = curmax / 2
             if (rmemmax > curmax):
-                cls.socket.setsockopt(socket.SOL_SOCKET,
-                                      socket.SO_RCVBUF,
-                                      rmemmax)
+                _io_apply(cls.socket.setsockopt, (socket.SOL_SOCKET,
+                                                  socket.SO_RCVBUF,
+                                                  rmemmax))
         except Exception:
             # FIXME: be more selective in catching exceptions
             pass
 
-        curmax = cls.socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
-        cls.readersockets = [cls.socket]
+        curmax = _io_apply(cls.socket.getsockopt,
+                           (socket.SOL_SOCKET, socket.SO_RCVBUF))
+        iosockets.append(cls.socket)
         curmax = curmax / 2
         # we throttle such that we never have no more outstanding packets than
         # our receive buffer should be able to handle
@@ -712,6 +813,14 @@ class Session(object):
         self._initsession()
         self._get_channel_auth_cap()
 
+    @classmethod
+    def pulltoqueue(cls, mysocket, queue):
+        while True:
+            rdata = _io_apply(_io_recvfrom, (mysocket, 3000))
+            if rdata is None:
+                break
+            queue.append(rdata)
+
     @classmethod
     def wait_for_rsp(cls, timeout=None, callout=True):
         """IPMI Session Event loop iteration
@@ -723,6 +832,7 @@ class Session(object):
         :param timeout: Maximum time to wait for data to come across.  If
                         unspecified, will autodetect based on earliest timeout
         """
+        global iosockets
         #Assume:
         #Instance A sends request to packet B
         #Then Instance C sends request to BMC D
@@ -762,34 +872,24 @@ class Session(object):
         while cls.iterwaiters:
             waiter = cls.iterwaiters.pop()
             waiter({'success': True})
+            # cause a quick exit from the event loop iteration for calling code
+            # to be able to reasonably set up for the next iteration before
+            # a long select comes along
+            if timeout is not None:
+                timeout = 0
         if timeout is None:
             return 0
-        rdylist, _, _ = select.select(cls.readersockets, (), (), timeout)
+        rdylist = _poller(iosockets, timeout=timeout)
         if len(rdylist) > 0:
             while _poller((cls.socket,)):  # if the somewhat lengthy
                         # queue # processing takes long enough for packets to
                         # come in, be eager
                 pktqueue = collections.deque([])
-                while _poller((cls.socket,)):  # looks rendundant, but
-                              # want # to queue and process packets to keep
-                              # things off RCVBUF
-                    rdata = cls.socket.recvfrom(3000)
-                    pktqueue.append(rdata)
+                cls.pulltoqueue(cls.socket, pktqueue)
                 while len(pktqueue):
                     (data, sockaddr) = pktqueue.popleft()
                     cls._route_ipmiresponse(sockaddr, data)
-                    while _poller((cls.socket,)):  # seems ridiculous,
-                         #but between every callback, check for packets again
-                        rdata = cls.socket.recvfrom(3000)
-                        pktqueue.append(rdata)
-            for handlepair in _poller(cls.readersockets):
-                if isinstance(handlepair, int):
-                    myhandle = handlepair
-                else:
-                    myhandle = handlepair.fileno()
-                if myhandle != cls.socket.fileno() and callout:
-                    myfile = cls._external_handlers[myhandle][1]
-                    cls._external_handlers[myhandle][0](myfile)
+                    cls.pulltoqueue(cls.socket, pktqueue)
         sessionstodel = []
         sessionstokeepalive = []
         for session, parms in cls.keepalive_sessions.iteritems():
@@ -820,28 +920,6 @@ class Session(object):
             return
         self.raw_command(netfn=6, command=1)
 
-    @classmethod
-    def register_handle_callback(cls, handle, callback):
-        """Add a handle to be watched by Session's event loop
-
-        In the event that an application would like IPMI Session event loop
-        to drive things while adding their own filehandle to watch for events,
-        this class method will register that.
-
-        :param handle: filehandle too watch for input
-        :param callback: function to call when input detected on the handle.
-                         will receive the handle as an argument
-        """
-        if isinstance(handle, int):
-            cls._external_handlers[handle] = (callback, handle)
-        else:
-            cls._external_handlers[handle.fileno()] = (callback, handle)
-        #If we don't have a socket yet, we need one for the code to behave
-        #correctly from this point forward
-        if not hasattr(Session, 'socket'):
-            cls._createsocket()
-        cls.readersockets += [handle]
-
     @classmethod
     def _route_ipmiresponse(cls, sockaddr, data):
         if not (data[0] == '\x06' and data[2:4] == '\xff\x07'):  # not ipmi
@@ -1236,7 +1314,8 @@ class Session(object):
                 _monotonic_time()
             return  # skip transmit, let retry timer do it's thing
         if self.sockaddr:
-            Session.socket.sendto(self.netpacket, self.sockaddr)
+            _io_apply(_io_sendto,
+                      (Session.socket, self.netpacket, self.sockaddr))
         else:  # he have not yet picked a working sockaddr for this connection,
               # try all the candidates that getaddrinfo provides
             self.allsockaddrs = []
@@ -1252,7 +1331,8 @@ class Session(object):
                         sockaddr = (newhost, sockaddr[1], 0, 0)
                     self.allsockaddrs.append(sockaddr)
                     Session.bmc_handlers[sockaddr] = self
-                    Session.socket.sendto(self.netpacket, sockaddr)
+                    _io_apply(_io_sendto, (Session.socket,
+                              self.netpacket, sockaddr))
             except socket.gaierror:
                 raise exc.IpmiException(
                     "Unable to transmit to specified address")

solconnect.py

@@ -19,12 +19,15 @@
 
 """A simple little script to exemplify/test ipmi.console module
 """
+import fcntl
 import os
+import select
 import sys
 import termios
 import tty
 
 from pyghmi.ipmi import console
+import threading
 
 tcattr = termios.tcgetattr(sys.stdin)
 newtcattr = tcattr
@@ -34,12 +37,32 @@ newtcattr[-1][termios.VSUSP] = 0
 termios.tcsetattr(sys.stdin, termios.TCSADRAIN, newtcattr)
 
 tty.setcbreak(sys.stdin.fileno())
+fcntl.fcntl(sys.stdin.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
 
 passwd = os.environ['IPMIPASSWORD']
 
+sol = None
+
+
+def _doinput():
+    while True:
+        select.select((sys.stdin,), (), (), 600)
+        try:
+            data = sys.stdin.read()
+        except OSError:
+            continue
+        sol.send_data(data)
+
+
+def _print(data):
+    sys.stdout.write(data)
+    sys.stdout.flush()
+
 try:
     sol = console.Console(bmc=sys.argv[1], userid=sys.argv[2], password=passwd,
-                          iohandler=(sys.stdin, sys.stdout), force=True)
+                          iohandler=_print, force=True)
+    inputthread = threading.Thread(target=_doinput)
+    inputthread.start()
     sol.main_loop()
 finally:
     termios.tcsetattr(sys.stdin, termios.TCSADRAIN, tcattr)