5

这可能是一罐蠕虫,我会尽力描述这个问题。我们有一个长期运行的数据处理工作。我们的行动数据库被添加到每晚并处理未完成的行动。处理夜间操作大约需要 15 分钟。在 Vapor 2 中,我们使用了大量原始查询来创建 PostgreSQL 游标并循环遍历它,直到它为空。

目前,我们通过命令行参数运行处理。将来我们希望它作为主服务器的一部分运行,以便在执行处理时可以检查进度。

func run(using context: CommandContext) throws -> Future<Void> {
    let table = "\"RecRegAction\""
    let cursorName = "\"action_cursor\""
    let chunkSize = 10_000


    return context.container.withNewConnection(to: .psql) { connection in
        return PostgreSQLDatabase.transactionExecute({ connection -> Future<Int> in

            return connection.simpleQuery("DECLARE \(cursorName) CURSOR FOR SELECT * FROM \(table)").map { result in
                var totalResults = 0
                var finished : Bool = false

                while !finished {
                    let results = try connection.raw("FETCH \(chunkSize) FROM \(cursorName)").all(decoding: RecRegAction.self).wait()
                    if results.count > 0 {
                        totalResults += results.count
                        print(totalResults)
                        // Obviously we do our processing here
                    }
                    else {
                        finished = true
                    }
                }

                return totalResults
            }
        }, on: connection)
    }.transform(to: ())
}

现在这不起作用,因为我正在调用wait()并且我收到错误“前提条件失败:在 EventLoop 上时不得调用 wait()”,这很公平。我面临的一个问题是,我不知道你是如何离开主事件循环以在后台线程上运行这样的事情的。我知道 BlockingIOThreadPool,但这似乎仍然在同一个 EventLoop 上运行并且仍然导致错误。虽然我能够提出越来越复杂的方法来实现这一目标,但我希望我错过了一个优雅的解决方案,也许对 SwiftNIO 和 Fluent 有更好了解的人可以提供帮助。

编辑:要清楚,这样做的目标显然不是汇总数据库中的操作数量。目标是使用光标同步处理每个动作。当我读入结果时,我会检测到动作的变化,然后将它们的批次扔到处理线程中。当所有线程都忙时,我不会再次从光标开始读取,直到它们完成。

这些动作有很多,一次运行多达 4500 万次。聚合承诺和递归似乎不是一个好主意,当我尝试它时,只是为了它,服务器挂起。

这是一个处理密集型任务,可以在单个线程上运行数天,所以我不关心创建新线程。问题是我无法弄清楚如何在命令中使用 wait() 函数,因为我需要一个容器来创建数据库连接,而我唯一可以访问的是context.container调用 wait() 会导致上述错误。

TIA

4

2 回答 2

10

好的,如您所知,问题出在以下几行:

while ... {
    ...
    try connection.raw("...").all(decoding: RecRegAction.self).wait()
    ...
}

您想等待许多结果,因此您使用while循环和.wait()所有中间结果。本质上,这是在事件循环上将异步代码转换为同步代码。这很可能会导致死锁,并且肯定会停止其他连接,这就是为什么 SwiftNIO 会尝试检测并给你那个错误的原因。我不会详细说明为什么它会停止其他连接,或者为什么这可能会导致这个答案出现死锁。

让我们看看我们有哪些选项可以解决这个问题:

  1. 正如您所说,我们可以将它.wait()放在另一个不是事件循环线程之一的线程上。为此,任何EventLoop线程都可以:要么 aDispatchQueue要么您可以使用BlockingIOThreadPool( 它不在 a 上运行EventLoop)
  2. 我们可以将您的代码重写为异步的

两种解决方案都可以使用,但 (1) 确实不可取,因为您会烧掉整个(内核)线程来等待结果。并且两者Dispatch都有BlockingIOThreadPool他们愿意产生的有限数量的线程,所以如果你经常这样做,你可能会用完线程,所以需要更长的时间。

因此,让我们看看如何在累积中间结果的同时多次调用异步函数。然后,如果我们累积了所有中间结果,则继续所有结果。

为了使事情更容易,让我们看一个与您的非常相似的函数。我们假设这个函数就像在你的代码中一样提供

/// delivers partial results (integers) and `nil` if no further elements are available
func deliverPartialResult() -> EventLoopFuture<Int?> {
    ...
}

我们现在想要的是一个新功能

func deliverFullResult() -> EventLoopFuture<[Int]>

请注意deliverPartialResult每次返回一个整数并deliverFullResult传递一个整数数组(即所有整数)。好的,那么我们如何在deliverFullResult 调用的情况下编写代码deliverPartialResult().wait()

那这个呢:

func accumulateResults(eventLoop: EventLoop,
                       partialResultsSoFar: [Int],
                       getPartial: @escaping () -> EventLoopFuture<Int?>) -> EventLoopFuture<[Int]> {
    // let's run getPartial once
    return getPartial().then { partialResult in
        // we got a partial result, let's check what it is
        if let partialResult = partialResult {
            // another intermediate results, let's accumulate and call getPartial again
            return accumulateResults(eventLoop: eventLoop,
                                     partialResultsSoFar: partialResultsSoFar + [partialResult],
                                     getPartial: getPartial)
        } else {
            // we've got all the partial results, yay, let's fulfill the overall future
            return eventLoop.newSucceededFuture(result: partialResultsSoFar)
        }
    }
}

鉴于accumulateResults,实施deliverFullResult不再太难:

func deliverFullResult() -> EventLoopFuture<[Int]> {
    return accumulateResults(eventLoop: myCurrentEventLoop,
                             partialResultsSoFar: [],
                             getPartial: deliverPartialResult)
}

但让我们更深入地了解accumulateResults它的作用:

  1. 它调用getPartial一次,然后当它回调它
  2. 检查我们是否有
    • 一个部分结果,在这种情况下,我们将它与另一个一起记住partialResultsSoFar并返回(1)
    • nil这意味着partialResultsSoFar我们所得到的一切,我们用迄今为止收集的一切返回一个新的成功的未来

这已经是真的了。我们在这里所做的是将同步循环变成异步递归。

好的,我们查看了很多代码,但现在这与您的函数有什么关系?

信不信由你,但这实际上应该有效(未经测试):

accumulateResults(eventLoop: el, partialResultsSoFar: []) {
    connection.raw("FETCH \(chunkSize) FROM \(cursorName)")
              .all(decoding: RecRegAction.self)
              .map { results -> Int? in
        if results.count > 0 {
            return results.count
        } else {
            return nil
        }
   }
}.map { allResults in
    return allResults.reduce(0, +)
}

所有这一切的结果将是一个EventLoopFuture<Int>带有所有中间值的总和result.count

当然,我们首先将您所有的计数收集到一个数组中,然后allResults.reduce(0, +)在最后将其相加(),这有点浪费,但也不是世界末日。我这样保留它是因为它accumulateResults可以在您想要在数组中累积部分结果的其他情况下使用。

现在最后一件事,一个真正的accumulateResults函数可能对元素类型是通用的,我们也可以消除partialResultsSoFar外部函数的参数。那这个呢?

func accumulateResults<T>(eventLoop: EventLoop,
                          getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
    // this is an inner function just to hide it from the outside which carries the accumulator
    func accumulateResults<T>(eventLoop: EventLoop,
                              partialResultsSoFar: [T] /* our accumulator */,
                              getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
        // let's run getPartial once
        return getPartial().then { partialResult in
            // we got a partial result, let's check what it is
            if let partialResult = partialResult {
                // another intermediate results, let's accumulate and call getPartial again
                return accumulateResults(eventLoop: eventLoop,
                                         partialResultsSoFar: partialResultsSoFar + [partialResult],
                                         getPartial: getPartial)
            } else {
                // we've got all the partial results, yay, let's fulfill the overall future
                return eventLoop.newSucceededFuture(result: partialResultsSoFar)
            }
        }
    }
    return accumulateResults(eventLoop: eventLoop, partialResultsSoFar: [], getPartial: getPartial)
}

编辑:编辑后,您的问题表明您实际上并不想累积中间结果。所以我的猜测是,您希望在收到每个中间结果后进行一些处理。如果这就是你想要做的,也许试试这个:

func processPartialResults<T, V>(eventLoop: EventLoop,
                                 process: @escaping (T) -> EventLoopFuture<V>,
                                 getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<V?> {
    func processPartialResults<T, V>(eventLoop: EventLoop,
                                     soFar: V?,
                                     process: @escaping (T) -> EventLoopFuture<V>,
                                     getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<V?> {
        // let's run getPartial once
        return getPartial().then { partialResult in
            // we got a partial result, let's check what it is
            if let partialResult = partialResult {
                // another intermediate results, let's call the process function and move on
                return process(partialResult).then { v in
                    return processPartialResults(eventLoop: eventLoop, soFar: v, process: process, getPartial: getPartial)
                }
            } else {
                // we've got all the partial results, yay, let's fulfill the overall future
                return eventLoop.newSucceededFuture(result: soFar)
            }
        }
    }
    return processPartialResults(eventLoop: eventLoop, soFar: nil, process: process, getPartial: getPartial)
}

这将(和以前一样)运行getPartial直到它返回nil,但它不会累积所有getPartial的结果,而是调用process它来获取部分结果并可以做一些进一步的处理。下一次getPartial调用将在EventLoopFuture process返回完成时发生。

这更接近你想要的吗?

注意:我在这里使用了 SwiftNIO 的EventLoopFuture类型,在 Vapor 中你只需要使用Future它,但其余代码应该是相同的。

于 2018-08-01T08:42:45.797 回答
0

这是通用解决方案,为 NIO 2.16/Vapor 4 重写,并作为 EventLoop 的扩展

extension EventLoop {

    func accumulateResults<T>(getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
        // this is an inner function just to hide it from the outside which carries the accumulator
        func accumulateResults<T>(partialResultsSoFar: [T] /* our accumulator */,
                                  getPartial: @escaping () -> EventLoopFuture<T?>) -> EventLoopFuture<[T]> {
            // let's run getPartial once
            return getPartial().flatMap { partialResult in
                // we got a partial result, let's check what it is
                if let partialResult = partialResult {
                    // another intermediate results, let's accumulate and call getPartial again
                    return accumulateResults(partialResultsSoFar: partialResultsSoFar + [partialResult],
                                             getPartial: getPartial)
                } else {
                    // we've got all the partial results, yay, let's fulfill the overall future
                    return self.makeSucceededFuture(partialResultsSoFar)
                }
            }
        }
        return accumulateResults(partialResultsSoFar: [], getPartial: getPartial)
    }
}
于 2020-05-13T15:46:04.023 回答