def _run_callback(self, callback: Callable[[], Any]) -> None:
        """Runs a callback with error handling.

        For use in subclasses.
        """
        try:
            ret = callback()
            if ret is not None:
                from tornado import gen

                # Functions that return Futures typically swallow all
                # exceptions and store them in the Future.  If a Future
                # makes it out to the IOLoop, ensure its exception (if any)
                # gets logged too.
                try:
                    ret = gen.convert_yielded(ret)
                except gen.BadYieldError:
                    # It's not unusual for add_callback to be used with
                    # methods returning a non-None and non-yieldable
                    # result, which should just be ignored.
                    pass
                else:
                    self.add_future(ret, self._discard_future_result)
        except Exception:
            app_log.error("Exception in callback %r", callback, exc_info=True)

 def _needs_document_lock_wrapper(self, *args, **kwargs):
     with (yield self._lock.acquire()):
         if self._pending_writes is not None:
             raise RuntimeError("internal class invariant violated: _pending_writes " + \
                                "should be None if lock is not held")
         self._pending_writes = []
         try:
             result = func(self, *args, **kwargs)
             while True:
                 try:
                     future = gen.convert_yielded(result)
                 except gen.BadYieldError:
                     # result is not a yieldable thing, we are done
                     break
                 else:
                     result = yield future
         finally:
             # we want to be very sure we reset this or we'll
             # keep hitting the RuntimeError above as soon as
             # any callback goes wrong
             pending_writes = self._pending_writes
             self._pending_writes = None
         for p in pending_writes:
             yield p
     raise gen.Return(result)

 def __init__(
     self,
     client: Optional[SimpleAsyncHTTPClient],
     request: HTTPRequest,
     release_callback: Callable[[], None],
     final_callback: Callable[[HTTPResponse], None],
     max_buffer_size: int,
     tcp_client: TCPClient,
     max_header_size: int,
     max_body_size: int,
 ) -> None:
     self.io_loop = IOLoop.current()
     self.start_time = self.io_loop.time()
     self.start_wall_time = time.time()
     self.client = client
     self.request = request
     self.release_callback = release_callback
     self.final_callback = final_callback
     self.max_buffer_size = max_buffer_size
     self.tcp_client = tcp_client
     self.max_header_size = max_header_size
     self.max_body_size = max_body_size
     self.code = None  # type: Optional[int]
     self.headers = None  # type: Optional[httputil.HTTPHeaders]
     self.chunks = []  # type: List[bytes]
     self._decompressor = None
     # Timeout handle returned by IOLoop.add_timeout
     self._timeout = None  # type: object
     self._sockaddr = None
     IOLoop.current().add_future(
         gen.convert_yielded(self.run()), lambda f: f.result()
     )

 def test_failure(self):
     @inlineCallbacks
     def fn():
         if False:
             yield
         1 / 0
     f = gen.convert_yielded(fn())
     with self.assertRaises(ZeroDivisionError):
         f.result()

    def start_serving(self, delegate: httputil.HTTPServerConnectionDelegate) -> None:
        """Starts serving requests on this connection.

        :arg delegate: a `.HTTPServerConnectionDelegate`
        """
        assert isinstance(delegate, httputil.HTTPServerConnectionDelegate)
        fut = gen.convert_yielded(self._server_request_loop(delegate))
        self._serving_future = fut
        # Register the future on the IOLoop so its errors get logged.
        self.stream.io_loop.add_future(fut, lambda f: f.result())

    def _on_access_token(self, future, response):
        if response.error:
            future.set_exception(AuthError("Could not fetch access token"))
            return

        access_token = _oauth_parse_response(response.body)
        fut = self._oauth_get_user_future(access_token)
        fut = gen.convert_yielded(fut)
        fut.add_done_callback(
            functools.partial(self._on_oauth_get_user, access_token, future))

 def test_success(self):
     @inlineCallbacks
     def fn():
         if False:
             # inlineCallbacks doesn't work with regular functions;
             # must have a yield even if it's unreachable.
             yield
         returnValue(42)
     f = gen.convert_yielded(fn())
     self.assertEqual(f.result(), 42)

 def invoke():
     # important to start the sleep before starting callback
     # so any initial time spent in callback "counts against"
     # the period.
     sleep_future = self.sleep()
     result = self._func()
     try:
         callback_future = gen.convert_yielded(result)
     except gen.BadYieldError:
         # result is not a yieldable thing
         return sleep_future
     else:
         return gen.multi([sleep_future, callback_future])

def to_asyncio_future(tornado_future):
    """Convert a Tornado yieldable object to an `asyncio.Future`.

    .. versionadded:: 4.1

    .. versionchanged:: 4.3
       Now accepts any yieldable object, not just
       `tornado.concurrent.Future`.
    """
    tornado_future = convert_yielded(tornado_future)
    af = asyncio.Future()
    tornado.concurrent.chain_future(tornado_future, af)
    return af

    def _handle_connection(self, connection: socket.socket, address: Any) -> None:
        if self.ssl_options is not None:
            assert ssl, "Python 2.6+ and OpenSSL required for SSL"
            try:
                connection = ssl_wrap_socket(
                    connection,
                    self.ssl_options,
                    server_side=True,
                    do_handshake_on_connect=False,
                )
            except ssl.SSLError as err:
                if err.args[0] == ssl.SSL_ERROR_EOF:
                    return connection.close()
                else:
                    raise
            except socket.error as err:
                # If the connection is closed immediately after it is created
                # (as in a port scan), we can get one of several errors.
                # wrap_socket makes an internal call to getpeername,
                # which may return either EINVAL (Mac OS X) or ENOTCONN
                # (Linux).  If it returns ENOTCONN, this error is
                # silently swallowed by the ssl module, so we need to
                # catch another error later on (AttributeError in
                # SSLIOStream._do_ssl_handshake).
                # To test this behavior, try nmap with the -sT flag.
                # https://github.com/tornadoweb/tornado/pull/750
                if errno_from_exception(err) in (errno.ECONNABORTED, errno.EINVAL):
                    return connection.close()
                else:
                    raise
        try:
            if self.ssl_options is not None:
                stream = SSLIOStream(
                    connection,
                    max_buffer_size=self.max_buffer_size,
                    read_chunk_size=self.read_chunk_size,
                )  # type: IOStream
            else:
                stream = IOStream(
                    connection,
                    max_buffer_size=self.max_buffer_size,
                    read_chunk_size=self.read_chunk_size,
                )

            future = self.handle_stream(stream, address)
            if future is not None:
                IOLoop.current().add_future(
                    gen.convert_yielded(future), lambda f: f.result()
                )
        except Exception:
            app_log.error("Error in connection callback", exc_info=True)

def to_asyncio_future(tornado_future):
    """Convert a Tornado yieldable object to an `asyncio.Future`.

    .. versionadded:: 4.1

    .. versionchanged:: 4.3
       Now accepts any yieldable object, not just
       `tornado.concurrent.Future`.

    .. deprecated:: 5.0
       Tornado ``Futures`` have been merged with `asyncio.Future`,
       so this method is now equivalent to `tornado.gen.convert_yielded`.
    """
    return convert_yielded(tornado_future)

def yield_for_all_futures(result):
    """ Converts result into a Future by collapsing any futures inside result.

    If result is a Future we yield until it's done, then if the value inside
    the Future is another Future we yield until it's done as well, and so on.
    """
    while True:
        try:
            future = gen.convert_yielded(result)
        except gen.BadYieldError:
            # result is not a yieldable thing, we are done
            break
        else:
            result = yield future
    raise gen.Return(result)

 def run():
     try:
         result = func()
         if result is not None:
             from tornado.gen import convert_yielded
             result = convert_yielded(result)
     except Exception:
         future_cell[0] = Future()
         future_set_exc_info(future_cell[0], sys.exc_info())
     else:
         if is_future(result):
             future_cell[0] = result
         else:
             future_cell[0] = Future()
             future_cell[0].set_result(result)
     self.add_future(future_cell[0], lambda future: self.stop())

    def _run_callback(self, callback, *args, **kwargs):
        """Runs the given callback with exception handling.

        If the callback is a coroutine, returns its Future. On error, aborts the
        websocket connection and returns None.
        """
        try:
            result = callback(*args, **kwargs)
        except Exception:
            app_log.error("Uncaught exception in %s",
                          getattr(self.request, 'path', None), exc_info=True)
            self._abort()
        else:
            if result is not None:
                result = gen.convert_yielded(result)
                self.stream.io_loop.add_future(result, lambda f: f.result())
            return result

    def _do_submit_job(self, job, run_times):
        def callback(f):
            try:
                events = f.result()
            except:
                self._run_job_error(job.id, *sys.exc_info()[1:])
            else:
                self._run_job_success(job.id, events)

        if iscoroutinefunction(job.func):
            f = run_coroutine_job(job, job._jobstore_alias, run_times, self._logger.name)
        else:
            f = self.executor.submit(run_job, job, job._jobstore_alias, run_times,
                                     self._logger.name)

        f = convert_yielded(f)
        f.add_done_callback(callback)

    def _write_body(self, start_read):
        if self.request.body is not None:
            self.connection.write(self.request.body)
        elif self.request.body_producer is not None:
            fut = self.request.body_producer(self.connection.write)
            if fut is not None:
                fut = gen.convert_yielded(fut)

                def on_body_written(fut):
                    fut.result()
                    self.connection.finish()
                    if start_read:
                        self._read_response()
                self.io_loop.add_future(fut, on_body_written)
                return
        self.connection.finish()
        if start_read:
            self._read_response()

        def invoke():
            # important to start the sleep before starting callback
            # so any initial time spent in callback "counts against"
            # the period.
            sleep_future = self.sleep()
            result = self._func()

            # This is needed for Tornado >= 4.5 where convert_yielded will no
            # longer raise BadYieldError on None
            if result is None:
                return sleep_future

            try:
                callback_future = gen.convert_yielded(result)
            except gen.BadYieldError:
                # result is not a yieldable thing
                return sleep_future
            else:
                return gen.multi([sleep_future, callback_future])

        def run() -> None:
            try:
                result = func()
                if result is not None:
                    from tornado.gen import convert_yielded

                    result = convert_yielded(result)
            except Exception:
                fut = Future()  # type: Future[Any]
                future_cell[0] = fut
                future_set_exc_info(fut, sys.exc_info())
            else:
                if is_future(result):
                    future_cell[0] = result
                else:
                    fut = Future()
                    future_cell[0] = fut
                    fut.set_result(result)
            assert future_cell[0] is not None
            self.add_future(future_cell[0], lambda future: self.stop())

def yield_for_all_futures(result):
    """ Converts result into a Future by collapsing any futures inside result.

    If result is a Future we yield until it's done, then if the value inside
    the Future is another Future we yield until it's done as well, and so on.
    """
    while True:

        # This is needed for Tornado >= 4.5 where convert_yielded will no
        # longer raise BadYieldError on None
        if result is None:
            break

        try:
            future = gen.convert_yielded(result)
        except gen.BadYieldError:
            # result is not a yieldable thing, we are done
            break
        else:
            result = yield future

    raise gen.Return(result)

 def wrapper(self, *args, **kwargs):
     result = method(self, *args, **kwargs)
     if result is not None:
         result = gen.convert_yielded(result)
         # If @asynchronous is used with @gen.coroutine, (but
         # not @gen.engine), we can automatically finish the
         # request when the future resolves.  Additionally,
         # the Future will swallow any exceptions so we need
         # to throw them back out to the stack context to finish
         # the request.
         def future_complete(f):
             f.result()
             # don't finish
         tornado.ioloop.IOLoop.current().add_future(result, future_complete)
         # Once we have done this, hide the Future from our
         # caller (i.e. RequestHandler._when_complete), which
         # would otherwise set up its own callback and
         # exception handler (resulting in exceptions being
         # logged twice).
         ret = yield result.result()
         return ret
     return result