# Copyright 2019, David Wilson
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# 3. Neither the name of the copyright holder nor the names of its contributors
# may be used to endorse or promote products derived from this software without
# specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

# !mitogen: minify_safe

import grp
import logging
import os
import pprint
import pwd
import stat
import sys
import threading

import mitogen.core
import mitogen.select
from mitogen.core import all
from mitogen.core import b
from mitogen.core import str_rpartition


LOG = logging.getLogger(__name__)

_pool = None
_pool_pid = None
#: Serialize pool construction.
_pool_lock = threading.Lock()


if mitogen.core.PY3:
    def func_code(func):
        return func.__code__
    def func_name(func):
        return func.__name__
else:
    def func_code(func):
        return func.func_code
    def func_name(func):
        return func.func_name


@mitogen.core.takes_router
def get_or_create_pool(size=None, router=None, context=None):
    global _pool
    global _pool_pid

    my_pid = os.getpid()
    if _pool is None or _pool.closed or my_pid != _pool_pid:
        # Avoid acquiring heavily contended lock if possible.
        _pool_lock.acquire()
        try:
            if _pool_pid != my_pid:
                if router is None:
                    # fallback to trying to get router from context if that exists
                    if context is not None:
                        router = context.router
                    else:
                        raise ValueError("Unable to create Pool! Missing router.")
                _pool = Pool(
                    router,
                    services=[],
                    size=size or 2,
                    overwrite=True,
                    recv=mitogen.core.Dispatcher._service_recv,
                )
                # In case of Broker shutdown crash, Pool can cause 'zombie'
                # processes.
                mitogen.core.listen(router.broker, 'shutdown',
                                    lambda: _pool.stop(join=True))
                _pool_pid = os.getpid()
        finally:
            _pool_lock.release()

    return _pool


def get_thread_name():
    thread = mitogen.core.threading__current_thread()
    return mitogen.core.threading__thread_name(thread)


def call(service_name, method_name, call_context=None, **kwargs):
    """
    Call a service registered with this pool, using the calling thread as a
    host.
    """
    if isinstance(service_name, mitogen.core.BytesType):
        service_name = service_name.encode('utf-8')
    elif not isinstance(service_name, mitogen.core.UnicodeType):
        service_name = service_name.name()  # Service.name()

    if call_context:
        return call_context.call_service(service_name, method_name, **kwargs)
    else:
        pool = get_or_create_pool(context=kwargs.get('context'))
        invoker = pool.get_invoker(service_name, msg=None)
        return getattr(invoker.service, method_name)(**kwargs)


def validate_arg_spec(spec, args):
    for name in spec:
        try:
            obj = args[name]
        except KeyError:
            raise mitogen.core.CallError(
                'Required argument %r missing.' % (name,)
            )

        if not isinstance(obj, spec[name]):
            raise mitogen.core.CallError(
                'Argument %r type incorrect, got %r, expected %r' % (
                    name,
                    type(obj),
                    spec[name]
                )
            )


def arg_spec(spec):
    """
    Annotate a method as requiring arguments with a specific type. This only
    validates required arguments. For optional arguments, write a manual check
    within the function.

    ::

        @mitogen.service.arg_spec({
            'path': str
        })
        def fetch_path(self, path, optional=None):
            ...

    :param dict spec:
        Mapping from argument name to expected type.
    """
    def wrapper(func):
        func.mitogen_service__arg_spec = spec
        return func
    return wrapper


def expose(policy):
    """
    Annotate a method to permit access to contexts matching an authorization
    policy. The annotation may be specified multiple times. Methods lacking any
    authorization policy are not accessible.

    ::

        @mitogen.service.expose(policy=mitogen.service.AllowParents())
        def unsafe_operation(self):
            ...

    :param mitogen.service.Policy policy:
        The policy to require.
    """
    def wrapper(func):
        func.mitogen_service__policies = (
            [policy] +
            getattr(func, 'mitogen_service__policies', [])
        )
        return func
    return wrapper


def no_reply():
    """
    Annotate a method as one that does not generate a response. Messages sent
    by the method are done so explicitly. This can be used for fire-and-forget
    endpoints where the requestee never receives a reply.
    """
    def wrapper(func):
        func.mitogen_service__no_reply = True
        return func
    return wrapper


class Error(Exception):
    """
    Raised when an error occurs configuring a service or pool.
    """
    pass  # cope with minify_source() bug.


class Policy(object):
    """
    Base security policy.
    """
    def is_authorized(self, service, msg):
        raise NotImplementedError()


class AllowAny(Policy):
    def is_authorized(self, service, msg):
        return True


class AllowParents(Policy):
    def is_authorized(self, service, msg):
        return (msg.auth_id in mitogen.parent_ids or
                msg.auth_id == mitogen.context_id)


class Activator(object):
    """
    """
    def is_permitted(self, mod_name, class_name, msg):
        return mitogen.core.has_parent_authority(msg)

    not_active_msg = (
        'Service %r is not yet activated in this context, and the '
        'caller is not privileged, therefore autoactivation is disabled.'
    )

    def activate(self, pool, service_name, msg):
        mod_name, _, class_name = str_rpartition(service_name, '.')
        if msg and not self.is_permitted(mod_name, class_name, msg):
            raise mitogen.core.CallError(self.not_active_msg, service_name)

        module = mitogen.core.import_module(mod_name)
        klass = getattr(module, class_name)
        service = klass(router=pool.router)
        pool.add(service)
        return service


class Invoker(object):
    def __init__(self, service):
        self.service = service

    def __repr__(self):
        return '%s(%s)' % (type(self).__name__, self.service)

    unauthorized_msg = (
        'Caller is not authorized to invoke %r of service %r'
    )

    def _validate(self, method_name, kwargs, msg):
        method = getattr(self.service, method_name, None)
        if method is None:
            raise mitogen.core.CallError('No such method: %r', method_name)

        policies = getattr(method, 'mitogen_service__policies', None)
        if not policies:
            raise mitogen.core.CallError('Method has no policies set.')

        if msg is not None:
            if not all(p.is_authorized(self.service, msg) for p in policies):
                raise mitogen.core.CallError(
                    self.unauthorized_msg,
                    method_name,
                    self.service.name()
                )

        required = getattr(method, 'mitogen_service__arg_spec', {})
        validate_arg_spec(required, kwargs)

    def _invoke(self, method_name, kwargs, msg):
        method = getattr(self.service, method_name)
        if 'msg' in func_code(method).co_varnames:
            kwargs['msg'] = msg  # TODO: hack

        no_reply = getattr(method, 'mitogen_service__no_reply', False)
        ret = None
        try:
            ret = method(**kwargs)
            if no_reply:
                return Service.NO_REPLY
            return ret
        except Exception:
            if no_reply:
                LOG.exception('While calling no-reply method %s.%s',
                              self.service.name(),
                              func_name(method))
            else:
                raise

    def invoke(self, method_name, kwargs, msg):
        self._validate(method_name, kwargs, msg)
        response = self._invoke(method_name, kwargs, msg)
        if response is not Service.NO_REPLY:
            msg.reply(response)


class SerializedInvoker(Invoker):
    def __init__(self, **kwargs):
        super(SerializedInvoker, self).__init__(**kwargs)
        self._lock = threading.Lock()
        self._queue = []
        self._running = False

    def _pop(self):
        self._lock.acquire()
        try:
            try:
                return self._queue.pop(0)
            except IndexError:
                self._running = False
        finally:
            self._lock.release()

    def _run(self):
        while True:
            tup = self._pop()
            if tup is None:
                return
            method_name, kwargs, msg = tup
            try:
                super(SerializedInvoker, self).invoke(method_name, kwargs, msg)
            except mitogen.core.CallError:
                e = sys.exc_info()[1]
                LOG.warning('%r: call error: %s: %s', self, msg, e)
                msg.reply(e)
            except Exception:
                LOG.exception('%r: while invoking %s()', self, method_name)
                msg.reply(mitogen.core.Message.dead())

    def invoke(self, method_name, kwargs, msg):
        self._lock.acquire()
        try:
            self._queue.append((method_name, kwargs, msg))
            first = not self._running
            self._running = True
        finally:
            self._lock.release()

        if first:
            self._run()
        return Service.NO_REPLY


class DeduplicatingInvoker(Invoker):
    """
    A service that deduplicates and caches expensive responses. Requests are
    deduplicated according to a customizable key, and the single expensive
    response is broadcast to all requestors.

    A side effect of this class is that processing of the single response is
    always serialized according to the result of :py:meth:`key_from_request`.

    Only one pool thread is blocked during generation of the response,
    regardless of the number of requestors.
    """
    def __init__(self, service):
        super(DeduplicatingInvoker, self).__init__(service)
        self._responses = {}
        self._waiters = {}
        self._lock = threading.Lock()

    def key_from_request(self, method_name, kwargs):
        """
        Generate a deduplication key from the request. The default
        implementation returns a string based on a stable representation of the
        input dictionary generated by :py:func:`pprint.pformat`.
        """
        return pprint.pformat((method_name, kwargs))

    def get_response(self, args):
        raise NotImplementedError()

    def _produce_response(self, key, response):
        self._lock.acquire()
        try:
            assert key not in self._responses
            assert key in self._waiters
            self._responses[key] = response
            for msg in self._waiters.pop(key):
                msg.reply(response)
        finally:
            self._lock.release()

    def _invoke(self, method_name, kwargs, msg):
        key = self.key_from_request(method_name, kwargs)
        self._lock.acquire()
        try:
            if key in self._responses:
                return self._responses[key]

            if key in self._waiters:
                self._waiters[key].append(msg)
                return Service.NO_REPLY

            self._waiters[key] = [msg]
        finally:
            self._lock.release()

        # I'm the unlucky thread that must generate the response.
        try:
            response = getattr(self, method_name)(**kwargs)
            self._produce_response(key, response)
        except mitogen.core.CallError:
            e = sys.exc_info()[1]
            self._produce_response(key, e)
        except Exception:
            e = sys.exc_info()[1]
            self._produce_response(key, mitogen.core.CallError(e))

        return Service.NO_REPLY


class Service(object):
    #: Sentinel object to suppress reply generation, since returning
    #: :data:`None` will trigger a response message containing the pickled
    #: :data:`None`.
    NO_REPLY = object()

    invoker_class = Invoker

    @classmethod
    def name(cls):
        return u'%s.%s' % (cls.__module__, cls.__name__)

    def __init__(self, router):
        self.router = router
        self.select = mitogen.select.Select()

    def __repr__(self):
        return '%s()' % (self.__class__.__name__,)

    def on_message(self, event):
        """
        Called when a message arrives on any of :attr:`select`'s registered
        receivers.

        :param mitogen.select.Event event:
        """
        pass

    def on_shutdown(self):
        """
        Called by Pool.shutdown() once the last worker thread has exitted.
        """
        pass


class Pool(object):
    """
    Manage a pool of at least one thread that will be used to process messages
    for a collection of services.

    Internally this is implemented by subscribing every :py:class:`Service`'s
    :py:class:`mitogen.core.Receiver` using a single
    :py:class:`mitogen.select.Select`, then arranging for every thread to
    consume messages delivered to that select.

    In this way the threads are fairly shared by all available services, and no
    resources are dedicated to a single idle service.

    There is no penalty for exposing large numbers of services; the list of
    exposed services could even be generated dynamically in response to your
    program's configuration or its input data.

    :param mitogen.core.Router router:
        :class:`mitogen.core.Router` to listen for
        :data:`mitogen.core.CALL_SERVICE` messages.
    :param list services:
        Initial list of services to register.
    :param mitogen.core.Receiver recv:
        :data:`mitogen.core.CALL_SERVICE` receiver to reuse. This is used by
        :func:`get_or_create_pool` to hand off a queue of messages from the
        Dispatcher stub handler while avoiding a race.
    """
    activator_class = Activator

    def __init__(self, router, services=(), size=1, overwrite=False,
                 recv=None):
        self.router = router
        self._activator = self.activator_class()
        self._ipc_latch = mitogen.core.Latch()
        self._receiver = mitogen.core.Receiver(
            router=router,
            handle=mitogen.core.CALL_SERVICE,
            overwrite=overwrite,
        )

        self._select = mitogen.select.Select(oneshot=False)
        self._select.add(self._receiver)
        self._select.add(self._ipc_latch)
        #: Serialize service construction.
        self._lock = threading.Lock()
        self._func_by_source = {
            self._receiver: self._on_service_call,
            self._ipc_latch: self._on_ipc_latch,
        }
        self._invoker_by_name = {}

        if recv is not None:
            # When inheriting from mitogen.core.Dispatcher, we must remove its
            # stub notification function before adding it to our Select. We
            # always overwrite this receiver since the standard service.Pool
            # handler policy differs from the one inherited from
            # core.Dispatcher.
            recv.notify = None
            self._select.add(recv)
            self._func_by_source[recv] = self._on_service_call

        for service in services:
            self.add(service)
        self._py_24_25_compat()
        self._threads = []
        for x in range(size):
            name = 'mitogen.Pool.%04x.%d' % (id(self) & 0xffff, x,)
            thread = threading.Thread(
                name=name,
                target=mitogen.core._profile_hook,
                args=('mitogen.service.pool', self._worker_main),
            )
            thread.start()
            self._threads.append(thread)
        LOG.debug('%r: initialized', self)

    def _py_24_25_compat(self):
        if sys.version_info < (2, 6):
            # import_module() is used to avoid dep scanner sending mitogen.fork
            # to all mitogen.service importers.
            os_fork = mitogen.core.import_module('mitogen.os_fork')
            os_fork._notice_broker_or_pool(self)

    @property
    def size(self):
        return len(self._threads)

    def add(self, service):
        name = service.name()
        if name in self._invoker_by_name:
            raise Error('service named %r already registered' % (name,))
        assert service.select not in self._func_by_source
        invoker = service.invoker_class(service=service)
        self._invoker_by_name[name] = invoker
        self._func_by_source[service.select] = service.on_message

    closed = False

    def stop(self, join=True):
        self.closed = True
        self._receiver.close()
        self._select.close()
        if join:
            self.join()

    def join(self):
        for th in self._threads:
            th.join()
        for invoker in self._invoker_by_name.values():
            invoker.service.on_shutdown()

    def get_invoker(self, name, msg):
        invoker = self._invoker_by_name.get(name)
        if invoker is None:
            # Avoid acquiring lock if possible.
            self._lock.acquire()
            try:
                invoker = self._invoker_by_name.get(name)
                if not invoker:
                    service = self._activator.activate(self, name, msg)
                    invoker = service.invoker_class(service=service)
                    self._invoker_by_name[name] = invoker
            finally:
                self._lock.release()

        return invoker

    def get_service(self, name):
        invoker = self.get_invoker(name, None)
        return invoker.service

    def _validate(self, msg):
        tup = msg.unpickle(throw=False)
        if not (isinstance(tup, tuple) and
                len(tup) == 3 and
                isinstance(tup[0], mitogen.core.AnyTextType) and
                isinstance(tup[1], mitogen.core.AnyTextType) and
                isinstance(tup[2], dict)):
            raise mitogen.core.CallError('Invalid message format.')

    def defer(self, func, *args, **kwargs):
        """
        Arrange for `func(*args, **kwargs)` to be invoked in the context of a
        service pool thread.
        """
        self._ipc_latch.put(lambda: func(*args, **kwargs))

    def _on_ipc_latch(self, event):
        event.data()

    def _on_service_call(self, event):
        msg = event.data
        service_name = None
        method_name = None
        try:
            self._validate(msg)
            service_name, method_name, kwargs = msg.unpickle()
            invoker = self.get_invoker(service_name, msg)
            return invoker.invoke(method_name, kwargs, msg)
        except mitogen.core.CallError:
            e = sys.exc_info()[1]
            LOG.warning('%r: call error: %s: %s', self, msg, e)
            msg.reply(e)
        except Exception:
            LOG.exception('%r: while invoking %r of %r',
                          self, method_name, service_name)
            e = sys.exc_info()[1]
            msg.reply(mitogen.core.CallError(e))

    def _worker_run(self):
        while not self.closed:
            try:
                event = self._select.get_event()
            except mitogen.core.LatchError:
                LOG.debug('thread %s exiting gracefully', get_thread_name())
                return
            except mitogen.core.ChannelError:
                LOG.debug('thread %s exiting with error: %s',
                          get_thread_name(), sys.exc_info()[1])
                return

            func = self._func_by_source[event.source]
            try:
                func(event)
            except Exception:
                LOG.exception('While handling %r using %r', event.data, func)

    def _worker_main(self):
        try:
            self._worker_run()
        except Exception:
            LOG.exception('%r: worker %r crashed', self, get_thread_name())
            raise

    def __repr__(self):
        return 'Pool(%04x, size=%d, th=%r)' % (
            id(self) & 0xffff,
            len(self._threads),
            get_thread_name(),
        )


class FileStreamState(object):
    def __init__(self):
        #: List of [(Sender, file object)]
        self.jobs = []
        self.completing = {}
        #: In-flight byte count.
        self.unacked = 0
        #: Lock.
        self.lock = threading.Lock()


class PushFileService(Service):
    """
    Push-based file service. Files are delivered and cached in RAM, sent
    recursively from parent to child. A child that requests a file via
    :meth:`get` will block until it has been delivered by a parent.

    This service will eventually be merged into FileService.
    """
    def __init__(self, **kwargs):
        super(PushFileService, self).__init__(**kwargs)
        self._lock = threading.Lock()
        self._cache = {}
        self._extra_sys_paths = set()
        self._waiters = {}
        self._sent_by_stream = {}

    def get(self, path):
        """
        Fetch a file from the cache.
        """
        assert isinstance(path, mitogen.core.UnicodeType)
        self._lock.acquire()
        try:
            if path in self._cache:
                return self._cache[path]
            latch = mitogen.core.Latch()
            waiters = self._waiters.setdefault(path, [])
            waiters.append(lambda: latch.put(None))
        finally:
            self._lock.release()

        LOG.debug('%r.get(%r) waiting for uncached file to arrive', self, path)
        latch.get()
        LOG.debug('%r.get(%r) -> %r', self, path, self._cache[path])
        return self._cache[path]

    def _forward(self, context, path):
        stream = self.router.stream_by_id(context.context_id)
        child = self.router.context_by_id(stream.protocol.remote_id)
        sent = self._sent_by_stream.setdefault(stream, set())
        if path in sent:
            if child.context_id != context.context_id:
                LOG.debug('requesting %s forward small file to %s: %s',
                          child, context, path)
                child.call_service_async(
                    service_name=self.name(),
                    method_name='forward',
                    path=path,
                    context=context
                ).close()
        else:
            LOG.debug('requesting %s cache and forward small file to %s: %s',
                      child, context, path)
            child.call_service_async(
                service_name=self.name(),
                method_name='store_and_forward',
                path=path,
                data=self._cache[path],
                context=context
            ).close()
            sent.add(path)

    @expose(policy=AllowParents())
    @arg_spec({
        'context': mitogen.core.Context,
        'paths': list,
        # 'modules': list, TODO, modules was passed into this func but it's not used yet
    })
    def propagate_paths_and_modules(self, context, paths, overridden_sources=None, extra_sys_paths=None):
        """
        One size fits all method to ensure a target context has been preloaded
        with a set of small files and Python modules.

        :param dict overridden_sources:
            Optional dict containing source code to override path's source code
        :param extra_sys_paths:
            Loads additional sys paths for use in finding modules; beneficial
            in situations like loading Ansible Collections because source code
            dependencies come from different file paths than where the source lives
        """
        for path in paths:
            overridden_source = None
            if overridden_sources is not None and path in overridden_sources:
                overridden_source = overridden_sources[path]
            self.propagate_to(context, mitogen.core.to_text(path), overridden_source)
        # self.router.responder.forward_modules(context, modules) TODO

        # NOTE: could possibly be handled by the above TODO, but not sure how forward_modules works enough
        #       to know for sure, so for now going to pass the sys paths themselves and have `propagate_to`
        #       load them up in sys.path for later import
        # ensure we don't add to sys.path the same path we've already seen
        for extra_path in extra_sys_paths:
            # store extra paths in cached set for O(1) lookup
            if extra_path not in self._extra_sys_paths:
                # not sure if it matters but we could prepend to sys.path instead if we need to
                sys.path.append(extra_path)
                self._extra_sys_paths.add(extra_path)

    @expose(policy=AllowParents())
    @arg_spec({
        'context': mitogen.core.Context,
        'path': mitogen.core.FsPathTypes,
    })
    def propagate_to(self, context, path, overridden_source=None):
        """
        If the optional parameter 'overridden_source' is passed, use
        that instead of the path's code as source code. This works around some bugs
        of source modules such as relative imports on unsupported Python versions
        """
        if path not in self._cache:
            LOG.debug('caching small file %s', path)
            if overridden_source is None:
                fp = open(path, 'rb')
                try:
                    self._cache[path] = mitogen.core.Blob(fp.read())
                finally:
                    fp.close()
            else:
                self._cache[path] = mitogen.core.Blob(overridden_source)
        self._forward(context, path)

    @expose(policy=AllowParents())
    @no_reply()
    @arg_spec({
        'path': mitogen.core.UnicodeType,
        'data': mitogen.core.Blob,
        'context': mitogen.core.Context,
    })
    def store_and_forward(self, path, data, context):
        LOG.debug('%r.store_and_forward(%r, %r, %r) %r',
                  self, path, data, context,
                  get_thread_name())
        self._lock.acquire()
        try:
            self._cache[path] = data
            waiters = self._waiters.pop(path, [])
        finally:
            self._lock.release()

        if context.context_id != mitogen.context_id:
            self._forward(context, path)
        for callback in waiters:
            callback()

    @expose(policy=AllowParents())
    @no_reply()
    @arg_spec({
        'path': mitogen.core.FsPathTypes,
        'context': mitogen.core.Context,
    })
    def forward(self, path, context):
        LOG.debug('%r.forward(%r, %r)', self, path, context)
        func = lambda: self._forward(context, path)

        self._lock.acquire()
        try:
            if path in self._cache:
                func()
            else:
                LOG.debug('%r: %r not cached yet, queueing', self, path)
                self._waiters.setdefault(path, []).append(func)
        finally:
            self._lock.release()


class FileService(Service):
    """
    Streaming file server, used to serve small and huge files alike. Paths must
    be registered by a trusted context before they will be served to a child.

    Transfers are divided among the physical streams that connect external
    contexts, ensuring each stream never has excessive data buffered in RAM,
    while still maintaining enough to fully utilize available bandwidth. This
    is achieved by making an initial bandwidth assumption, enqueueing enough
    chunks to fill that assumed pipe, then responding to delivery
    acknowledgements from the receiver by scheduling new chunks.

    Transfers proceed one-at-a-time per stream. When multiple contexts exist on
    a stream (e.g. one is the SSH account, another is a sudo account, and a
    third is a proxied SSH connection), each request is satisfied in turn
    before subsequent requests start flowing. This ensures when a stream is
    contended, priority is given to completing individual transfers rather than
    potentially aborting many partial transfers, causing the bandwidth to be
    wasted.

    Theory of operation:
        1. Trusted context (i.e. WorkerProcess) calls register(), making a
           file available to any untrusted context.
        2. Requestee context creates a mitogen.core.Receiver() to receive
           chunks, then calls fetch(path, recv.to_sender()), to set up the
           transfer.
        3. fetch() replies to the call with the file's metadata, then
           schedules an initial burst up to the window size limit (1MiB).
        4. Chunks begin to arrive in the requestee, which calls acknowledge()
           for each 128KiB received.
        5. The acknowledge() call arrives at FileService, which scheduled a new
           chunk to refill the drained window back to the size limit.
        6. When the last chunk has been pumped for a single transfer,
           Sender.close() is called causing the receive loop in
           target.py::_get_file() to exit, allowing that code to compare the
           transferred size with the total file size from the metadata.
        7. If the sizes mismatch, _get_file()'s caller is informed which will
           discard the result and log/raise an error.

    Shutdown:
        1. process.py calls service.Pool.shutdown(), which arranges for the
           service pool threads to exit and be joined, guranteeing no new
           requests can arrive, before calling Service.on_shutdown() for each
           registered service.
        2. FileService.on_shutdown() walks every in-progress transfer and calls
           Sender.close(), causing Receiver loops in the requestees to exit
           early. The size check fails and any partially downloaded file is
           discarded.
        3. Control exits _get_file() in every target, and graceful shutdown can
           proceed normally, without the associated thread needing to be
           forcefully killed.
    """
    unregistered_msg = 'Path %r is not registered with FileService.'
    context_mismatch_msg = 'sender= kwarg context must match requestee context'

    #: Burst size. With 1MiB and 10ms RTT max throughput is 100MiB/sec, which
    #: is 5x what SSH can handle on a 2011 era 2.4Ghz Core i5.
    window_size_bytes = 1048576

    def __init__(self, router):
        super(FileService, self).__init__(router)
        #: Set of registered paths.
        self._paths = set()
        #: Set of registered directory prefixes.
        self._prefixes = set()
        #: Mapping of Stream->FileStreamState.
        self._state_by_stream = {}

    def _name_or_none(self, func, n, attr):
        try:
            return getattr(func(n), attr)
        except KeyError:
            return None

    @expose(policy=AllowParents())
    @arg_spec({
        'path': mitogen.core.FsPathTypes,
    })
    def register(self, path):
        """
        Authorize a path for access by children. Repeat calls with the same
        path has no effect.

        :param str path:
            File path.
        """
        if path not in self._paths:
            LOG.debug('%r: registering %r', self, path)
            self._paths.add(path)

    @expose(policy=AllowParents())
    @arg_spec({
        'path': mitogen.core.FsPathTypes,
    })
    def register_prefix(self, path):
        """
        Authorize a path and any subpaths for access by children. Repeat calls
        with the same path has no effect.

        :param str path:
            File path.
        """
        if path not in self._prefixes:
            LOG.debug('%r: registering prefix %r', self, path)
            self._prefixes.add(path)

    def _generate_stat(self, path):
        st = os.stat(path)
        if not stat.S_ISREG(st.st_mode):
            raise IOError('%r is not a regular file.' % (path,))

        return {
            u'size': st.st_size,
            u'mode': st.st_mode,
            u'owner': self._name_or_none(pwd.getpwuid, 0, 'pw_name'),
            u'group': self._name_or_none(grp.getgrgid, 0, 'gr_name'),
            u'mtime': float(st.st_mtime),  # Python 2.4 uses int.
            u'atime': float(st.st_atime),  # Python 2.4 uses int.
        }

    def on_shutdown(self):
        """
        Respond to shutdown by sending close() to every target, allowing their
        receive loop to exit and clean up gracefully.
        """
        LOG.debug('%r.on_shutdown()', self)
        for stream, state in self._state_by_stream.items():
            state.lock.acquire()
            try:
                for sender, fp in reversed(state.jobs):
                    sender.close()
                    fp.close()
                    state.jobs.pop()
            finally:
                state.lock.release()

    # The IO loop pumps 128KiB chunks. An ideal message is a multiple of this,
    # odd-sized messages waste one tiny write() per message on the trailer.
    # Therefore subtract 10 bytes pickle overhead + 24 bytes header.
    IO_SIZE = mitogen.core.CHUNK_SIZE - (mitogen.core.Message.HEADER_LEN + (
        len(
            mitogen.core.Message.pickled(
                mitogen.core.Blob(b(' ') * mitogen.core.CHUNK_SIZE)
            ).data
        ) - mitogen.core.CHUNK_SIZE
    ))

    def _schedule_pending_unlocked(self, state):
        """
        Consider the pending transfers for a stream, pumping new chunks while
        the unacknowledged byte count is below :attr:`window_size_bytes`. Must
        be called with the FileStreamState lock held.

        :param FileStreamState state:
            Stream to schedule chunks for.
        """
        while state.jobs and state.unacked < self.window_size_bytes:
            sender, fp = state.jobs[0]
            s = fp.read(self.IO_SIZE)
            if s:
                state.unacked += len(s)
                sender.send(mitogen.core.Blob(s))
            else:
                # File is done. Cause the target's receive loop to exit by
                # closing the sender, close the file, and remove the job entry.
                sender.close()
                fp.close()
                state.jobs.pop(0)

    def _prefix_is_authorized(self, path):
        """
        Return the set of all possible directory prefixes for `path`.
        :func:`os.path.abspath` is used to ensure the path is absolute.

        :param str path:
            The path.
        :returns: Set of prefixes.
        """
        path = os.path.abspath(path)
        while True:
            if path in self._prefixes:
                return True
            if path == '/':
                break
            path = os.path.dirname(path)
        return False

    @expose(policy=AllowAny())
    @no_reply()
    @arg_spec({
        'path': mitogen.core.FsPathTypes,
        'sender': mitogen.core.Sender,
    })
    def fetch(self, path, sender, msg):
        """
        Start a transfer for a registered path.

        :param str path:
            File path.
        :param mitogen.core.Sender sender:
            Sender to receive file data.
        :returns:
            Dict containing the file metadata:

            * ``size``: File size in bytes.
            * ``mode``: Integer file mode.
            * ``owner``: Owner account name on host machine.
            * ``group``: Owner group name on host machine.
            * ``mtime``: Floating point modification time.
            * ``ctime``: Floating point change time.
        :raises Error:
            Unregistered path, or Sender did not match requestee context.
        """
        if (
            (path not in self._paths) and
            (not self._prefix_is_authorized(path)) and
            (not mitogen.core._has_parent_authority(msg.auth_id))
        ):
            msg.reply(mitogen.core.CallError(
                Error(self.unregistered_msg % (path,))
            ))
            return

        if msg.src_id != sender.context.context_id:
            msg.reply(mitogen.core.CallError(
                Error(self.context_mismatch_msg)
            ))
            return

        LOG.debug('Serving %r', path)

        # Response must arrive first so requestee can begin receive loop,
        # otherwise first ack won't arrive until all pending chunks were
        # delivered. In that case max BDP would always be 128KiB, aka. max
        # ~10Mbit/sec over a 100ms link.
        try:
            fp = open(path, 'rb', self.IO_SIZE)
            msg.reply(self._generate_stat(path))
        except IOError:
            msg.reply(mitogen.core.CallError(
                sys.exc_info()[1]
            ))
            return

        stream = self.router.stream_by_id(sender.context.context_id)
        state = self._state_by_stream.setdefault(stream, FileStreamState())
        state.lock.acquire()
        try:
            state.jobs.append((sender, fp))
            self._schedule_pending_unlocked(state)
        finally:
            state.lock.release()

    @expose(policy=AllowAny())
    @no_reply()
    @arg_spec({
        'size': int,
    })
    @no_reply()
    def acknowledge(self, size, msg):
        """
        Acknowledge bytes received by a transfer target, scheduling new chunks
        to keep the window full. This should be called for every chunk received
        by the target.
        """
        stream = self.router.stream_by_id(msg.src_id)
        state = self._state_by_stream[stream]
        state.lock.acquire()
        try:
            if state.unacked < size:
                LOG.error('%r.acknowledge(src_id %d): unacked=%d < size %d',
                          self, msg.src_id, state.unacked, size)
            state.unacked -= min(state.unacked, size)
            self._schedule_pending_unlocked(state)
        finally:
            state.lock.release()

    @classmethod
    def get(cls, context, path, out_fp):
        """
        Streamily download a file from the connection multiplexer process in
        the controller.

        :param mitogen.core.Context context:
            Reference to the context hosting the FileService that will be used
            to fetch the file.
        :param bytes path:
            FileService registered name of the input file.
        :param bytes out_path:
            Name of the output path on the local disk.
        :returns:
            Tuple of (`ok`, `metadata`), where `ok` is :data:`True` on success,
            or :data:`False` if the transfer was interrupted and the output
            should be discarded.

            `metadata` is a dictionary of file metadata as documented in
            :meth:`fetch`.
        """
        LOG.debug('get_file(): fetching %r from %r', path, context)
        t0 = mitogen.core.now()
        recv = mitogen.core.Receiver(router=context.router)
        metadata = context.call_service(
            service_name=cls.name(),
            method_name='fetch',
            path=path,
            sender=recv.to_sender(),
        )

        received_bytes = 0
        for chunk in recv:
            s = chunk.unpickle()
            LOG.debug('get_file(%r): received %d bytes', path, len(s))
            context.call_service_async(
                service_name=cls.name(),
                method_name='acknowledge',
                size=len(s),
            ).close()
            out_fp.write(s)
            received_bytes += len(s)

        ok = received_bytes == metadata['size']
        if received_bytes < metadata['size']:
            LOG.error('get_file(%r): receiver was closed early, controller '
                      'may be shutting down, or the file was truncated '
                      'during transfer. Expected %d bytes, received %d.',
                      path, metadata['size'], received_bytes)
        elif received_bytes > metadata['size']:
            LOG.error('get_file(%r): the file appears to have grown '
                      'while transfer was in progress. Expected %d '
                      'bytes, received %d.',
                      path, metadata['size'], received_bytes)

        LOG.debug('target.get_file(): fetched %d bytes of %r from %r in %dms',
                  metadata['size'], path, context,
                  1000 * (mitogen.core.now() - t0))
        return ok, metadata