Receiving a file sometimes gets timeout on async io loop in active mode

[voegtlel]

Hi there!

Operating System: Windows 10 / 1909
Python Version: 3.6.8
pyftpdlib Version: 1.5.6

lets start with the code that produces this issue:

import os
from io import BytesIO
from typing import List, Tuple

from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.filesystems import AbstractedFS
from pyftpdlib.handlers import DTPHandler, FTPHandler
from pyftpdlib.servers import FTPServer, ThreadedFTPServer


class MemoryFS(AbstractedFS):

    def getsize(self, path):
        return 0

    def stat(self, path):
        return os.stat_result(tuple([0, 0, 0, 0, 0, 0, 0, 0, 0, 0]))

    def chmod(self, path, mode):
        pass

    def listdir(self, path):
        return []

    def listdirinfo(self, path):
        return []

    def open(self, filename, mode):
        file = BytesIO()
        file.name = os.path.basename(filename)
        return file


class VirtualFTPServer(FTPServer):

    def __init__(
            self,
            logins: List[Tuple[str, str]],
            address_tuple: Tuple[str, int] = ("0.0.0.0", 21),
            receive_timeout: float = 10,
    ):
        dummy_authorizer = DummyAuthorizer()
        for username, password in logins:
            dummy_authorizer.add_user(username, password, ".", perm="elradfmw")

        class VirtualDTPHandler(DTPHandler):
            timeout = receive_timeout

        class VirtualFTPServerHandler(FTPHandler):

            dtp_handler = VirtualDTPHandler
            authorizer = dummy_authorizer
            abstracted_fs = MemoryFS

        super().__init__(address_tuple, VirtualFTPServerHandler)

    def collect_incoming_data(self, data):
        super(VirtualFTPServer, self).collect_incoming_data(data)

    def found_terminator(self):
        super(VirtualFTPServer, self).found_terminator()


if __name__ == '__main__':
    srv = VirtualFTPServer([
        ('Test', 'Test')
    ])
    srv.serve_forever()

Here is the code I use for shooting it with data:

import ftplib
import threading
import time
from io import BytesIO


def send_file(ftp_connection, data, target_name):
    ftp_connection.storbinary(f'STOR {target_name}', BytesIO(data))
    print(f"Sent {target_name}")


if __name__ == '__main__':
    ftp_connection_1 = ftplib.FTP('127.0.0.1', 'Test', 'Test')
    ftp_connection_2 = ftplib.FTP('127.0.0.1', 'Test', 'Test')
    ftp_connection_1.set_pasv(0)
    ftp_connection_2.set_pasv(0)

    # 5mb of data
    data = 5 * 1024 * 1024 * b'0'

    idx = 0
    while True:
        print(f"Sending {idx}")
        t1 = threading.Thread(target=send_file, args=[ftp_connection_1, data, f'c1_{idx}.jpg'])
        t2 = threading.Thread(target=send_file, args=[ftp_connection_2, data, f'c2_{idx}.jpg'])
        t1.start()
        t2.start()
        t1.join()
        t2.join()
        print(f"Sent {idx}")
        idx += 1
        time.sleep(0.1)

This works absolutely fine, as long as there is only one client connected. If you connect two clients at once, it sometimes gets stuck when transferring a file. That means, it'll log:

[I 2021-03-18 17:58:41] 127.0.0.1:50591-[Test] STOR c2_16.jpg completed=1 bytes=5242880 seconds=0.031
[I 2021-03-18 17:58:58] 127.0.0.1:50590-[Test] STOR c1_16.jpg completed=0 bytes=147456 seconds=17.141

(here after 31 completed files, one times out when receiving, although some bytes are received). I don't believe it's an error in the client, as it was first observed with two non-python clients. Also, I didn't observe this when using the ThreadedFTPServer instead of the FTPServer.

The error in general is reproducible and usually happens with less than 50 tries.

Thanks!