FM / internet

/usr/lib/python3/dist-packages/twisted/internet/

/usr/lib/python3/dist-packages/twisted/internet

Name	Size	Mode	Actions
iocpreactor/	-	0755	rm
test/	-	0755	rm
__pycache__/	-	0755	rm
abstract.py	19274	0644	edit dl rm
address.py	5246	0644	edit dl rm
asyncioreactor.py	10553	0644	edit dl rm
base.py	44260	0644	edit dl rm
cfreactor.py	17501	0644	edit dl rm
default.py	1942	0644	edit dl rm
defer.py	71193	0644	edit dl rm
endpoints.py	77207	0644	edit dl rm
epollreactor.py	8493	0644	edit dl rm
error.py	12645	0644	edit dl rm
fdesc.py	3223	0644	edit dl rm
gireactor.py	6123	0644	edit dl rm
glib2reactor.py	1115	0644	edit dl rm
gtk2reactor.py	3614	0644	edit dl rm
gtk3reactor.py	2252	0644	edit dl rm
inotify.py	14691	0644	edit dl rm
interfaces.py	96515	0644	edit dl rm
kqreactor.py	10292	0644	edit dl rm
main.py	1054	0644	edit dl rm
pollreactor.py	6026	0644	edit dl rm
posixbase.py	26357	0644	edit dl rm
process.py	38896	0644	edit dl rm
protocol.py	27134	0644	edit dl rm
pyuisupport.py	817	0644	edit dl rm
reactor.py	1863	0644	edit dl rm
selectreactor.py	6215	0644	edit dl rm
serialport.py	2317	0644	edit dl rm
ssl.py	8444	0644	edit dl rm
stdio.py	1045	0644	edit dl rm
task.py	31119	0644	edit dl rm
tcp.py	54908	0644	edit dl rm
threads.py	3953	0644	edit dl rm
tksupport.py	2045	0644	edit dl rm
udp.py	18569	0644	edit dl rm
unix.py	21965	0644	edit dl rm
utils.py	7876	0644	edit dl rm
win32eventreactor.py	15193	0644	edit dl rm
wxreactor.py	5263	0644	edit dl rm
wxsupport.py	1363	0644	edit dl rm
_baseprocess.py	1913	0644	edit dl rm
_dumbwin32proc.py	13107	0644	edit dl rm
_glibbase.py	12813	0644	edit dl rm
_idna.py	1397	0644	edit dl rm
_newtls.py	9332	0644	edit dl rm
_pollingfile.py	8982	0644	edit dl rm
_posixserialport.py	1995	0644	edit dl rm
_posixstdio.py	4687	0644	edit dl rm
_producer_helpers.py	3788	0644	edit dl rm
_resolver.py	8530	0644	edit dl rm
_signals.py	2718	0644	edit dl rm
_sslverify.py	71770	0644	edit dl rm
_threadedselect.py	11764	0644	edit dl rm
_win32serialport.py	4744	0644	edit dl rm
_win32stdio.py	3204	0644	edit dl rm
__init__.py	521	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/internet/win32eventreactor.py (15193B)

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
A win32event based implementation of the Twisted main loop.

This requires pywin32 (formerly win32all) or ActivePython to be installed.

To install the event loop (and you should do this before any connections,
listeners or connectors are added)::

from twisted.internet import win32eventreactor
    win32eventreactor.install()

LIMITATIONS:
 1. WaitForMultipleObjects and thus the event loop can only handle 64 objects.
 2. Process running has some problems (see L{twisted.internet.process} docstring).

TODO:
 1. Event loop handling of writes is *very* problematic (this is causing failed tests).
    Switch to doing it the correct way, whatever that means (see below).
 2. Replace icky socket loopback waker with event based waker (use dummyEvent object)
 3. Switch everyone to using Free Software so we don't have to deal with proprietary APIs.

ALTERNATIVE SOLUTIONS:
 - IIRC, sockets can only be registered once. So we switch to a structure
   like the poll() reactor, thus allowing us to deal with write events in
   a decent fashion. This should allow us to pass tests, but we're still
   limited to 64 events.

Or:

- Instead of doing a reactor, we make this an addon to the select reactor.
   The WFMO event loop runs in a separate thread. This means no need to maintain
   separate code for networking, 64 event limit doesn't apply to sockets,
   we can run processes and other win32 stuff in default event loop. The
   only problem is that we're stuck with the icky socket based waker.
   Another benefit is that this could be extended to support >64 events
   in a simpler manner than the previous solution.

The 2nd solution is probably what will get implemented.
"""

# System imports
import time
import sys
from threading import Thread
from weakref import WeakKeyDictionary

from zope.interface import implementer

# Win32 imports
from win32file import FD_READ, FD_CLOSE, FD_ACCEPT, FD_CONNECT, WSAEventSelect
try:
    # WSAEnumNetworkEvents was added in pywin32 215
    from win32file import WSAEnumNetworkEvents
except ImportError:
    import warnings
    warnings.warn(
        'Reliable disconnection notification requires pywin32 215 or later',
        category=UserWarning)
    def WSAEnumNetworkEvents(fd, event):
        return set([FD_READ])

from win32event import CreateEvent, MsgWaitForMultipleObjects
from win32event import WAIT_OBJECT_0, WAIT_TIMEOUT, QS_ALLINPUT

import win32gui

# Twisted imports
from twisted.internet import posixbase
from twisted.python import log, threadable, failure
from twisted.internet.interfaces import IReactorFDSet
from twisted.internet.interfaces import IReactorWin32Events
from twisted.internet.threads import blockingCallFromThread

@implementer(IReactorFDSet, IReactorWin32Events)
class Win32Reactor(posixbase.PosixReactorBase):
    """
    Reactor that uses Win32 event APIs.

@ivar _reads: A dictionary mapping L{FileDescriptor} instances to a
        win32 event object used to check for read events for that descriptor.

@ivar _writes: A dictionary mapping L{FileDescriptor} instances to a
        arbitrary value.  Keys in this dictionary will be given a chance to
        write out their data.

@ivar _events: A dictionary mapping win32 event object to tuples of
        L{FileDescriptor} instances and event masks.

@ivar _closedAndReading: Along with C{_closedAndNotReading}, keeps track of
        descriptors which have had close notification delivered from the OS but
        which we have not finished reading data from.  MsgWaitForMultipleObjects
        will only deliver close notification to us once, so we remember it in
        these two dictionaries until we're ready to act on it.  The OS has
        delivered close notification for each descriptor in this dictionary, and
        the descriptors are marked as allowed to handle read events in the
        reactor, so they can be processed.  When a descriptor is marked as not
        allowed to handle read events in the reactor (ie, it is passed to
        L{IReactorFDSet.removeReader}), it is moved out of this dictionary and
        into C{_closedAndNotReading}.  The descriptors are keys in this
        dictionary.  The values are arbitrary.
    @type _closedAndReading: C{dict}

@ivar _closedAndNotReading: These descriptors have had close notification
        delivered from the OS, but are not marked as allowed to handle read
        events in the reactor.  They are saved here to record their closed
        state, but not processed at all.  When one of these descriptors is
        passed to L{IReactorFDSet.addReader}, it is moved out of this dictionary
        and into C{_closedAndReading}.  The descriptors are keys in this
        dictionary.  The values are arbitrary.  This is a weak key dictionary so
        that if an application tells the reactor to stop reading from a
        descriptor and then forgets about that descriptor itself, the reactor
        will also forget about it.
    @type _closedAndNotReading: C{WeakKeyDictionary}
    """
    dummyEvent = CreateEvent(None, 0, 0, None)

def __init__(self):
        self._reads = {}
        self._writes = {}
        self._events = {}
        self._closedAndReading = {}
        self._closedAndNotReading = WeakKeyDictionary()
        posixbase.PosixReactorBase.__init__(self)

def _makeSocketEvent(self, fd, action, why):
        """
        Make a win32 event object for a socket.
        """
        event = CreateEvent(None, 0, 0, None)
        WSAEventSelect(fd, event, why)
        self._events[event] = (fd, action)
        return event

def addEvent(self, event, fd, action):
        """
        Add a new win32 event to the event loop.
        """
        self._events[event] = (fd, action)

def removeEvent(self, event):
        """
        Remove an event.
        """
        del self._events[event]

def addReader(self, reader):
        """
        Add a socket FileDescriptor for notification of data available to read.
        """
        if reader not in self._reads:
            self._reads[reader] = self._makeSocketEvent(
                reader, 'doRead', FD_READ | FD_ACCEPT | FD_CONNECT | FD_CLOSE)
            # If the reader is closed, move it over to the dictionary of reading
            # descriptors.
            if reader in self._closedAndNotReading:
                self._closedAndReading[reader] = True
                del self._closedAndNotReading[reader]

def addWriter(self, writer):
        """
        Add a socket FileDescriptor for notification of data available to write.
        """
        if writer not in self._writes:
            self._writes[writer] = 1

def removeReader(self, reader):
        """Remove a Selectable for notification of data available to read.
        """
        if reader in self._reads:
            del self._events[self._reads[reader]]
            del self._reads[reader]

# If the descriptor is closed, move it out of the dictionary of
            # reading descriptors into the dictionary of waiting descriptors.
            if reader in self._closedAndReading:
                self._closedAndNotReading[reader] = True
                del self._closedAndReading[reader]

def removeWriter(self, writer):
        """Remove a Selectable for notification of data available to write.
        """
        if writer in self._writes:
            del self._writes[writer]

def removeAll(self):
        """
        Remove all selectables, and return a list of them.
        """
        return self._removeAll(self._reads, self._writes)

def getReaders(self):
        return list(self._reads.keys())

def getWriters(self):
        return list(self._writes.keys())

def doWaitForMultipleEvents(self, timeout):
        log.msg(channel='system', event='iteration', reactor=self)
        if timeout is None:
            timeout = 100

# Keep track of whether we run any application code before we get to the
        # MsgWaitForMultipleObjects.  If so, there's a chance it will schedule a
        # new timed call or stop the reactor or do something else that means we
        # shouldn't block in MsgWaitForMultipleObjects for the full timeout.
        ranUserCode = False

# If any descriptors are trying to close, try to get them out of the way
        # first.
        for reader in list(self._closedAndReading.keys()):
            ranUserCode = True
            self._runAction('doRead', reader)

for fd in list(self._writes.keys()):
            ranUserCode = True
            log.callWithLogger(fd, self._runWrite, fd)

if ranUserCode:
            # If application code *might* have scheduled an event, assume it
            # did.  If we're wrong, we'll get back here shortly anyway.  If
            # we're right, we'll be sure to handle the event (including reactor
            # shutdown) in a timely manner.
            timeout = 0

if not (self._events or self._writes):
            # sleep so we don't suck up CPU time
            time.sleep(timeout)
            return

handles = list(self._events.keys()) or [self.dummyEvent]
        timeout = int(timeout * 1000)
        val = MsgWaitForMultipleObjects(handles, 0, timeout, QS_ALLINPUT)
        if val == WAIT_TIMEOUT:
            return
        elif val == WAIT_OBJECT_0 + len(handles):
            exit = win32gui.PumpWaitingMessages()
            if exit:
                self.callLater(0, self.stop)
                return
        elif val >= WAIT_OBJECT_0 and val < WAIT_OBJECT_0 + len(handles):
            event = handles[val - WAIT_OBJECT_0]
            fd, action = self._events[event]

if fd in self._reads:
                # Before anything, make sure it's still a valid file descriptor.
                fileno = fd.fileno()
                if fileno == -1:
                    self._disconnectSelectable(fd, posixbase._NO_FILEDESC, False)
                    return

# Since it's a socket (not another arbitrary event added via
                # addEvent) and we asked for FD_READ | FD_CLOSE, check to see if
                # we actually got FD_CLOSE.  This needs a special check because
                # it only gets delivered once.  If we miss it, it's gone forever
                # and we'll never know that the connection is closed.
                events = WSAEnumNetworkEvents(fileno, event)
                if FD_CLOSE in events:
                    self._closedAndReading[fd] = True
            log.callWithLogger(fd, self._runAction, action, fd)

def _runWrite(self, fd):
        closed = 0
        try:
            closed = fd.doWrite()
        except:
            closed = sys.exc_info()[1]
            log.deferr()

if closed:
            self.removeReader(fd)
            self.removeWriter(fd)
            try:
                fd.connectionLost(failure.Failure(closed))
            except:
                log.deferr()
        elif closed is None:
            return 1

def _runAction(self, action, fd):
        try:
            closed = getattr(fd, action)()
        except:
            closed = sys.exc_info()[1]
            log.deferr()
        if closed:
            self._disconnectSelectable(fd, closed, action == 'doRead')

doIteration = doWaitForMultipleEvents

class _ThreadFDWrapper(object):
    """
    This wraps an event handler and translates notification in the helper
    L{Win32Reactor} thread into a notification in the primary reactor thread.

@ivar _reactor: The primary reactor, the one to which event notification
        will be sent.

@ivar _fd: The L{FileDescriptor} to which the event will be dispatched.

@ivar _action: A C{str} giving the method of C{_fd} which handles the event.

@ivar _logPrefix: The pre-fetched log prefix string for C{_fd}, so that
        C{_fd.logPrefix} does not need to be called in a non-main thread.
    """
    def __init__(self, reactor, fd, action, logPrefix):
        self._reactor = reactor
        self._fd = fd
        self._action = action
        self._logPrefix = logPrefix

def logPrefix(self):
        """
        Return the original handler's log prefix, as it was given to
        C{__init__}.
        """
        return self._logPrefix

def _execute(self):
        """
        Callback fired when the associated event is set.  Run the C{action}
        callback on the wrapped descriptor in the main reactor thread and raise
        or return whatever it raises or returns to cause this event handler to
        be removed from C{self._reactor} if appropriate.
        """
        return blockingCallFromThread(
            self._reactor, lambda: getattr(self._fd, self._action)())

def connectionLost(self, reason):
        """
        Pass through to the wrapped descriptor, but in the main reactor thread
        instead of the helper C{Win32Reactor} thread.
        """
        self._reactor.callFromThread(self._fd.connectionLost, reason)

@implementer(IReactorWin32Events)
class _ThreadedWin32EventsMixin(object):
    """
    This mixin implements L{IReactorWin32Events} for another reactor by running
    a L{Win32Reactor} in a separate thread and dispatching work to it.

@ivar _reactor: The L{Win32Reactor} running in the other thread.  This is
        L{None} until it is actually needed.

@ivar _reactorThread: The L{threading.Thread} which is running the
        L{Win32Reactor}.  This is L{None} until it is actually needed.
    """

_reactor = None
    _reactorThread = None

def _unmakeHelperReactor(self):
        """
        Stop and discard the reactor started by C{_makeHelperReactor}.
        """
        self._reactor.callFromThread(self._reactor.stop)
        self._reactor = None

def _makeHelperReactor(self):
        """
        Create and (in a new thread) start a L{Win32Reactor} instance to use for
        the implementation of L{IReactorWin32Events}.
        """
        self._reactor = Win32Reactor()
        # This is a helper reactor, it is not the global reactor and its thread
        # is not "the" I/O thread.  Prevent it from registering it as such.
        self._reactor._registerAsIOThread = False
        self._reactorThread = Thread(
            target=self._reactor.run, args=(False,))
        self.addSystemEventTrigger(
            'after', 'shutdown', self._unmakeHelperReactor)
        self._reactorThread.start()

def addEvent(self, event, fd, action):
        """
        @see: L{IReactorWin32Events}
        """
        if self._reactor is None:
            self._makeHelperReactor()
        self._reactor.callFromThread(
            self._reactor.addEvent,
            event, _ThreadFDWrapper(self, fd, action, fd.logPrefix()),
            "_execute")

def removeEvent(self, event):
        """
        @see: L{IReactorWin32Events}
        """
        self._reactor.callFromThread(self._reactor.removeEvent, event)

def install():
    threadable.init(1)
    r = Win32Reactor()
    from . import main
    main.installReactor(r)

__all__ = ["Win32Reactor", "install"]

cmd: