def _do_ssl_handshake(self):
     # Based on code from test_ssl.py in the python stdlib
     try:
         self._handshake_reading = False
         self._handshake_writing = False
         self.socket.do_handshake()
     except ssl.SSLError, err:
         if err.args[0] == ssl.SSL_ERROR_WANT_READ:
             self._handshake_reading = True
             return
         elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
             self._handshake_writing = True
             return
         elif err.args[0] in (ssl.SSL_ERROR_EOF,
                              ssl.SSL_ERROR_ZERO_RETURN):
             return self.close()
         elif err.args[0] == ssl.SSL_ERROR_SSL:
             try:
                 peer = self.socket.getpeername()
             except:
                 peer = '(not connected)'
             gen_log.warning("SSL Error on %d %s: %s",
                             self.socket.fileno(), peer, err)
             return self.close()
         raise

 def to_json(self, content):
     try:
         if content:
             content = json.loads(content)
     except Exception as e:
         gen_log.warning("HttpHelper to json Error: " + str(e))
     return content

    def get_authenticated_user(self, callback, http_client=None):
        """Gets the OAuth authorized user and access token on callback.

        This method should be called from the handler for your registered
        OAuth Callback URL to complete the registration process. We call
        callback with the authenticated user, which in addition to standard
        attributes like 'name' includes the 'access_key' attribute, which
        contains the OAuth access you can use to make authorized requests
        to this service on behalf of the user.

        """
        request_key = escape.utf8(self.get_argument("oauth_token"))
        oauth_verifier = self.get_argument("oauth_verifier", None)
        request_cookie = self.get_cookie("_oauth_request_token")
        if not request_cookie:
            gen_log.warning("Missing OAuth request token cookie")
            callback(None)
            return
        self.clear_cookie("_oauth_request_token")
        cookie_key, cookie_secret = [base64.b64decode(escape.utf8(i)) for i in request_cookie.split("|")]
        if cookie_key != request_key:
            gen_log.info((cookie_key, request_key, request_cookie))
            gen_log.warning("Request token does not match cookie")
            callback(None)
            return
        token = dict(key=cookie_key, secret=cookie_secret)
        if oauth_verifier:
            token["verifier"] = oauth_verifier
        if http_client is None:
            http_client = self.get_auth_http_client()
        http_client.fetch(self._oauth_access_token_url(token),
                          self.async_callback(self._on_access_token, callback))

 def _do_ssl_handshake(self):
     # Based on code from test_ssl.py in the python stdlib
     try:
         self._handshake_reading = False
         self._handshake_writing = False
         self.socket.do_handshake()
     except ssl.SSLError as err:
         if err.args[0] == ssl.SSL_ERROR_WANT_READ:
             self._handshake_reading = True
             return
         elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
             self._handshake_writing = True
             return
         elif err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN):
             return self.close(exc_info=True)
         elif err.args[0] == ssl.SSL_ERROR_SSL:
             try:
                 peer = self.socket.getpeername()
             except:
                 peer = "(not connected)"
             gen_log.warning("SSL Error on %d %s: %s", self.socket.fileno(), peer, err)
             return self.close(exc_info=True)
         raise
     except socket.error as err:
         if err.args[0] in (errno.ECONNABORTED, errno.ECONNRESET):
             return self.close(exc_info=True)
     else:
         self._ssl_accepting = False
         if self._ssl_connect_callback is not None:
             callback = self._ssl_connect_callback
             self._ssl_connect_callback = None
             self._run_callback(callback)

def _auth_future_to_callback(callback, future):
	try:
		result = future.result()
	except AuthError as e:
		gen_log.warning(str(e))
		result = None
	callback(result)

 def _on_friendfeed_request(self, callback, response):
     if response.error:
         gen_log.warning("Error response %s fetching %s", response.error,
                         response.request.url)
         callback(None)
         return
     callback(escape.json_decode(response.body))

def parse_body_arguments(content_type, body, arguments, files):
    """Parses a form request body.

    Supports ``application/x-www-form-urlencoded`` and
    ``multipart/form-data``.  The ``content_type`` parameter should be
    a string and ``body`` should be a byte string.  The ``arguments``
    and ``files`` parameters are dictionaries that will be updated
    with the parsed contents.
    """
    if content_type.startswith("application/x-www-form-urlencoded"):
        try:
            uri_arguments = parse_qs_bytes(native_str(body), keep_blank_values=True)
        except Exception as e:
            gen_log.warning('Invalid x-www-form-urlencoded body: %s', e)
            uri_arguments = {}
        for name, values in uri_arguments.items():
            if values:
                arguments.setdefault(name, []).extend(values)
    elif content_type.startswith("multipart/form-data"):
        fields = content_type.split(";")
        for field in fields:
            k, sep, v = field.strip().partition("=")
            if k == "boundary" and v:
                parse_multipart_form_data(utf8(v), body, arguments, files)
                break
        else:
            gen_log.warning("Invalid multipart/form-data")

    def initialize(self, io_loop=None, max_clients=10):
        self.io_loop = io_loop
        self._multi = pycurl.CurlMulti()
        self._multi.setopt(pycurl.M_TIMERFUNCTION, self._set_timeout)
        self._multi.setopt(pycurl.M_SOCKETFUNCTION, self._handle_socket)
        self._curls = [_curl_create() for i in xrange(max_clients)]
        self._free_list = self._curls[:]
        self._requests = collections.deque()
        self._fds = {}
        self._timeout = None

        try:
            self._socket_action = self._multi.socket_action
        except AttributeError:
            # socket_action is found in pycurl since 7.18.2 (it's been
            # in libcurl longer than that but wasn't accessible to
            # python).
            gen_log.warning("socket_action method missing from pycurl; "
                            "falling back to socket_all. Upgrading "
                            "libcurl and pycurl will improve performance")
            self._socket_action = \
                lambda fd, action: self._multi.socket_all()

        # libcurl has bugs that sometimes cause it to not report all
        # relevant file descriptors and timeouts to TIMERFUNCTION/
        # SOCKETFUNCTION.  Mitigate the effects of such bugs by
        # forcing a periodic scan of all active requests.
        self._force_timeout_callback = ioloop.PeriodicCallback(
            self._handle_force_timeout, 1000, io_loop=io_loop)
        self._force_timeout_callback.start()

 def _handle_read(self):
     try:
         try:
             # Pretend to have a pending callback so that an EOF in
             # _read_to_buffer doesn't trigger an immediate close
             # callback.  At the end of this method we'll either
             # estabilsh a real pending callback via
             # _read_from_buffer or run the close callback.
             #
             # We need two try statements here so that
             # pending_callbacks is decremented before the `except`
             # clause below (which calls `close` and does need to
             # trigger the callback)
             self._pending_callbacks += 1
             while True:
                 # Read from the socket until we get EWOULDBLOCK or equivalent.
                 # SSL sockets do some internal buffering, and if the data is
                 # sitting in the SSL object's buffer select() and friends
                 # can't see it; the only way to find out if it's there is to
                 # try to read it.
                 if self._read_to_buffer() == 0:
                     break
         finally:
             self._pending_callbacks -= 1
     except Exception:
         gen_log.warning("error on read", exc_info=True)
         self.close()
         return
     if self._read_from_buffer():
         return
     else:
         self._maybe_run_close_callback()

 def _handle_write(self):
     while self._write_buffer:
         try:
             if not self._write_buffer_frozen:
                 # On windows, socket.send blows up if given a
                 # write buffer that's too large, instead of just
                 # returning the number of bytes it was able to
                 # process.  Therefore we must not call socket.send
                 # with more than 128KB at a time.
                 _merge_prefix(self._write_buffer, 128 * 1024)
             num_bytes = self.write_to_fd(self._write_buffer[0])
             if num_bytes == 0:
                 # With OpenSSL, if we couldn't write the entire buffer,
                 # the very same string object must be used on the
                 # next call to send.  Therefore we suppress
                 # merging the write buffer after an incomplete send.
                 # A cleaner solution would be to set
                 # SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER, but this is
                 # not yet accessible from python
                 # (http://bugs.python.org/issue8240)
                 self._write_buffer_frozen = True
                 break
             self._write_buffer_frozen = False
             _merge_prefix(self._write_buffer, num_bytes)
             self._write_buffer.popleft()
         except socket.error, e:
             if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
                 self._write_buffer_frozen = True
                 break
             else:
                 gen_log.warning("Write error on %d: %s",
                                 self.fileno(), e)
                 self.close()
                 return

    def connect(self, address, callback=None):
        """Connects the socket to a remote address without blocking.

        May only be called if the socket passed to the constructor was
        not previously connected.  The address parameter is in the
        same format as for socket.connect, i.e. a (host, port) tuple.
        If callback is specified, it will be called when the
        connection is completed.

        Note that it is safe to call IOStream.write while the
        connection is pending, in which case the data will be written
        as soon as the connection is ready.  Calling IOStream read
        methods before the socket is connected works on some platforms
        but is non-portable.
        """
        self._connecting = True
        try:
            self.socket.connect(address)
        except socket.error, e:
            # In non-blocking mode we expect connect() to raise an
            # exception with EINPROGRESS or EWOULDBLOCK.
            #
            # On freebsd, other errors such as ECONNREFUSED may be
            # returned immediately when attempting to connect to
            # localhost, so handle them the same way as an error
            # reported later in _handle_connect.
            if e.args[0] not in (errno.EINPROGRESS, errno.EWOULDBLOCK):
                gen_log.warning("Connect error on fd %d: %s",
                                self.socket.fileno(), e)
                self.close()
                return

    def _verify_cert(self, peercert):
        """Returns True if peercert is valid according to the configured
        validation mode and hostname.

        The ssl handshake already tested the certificate for a valid
        CA signature; the only thing that remains is to check
        the hostname.
        """
        if isinstance(self._ssl_options, dict):
            verify_mode = self._ssl_options.get('cert_reqs', ssl.CERT_NONE)
        elif isinstance(self._ssl_options, ssl.SSLContext):
            verify_mode = self._ssl_options.verify_mode
        assert verify_mode in (ssl.CERT_NONE, ssl.CERT_REQUIRED, ssl.CERT_OPTIONAL)
        if verify_mode == ssl.CERT_NONE or self._server_hostname is None:
            return True
        cert = self.socket.getpeercert()
        if cert is None and verify_mode == ssl.CERT_REQUIRED:
            gen_log.warning("No SSL certificate given")
            return False
        try:
            ssl_match_hostname(peercert, self._server_hostname)
        except SSLCertificateError:
            gen_log.warning("Invalid SSL certificate", exc_info=True)
            return False
        else:
            return True

    def _handle_events(self, fd, events):
        """This method is the actual handler for IOLoop, that gets called whenever
        an event on my socket is posted. It dispatches to _handle_recv, etc."""
        # print "handling events"
        if not self.socket:
            gen_log.warning("Got events for closed stream %s", fd)
            return
        try:
            # dispatch events:
            if events & IOLoop.ERROR:
                gen_log.error("got POLLERR event on ZMQStream, which doesn't make sense")
                return
            if events & IOLoop.READ:
                self._handle_recv()
                if not self.socket:
                    return
            if events & IOLoop.WRITE:
                self._handle_send()
                if not self.socket:
                    return

            # rebuild the poll state
            self._rebuild_io_state()
        except:
            gen_log.error("Uncaught exception, closing connection.",
                          exc_info=True)
            self.close()
            raise

    def _on_authentication_verified(self, callback, response):
        if response.error or b("is_valid:true") not in response.body:
            gen_log.warning("Invalid OpenID response: %s", response.error or
                            response.body)
            callback(None)
            return

        # Make sure we got back at least an email from attribute exchange
        ax_ns = None
        for name in self.request.arguments.iterkeys():
            if name.startswith("openid.ns.") and \
               self.get_argument(name) == u"http://openid.net/srv/ax/1.0":
                ax_ns = name[10:]
                break

        def get_ax_arg(uri):
            if not ax_ns:
                return u""
            prefix = "openid." + ax_ns + ".type."
            ax_name = None
            for name in self.request.arguments.iterkeys():
                if self.get_argument(name) == uri and name.startswith(prefix):
                    part = name[len(prefix):]
                    ax_name = "openid." + ax_ns + ".value." + part
                    break
            if not ax_name:
                return u""
            return self.get_argument(ax_name, u"")

        email = get_ax_arg("http://axschema.org/contact/email")
        name = get_ax_arg("http://axschema.org/namePerson")
        first_name = get_ax_arg("http://axschema.org/namePerson/first")
        last_name = get_ax_arg("http://axschema.org/namePerson/last")
        username = get_ax_arg("http://axschema.org/namePerson/friendly")
        locale = get_ax_arg("http://axschema.org/pref/language").lower()
        user = dict()
        name_parts = []
        if first_name:
            user["first_name"] = first_name
            name_parts.append(first_name)
        if last_name:
            user["last_name"] = last_name
            name_parts.append(last_name)
        if name:
            user["name"] = name
        elif name_parts:
            user["name"] = u" ".join(name_parts)
        elif email:
            user["name"] = email.split("@")[0]
        if email:
            user["email"] = email
        if locale:
            user["locale"] = locale
        if username:
            user["username"] = username
        claimed_id = self.get_argument("openid.claimed_id", None)
        if claimed_id:
            user["claimed_id"] = claimed_id
        callback(user)

 def submit(self, fn, *args, **kwargs):
     gen_log.warning("opps~~can i use the threading pool ones?")
     future = TracebackFuture()
     try:
         future.set_result(fn(*args, **kwargs))
     except Exception:
         future.set_exc_info(sys.exc_info())
     return future

 def log_exception(self, typ, value, tb):
     if isinstance(value, HTTPError):
         if value.log_message:
             info = "%d %s"%(value.status_code,value.log_message)
             gen_log.warning(info)
     else:
         app_log.error("Uncaught exception %s\n%r", self._request_summary(),
                       self.request, exc_info=(typ, value, tb))

def _add_watching(path):
    try:
        last_modified = os.stat(path).st_mtime
    except:
        gen_log.warning("failed to watch %s" % path)
        traceback.print_exc()
        return
    _watched_files[path] = last_modified

    def log_stack(self, signal, frame):
        """信号处理程序来记录当前线程的堆栈跟踪。

        For use with `set_blocking_signal_threshold`.
        """
        gen_log.warning('IOLoop blocked for %f seconds in\n%s',
                        self._blocking_signal_threshold,
                        ''.join(traceback.format_stack(frame)))

    def log_stack(self, signal, frame):
        """Signal handler to log the stack trace of the current thread.

        For use with `set_blocking_signal_threshold`.
        """
        gen_log.warning('IOLoop blocked for %f seconds in\n%s',
                        self._blocking_signal_threshold,
                        ''.join(traceback.format_stack(frame)))

def start(io_loop=None, check_time=500):
    io_loop = io_loop or ioloop.IOLoop.current()
    if io_loop in _io_loops:
        return
    _io_loops[io_loop] = True
    if len(_io_loops) > 1:
        gen_log.warning("watcher started more than once in the same process")
    scheduler = ioloop.PeriodicCallback(_check_files, check_time, io_loop=io_loop)
    scheduler.start()