asyncio.base_futures.InvalidStateError: invalid state

[xujiasci]

Describe the bug

I use aiohttp as client to emulate ddos attack, I use multi-process patten, and each process use an asyncio looper to perform
serveral http requests in parallel.

I got exception

Exception in callback BaseSelectorEventLoop._sock_connect_cb(<Future finished result=None>, <socket.socke...0.0.1', 8080)>, ('127.0.0.1', 8080))
handle: <Handle BaseSelectorEventLoop._sock_connect_cb(<Future finished result=None>, <socket.socke...0.0.1', 8080)>, ('127.0.0.1', 8080))>
Traceback (most recent call last):
  File "/usr/lib64/python3.6/asyncio/events.py", line 145, in _run
    self._callback(*self._args)
  File "/usr/lib64/python3.6/asyncio/selector_events.py", line 476, in _sock_connect_cb
    fut.set_result(None)
asyncio.base_futures.InvalidStateError: invalid state
Exception in callback BaseSelectorEventLoop._sock_connect_cb(<Future finished result=None>, <socket.socke...0.0.1', 8080)>, ('127.0.0.1', 8080))
handle: <Handle BaseSelectorEventLoop._sock_connect_cb(<Future finished result=None>, <socket.socke...0.0.1', 8080)>, ('127.0.0.1', 8080))>
Traceback (most recent call last):
  File "/usr/lib64/python3.6/asyncio/events.py", line 145, in _run
    self._callback(*self._args)
  File "/usr/lib64/python3.6/asyncio/selector_events.py", line 476, in _sock_connect_cb
    fut.set_result(None)
asyncio.base_futures.InvalidStateError: invalid state
Hello, Anonymous
Hello, Anonymous

To Reproduce

1. server code

from aiohttp import web


async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)

app = web.Application()
app.add_routes([web.get('/', handle),
                web.get('/{name}', handle)])

if __name__ == '__main__':
    web.run_app(app)

2. client code

import asyncio
from aiohttp import ClientSession
import argparse
import os
import signal
import functools
from concurrent.futures import ThreadPoolExecutor
import argparse

url=""
process_count = 0
concurr_count = 0
run_times = 1

running = False
child_process = []

async def requrl(url):
    async with ClientSession() as session:
        async with session.get(url) as response:
            res = await response.text()
            print(res) 
            pass

async def do_attack():
    global url, concurr_count, run_times, running
    looper = asyncio.get_event_loop()
    while running:
        task_list = []
        for i in range(concurr_count):
            task = looper.create_task(requrl(url))
            task_list.append(task)
        await asyncio.wait(task_list, timeout=None)
        run_times = run_times - 1
        if run_times == 0:
            running = False


def father_sighandler(signum, frame):
    global child_process
    for pid in child_process:
        os.system("kill -9 " + str(pid))
    child_process = []
    asyncio.get_event_loop().stop()
    exit()


def childProcessWork():
    global running
    running = True
    looper = asyncio.get_event_loop()
    looper.set_default_executor(ThreadPoolExecutor(10))
    looper.run_until_complete(do_attack())

if __name__ == '__main__':
    main_pid = os.getpid()
    looper = asyncio.get_event_loop()
    parser = argparse.ArgumentParser()
    parser.add_argument("url", type=str)
    parser.add_argument("process_count", type=int)
    parser.add_argument("concurr_count", type=int)
    parser.add_argument("run_times", type=int)
 
    args = parser.parse_args()
    url = args.url
    process_count = args.process_count
    concurr_count = args.concurr_count
    run_times = args.run_times

    for i in range(process_count):
        pid = os.fork()
        if (pid == 0):
            childProcessWork()
            exit()
        else:
            child_process.append(pid)
    looper = asyncio.get_event_loop()
    looper.add_signal_handler(getattr(signal, 'SIGINT'),
                              functools.partial(father_sighandler, 'SIGINT'),
                              None)
    looper.add_signal_handler(getattr(signal, 'SIGTERM'),
                              functools.partial(father_sighandler, 'SIGTERM'),
                              None)
    for id in child_process:
        os.waitid(os.P_ALL, id, os.WSTOPPED | os.WEXITED)

2. run command
   python3 attack.py http://127.0.0.1:8080 2 1 1

Expected behavior

No expetion

Logs/tracebacks

N/A

Python Version

$ python --version
Python 3.6.8

aiohttp Version

$ python -m pip show aiohttp
Name: aiohttp
Version: 3.8.1
Summary: Async http client/server framework (asyncio)
Home-page: https://github.com/aio-libs/aiohttp
Author: 
Author-email: 
License: Apache 2
Location: /home/jacen/.local/lib/python3.6/site-packages
Requires: aiosignal, async-timeout, asynctest, attrs, charset-normalizer, frozenlist, idna-ssl, multidict, typing-extensions, yarl
Required-by:

multidict Version

$ python -m pip show multidict
Name: multidict
Version: 5.2.0
Summary: multidict implementation
Home-page: https://github.com/aio-libs/multidict
Author: Andrew Svetlov
Author-email: andrew.svetlov@gmail.com
License: Apache 2
Location: /home/jacen/.local/lib/python3.6/site-packages
Requires: 
Required-by: aiohttp, yarl

yarl Version

$ python -m pip show yarl
Name: yarl
Version: 1.7.2
Summary: Yet another URL library
Home-page: https://github.com/aio-libs/yarl/
Author: Andrew Svetlov
Author-email: andrew.svetlov@gmail.com
License: Apache 2
Location: /home/jacen/.local/lib/python3.6/site-packages
Requires: idna, multidict, typing-extensions
Required-by: aiohttp

OS

CentOS Linux release 8.4.2105

Related component

Client

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[AtomsForPeace]

Doesn't throw this error in Python 3.10.0 but can confirm it in Python 3.6.9