如何使用龙卷风进行异步ping？

Question

一个非常简单的tornado应用，当服务器收到一个HTTP get请求时，它ping -c 2 www.google.com，然后返回结果。我想使用龙卷风。这是一篇文章的代码。

class AsyncTaskHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def get(self, *args, **kwargs):
        response = yield tornado.gen.Task(self.ping, ' www.google.com')
        print 'response', response
        self.finish('hello')

    @tornado.gen.coroutine
    def ping(self, url):
        os.system("ping -c 2 {}".format(url))
        return 'after'

作者说ab测试结果很棒。 ab -c 5 -n 5 http://127.0.0.1:5000/async

Document Path:          /async
Document Length:        5 bytes

Concurrency Level:      5
Time taken for tests:   0.009 seconds
Complete requests:      5
Failed requests:        0
Total transferred:      985 bytes
HTML transferred:       25 bytes
Requests per second:    556.92 [#/sec] (mean)
Time per request:       8.978 [ms] (mean)
Time per request:       1.796 [ms] (mean, across all concurrent requests)
Transfer rate:          107.14 [Kbytes/sec] received

但实际上我使用的是相同的代码，在我的测试中，每秒请求数为 0.77！ 我寻找原因。我找到了这个版本：

class IndexHandler(tornado.web.RequestHandler):
    executor = ThreadPoolExecutor(10)

    @tornado.gen.coroutine
    def get(self):
        print "begin"
        response = yield self.pin()
        print response
        self.finish()

    @run_on_executor
    def pin(self):
        return os.system("ping -c 2 www.google.com")

测试结果，每秒请求数为 0.85。 我想使用 tornado 协程使 1000 个或更多 ping 命令无阻塞。我该如何编码？非常感谢！

Answer 1

使用tornado.process.Subprocess，而不是ThreadPoolExecutor 和os.system，效率更高：

class IndexHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def get(self):
        print "begin"
        response = yield self.ping()
        print response
        self.finish()

    @tornado.gen.coroutine
    def ping(self):
        proc = tornado.process.Subprocess("ping -c 2 www.google.com")
        return yield proc.wait_for_exit()

但是，由于ping 仍在启动一个单独的进程，在这种情况下它并不比线程池好多少，并且线程池对于限制并发 ping 进程的数量很有用。

Answer 2

代码的第一个版本一次只运行一个“ping”。 IOLoop 被阻塞，直到os.system 调用返回。

第二个版本使用run_on_executor 将os.system 调用推迟到线程池中的线程，使其成为非阻塞并允许并发调用。

如果（出于某种奇怪的原因？）您想要运行 1000 个并发 ping，则需要扩展 ThreadPoolExecutor 的默认线程数：

thread_pool = ThreadPoolExecutor(1000)

def ping_blocking():
    os.system("ping -c 2 www.google.com")

@gen.coroutine
def ping_many():
    futures = [thread_pool.submit(ping_blocking)
               for _ in range(1000)]
    yield futures

有关更多信息，请参阅the Tornado documentation on calling blocking functions。有关并行执行的文档也在该页面上。