【问题标题】:AIOHTTP - Application.make_handler(...) is deprecated - Adding MultiprocessingAIOHTTP - Application.make_handler(...) 已弃用 - 添加多处理
【发布时间】:2020-02-03 22:56:33
【问题描述】:

我经历了“我可以从 Python 网络服务器中挤出多少性能?”的旅程。这导致我使用 AIOHTTP 和 uvloop。不过,我可以看到 AIOHTTP 并没有充分利用我的 CPU。我开始使用 AIOHTTP 的多处理。我了解到有一个 Linux 内核特性允许多个进程共享同一个 TCP 端口。这导致我开发了以下代码(效果很好):

import asyncio
import os
import socket
import time
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count

CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)


def mk_socket(host="127.0.0.1", port=8000, reuseport=False):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if reuseport:
        SO_REUSEPORT = 15
        sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
    sock.bind((host, port))
    return sock


async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    pid = os.getpid()
    text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
        time.time(), name, pid)
    #time.sleep(5)  # intentionally blocking sleep to simulate CPU load
    return web.Response(text=text)


def start_server():
    host = "127.0.0.1"
    port=8000
    reuseport = True
    app = web.Application()
    sock = mk_socket(host, port, reuseport=reuseport)
    app.add_routes([web.get('/', handle),
                    web.get('/{name}', handle)])
    loop = asyncio.get_event_loop()
    coro = loop.create_server(
            protocol_factory=app.make_handler(),
            sock=sock,
        )
    srv = loop.run_until_complete(coro)
    loop.run_forever()


if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        for i in range(0, CPU_COUNT):
            executor.submit(start_server)

在应用此代码之前对我的网站进行 wrk 基准测试:

Running 30s test @ http://127.0.0.1:8000/
  12 threads and 400 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    54.33ms    6.54ms 273.24ms   89.95%
    Req/Sec   608.68    115.97     2.27k    83.63%
  218325 requests in 30.10s, 41.23MB read
  Non-2xx or 3xx responses: 218325
Requests/sec:   7254.17
Transfer/sec:      1.37MB

wrk 基准测试后:

Running 30s test @ http://127.0.0.1:8000/
  12 threads and 400 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    15.96ms    7.27ms  97.29ms   84.78%
    Req/Sec     2.11k   208.30     4.45k    75.50%
  759290 requests in 30.08s, 153.51MB read
Requests/sec:  25242.39
Transfer/sec:      5.10MB

哇!但是有一个问题:

DeprecationWarning: Application.make_handler(...) is deprecated, use AppRunner API instead
  protocol_factory=app.make_handler()

所以我尝试了这个:

import asyncio
import os
import socket
import time
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count

CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)

def mk_socket(host="127.0.0.1", port=8000, reuseport=False):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if reuseport:
        SO_REUSEPORT = 15
        sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
    sock.bind((host, port))
    return sock

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    pid = os.getpid()
    text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
        time.time(), name, pid)
    #time.sleep(5)  # intentionally blocking sleep to simulate CPU load
    return web.Response(text=text)

async def start_server():
    host = "127.0.0.1"
    port=8000
    reuseport = True
    app = web.Application()
    sock = mk_socket(host, port, reuseport=reuseport)
    app.add_routes([web.get('/', handle),
                    web.get('/{name}', handle)])

    coro = loop.create_server(
            protocol_factory=app.make_handler(),
            sock=sock,
        )
    runner = web.AppRunner(app)
    await runner.setup()
    srv = web.TCPSite(runner, 'localhost', 8000)
    await srv.start()
    print('Server started at http://127.0.0.1:8000')
    return coro, app, runner

async def finalize(srv, app, runner):
    sock = srv.sockets[0]
    app.loop.remove_reader(sock.fileno())
    sock.close()

    #await handler.finish_connections(1.0)
    await runner.cleanup()
    srv.close()
    await srv.wait_closed()
    await app.finish()

def init():
    loop = asyncio.get_event_loop()
    srv, app, runner = loop.run_until_complete(init)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        loop.run_until_complete((finalize(srv, app, runner)))


if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        for i in range(0, CPU_COUNT):
            executor.submit(init)

这显然是不完整的,因为 coro 没有被使用。我不确定在哪里将套接字与 AppRunner 集成。答案应显示修改为使用 App Runner 的原始示例。

【问题讨论】:

标签: python python-3.x multiprocessing python-asyncio aiohttp


【解决方案1】:

由于这是我第一次使用协程和 aiohttp,我可能错了,但它似乎可以与 SockSite 一起使用:

#!/usr/bin/env python

import asyncio
import os
import socket
import time
import traceback
from aiohttp import web
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import cpu_count

CPU_COUNT = cpu_count()
print("CPU Count:", CPU_COUNT)

def mk_socket(host="127.0.0.1", port=9090, reuseport=False):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if reuseport:
        SO_REUSEPORT = 15
        sock.setsockopt(socket.SOL_SOCKET, SO_REUSEPORT, 1)
    sock.bind((host, port))
    return sock

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    pid = os.getpid()
    text = "{:.2f}: Hello {}! Process {} is treating you\n".format(
        time.time(), name, pid)
    #time.sleep(5)  # intentionally blocking sleep to simulate CPU load
    return web.Response(text=text)

async def start_server():
    try:
        host = "127.0.0.1"
        port=9090
        reuseport = True
        app = web.Application()
        app.add_routes([web.get('/', handle),
                        web.get('/{name}', handle)])
        runner = web.AppRunner(app)
        await runner.setup()
        sock = mk_socket(host, port, reuseport=reuseport)
        srv = web.SockSite(runner, sock)
        await srv.start()
        print('Server started at http://127.0.0.1:9090')
        return srv, app, runner
    except Exception:
        traceback.print_exc()
        raise

async def finalize(srv, app, runner):
    sock = srv.sockets[0]
    app.loop.remove_reader(sock.fileno())
    sock.close()

    #await handler.finish_connections(1.0)
    await runner.cleanup()
    srv.close()
    await srv.wait_closed()
    await app.finish()

def init():
    loop = asyncio.get_event_loop()
    srv, app, runner = loop.run_until_complete(start_server())
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        loop.run_until_complete((finalize(srv, app, runner)))


if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        for i in range(0, CPU_COUNT):
            executor.submit(init)

最终:

>curl http://127.0.0.1:9090
1580741746.47: Hello Anonymous! Process 54623 is treating you
>curl http://127.0.0.1:9090
1580741747.05: Hello Anonymous! Process 54620 is treating you
>curl http://127.0.0.1:9090
1580741747.77: Hello Anonymous! Process 54619 is treating you
>curl http://127.0.0.1:9090
1580741748.36: Hello Anonymous! Process 54621 is treating you

我还在 finalize 例程中添加了一个日志,它似乎正确触发了。

编辑:而且,出于好奇,我在较旧的内核上试了一下,并确认在启用该选项时它不起作用(它适用于 False)。

【讨论】:

  • 谢谢。这似乎大部分工作并且可以接受。但是,在将它应用到我当前的代码并运行 wrk 压力测试 (./wrk -t12 -c400 -d30s http://127.0.0.1:9090) 之后,测试似乎运行良好,但是一旦测试结束(可能是最终连接断开时),我看到我的服务器输出一堆行:module 'asyncio.streams' has no attribute 'IncompleteReadError'。这在单线程版本中不会发生。任何想法?如果没有,仍然会接受答案。只是想弄清楚。
  • 我认为这可能与我在此设置中使用 aiomysql 有关。
  • 发现问题。如果其他人遇到它,那是因为 aiomysql 还没有为 python 3.8 做好准备。您只能大规模地看到它(许多连接)。必须通过 pip install git+https://github.com/dongweiming/aiomysql.git@python38#egg=aiomysql 手动安装一个 fork,直到他们将其合并到主仓库。
猜你喜欢
  • 2011-12-24
  • 2022-11-20
  • 1970-01-01
  • 2012-03-14
  • 2020-02-26
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2013-09-04
相关资源
最近更新 更多