如何从 SFTPFile 重构 readChunk 以停止使用 inlineCallbacks？

Question

我正在尝试通过ISFTPFile 从文件中读取数据，并且我想避免在这种情况下使用@inlinceCallbacks？

或者也许有更好的方法来读/写ISFTPFile？

@defer.inlineCallbacks
def calculate_checksum(open_file):
    hasher = hashlib.sha256()

    offset = 0
    try:
        while True:
            d = yield open_file.readChunk(offset, chunk_size)
            offset += chunk_size
            hasher.update(d)

    except EOFError:
        pass

    target_checksum = hasher.hexdigest()
    defer.returnValue(target_checksum)


client_file = client.openFile(
    filename=target, flags=FXF_READ, attrs={})
checksum = yield client_file.addCallback(calculate_checksum)

Answer 1

您实际上希望将 sha256.update 映射到文件块的迭代器上：

hasher = hashlib.sha256()
chunks = read_those_chunks()
map(hasher.update, chunks)
return hasher.hexdigest()

请注意，来自原始 calculate_checksums 的显式迭代（使用 while 循环）现在隐藏在 map 内。基本上，map 已经取代了迭代。

障碍是你想避免read_those_chunks 将整个文件加载到内存中（大概）。因此，作为第一步，实现该部分：

def read_those_chunks(open_file, chunk_size):
    offset = 0
    while True:
        yield open_file.readChunk(offset, chunk_size)
        offset += chunk_size

有一个生成器会产生Deferreds，它会随着后续的块（或EOFError）触发。不幸的是，您不能将它与map 一起使用。所以现在实现一个可以处理这个的类似地图：

def async_map(function, iterable):
    try:
        d = next(iterable)
    except StopIteration:
        return

    d.addCallback(function)
    d.addCallback(lambda ignored: async_map(function, iterable))
    return d

由于async_map 将替换map 和map 替换原始实现中的迭代，async_map 仍然负责确保我们访问可迭代的每个块。但是，迭代（使用for 或while）不能很好地与Deferred 混合（混合它们通常是在您退出inlineCallbacks 时）。所以async_map 不会迭代。它递归 - 迭代的常见替代方案。每个递归调用都对迭代的下一个元素进行操作，直到没有更多元素为止（或者直到 Deferred 失败，在这种情况下由于 EOFError 而发生）。

递归比Deferred 的迭代效果更好，因为递归对函数和函数调用进行操作。 Deferred 可以处理函数和函数调用 - 将函数传递给 addCallback 和 Deferred 最终将调用该函数。迭代由函数的一小部分组成（有时称为“块”或“套件”），Deferred 无法处理这些。您不能将块传递给addCallback。

现在使用这两个创建一个Deferred，在计算摘要时触发：

def calculate_checksum(open_file, chunk_size):
    hasher = hashlib.sha256()
    chunks = read_those_chunks(open_file, chunk_size)
    d = async_map(hasher.update, chunks)
    d.addErrback(lambda err: err.trap(EOFError))
    d.addCallback(lambda ignored: hasher.hexdigest())
    return d

您可能还注意到async_map 与map 的不同之处在于它不会生成它所进行的函数调用的结果列表。或许更像reduce：

def async_reduce(function, iterable, lhs):
    try:
        d = next(iterable)
    except StopIteration:
        return lhs

    d.addCallback(lambda rhs: function(lhs, rhs))
    d.addCallback(lambda lhs: async_reduce(function, iterable, lhs))
    return d

当然，它仍然是递归而不是迭代。

计算十六进制摘要的归约函数如下：

def update_hash(hasher, s):
    hasher.update(s)
    return hasher

所以calculate_checksum 变成：

def calculate_checksum(open_file, chunk_size):
    chunks = read_those_chunks(open_file, chunk_size)
    d = async_reduce(update_hash, hashlib.sha256(), "")
    d.addErrback(lambda err: err.trap(EOFError))
    d.addCallback(lambda hasher: hasher.hexdigest())
    return d

没有hasher 闭包会更好。

当然，还有很多其他方法可以重写这个函数来避免inlineCallbacks。我选择的方式并没有消除生成器函数的使用，所以如果你想逃避它并没有真正帮助。如果是这样，也许您可​​以像我在此处所做的那样将问题分解为不同的部分，其中不涉及生成器。