/usr/lib/python3/dist-packages/autobahn/twisted/choosereactor.py is in python3-autobahn 17.10.1+dfsg1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | ###############################################################################
#
# The MIT License (MIT)
#
# Copyright (c) Crossbar.io Technologies GmbH
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
###############################################################################
from __future__ import absolute_import
import traceback
import txaio
txaio.use_twisted()
__all__ = (
'install_optimal_reactor',
'install_reactor'
)
def install_optimal_reactor(verbose=False):
"""
Try to install the optimal Twisted reactor for this platform.
:param verbose: If ``True``, print what happens.
:type verbose: bool
"""
log = txaio.make_logger()
import sys
from twisted.python import reflect
# determine currently installed reactor, if any
##
if 'twisted.internet.reactor' in sys.modules:
current_reactor = reflect.qual(sys.modules['twisted.internet.reactor'].__class__).split('.')[-1]
else:
current_reactor = None
# depending on platform, install optimal reactor
##
if 'bsd' in sys.platform or sys.platform.startswith('darwin'):
# *BSD and MacOSX
##
if current_reactor != 'KQueueReactor':
try:
from twisted.internet import kqreactor
kqreactor.install()
except:
log.critical("Running on *BSD or MacOSX, but cannot install kqueue Twisted reactor")
log.warn("{tb}", tb=traceback.format_exc())
else:
log.debug("Running on *BSD or MacOSX and optimal reactor (kqueue) was installed.")
else:
log.debug("Running on *BSD or MacOSX and optimal reactor (kqueue) already installed.")
elif sys.platform in ['win32']:
# Windows
##
if current_reactor != 'IOCPReactor':
try:
from twisted.internet.iocpreactor import reactor as iocpreactor
iocpreactor.install()
except:
log.critical("Running on Windows, but cannot install IOCP Twisted reactor")
log.warn("{tb}", tb=traceback.format_exc())
else:
log.debug("Running on Windows and optimal reactor (ICOP) was installed.")
else:
log.debug("Running on Windows and optimal reactor (ICOP) already installed.")
elif sys.platform.startswith('linux'):
# Linux
##
if current_reactor != 'EPollReactor':
try:
from twisted.internet import epollreactor
epollreactor.install()
except:
log.critical("Running on Linux, but cannot install Epoll Twisted reactor")
log.warn("{tb}", tb=traceback.format_exc())
else:
log.debug("Running on Linux and optimal reactor (epoll) was installed.")
else:
log.debug("Running on Linux and optimal reactor (epoll) already installed.")
else:
try:
from twisted.internet import default as defaultreactor
defaultreactor.install()
except:
log.critical("Could not install default Twisted reactor for this platform")
log.warn("{tb}", tb=traceback.format_exc())
from twisted.internet import reactor
txaio.config.loop = reactor
def install_reactor(explicit_reactor=None, verbose=False):
"""
Install Twisted reactor.
:param explicit_reactor: If provided, install this reactor. Else, install
the optimal reactor.
:type explicit_reactor: obj
:param verbose: If ``True``, print what happens.
:type verbose: bool
"""
import sys
log = txaio.make_logger()
if explicit_reactor:
# install explicitly given reactor
##
from twisted.application.reactors import installReactor
log.info("Trying to install explicitly specified Twisted reactor '{reactor}'", reactor=explicit_reactor)
try:
installReactor(explicit_reactor)
except:
log.failure("Could not install Twisted reactor {reactor}\n{log_failure.value}",
reactor=explicit_reactor)
sys.exit(1)
else:
# automatically choose optimal reactor
##
log.debug("Automatically choosing optimal Twisted reactor")
install_optimal_reactor(verbose)
# now the reactor is installed, import it
from twisted.internet import reactor
txaio.config.loop = reactor
if verbose:
from twisted.python.reflect import qual
log.debug("Running Twisted reactor {reactor}", reactor=qual(reactor.__class__))
return reactor
|