function thunkify(fn) {
    assert('function' == typeof fn, 'function required');    

    return function () {
        var args = slice.call(arguments);    
        var ctx = this;
        return function (done) {   // done 是回调函数
            var called;
            args.push(function() {
                if (called) return;
                called = true;    
                done.apply(null, arguments); 
            }); // 将回调处理加入参数列表

            try {
                fn.apply(ctx, args);   // 函数处理 
            } catch (err) {
                done(err);   // 异常获取并用回调处理 
            }
        }
    }
};

var thunkify = require('thunkify');
var fs = require('fs');


var read = thunkify(fs.readFile);

read('package.json', 'utf8')(function(err, str){
});

var koa = require('koa');
var app = koa();

app.use(function *(){
    this.body = 'Hello World';    
});

app.listen(3000);

app.listen = function () {
    var server = http.createServer(this.callback()) ;  
    return server.listen.apply(server, arguments);
}

app.callback = function () {
    var mw = [respond].concat(this.middleware);   // 实际上 respond 就是最后一个需要执行的中间件， 负责返回数据的 
    var gen = compose(mw); 
    var fn = co(gen);  // co 和 compose 就是将一个个middle，按序执行的操作, 此时middle还没有运行
    var self = this;
    
    if (!this.listeners('error').length) this.on('error', this.onerror);
    
    return function (req, res) {
        res.statusCode = 404;    
        var ctx = self.createContext(req, res);
        finished(ctx, ctx.onerror);
        fn.call(ctx, ctx.onerror); // 这步运行了说明服务器接收到了请求， 然后一步步的运行middle 和 请求业务处理
    }
}

Lazy([1, 2, 4]) // in [1, 2, 4] -> out instanceof Lazy.ArrayLikeSequence 
Lazy({ a: 'b'})
Lazy("hello world")
Lazy()
Lazy(null)

var seq = Lazy(source) // 1st sequence
        .map(func) // 2nd MappedSequence
       .filter(pred) // 3rd FilteredSequence
        .reverse() // 4th ReversedSequence
seq.each(function(x) { console.log(x); })

function Sequence() {} # 创建Sequence构造函数

function dispense() {
    var obj = null,
    objWithTimeout = null,
    err = null,
    clientCb = null,
    waitingCount = waitingClients.size();
    log("...");
    if (waitingCount > 0) { //此时waitingCount = 1
        while(availableObjects.length > 0) {
            log("...");
            objWithTimeout = avaiableObjects[0];
            if (!factory.validate(objWithTimeout.obj)) {
                me.destroy(objWithTimeout.obj);
                continue;
            } // 这一步验证对象
            avaiableObjects.shift(); // LIFO
            clientCb = waitingClients.dequeue(); // dequeue
            return clientCb(err, objWithTimeout.obj); // callback
        }
        if (count < factory.max) {
                createResource();
        }
    }
}

me.release = function(obj) {
    // 确保对象已经被释放
    if (availableObjects.some(function(objWithTimeout) {
        return (objWithTimeout.obj === obj);
        })) {
        log("release called twice for the same resource")
        return;
    }
    var objWithTimeout = { obj: obj, timeout: (new Date().getTime() + idleTimeoutMillis) };
    availableObjects.push(objWithTimeout); // 此时 availableObjects.size()=1
    log("timeout:..");
    dispense();
    scheduleRemoveIdle(); // 计划移走idle items(空闲项目) 延迟1000ms 执行 removeIdle
}

 function removeIdle(){
    var toRemove = [],
        now = new Date().getTime(),
        i,
        al, tr,
        timeout;
    removeIdleScheduled = false;
    // go through the available(idle) items, check if they have timed out;
    for (i=0, al=availableObjects.length; i< al &&
        (refreshIdle) || (count - factory.min > toRemove.length));i+=1
        ){
        timeout = availableObjects[i].timeout;
        if (now >= timeout) {
           //client timed out, so destroy it 
           toRemove.push(availableObjects[i].obj);
            }
        }
        for (i=0, tr=toRemove.length; i<tr; i+=1) {
            me.destroy(toRemove[i]);
        }
        al = availableObjects.length;
        if (al > 0) {
           log("..");
           scheduleRemoveIdle();
        } else {
            log("removeIdle() all objects removed", 'verbose');
        }
     }

 me.destroy = function(obj) {
    count -= 1;
    availableOjbects = availableObjects.filter(function(objWithTimeout) {
       return (objWithTimeout.obj != obj)
        });//过滤掉return true 的对象
    factory.destroy(obj);
    ensureMinimum();
     }

"""
如果有 concurrent.futures 可用，那就用它; 否则用 _DummyFuture 
"""
if futures is None:
    Future = _DummyFuture
else:
    Future = futures.Future

"""
先检查有没有完成 ，没有完成 ,直接抛出异常!如果有异常就raise 异常，否则返回 result
"""
def result(self, timeout=None):
  self._check_done()
  if self._exception:
      raise self._exception
  return self._result

"""
给Future加入一个回调。当它完成将会 以Future为参数调用回调函数. 
"""
def add_done_callback(self, fn):
  if self.done:
      fn(self)
  else:
      self._callbacks.append(fn)

class DummyExecutor(object):
  def submit(self, fn, *args, **kwargs):
      future = TracebackFuture()
      try:
          future.set_result(fn(*args, **kwargs))
      except Exception:
          future.set_exc_info(sys.exc_info())
      return future
  def shutdown(self, wait=True):
      pass

def run_on_executor(fn):
    """
    异步的跑同步方法
    """
    @functools.wraps(fn):
    def wrapper(self, *args, **kwargs):
        callback = kwargs.pop("callback", None)
        future = self.executor.submit(fn, self, *args, **kwargs) #self.executor  理解为线程池
        if callback:
            self.io_loop.add_future(future, \
                lambda future: callback(future.result()))
        return future
    return wrapper

class ReturnFutureTest(AsyncTestCase):
    @return_future
    def sync_future(self, callback): # 同步future
        print (callback, '+++++++++++++++') # d -> 对应log 的d
        callback(42)

    @return_future
    def async_future(self, callback): #异步future
        print (callback, '+++++++++++++++') # d -> 对应 log 的 d
        self.io_loop.add_callback(callback, 42)

    def test_sync_future(self): #测试同步 
        future = self.sync_future()
        self.assertEqual(future.result(), 42)

    def test_async_future(self): #测试异步
        future = self.async_future()
        self.assertFalse(future.done())
        self.io_loop.add_future(future, self.stop) #add_future
        future2 = self.wait()
        self.assertIs(future, future2)
        self.assertEqual(future.result(), 42)

def return_future(f):
    replacer = ArgReplacer(f, 'callback') # 1 

    @fuctools.wraps(f)
    def wrapper(*args, **kwrags):
        print (args, kwrags, "argsssssssss") #a 对应下面的 log -> a
        future = TracebackFuture() # 2
        callback, args, kwargs = replacer.replace(
            lambda value=_NO_RESULT: future.set_result(value),
            args, kwargs) # 1  替代 f 函数的 `callback`,  
        print (callback, args, kwargs, 'callback') #b 对应下面的 log -> b

        def handle_error(typ, value, tb):
            future.set_exc_info((typ, value, tb))
            return True

        exc_info = None
        with ExceptionStackContext(handle_error): # 4
            try:
                result = f(*args, **kwargs)
                print (result, 'result-------------------') #c 对应下面的 log -> c
                if result is not None:
                        raise ReturnValueIgoredError(
                            "@return_future should not be used with functions"
                            "that return values")
            except:
                exc_info = sys.exc_info()
                raise
        if exc_info is not None:
            raise_exc_info(exc_info)

        if callback is not None:
            def run_callback(future):
                result = future.result()
                print (future, "+__+_+_+_+_") #e -> 对应下面 ＬＯＧ -> e
                if result is _NO_RESULT:
                    callback()
                else:
                    callback(future.result())
            future.add_done_callback(wrap(run_callback)) # 6
        return future
    return wrapper

@classmethod
def configurable_base(cls):
  return IOLoop

@classmethod
def configurable_default(cls):
  if hasattr(select, "epoll"):
      from tornado.platform.epoll import EPollIOLoop
      return EPollIOLoop
  if hasattr(select, "kqueue"):
      #python 2.6+ on BSD or Mac
      from tornado.platform.kqueue import KQueueIOLoop
      return KQueueIOLoop
  from tornado.platform.select import SelectIOLoop
  return SelectIOLoop

def __new__(cls, **kwargs):
  base = cls.configurable_base()
  args = {}
  if cls is base:
      impl = cls.configured_class()
      if base.__impl_kwargs:
          args.update(base.__impl_kwargs)
  else:
      impl = cls
  args.update(kwargs)
  instance = super(Configurable, cls).__new__(impl)
  instance.initialize(**args)
  return instance

base = cls.configurable_base()
if cls.__impl_class is None:
  base._impl_class = cls.configurable_default()
return base.__impl_class

 def add_handler(self, fd, handler, events):
  self._handlers[fd] = statck_context.wrap(handler)
  self._impl.register(fd, events | self.ERROR)

def start(self):
  if not logging.getLogger().handlers:
      logging.basicConfig
  if self._stopped:
      self._stopped = False
      return
  old_current = getattr(IOLoop._current, "instance", None)
  IOLoop._current.instance = self
  self._thread_ident = thread.get_ident() #Return the ‘thread identifier’ of the current thread. This is a nonzero integer
  self._running = True

  old_wakeup_fd = None
  if hasattr(singal, 'set_wakeup_fd') and os.name == 'posix':
      try:
          old_wakeup_fd = signal.set_wakeup_fd(self._waker.write_fileno())
          if old_wakeup_fd != -1:
              signal.set_walkeup_fd(old_wakeup_fd)
              old_walkup_fd = None
      except ValueError:
          pass

while True:
  poll_timeout = _POLL_TIMEOUT
  #Prevent IO event starvation by delaying new callbacks 
  # to the next iteration of the event loop.
  with self._callback_lock:
      callbacks = self._callbacks
      self._callbacks = []
  for callback in callbacks:
      self._run_callback(callback)
  # Closures may be holding on to a lot of memory, so allow
  # them to be freed before we go into our poll wait.
  callbacks = callback = None

  if self._timeouts:
      now = self.time()
      while self_timeouts:
          if self._timesouts[0].callback is None:
              # the timeout was cancelled
              heapq.heappop(self._timeouts)
              self._cancellations -= 1
          elif self._timeouts[0].deadline <= now:
              timeout = heapq.heappop(self._timeouts)
              self._run_callback(timeout.callback)
              del timeout
          else:
              seconds = self._timeouts[0].deadline - now
              poll_timeout = min(seconds, poll_timeout)
              break

TCPServer(ssl_options={
  "certfile": os.path.join(data_dir, "mydomain.crt"),
  "keyfile": os.path.join(data_dir, "mydomain.key"),
})

server = TCPServer()
server.listen(8888)
IOLoop.instance().start()

server = TCPServer()
server.bind(8888)
server.start(0) #Forks multiple sub-process
IOLoop.instance().start()

sockets = bind_sockets(8888)
tornaado.process.fork_processes(0)
server = TCPServer()
server.add_sockets(sockets)
IOLoop.instance().start()

def __init__(self, io_loop=None, ssl_options=None, max_buffer_size=None):
  self.io_loop = io_loop
  self.ssl_options = ssl_options
  self.sockets = {}
  self._pending_sockets = []
  self._started = False
  self.max_buffer_size = max_buffer_size
  #下面是检验 SSL 文件，此处略

sockets = bind_sockets(port, address=address)
self.add_sockets(sockets)

if self.io_loop is None:
  self.io_loop = IOLoop.current()

for sock in sockets:
  self._sockets[sock.fileno()] = sock
  add_accept_handler(sock, self._handle_connection, io_loop=self.io.loop) 

#如果self.ssl_options不为空，处理ssl代码略
try:
  if self.ssl_options in not None:
      stream = SSLTOSTREAM(connection, io_loop=self.io_loop, max_buffer_size=self.max_buffer_size)
  else:
      stream = IOSTream(connection, io_loop=self.io_loop, max_buffer_size=self.max_buffer_size)
  self.handle_stream(stream, address)
except Excetpion:
  app_log.error('Error in connection callback', exc_info=True)

sockets = []
if address == "":
  address = None
if not socket.has_ipv6 and family == socket.AF_UNSPEC:
  family = socket.AF_INET
if flags is None:
  flags = socket.AI_PASSIVE
  #见 [getaddrinfo](https://github.com/zs1621/pythostudy/wiki/socket)
  #`getaddrinfo`会返回服务器的所有网卡信息， 每个网卡都要监听客户端的请求并返回创建的sockets
for res in set(socket.getaddrinfo(address, port, family, socket.SOCK.STREAM, 0, flags)):
  af, socktype, proto, canoname, sockaddr = res
  try:
      #创建套节字
      sock = socket.socket(af, socktype, proto)
  except socket.error as e:
      if e.args[0] == errno.EAFNOSUPPORT:
          continue
      raise
  set_close_exec(sock.fileno())
  if os.name != 'nt':
      #TCP连接中，recv等函数默认为阻塞模式(block)，
      #即直到有数据到来之前函数不会返回，
      #而我们有时则需要一种超时机制使其在一定时间后返回而不管是否有数据到来，
      #这里我们就会用到setsockopt()函数 见 [setsockopt](http://blog.chinaunix.net/uid-25749806-id-348637.html)
      sock.setsockopt(socket.SOL_SOCKET, socket.SOREUSERADDR, 1)
  #在linux ipv6 sockets 也接受 ipv4，这样的话不能同时绑定 ipv4 和 ipv6.为了方便，在ipv6 sockets中总是禁用ipv4 
  if af == socket.AF_INET6:
      if hasattr(socket, "IPPROTO_IPV6"):
          sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY)
  sock.setblocking(0)
  sock.bind(sockaddr)
  #默认设定等待被处理的连接最大个数
  sock.listen(backlog)
  sockets.append(sock)
return sockets

if io_loop is None:
  io_loop = IOLoop.current()
def accept_handler(fd, events):
  while True:
      try:
          connection, address = sock.accept()
      except socket.error as e:
          #EWOULDBLOCK 和 EAGAIN 表明我们
          #已经接受了每个可以接受的连接
          #具体见[EWOULDBLOCK](http://stackoverflow.com/questions/3647539/socket-error-errno-ewouldblock)
          if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
              return
          #ECONNABORTED 表明有个连接还在接受队列时被关闭了
          if e.args[0] == errno.ECONNABORTED:
              continue
          raise
      callback(connection, address)
io_loop.add_handler(sock.fileno(), accept_handler, IOLoop.READ)

exports = module.exports = send;

function send(req, path, options) {
  return new SendStream(req, path, options);
}

function SendStream(req, path, options) {
  ....
}

SendStream.prototype.__proto__ = Stream.prototype;

var self = this
, args = arguments
, path = this.path
, root = this._root

this.res = res;
//判断uri的有效性
path = utils.decode(path);
if (-1 == path) return this.error(400);

//如果`path`里包含`\0` ，表明`path`为空
if (~path.indexOf('\0')) return this.error(400);

//连接 this._root 和 path
if (root) path = normalize(join(this._root, path));

//如果异常路径禁止访问
if (this.isMalicious()) return this.error(403);

//此时`path`已经是 `this._root`和 `path`的结合, 所以此种情况不会出现,如果有root
if (root && 0 != path.indexOf(root)) return this.error(403)

//如果_hidden为`false`,那么不支持隐藏文件，此时文件如果以`.`开头那么就不支持 
if (!this._hidden && this.hasLeadingDot()) return this.error(404);

//index 文件支持
if (this._index && this.hasTrailingSlash()) path += this._index

//最后看path 的信息
fs.stat(path, function(err, stat){
  if (err) return self.onStatError(err); //上面已介绍
  if (stat.isDirectory) return self.redirect(self.path); //见上
  self.emit('file', path, stat); //见下
  self.send(path, stat);//见下
});

var options = this.options;
var len = stat.size;
var res = this.res;
var req = this.req;
var ranges = req.headers.range; //截取部分文件之用
var offset = options.start || 0;

//赋值header
this.setHeader(stat);

//赋值 content-type
this.type(path);

//条件 GET 支持, isFresh() 见 [fresh](https://github.com/visionmedia/node-fresh/blob/master/index.js)
if (this.isConditionGET()
&& this.isCachable()
&& this.isFresh()) {
  return this.notModified();
}

//
len = Math.max(0, len - offset);
if (options.end != undefined) {
  var bytes = options.end - offset + 1;
  if (len > bytes) len = bytes;
}

// Range 支持
if (ranges) {
  ... 
}

// content-length
res.setHeader('Content-Length', len);

// HEAD support
if ('HEAD' == req.method) return res.end();

this.stream(path, options); //见下

var self = this;
var res = this.res;
var req = this.req;

//pipe 把流数据加入 `res`管道
var stream = fs.createReadStream(path, options);
this.emit('stream', stream);
stream.pipe(res);

//socket 关闭， 

req.on('close', stream.destroy.bind(stream));

//error处理
steam.on('error', function(err){
  //不回复
  if (res._header) {
      console.error(err.stack);
      req.destroy();
      return;
  }
  erro.status = 500;
  self.emit('error', err);
});

//end
stream.on('end', function(){
  self.emit('end');   
});

application = web.Application([
(r"/", MainPageHandler),  
])
http_server = httpserver.HTTPServer(application)
http_server.listen(8080)
ioloop.IOLoop.instance().start()

def __init__(self, request_callback, no_keep_alive=False, io_loop=None,
                 xheaders=False, ssl_options=None, protocol=None, **kwargs):
        self.request_callback = request_callback
        self.no_keep_alive = no_keep_alive
        self.xheaders = xheaders
        self.protocol = protocol
        TCPServer.__init__(self, io_loop=io_loop, ssl_options=ssl_options,
                           **kwargs) #HTTPServer 继承自 TCPServer, 初始化TCPServer

HTTPServer(applicaton, ssl_options={
   "certfile": os.path.join(data_dir, "mydomain.crt"),
   "keyfile": os.path.join(data_dir, "mydomain.key"),
  })