def killsock(sock):
    """Attempts to cleanly shutdown a socket. Regardless of cleanliness,
    ensures that upon return, the socket is fully closed, catching any
    exceptions along the way. A safe and prompt way to dispose of the
    socket, freeing system resources.
    """
    if hasattr(sock, '_sock'):
        ml.ld("Killing socket {0}/FD {1}", id(sock), sock._sock.fileno())
    else:
        ml.ld("Killing socket {0}", id(sock))
    try:
        # TODO: better ideas for how to get SHUT_RDWR constant?
        sock.shutdown(gevent.socket.SHUT_RDWR)
    except gevent.socket.error:
        pass  # just being nice to the server, don't care if it fails
    except Exception as e:
        log_rec = context.get_context().log.info("SOCKET", "SHUTDOWN")
        log_rec.failure('error ({exc}) shutting down socket: {socket}',
                        socket=sock, exc=e)
    try:
        sock.close()
    except gevent.socket.error:
        pass  # just being nice to the server, don't care if it fails
    except Exception as e:
        log_rec = context.get_context().log.info("SOCKET", "CLOSE")
        log_rec.failure('error ({exc}) closing socket: {socket}',
                        socket=sock, exc=e)

 def apply(self, func, args, kwargs):
     done = gevent.event.Event()
     self.in_q.append((done, func, args, kwargs, curtime()))
     context.get_context().stats['cpu_bound.depth'].add(1 + len(self.in_q))
     while not self.in_async:
         gevent.sleep(0.01)  # poll until worker thread has initialized
     self.in_async.send()
     done.wait()
     res = self.results[done]
     del self.results[done]
     if isinstance(res, self._Caught):
         raise res.err
     return res

def get_sampro_data():
    processed = defaultdict(int)
    profiler = context.get_context().profiler
    if profiler is None:
        return "(sampling disabled; enable with infra.get_context().set_sampling(True))"
    for k, v in context.get_context().profiler.live_data_copy().iteritems():
        code, lineno, parentcode = k
        if parentcode is None:
            # TODO: shorten filenames up to the .git directory
            key = (code.co_name + "(" + code.co_filename +
                   ", " + str(code.co_firstlineno) + ")")
            processed[key] += v
    processed = reversed(sorted([(v, k) for k, v in processed.items()]))
    return dict(processed)

def _make_server_sock(address, socket_type=gevent.socket.socket):
    ml.ld("about to bind to {0!r}", address)
    ml2.info('listen_prep').success('about to bind to {addr}', addr=address)
    if isinstance(address, basestring):
        if not hasattr(socket, "AF_UNIX"):
            raise ValueError(
                "attempted to bind to Unix Domain Socket {0:r} "
                "on system without UDS support".format(address))
        if os.path.exists(address):
            os.unlink(address)
        sock = socket_type(socket.AF_UNIX)
    else:
        sock = socket_type()
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(address)

    # NOTE: this is a "hint" to the OS than a strict rule about backlog size
    with context.get_context().log.critical('LISTEN') as _log:
        sock.listen(DEFAULT_SOCKET_LISTEN_SIZE)
        if ufork is not None:
            # we may fork, so protect ourselves
            flags = fcntl.fcntl(sock.fileno(), fcntl.F_GETFD)
            fcntl.fcntl(sock.fileno(), fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)
        #ml.la("Listen to {0!r} gave this socket {1!r}", address, sock)
        _log.success('Listen to {addr} gave {sock}', addr=address, sock=sock)
    return sock

 def _run(self):
     # in_cpubound_thread is sentinel to prevent double thread dispatch
     context.get_context().thread_locals.in_cpubound_thread = True
     try:
         self.in_async = gevent.get_hub().loop.async()
         self.in_q_has_data = gevent.event.Event()
         self.in_async.start(self.in_q_has_data.set)
         while not self.stopping:
             if not self.in_q:
                 # wait for more work
                 self.in_q_has_data.clear()
                 self.in_q_has_data.wait()
                 continue
             # arbitrary non-preemptive service discipline can go here
             # FIFO for now, but we should experiment with others
             jobid, func, args, kwargs, enqueued = self.in_q.popleft()
             started = curtime()
             try:
                 ret = self.results[jobid] = func(*args, **kwargs)
             except Exception as e:
                 ret = self.results[jobid] = self._Caught(e)
             self.out_q.append(jobid)
             self.out_async.send()
             # keep track of some statistics
             queued, duration = started - enqueued, curtime() - started
             size = None
             # ret s set up above before async send
             if hasattr(ret, '__len__') and callable(ret.__len__):
                 size = len(ret)
             _queue_stats('cpu_bound', func.__name__, queued, duration, size)
     except:
         self._error()

        def g(*a, **kw):
            enqueued = curtime()
            ctx = context.get_context()
            started = []

            def in_thread(*a, **kw):
                ml.ld3("In thread {0}", f.__name__)
                started.append(curtime())
                return f(*a, **kw)

            # some modules import things lazily; it is too dangerous
            # to run a function in another thread if the import lock is
            # held by the current thread (this happens rarely -- only
            # if the thread dispatched function is being executed at
            # the import time of a module)
            if not ctx.cpu_thread_enabled or imp.lock_held():
                ret = in_thread(*a, **kw)
            elif in_threadpool() is self.pool:
                ret = in_thread(*a, **kw)
            else:
                ctx.stats[self.name + '.depth'].add(1 + len(self.pool))
                ret = self.pool.apply_e((Exception,), in_thread, a, kw)
                ml.ld3("Enqueued to thread {0}/depth {1}", f.__name__, len(pool))
            start = started[0]
            duration = curtime() - start
            queued = start - enqueued
            if hasattr(ret, '__len__') and callable(ret.__len__):
                prsize = ret.__len__()  # parameter-or-return size
            elif a and hasattr(a[0], '__len__') and callable(a[0].__len__):
                prsize = a[0].__len__()
            else:
                prsize = None
            _queue_stats(name, f.__name__, queued, duration, prsize)
            return ret

def close_threadpool():
    tlocals = context.get_context().thread_locals
    if hasattr(tlocals, 'cpu_bound_thread'):
        ml.ld2("Closing thread pool {0}", id(tlocals.cpu_thread))
        cpu_thread = tlocals.cpu_bound_thread
        cpu_thread.stopping = True
        del tlocals.cpu_bound_thread

 def _error(self):
     # TODO: something better, but this is darn useful for debugging
     import traceback
     traceback.print_exc()
     ctx = context.get_context()
     tl = ctx.thread_locals
     if hasattr(tl, 'cpu_bound_thread') and tl.cpu_bound_thread is self:
         del tl.cpu_bound_thread

def staggered_retries(run, *a, **kw):
    """
    A version of spawn that will block will it is done
    running the function, and which will call the function
    repeatedly as time progresses through the timeouts list.

    Best used for idempotent network calls (e.g. HTTP GETs).

    e.g.::

        user_data = async.staggered_retries(get_data, max_results,
                                            latent_data_ok, public_credential_load,
                                            timeouts_secs=[0.1, 0.5, 1, 2])

    returns None on timeout.
    """
    ctx = context.get_context()
    ready = gevent.event.Event()
    ready.clear()

    def callback(source):
        if source.successful():
            ready.set()

    if 'timeouts_secs' in kw:
        timeouts_secs = kw.pop('timeouts_secs')
    else:
        timeouts_secs = [0.05, 0.1, 0.15, 0.2]
    if timeouts_secs[0] > 0:
        timeouts_secs.insert(0, 0)
    gs = gevent.spawn(run, *a, **kw)
    gs.link_value(callback)
    running = [gs]
    for i in range(1, len(timeouts_secs)):
        this_timeout = timeouts_secs[i] - timeouts_secs[i - 1]
        if ctx.dev:
            this_timeout = this_timeout * 5.0
        ml.ld2("Using timeout {0}", this_timeout)
        try:
            with gevent.Timeout(this_timeout):
                ready.wait()
                break
        except gevent.Timeout:
            ml.ld2("Timed out!")
            log_rec = ctx.log.critical('ASYNC.STAGGER', run.__name__)
            log_rec.failure('timed out after {timeout}',
                            timeout=this_timeout)
            gs = gevent.spawn(run, *a, **kw)
            gs.link_value(callback)
            running.append(gs)
    vals = [l.value for l in running if l.successful()]
    for g in running:
        g.kill()
    if vals:
        return vals[0]
    else:
        return None

 def __init__(self, socket, address, server, rfile=None):
     if rfile is None and hasattr(socket, "_makefile_refs"):
         rfile = socket.makefile()
         # restore gEvent's work-around of not letting wsgi.environ['input']
         # keep the socket alive in CLOSE_WAIT state after client is gone
         # to work with async.SSLSocket
         socket._makefile_refs -= 1
     self.state = context.get_context().markov_stats['wsgi_handler'].make_transitor('new')
     super(MakeFileCloseWSGIHandler, self).__init__(socket, address, server, rfile)

def get_web_logs():
    sgrp = context.get_context().server_group
    if not sgrp:
        raise EnvironmentError("context.server_group unset (infra.serve() not called)")
    result = []
    for serv in sgrp.servers:
        if hasattr(serv, "log"):
            result.extend(serv.log.msgs)
    return result

 def start_accepting(self):
     # NB: This is called via BaseServer.start, which is invoked
     # *only* in post_fork.  It is *criticial* this thread is *not*
     # started prior to forking, lest it die.
     if self._watcher is None:
         accept_maxlen = context.get_context().accept_queue_maxlen
         self._watcher = ThreadWatcher(self.socket, self.loop,
                                       accept_maxlen)
         self._watcher.start(self._do_read)

    def do_read(self):
        # invoked via BaseServer._do_read.  Whereas
        # StreamServer.do_read calls self.socket.accept, we just
        # need to pop off our queue
        if not self._watcher:
            return
        if not self._watcher.queue:
            raise RuntimeError('QUEUE DISAPPEARED')
        client_socket, address, exc, pushed_at = self._watcher.queue.pop()

        age = nanotime() - pushed_at
        context.get_context().stats[CONN_AGE_STATS].add(age / 1e6)

        if exc is not None:
            # raise the Exception
            raise exc

        return gevent.socket.socket(_sock=client_socket), address

def get_connections():
    ctx = context.get_context()
    ret = {}
    for model in ctx.connection_mgr.server_models.values():
        ret[model.address] = {
            "last_error": datetime.datetime.fromtimestamp(
                int(model.last_error)).strftime('%Y-%m-%d %H:%M:%S'),
            "sockets": [repr(s) for s in model.active_connections.values()]
        }
    return ret

def _filter_stats(prefix):
    out = {}
    ctx = context.get_context()
    # pick up numerical stats
    stat_dict_names = (
        "stats", "durations", "intervals", "volatile_stats", "markov_stats", "sketches")
    for stat_dict_name in stat_dict_names:
        stats = getattr(ctx, stat_dict_name)
        out.update([(k, v) for k,v in stats.items() if prefix in k])
    return out

 def do_open(self, conn_type, req):
     get_log_record = getattr(context.get_context().log, self.LOG_LEVEL)
     with get_log_record(**self.get_log_kwargs(req)) as log_record:
         self.pre_request(log_record, req)
         log_record['full_url'] = req.get_full_url()
         resp = urllib2.AbstractHTTPHandler.do_open(self, conn_type, req)
         log_record['status_code'] = resp.getcode()
         log_record.success('{record_name} got {status_code}')
         self.post_request(log_record, req, resp)
         return resp

def get_warnings(path=None):
    warns = context.get_context().get_warnings()
    if path:
        path_segs = path.split('.')
        for seg in path_segs:
            if seg in warns:
                warns = warns[seg]
            else:
                return "(none)"
    warns = _dict_map(warns, _transform)
    return warns

def get_config_dict():
    'Returns information about the current environment in a dictionary'
    ctx = context.get_context()
    data = []
    keys_handled_directly = ['protected', 'ssl_contexts']
    for k in ctx.__dict__:
        if k not in keys_handled_directly:
            data.append([k, getattr(ctx, k)])
    # TODO: log and ssl_context info

    return dict([(e[0], e[1:]) for e in data])

    def wrap_socket_and_handle(self, client_socket, address):
        ctx = context.get_context()
        ctx.client_sockets[client_socket] = 1

        if not self.ssl_args:
            raise ValueError('https server requires server-side'
                             ' SSL certificate')
        protocol = _socket_protocol(client_socket)
        if protocol == "ssl":
            with ctx.log.info('HANDLE.SSL', str(address[0])):
                ssl_socket = async.wrap_socket_context(
                    client_socket, **self.ssl_args)
            ctx.sketches['http.ssl.client_ips'].add(address[0])
            return self.handle(ssl_socket, address)
        elif protocol == "http":
            with ctx.log.info('HANDLE.NOSSL', str(address[0])):
                self._no_ssl(client_socket, address)
            ctx.sketches['http.client_ips'].add(address[0])
        else:
            context.get_context().intervals["server.pings"].tick()

def _queue_stats(qname, fname, queued_ns, duration_ns, size_B=None):
    ctx = context.get_context()
    fprefix = qname + '.' + fname
    ctx.stats[fprefix + '.queued(ms)'].add(queued_ns * 1000)
    ctx.stats[fprefix + '.duration(ms)'].add(duration_ns * 1000)
    ctx.stats[qname + '.queued(ms)'].add(queued_ns * 1000)
    ctx.stats[qname + '.duration(ms)'].add(duration_ns * 1000)
    if size_B is not None:
        ctx.stats[fprefix + '.len'].add(size_B)
        if duration_ns:  # may be 0
            ctx.stats[fprefix + '.rate(B/ms)'].add(size_B / (duration_ns * 1000.0))