@ -41,6 +41,7 @@ import getpass
import heapq
import inspect
import logging
import logging
import os
import re
import signal
@ -65,9 +66,22 @@ except ImportError:
import mitogen . core
from mitogen . core import b
from mitogen . core import bytes_partition
from mitogen . core import LOG
from mitogen . core import IOLOG
LOG = logging . getLogger ( __name__ )
# #410: we must avoid the use of socketpairs if SELinux is enabled.
try :
fp = open ( ' /sys/fs/selinux/enforce ' , ' rb ' )
try :
SELINUX_ENABLED = bool ( int ( fp . read ( ) ) )
finally :
fp . close ( )
except IOError :
SELINUX_ENABLED = False
try :
next
except NameError :
@ -91,6 +105,10 @@ try:
except ValueError :
SC_OPEN_MAX = 1024
BROKER_SHUTDOWN_MSG = (
' Connection cancelled because the associated Broker began to shut down. '
)
OPENPTY_MSG = (
" Failed to create a PTY: %s . It is likely the maximum number of PTYs has "
" been reached. Consider increasing the ' kern.tty.ptmx_max ' sysctl on OS "
@ -143,7 +161,7 @@ _core_source_partial = None
def get_log_level ( ) :
return ( LOG . level or logging . getLogger ( ) . level or logging . INFO )
return ( LOG . getEffectiveLevel( ) or logging . INFO )
def get_sys_executable ( ) :
@ -270,6 +288,41 @@ def create_socketpair(size=None):
return parentfp , childfp
def create_best_pipe ( escalates_privilege = False ) :
"""
By default we prefer to communicate with children over a UNIX socket , as a
single file descriptor can represent bidirectional communication , and a
cross - platform API exists to align buffer sizes with the needs of the
library .
SELinux prevents us setting up a privileged process to inherit an AF_UNIX
socket , a facility explicitly designed as a better replacement for pipes ,
because at some point in the mid 90 s it might have been commonly possible
for AF_INET sockets to end up undesirably connected to a privileged
process , so let ' s make up arbitrary rules breaking all sockets instead.
If SELinux is detected , fall back to using pipes .
: param bool escalates_privilege :
If : data : ` True ` , the target program may escalate privileges , causing
SELinux to disconnect AF_UNIX sockets , so avoid those .
: returns :
` ( parent_rfp , child_wfp , child_rfp , parent_wfp ) `
"""
if ( not escalates_privilege ) or ( not SELINUX_ENABLED ) :
parentfp , childfp = create_socketpair ( )
return parentfp , childfp , childfp , parentfp
parent_rfp , child_wfp = mitogen . core . pipe ( )
try :
child_rfp , parent_wfp = mitogen . core . pipe ( )
return parent_rfp , child_wfp , child_rfp , parent_wfp
except :
parent_rfp . close ( )
child_wfp . close ( )
raise
def popen ( * * kwargs ) :
"""
Wrap : class : ` subprocess . Popen ` to ensure any global : data : ` _preexec_hook `
@ -284,7 +337,8 @@ def popen(**kwargs):
return subprocess . Popen ( preexec_fn = preexec_fn , * * kwargs )
def create_child ( args , merge_stdio = False , stderr_pipe = False , preexec_fn = None ) :
def create_child ( args , merge_stdio = False , stderr_pipe = False ,
escalates_privilege = False , preexec_fn = None ) :
"""
Create a child process whose stdin / stdout is connected to a socket .
@ -293,27 +347,30 @@ def create_child(args, merge_stdio=False, stderr_pipe=False, preexec_fn=None):
: param bool merge_stdio :
If : data : ` True ` , arrange for ` stderr ` to be connected to the ` stdout `
socketpair , rather than inherited from the parent process . This may be
necessary to ensure that not TTY is connected to any stdio handle , for
necessary to ensure that no TTY is connected to any stdio handle , for
instance when using LXC .
: param bool stderr_pipe :
If : data : ` True ` and ` merge_stdio ` is : data : ` False ` , arrange for
` stderr ` to be connected to a separate pipe , to allow any ongoing debug
logs generated by e . g . SSH to be outpu as the session progresses ,
logs generated by e . g . SSH to be outpu t as the session progresses ,
without interfering with ` stdout ` .
: param bool escalates_privilege :
If : data : ` True ` , the target program may escalate privileges , causing
SELinux to disconnect AF_UNIX sockets , so avoid those .
: param function preexec_fn :
If not : data : ` None ` , a function to run within the post - fork child
before executing the target program .
: returns :
: class : ` Process ` instance .
"""
parentfp , childfp = create_socketpair ( )
# When running under a monkey patches-enabled gevent, the socket module
# yields descriptors who already have O_NONBLOCK, which is persisted across
# fork, totally breaking Python. Therefore, drop O_NONBLOCK from Python's
# future stdin fd.
mitogen . core . set_block ( childfp . fileno ( ) )
parent_rfp , child_wfp , child_rfp , parent_wfp = create_best_pipe (
escalates_privilege = escalates_privilege
)
stderr = None
stderr_r = None
if merge_stdio :
stderr = child fp
stderr = child _w fp
elif stderr_pipe :
stderr_r , stderr = mitogen . core . pipe ( )
mitogen . core . set_cloexec ( stderr_r . fileno ( ) )
@ -321,27 +378,33 @@ def create_child(args, merge_stdio=False, stderr_pipe=False, preexec_fn=None):
try :
proc = popen (
args = args ,
stdin = child fp,
stdout = child fp,
stdin = child _r fp,
stdout = child _w fp,
stderr = stderr ,
close_fds = True ,
preexec_fn = preexec_fn ,
)
except :
childfp . close ( )
parentfp . close ( )
child_rfp . close ( )
child_wfp . close ( )
parent_rfp . close ( )
parent_wfp . close ( )
if stderr_pipe :
stderr . close ( )
stderr_r . close ( )
raise
childfp . close ( )
child_rfp . close ( )
child_wfp . close ( )
if stderr_pipe :
stderr . close ( )
LOG . debug ( ' create_child() child %d fd %d , parent %d , cmd: %s ' ,
proc . pid , parentfp . fileno ( ) , os . getpid ( ) , Argv ( args ) )
return PopenProcess ( proc , stdin = parentfp , stdout = parentfp , stderr = stderr_r )
return PopenProcess (
proc = proc ,
stdin = parent_wfp ,
stdout = parent_rfp ,
stderr = stderr_r ,
)
def _acquire_controlling_tty ( ) :
@ -453,17 +516,28 @@ def tty_create_child(args):
raise
slave_fp . close ( )
LOG . debug ( ' tty_create_child() child %d fd %d , parent %d , cmd: %s ' ,
proc . pid , master_fp . fileno ( ) , os . getpid ( ) , Argv ( args ) )
return PopenProcess ( proc , stdin = master_fp , stdout = master_fp )
return PopenProcess (
proc = proc ,
stdin = master_fp ,
stdout = master_fp ,
)
def hybrid_tty_create_child ( args :
def hybrid_tty_create_child ( args , escalates_privilege = False ):
"""
Like : func : ` tty_create_child ` , except attach stdin / stdout to a socketpair
like : func : ` create_child ` , but leave stderr and the controlling TTY
attached to a TTY .
This permits high throughput communication with programs that are reached
via some program that requires a TTY for password input , like many
configurations of sudo . The UNIX TTY layer tends to have tiny ( no more than
14 KiB ) buffers , forcing many IO loop iterations when transferring bulk
data , causing significant performance loss .
: param bool escalates_privilege :
If : data : ` True ` , the target program may escalate privileges , causing
SELinux to disconnect AF_UNIX sockets , so avoid those .
: param list args :
Program argument vector .
: returns :
@ -471,20 +545,25 @@ def hybrid_tty_create_child(args):
"""
master_fp , slave_fp = openpty ( )
try :
parentfp , childfp = create_socketpair ( )
parent_rfp , child_wfp , child_rfp , parent_wfp = create_best_pipe (
escalates_privilege = escalates_privilege ,
)
try :
mitogen . core . set_block ( childfp )
mitogen . core . set_block ( child_rfp )
mitogen . core . set_block ( child_wfp )
proc = popen (
args = args ,
stdin = child fp,
stdout = child fp,
stdin = child _r fp,
stdout = child _w fp,
stderr = slave_fp ,
preexec_fn = _acquire_controlling_tty ,
close_fds = True ,
)
except :
parentfp . close ( )
childfp . close ( )
parent_rfp . close ( )
child_wfp . close ( )
parent_wfp . close ( )
child_rfp . close ( )
raise
except :
master_fp . close ( )
@ -492,17 +571,23 @@ def hybrid_tty_create_child(args):
raise
slave_fp . close ( )
childfp . close ( )
LOG . debug ( ' hybrid_tty_create_child() pid= %d stdio= %d , tty= %d , cmd: %s ' ,
proc . pid , parentfp . fileno ( ) , master_fp . fileno ( ) , Argv ( args ) )
return PopenProcess ( proc , stdin = parentfp , stdout = parentfp , stderr = master_fp )
child_rfp . close ( )
child_wfp . close ( )
return PopenProcess (
proc = proc ,
stdin = parent_wfp ,
stdout = parent_rfp ,
stderr = master_fp ,
)
class Timer ( object ) :
"""
Represents a future event .
"""
cancelled = False
#: Set to :data:`False` if :meth:`cancel` has been called, or immediately
#: prior to being executed by :meth:`TimerList.expire`.
active = True
def __init__ ( self , when , func ) :
self . when = when
@ -525,7 +610,7 @@ class Timer(object):
Cancel this event . If it has not yet executed , it will not execute
during any subsequent : meth : ` TimerList . expire ` call .
"""
self . cancelled = Tru e
self . active = Fals e
class TimerList ( object ) :
@ -561,7 +646,7 @@ class TimerList(object):
Floating point delay , or 0.0 , or : data : ` None ` if no events are
scheduled .
"""
while self . _lst and self . _lst [ 0 ] . c an celled :
while self . _lst and not self . _lst [ 0 ] . tiv e:
heapq . heappop ( self . _lst )
if self . _lst :
return max ( 0 , self . _lst [ 0 ] . when - self . _now ( ) )
@ -589,7 +674,8 @@ class TimerList(object):
now = self . _now ( )
while self . _lst and self . _lst [ 0 ] . when < = now :
timer = heapq . heappop ( self . _lst )
if not timer . cancelled :
if timer . active :
timer . active = False
timer . func ( )
@ -659,7 +745,7 @@ def _upgrade_broker(broker):
root . setLevel ( old_level )
broker . timers = TimerList ( )
LOG . debug ( ' replac ed %r with %r (new: %d readers, %d writers; '
LOG . debug ( ' upgrad ed %r with %r (new: %d readers, %d writers; '
' old: %d readers, %d writers) ' , old , new ,
len ( new . readers ) , len ( new . writers ) ,
len ( old . readers ) , len ( old . writers ) )
@ -691,7 +777,7 @@ def get_connection_class(name):
def _proxy_connect ( name , method_name , kwargs , econtext ) :
"""
Implements the target portion of Router . _proxy_connect ( ) by upgrading the
local context to a parent if it was not already , then calling back into
local process to a parent if it was not already , then calling back into
Router . _connect ( ) using the arguments passed to the parent ' s
Router . connect ( ) .
@ -737,13 +823,21 @@ def returncode_to_str(n):
class EofError ( mitogen . core . StreamError ) :
"""
Raised by : func: ` iter_read ` and : func : ` write_all ` when EOF is detected by
the child proces s.
Raised by : class: ` Connection ` when an empty read is detected from the
remote process before bootstrap complete s.
"""
# inherits from StreamError to maintain compatibility.
pass
class CancelledError ( mitogen . core . StreamError ) :
"""
Raised by : class : ` Connection ` when : meth : ` mitogen . core . Broker . shutdown ` is
called before bootstrap completes .
"""
pass
class Argv ( object ) :
"""
Wrapper to defer argv formatting when debug logging is disabled .
@ -1064,7 +1158,6 @@ class RegexProtocol(LineLoggingProtocolMixin, mitogen.core.DelimitedProtocol):
falling back to : meth : ` on_unrecognized_line_received ` and
: meth : ` on_unrecognized_partial_line_received ` .
"""
#: A sequence of 2-tuples of the form `(compiled pattern, method)` for
#: patterns that should be matched against complete (delimited) messages,
#: i.e. full lines.
@ -1105,10 +1198,10 @@ class RegexProtocol(LineLoggingProtocolMixin, mitogen.core.DelimitedProtocol):
class BootstrapProtocol ( RegexProtocol ) :
"""
Respond to stdout of a child during bootstrap . Wait for EC0_MARKER to be
by the first stage to indicate it can receive the bootstrap , then
await EC1_MARKER to indicate success , and : class : ` MitogenProtocol ` can be
enabled .
Respond to stdout of a child during bootstrap . Wait for : attr : ` EC0_MARKER `
to be written by the first stage to indicate it can receive the bootstrap ,
then await : attr : ` EC1_MARKER ` to indicate success , and
: class : ` MitogenProtocol ` can be enabled .
"""
#: Sentinel value emitted by the first stage to indicate it is ready to
#: receive the compressed bootstrap. For :mod:`mitogen.ssh` this must have
@ -1129,7 +1222,7 @@ class BootstrapProtocol(RegexProtocol):
self . _writer . write ( self . stream . conn . get_preamble ( ) )
def _on_ec1_received ( self , line , match ) :
LOG . debug ( ' %r : first stage received bootstrap ' , self )
LOG . debug ( ' %r : first stage received mitogen.core source ' , self )
def _on_ec2_received ( self , line , match ) :
LOG . debug ( ' %r : new child booted successfully ' , self )
@ -1236,6 +1329,10 @@ class Options(object):
class Connection ( object ) :
"""
Manage the lifetime of a set of : class : ` Streams < Stream > ` connecting to a
remote Python interpreter , including bootstrap , disconnection , and external
tool integration .
Base for streams capable of starting children .
"""
options_class = Options
@ -1278,7 +1375,22 @@ class Connection(object):
#: Prefix given to default names generated by :meth:`connect`.
name_prefix = u ' local '
timer = None
#: :class:`Timer` that runs :meth:`_on_timer_expired` when connection
#: timeout occurs.
_timer = None
#: When disconnection completes, instance of :class:`Reaper` used to wait
#: on the exit status of the subprocess.
_reaper = None
#: On failure, the exception object that should be propagated back to the
#: user.
exception = None
#: Extra text appended to :class:`EofError` if that exception is raised on
#: a failed connection attempt. May be used in subclasses to hint at common
#: problems with a particular connection method.
eof_error_hint = None
def __init__ ( self , options , router ) :
#: :class:`Options`
@ -1405,6 +1517,7 @@ class Connection(object):
def start_child ( self ) :
args = self . get_boot_command ( )
LOG . debug ( ' command line for %r : %s ' , self , Argv ( args ) )
try :
return self . create_child ( args = args , * * self . create_child_args )
except OSError :
@ -1412,8 +1525,6 @@ class Connection(object):
msg = ' Child start failed: %s . Command was: %s ' % ( e , Argv ( args ) )
raise mitogen . core . StreamError ( msg )
eof_error_hint = None
def _adorn_eof_error ( self , e ) :
"""
Subclasses may provide additional information in the case of a failed
@ -1422,11 +1533,11 @@ class Connection(object):
if self . eof_error_hint :
e . args = ( ' %s \n \n %s ' % ( e . args [ 0 ] , self . eof_error_hint ) , )
exception = None
def _complete_connection ( self ) :
self . . cancel ( )
self . _timer . cancel ( )
if not self . exception :
mitogen . core . unlisten ( self . _router . broker , ' shutdown ' ,
self . _on_broker_shutdown )
self . _router . register ( self . context , self . stdio_stream )
self . stdio_stream . set_protocol (
MitogenProtocol (
@ -1440,11 +1551,13 @@ class Connection(object):
"""
Fail the connection attempt .
"""
LOG . debug ( ' %s : failing connection due to %r ' ,
self . stdio_stream . name , exc )
LOG . debug ( ' %s due to %r ' ,
self . stdio_stream and self . stdio_stream . name , exc )
if self . exception is None :
self . _adorn_eof_error ( exc )
self . exception = exc
mitogen . core . unlisten ( self . _router . broker , ' shutdown ' ,
self . _on_broker_shutdown )
for stream in self . stdio_stream , self . stderr_stream :
if stream and not stream . receive_side . closed :
stream . on_disconnect ( self . _router . broker )
@ -1478,33 +1591,43 @@ class Connection(object):
def _on_streams_disconnected ( self ) :
"""
When disconnection has been detected for both our streams, cancel the
When disconnection has been detected for both , cancel the
connection timer , mark the connection failed , and reap the child
process . Do nothing if the timer has already been cancelled , indicating
some existing failure has already been noticed .
"""
if not self . timer . cancelled :
self . . cancel ( )
if self . _timer . active :
self . _ timer. cancel ( )
self . _fail_connection ( EofError (
self . eof_error_msg + get_history (
[ self . stdio_stream , self . stderr_stream ]
)
) )
self . proc . _async_reap ( self , self . _router )
def _start_timer ( self ) :
self . timer = self . _router . broker . timers . schedule (
when = self . options . connect_deadline ,
func = self . _on_timer_expired ,
)
if self . _reaper :
return
def _on_timer_expired ( self ) :
self . _fail_connection (
mitogen . core . TimeoutError (
' Failed to setup connection after %.2f seconds ' ,
self . options . connect_timeout ,
)
self . _reaper = Reaper (
broker = self . _router . broker ,
proc = self . proc ,
kill = not (
( self . detached and self . child_is_immediate_subprocess ) or
# Avoid killing so child has chance to write cProfile data
self . _router . profiling
) ,
# Don't delay shutdown waiting for a detached child, since the
# detached child may expect to live indefinitely after its parent
# exited.
wait_on_shutdown = ( not self . detached ) ,
)
self . _reaper . reap ( )
def _on_broker_shutdown ( self ) :
"""
Respond to broker . shutdown ( ) being called by failing the connection
attempt .
"""
self . _fail_connection ( CancelledError ( BROKER_SHUTDOWN_MSG ) )
def stream_factory ( self ) :
return self . stream_protocol_class . build_stream (
@ -1534,8 +1657,36 @@ class Connection(object):
self . _router . broker . start_receive ( stream )
return stream
def _on_timer_expired ( self ) :
self . _fail_connection (
mitogen . core . TimeoutError (
' Failed to setup connection after %.2f seconds ' ,
self . options . connect_timeout ,
)
)
def _async_connect ( self ) :
self . _start_timer ( )
LOG . debug ( ' creating connection to context %d using %s ' ,
self . context . context_id , self . __class__ . __module__ )
mitogen . core . listen ( self . _router . broker , ' shutdown ' ,
self . _on_broker_shutdown )
self . _timer = self . _router . broker . timers . schedule (
when = self . options . connect_deadline ,
func = self . _on_timer_expired ,
)
try :
self . proc = self . start_child ( )
except Exception :
self . _fail_connection ( sys . exc_info ( ) [ 1 ] )
return
LOG . debug ( ' child for %r started: pid: %r stdin: %r stdout: %r stderr: %r ' ,
self , self . proc . pid ,
self . proc . stdin . fileno ( ) ,
self . proc . stdout . fileno ( ) ,
self . proc . stderr and self . proc . stderr . fileno ( ) )
self . stdio_stream = self . _setup_stdio_stream ( )
if self . context . name is None :
self . context . name = self . stdio_stream . name
@ -1544,15 +1695,7 @@ class Connection(object):
self . stderr_stream = self . _setup_stderr_stream ( )
def connect ( self , context ) :
LOG . debug ( ' %r .connect() ' , self )
self . context = context
self . proc = self . start_child ( )
LOG . debug ( ' %r .connect(): pid: %r stdin: %r stdout: %r stderr: %r ' ,
self , self . proc . pid ,
self . proc . stdin . fileno ( ) ,
self . proc . stdout . fileno ( ) ,
self . proc . stderr and self . proc . stderr . fileno ( ) )
self . latch = mitogen . core . Latch ( )
self . _router . broker . defer ( self . _async_connect )
self . latch . get ( )
@ -1561,12 +1704,28 @@ class Connection(object):
class ChildIdAllocator ( object ) :
"""
Allocate new context IDs from a block of unique context IDs allocated by
the master process .
"""
def __init__ ( self , router ) :
self . router = router
self . lock = threading . Lock ( )
self . it = iter ( xrange ( 0 ) )
def allocate ( self ) :
"""
Allocate an ID , requesting a fresh block from the master if the
existing block is exhausted .
: returns :
The new context ID .
. . warning : :
This method is not safe to call from the : class : ` Broker ` thread , as
it may block on IO of its own .
"""
self . lock . acquire ( )
try :
for id_ in self . it :
@ -1733,7 +1892,9 @@ class CallChain(object):
pipelining is disabled , the exception will be logged to the target
context ' s logging framework.
"""
LOG . debug ( ' %r .call_no_reply(): %r ' , self , CallSpec ( fn , args , kwargs ) )
LOG . debug ( ' starting no-reply function call to %r : %r ' ,
self . context . name or self . context . context_id ,
CallSpec ( fn , args , kwargs ) )
self . context . send ( self . make_msg ( fn , * args , * * kwargs ) )
def call_async ( self , fn , * args , * * kwargs ) :
@ -1789,7 +1950,9 @@ class CallChain(object):
contexts and consumed as they complete using
: class : ` mitogen . select . Select ` .
"""
LOG . debug ( ' %r .call_async(): %r ' , self , CallSpec ( fn , args , kwargs ) )
LOG . debug ( ' starting function call to %s : %r ' ,
self . context . name or self . context . context_id ,
CallSpec ( fn , args , kwargs ) )
return self . context . send_async ( self . make_msg ( fn , * args , * * kwargs ) )
def call ( self , fn , * args , * * kwargs ) :
@ -1911,15 +2074,16 @@ class RouteMonitor(object):
RouteMonitor lives entirely on the broker thread , so its data requires no
locking .
: param router :
: param mitogen. master . Router router :
Router to install handlers on .
: param parent :
: param mitogen. core . Context parent :
: data : ` None ` in the master process , or reference to the parent context
we should propagate route updates towards .
"""
def __init__ ( self , router , parent = None ) :
self . router = router
self . parent = parent
self . _log = logging . getLogger ( ' mitogen.route_monitor ' )
#: Mapping of Stream instance to integer context IDs reachable via the
#: stream; used to cleanup routes during disconnection.
self . _routes_by_stream = { }
@ -2008,8 +2172,8 @@ class RouteMonitor(object):
def notice_stream ( self , stream ) :
"""
When this parent is responsible for a new directly connected child
stream , we ' re also responsible for broadcasting DEL_ROUTE upstream
if / when that child disconnects .
stream , we ' re also responsible for broadcasting
: data : ` mitogen . core . DEL_ROUTE ` upstream when that child disconnects .
"""
self . _routes_by_stream [ stream ] = set ( [ stream . protocol . remote_id ] )
self . _propagate_up ( mitogen . core . ADD_ROUTE , stream . protocol . remote_id ,
@ -2040,8 +2204,8 @@ class RouteMonitor(object):
if routes is None :
return
LOG . debug ( ' %r : %r is gone; propagating DEL_ROUTE for %r ' ,
self , stream , routes )
self . _log . debug ( ' stream %s is gone; propagating DEL_ROUTE for %r ' ,
stream . name , routes )
for target_id in routes :
self . router . del_route ( target_id )
self . _propagate_up ( mitogen . core . DEL_ROUTE , target_id )
@ -2067,12 +2231,12 @@ class RouteMonitor(object):
stream = self . router . stream_by_id ( msg . auth_id )
current = self . router . stream_by_id ( target_id )
if current and current . protocol . remote_id != mitogen . parent_id :
LOG . error ( ' Cannot add duplicate route to %r via %r , '
' already have existing route via %r ' ,
target_id , stream , current )
self . _log . error ( ' Cannot add duplicate route to %r via %r , '
' already have existing route via %r ' ,
target_id , stream , current )
return
LOG . debug ( ' Adding route to %d via %r ' , target_id , stream )
self . _log . debug ( ' Adding route to %d via %r ' , target_id , stream )
self . _routes_by_stream [ stream ] . add ( target_id )
self . router . add_route ( target_id , stream )
self . _propagate_up ( mitogen . core . ADD_ROUTE , target_id , target_name )
@ -2094,16 +2258,16 @@ class RouteMonitor(object):
stream = self . router . stream_by_id ( msg . auth_id )
if registered_stream != stream :
LOG . error ( ' %r : received DEL_ROUTE for %d from %r , expected %r ' ,
self , target_id , stream , registered_stream )
self . _log . error ( ' %d from %r , expected %r ' ,
target_id , stream , registered_stream )
return
context = self . router . context_by_id ( target_id , create = False )
if context :
LOG . debug ( ' %r : firing local disconnect for %r ' , self , context )
self . _log . debug ( ' signal for %r ' , context )
mitogen . core . fire ( context , ' disconnect ' )
LOG . debug ( ' %r : deleting route to %d via %r ' , self , target_id , stream )
self . _log . debug ( ' %d via %r ' , target_id , stream )
routes = self . _routes_by_stream . get ( stream )
if routes :
routes . discard ( target_id )
@ -2125,7 +2289,7 @@ class Router(mitogen.core.Router):
route_monitor = None
def upgrade ( self , importer , parent ) :
LOG . debug ( ' %r .upgrade() ' , self )
LOG . debug ( ' upgrading %r with capabilities to start new children ' , self )
self . id_allocator = ChildIdAllocator ( router = self )
self . responder = ModuleForwarder (
router = self ,
@ -2152,7 +2316,8 @@ class Router(mitogen.core.Router):
def get_streams ( self ) :
"""
Return a snapshot of all streams in existence at time of call .
Return an atomic snapshot of all streams in existence at time of call .
This is safe to call from any thread .
"""
self . _write_lock . acquire ( )
try :
@ -2179,13 +2344,21 @@ class Router(mitogen.core.Router):
def add_route ( self , target_id , stream ) :
"""
Arrange for messages whose ` dst_id ` is ` target_id ` to be forwarded on
the directly connected stream for ` via_id ` . This method is called
automatically in response to : data : ` mitogen . core . ADD_ROUTE ` messages ,
but remains public while the design has not yet settled , and situations
may arise where routing is not fully automatic .
Arrange for messages whose ` dst_id ` is ` target_id ` to be forwarded on a
directly connected : class : ` Stream ` . Safe to call from any thread .
This is called automatically by : class : ` RouteMonitor ` in response to
: data : ` mitogen . core . ADD_ROUTE ` messages , but remains public while the
design has not yet settled , and situations may arise where routing is
not fully automatic .
: param int target_id :
Target context ID to add a route for .
: param mitogen . core . Stream stream :
Stream over which messages to the target should be routed .
"""
LOG . debug ( ' %r .add_route( %r , %r ) ' , self , target_id , stream )
LOG . debug ( ' %r : adding route to context %r via %r ' ,
self , target_id , stream )
assert isinstance ( target_id , int )
assert isinstance ( stream , mitogen . core . Stream )
@ -2196,6 +2369,19 @@ class Router(mitogen.core.Router):
self . _write_lock . release ( )
def del_route ( self , target_id ) :
"""
Delete any route that exists for ` target_id ` . It is not an error to
delete a route that does not currently exist . Safe to call from any
thread .
This is called automatically by : class : ` RouteMonitor ` in response to
: data : ` mitogen . core . DEL_ROUTE ` messages , but remains public while the
design has not yet settled , and situations may arise where routing is
not fully automatic .
: param int target_id :
Target context ID to delete route for .
"""
LOG . debug ( ' %r : deleting route to %r ' , self , target_id )
# DEL_ROUTE may be sent by a parent if it knows this context sent
# messages to a peer that has now disconnected, to let us raise
@ -2314,82 +2500,175 @@ class Router(mitogen.core.Router):
return self . connect ( u ' ssh ' , * * kwargs )
class Process ( object ) :
_delays = [ 0.05 , 0.15 , 0.3 , 1.0 , 5.0 , 10.0 ]
name = None
class Reaper ( object ) :
"""
Asynchronous logic for reaping : class : ` Process ` objects . This is necessary
to prevent uncontrolled buildup of zombie processes in long - lived parents
that will eventually reach an OS limit , preventing creation of new threads
and processes , and to log the exit status of the child in the case of an
error .
def __init__ ( self , pid , stdin , stdout , stderr = None ) :
self . pid = pid
self . stdin = stdin
self . stdout = stdout
self . stderr = stderr
self . _returncode = None
self . _reap_count = 0
To avoid modifying process - global state such as with
: func : ` signal . set_wakeup_fd ` or installing a : data : ` signal . SIGCHLD ` handler
that might interfere with the user ' s ability to use those facilities,
Reaper polls for exit with backoff using timers installed on an associated
: class : ` Broker ` .
def __repr__ ( self ) :
return ' %s %s pid %d ' % (
type ( self ) . __name__ ,
self . name ,
self . pid ,
)
: param mitogen . core . Broker broker :
The : class : ` Broker ` on which to install timers
: param Process proc :
The process to reap .
: param bool kill :
If : data : ` True ` , send ` ` SIGTERM ` ` and ` ` SIGKILL ` ` to the process .
: param bool wait_on_shutdown :
If : data : ` True ` , delay : class : ` Broker ` shutdown if child has not yet
exited . If : data : ` False ` simply forget the child .
"""
#: :class:`Timer` that invokes :meth:`reap` after some polling delay.
_timer = None
def poll ( self ) :
raise NotImplementedError ( )
def __init__ ( self , broker , proc , kill , wait_on_shutdown ) :
self . broker = broker
self . proc = proc
self . kill = kill
self . wait_on_shutdown = wait_on_shutdown
self . _tries = 0
def _signal_child ( self , signum ) :
# For processes like sudo we cannot actually send sudo a signal,
# because it is setuid, so this is best-effort only.
LOG . debug ( ' %r : child process still alive, sending %s ' ,
self , SIGNAL_BY_NUM [ signum ] )
LOG . debug ( ' %r : sending %s ' , self . proc , SIGNAL_BY_NUM [ signum ] )
try :
os . kill ( self . p id, signum )
os . kill ( self . p roc. p id, signum )
except OSError :
e = sys . exc_info ( ) [ 1 ]
if e . args [ 0 ] != errno . EPERM :
raise
def _ async_reap( self , conn , router ) :
def _ calc_delay( self , count ) :
"""
Reap the child process during disconnection .
Calculate a poll delay given ` count ` attempts have already been made .
These constants have no principle , they just produce rapid but still
relatively conservative retries .
"""
if self . _returncode is not None :
# on_disconnect() may be invoked more than once, for example, if
# there is still a pending message to be sent after the first
# on_disconnect() call.
return
delay = 0.05
for _ in xrange ( count ) :
delay * = 1.72
return delay
if conn . detached and conn . child_is_immediate_subprocess :
LOG . debug ( ' %r : immediate child is detached, won \' t reap it ' , self )
return
def _on_broker_shutdown ( self ) :
"""
Respond to : class : ` Broker ` shutdown by cancelling the reap timer if
: attr : ` Router . await_children_at_shutdown ` is disabled . Otherwise
shutdown is delayed for up to : attr : ` Broker . shutdown_timeout ` for
subprocesses may have no intention of exiting any time soon .
"""
if not self . wait_on_shutdown :
self . _timer . cancel ( )
if router . profiling :
LOG . info ( ' %r : wont kill child because profiling=True ' , self )
return
def _install_timer ( self , delay ) :
new = self . _timer is None
self . _timer = self . broker . timers . schedule (
when = time . time ( ) + delay ,
func = self . reap ,
)
if new :
mitogen . core . listen ( self . broker , ' shutdown ' ,
self . _on_broker_shutdown )
def _remove_timer ( self ) :
if self . _timer and self . _timer . active :
self . _timer . cancel ( )
mitogen . core . unlisten ( self . broker , ' shutdown ' ,
self . _on_broker_shutdown )
self . _reap_count + = 1
status = self . poll ( )
def reap ( self ) :
"""
Reap the child process during disconnection .
"""
status = self . proc . poll ( )
if status is not None :
LOG . debug ( ' %r : %s ' , self , returncode_to_str ( status ) )
LOG . debug ( ' %r : %s ' , self . proc , returncode_to_str ( status ) )
self . _remove_timer ( )
return
i = self . _reap_count - 1
if i > = len ( self . _delays ) :
LOG . warning ( ' %r : child will not die, abandoning it ' , self )
self . _tries + = 1
if self . _tries > 20 :
LOG . warning ( ' %r : child will not exit, giving up ' , self )
self . _remove_timer ( )
return
elif i == 0 :
delay = self . _calc_delay ( self . _tries - 1 )
LOG . debug ( ' %r still running after IO disconnect, recheck in %.03f s ' ,
self . proc , delay )
self . _install_timer ( delay )
if not self . kill :
pass
elif self . _tries == 1 :
self . _signal_child ( signal . SIGTERM )
elif i == 1 :
elif self . _tries == 5 : # roughly 4 seconds
self . _signal_child ( signal . SIGKILL )
router . broker . timers . schedule (
when = time . time ( ) + self . _delays [ i ] ,
func = lambda : self . _async_reap ( conn , router ) ,
class Process ( object ) :
"""
Process objects provide a uniform interface to the : mod : ` subprocess ` and
: mod : ` mitogen . fork ` . This class is extended by : class : ` PopenProcess ` and
: class : ` mitogen . fork . Process ` .
: param int pid :
The process ID .
: param file stdin :
File object attached to standard input .
: param file stdout :
File object attached to standard output .
: param file stderr :
File object attached to standard error , or : data : ` None ` .
"""
#: Name of the process used in logs. Set to the stream/context name by
#: :class:`Connection`.
name = None
def __init__ ( self , pid , stdin , stdout , stderr = None ) :
#: The process ID.
self . pid = pid
#: File object attached to standard input.
self . stdin = stdin
#: File object attached to standard output.
self . stdout = stdout
#: File object attached to standard error.
self . stderr = stderr
def __repr__ ( self ) :
return ' %s %s pid %d ' % (
type ( self ) . __name__ ,
self . name ,
self . pid ,
)
def poll ( self ) :
"""
Fetch the child process exit status , or : data : ` None ` if it is still
running . This should be overridden by subclasses .
: returns :
Exit status in the style of the : attr : ` subprocess . Popen . returncode `
attribute , i . e . with signals represented by a negative integer .
"""
raise NotImplementedError ( )
class PopenProcess ( Process ) :
"""
: class : ` Process ` subclass wrapping a : class : ` subprocess . Popen ` object .
: param subprocess . Popen proc :
The subprocess .
"""
def __init__ ( self , proc , stdin , stdout , stderr = None ) :
super ( PopenProcess , self ) . __init__ ( proc . pid , stdin , stdout , stderr )
#: The subprocess.
self . proc = proc
def poll ( self ) :
@ -2398,8 +2677,9 @@ class PopenProcess(Process):
class ModuleForwarder ( object ) :
"""
Respond to GET_MODULE requests in a slave by forwarding the request to our
parent context , or satisfying the request from our local Importer cache .
Respond to : data : ` mitogen . core . GET_MODULE ` requests in a child by
forwarding the request to our parent context , or satisfying the request
from our local Importer cache .
"""
def __init__ ( self , router , parent_context , importer ) :
self . router = router
@ -2454,7 +2734,7 @@ class ModuleForwarder(object):
return
fullname = msg . data . decode ( ' utf-8 ' )
LOG . debug ( ' %r : %s requested by %d ' , self , fullname , msg . src_id )
LOG . debug ( ' %r : %s requested by context %d ' , self , fullname , msg . src_id )
callback = lambda : self . _on_cache_callback ( msg , fullname )
self . importer . _request_module ( fullname , callback )
David Wilson
5 years ago