扭曲的大文件传输

Question

我这样编写客户端-服务器应用程序： 客户端（c#） 服务器（扭曲；ftp 代理和附加功能） ftp 服务器

Server 有两个类：我自己的类协议继承自 LineReceiever 协议和 FTPClient 继承自 twisted.protocols.ftp。

但是当客户端发送或获取大文件（10 Gb - 20 Gb）时，服务器会捕获 MemoryError。我的代码中没有使用任何缓冲区。当调用 transport.write(data) 数据附加到反应器编写器的内部缓冲区时会发生这种情况（如果我错了，请纠正我）。

我应该使用什么来避免这个问题？还是我应该改变解决问题的方法？

我发现对于大流，我应该使用 IConsumer 和 IProducer 接口。但最后它会调用 transfer.write 方法，效果是一样的。还是我错了？

UPD：

这是文件下载/上传的逻辑（从 ftp 通过 Twisted 服务器到 Windows 上的客户端）：

客户端向 Twisted 服务器发送一些标头，然后开始发送文件。 Twisted 服务器接收标头，然后（如果需要）调用setRawMode()，打开 ftp 连接并从/向客户端接收/发送字节，并在所有关闭连接之后。下面是上传文件的部分代码：

FTPManager 类

def _ftpCWDSuccees(self, protocol, fileName):
        self._ftpClientAsync.retrieveFile(fileName, FileReceiver(protocol))



class FileReceiver(Protocol):
    def __init__(self, proto):
        self.__proto = proto

    def dataReceived(self, data):
        self.__proto.transport.write(data)

    def connectionLost(self, why = connectionDone):
        self.__proto.connectionLost(why)

主代理服务器类：

class SSDMProtocol(LineReceiver)
...

在 SSDMProtocol 对象（调用 obSSDMProtocol）解析标头后，它调用打开 ftp 连接的方法（FTPClient 来自 twisted.protocols.ftp）并设置 FTPManager 字段的对象 _ftpClientAsync 并调用 _ftpCWDSuccees(self, protocol, fileName) 和 protocol = obSSDMProtocol 以及何时文件的字节recieved 调用 FileReceiver 对象的dataReceived(self, data)。

当self.__proto.transport.write(data) 被调用时，数据附加到内部缓冲区比发送回客户端更快，因此内存耗尽。当缓冲区达到一定大小时我可以停止读取并在缓冲区全部发送到客户端后恢复读取？或类似的东西？

Answer 1

如果您将 20 GB（千兆位？）字符​​串传递给 transport.write，您将需要至少 20 GB（千兆位？）的内存 - 由于需要额外的复制，可能更像是 40 或 60在 Python 中处理字符串时。

即使您从未将单个字符串传递给 transport.write，即 20 GB（千兆位？），如果您以超出网络处理能力的速度反复使用短字符串调用 transport.write，发送缓冲区最终会增长太大而无法放入内存，您会遇到MemoryError。

解决这两个问题的方法是生产者/消费者系统。使用IProducer 和IConsumer 的好处是你永远不会有一个20 GB（千兆位？）的字符串，你永远不会用太多较短的字符串填满发送缓冲区。网络将受到限制，因此字节的读取速度不会超过您的应用程序处理它们并忘记它们的速度。您的字符串最终将达到 16kB - 64kB 的大小，应该很容易放入内存中。

您只需要调整您对FileReceiver 的使用，以将传入连接注册为传出连接的生产者：

class FileReceiver(Protocol):
    def __init__(self, outgoing):
        self._outgoing = outgoing

    def connectionMade(self):
        self._outgoing.transport.registerProducer(self.transport, streaming=True)

    def dataReceived(self, data):
        self._outgoing.transport.write(data)

现在每当self._outgoing.transport 的发送缓冲区填满时，它就会告诉self.transport 暂停。一旦发送缓冲区清空，它将告诉self.transport 恢复。 self.transport 现在知道如何在 TCP 级别执行这些操作，这样进入服务器的数据也会减慢。