Clean up use of EventLoopFutureQueue (#453)

* Replace all uses of EventLoopFutureQueue with the more straightforward sequencedFlatMapEach(). Should noticeably reduce contention and memory allocation traffic around large numbers of migrations in particular. Requires latest AsyncKit for updated utility API.

* Update CI with 5.4 and 5.5 images

.github/workflows/test.yml

@@ -4,7 +4,6 @@ on:
   push: 
     branches: 
       - main
-      - master
 env:
   LOG_LEVEL: info
 
@@ -43,7 +42,7 @@ jobs:
         image: mongo:latest
       mongo-b:
         image: mongo:latest
-    container: swift:5.3-focal
+    container: swift:5.4-focal
     strategy:
       fail-fast: false
       matrix:
@@ -91,9 +90,12 @@ jobs:
           - swift:5.3-bionic
           - swift:5.3-focal
           - swift:5.3-amazonlinux2
-          - swiftlang/swift:nightly-5.3-bionic
-          - swiftlang/swift:nightly-5.3-focal
-          - swiftlang/swift:nightly-5.3-amazonlinux2
+          - swift:5.4-bionic
+          - swift:5.4-focal
+          - swift:5.4-amazonlinux2
+          - swiftlang/swift:nightly-5.5-bionic
+          - swiftlang/swift:nightly-5.5-focal
+          - swiftlang/swift:nightly-5.5-amazonlinux2
     container: ${{ matrix.image }}
     steps:
       - name: Checkout code

Package.swift

@@ -17,7 +17,7 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.0.0"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
         .package(url: "https://github.com/vapor/sql-kit.git", from: "3.1.0"),
-        .package(url: "https://github.com/vapor/async-kit.git", from: "1.0.0"),
+        .package(url: "https://github.com/vapor/async-kit.git", from: "1.4.0"),
     ],
     targets: [
         .target(name: "FluentKit", dependencies: [

Sources/FluentBenchmark/SolarSystem/SolarSystem.swift

@@ -34,7 +34,7 @@ public struct SolarSystem: Migration {
             all = migrations
         }
 
-        return EventLoopFutureQueue(eventLoop: database.eventLoop).append(each: all) { $0.prepare(on: database) }
+        return all.sequencedFlatMapEach(on: database.eventLoop) { $0.prepare(on: database) }
     }
 
     public func revert(on database: Database) -> EventLoopFuture<Void> {
@@ -45,6 +45,6 @@ public struct SolarSystem: Migration {
             all = migrations
         }
 
-        return EventLoopFutureQueue(eventLoop: database.eventLoop).append(each: all.reversed()) { $0.revert(on: database) }
+        return all.reversed().sequencedFlatMapEach(on: database.eventLoop) { $0.revert(on: database) }
     }
 }

Sources/FluentKit/Migration/Migrations.swift

@@ -1,18 +1,12 @@
 public final class Migrations {
-    struct Item {
-        var id: DatabaseID?
-        var migration: Migration
-    }
-
-    var storage: [Item]
-    var databases: Set<DatabaseID?> { Set(self.storage.map(\.id)) }
+    var storage: [DatabaseID?: [Migration]]
 
     public init() {
-        self.storage = []
+        self.storage = [:]
     }
 
     public func add(_ migration: Migration, to id: DatabaseID? = nil) {
-        self.storage.append(.init(id: id, migration: migration))
+        self.storage[id, default: []].append(migration)
     }
 
     @inlinable
@@ -21,6 +15,6 @@ public final class Migrations {
     }
 
     public func add(_ migrations: [Migration], to id: DatabaseID? = nil) {
-        self.storage.append(contentsOf: migrations.map { .init(id: id, migration: $0) })
+        self.storage[id, default: []].append(contentsOf: migrations)
     }
 }

Sources/FluentKit/Migration/Migrator.swift

@@ -108,15 +108,8 @@ public struct Migrator {
     private func migrators<Result>(
         _ handler: (DatabaseMigrator) -> EventLoopFuture<Result>
     ) -> EventLoopFuture<[Result]> {
-        return self.migrations.databases.map { id in
-            let migrations = self.migrations.storage.compactMap { item -> Migration? in
-                guard item.id == id else { return nil }
-                return item.migration
-            }
-
-            let migrator = DatabaseMigrator(id: id, database: self.databaseFactory(id), migrations: migrations)
-            return handler(migrator)
-        }.flatten(on: self.eventLoop)
+        return self.migrations.storage.map { handler(.init(id: $0, database: self.databaseFactory($0), migrations: $1)) }
+            .flatten(on: self.eventLoop)
     }
 }
 
@@ -211,12 +204,8 @@ private final class DatabaseMigrator {
     // MARK: Prepare
 
     func prepareBatch() -> EventLoopFuture<Void> {
-        return self.unpreparedMigrations().flatMap { migrations in
-            return self.lastBatchNumber().and(value: migrations)
-        }.flatMap { batch, migrations in
-            return EventLoopFutureQueue(eventLoop: self.database.eventLoop).append(each: migrations) { migration in
-                self.prepare(migration, batch: batch + 1)
-            }
+        return self.lastBatchNumber().flatMap { batch in
+            self.unpreparedMigrations().sequencedFlatMapEach { self.prepare($0, batch: batch + 1) }
         }
     }
 
@@ -227,15 +216,11 @@ private final class DatabaseMigrator {
     }
 
     func revertBatch(number: Int) -> EventLoopFuture<Void> {
-        return self.preparedMigrations(batch: number).flatMap { migrations in
-            return EventLoopFutureQueue(eventLoop: self.database.eventLoop).append(each: migrations, self.revert)
-        }
+        return self.preparedMigrations(batch: number).sequencedFlatMapEach(self.revert)
     }
 
     func revertAllBatches() -> EventLoopFuture<Void> {
-        return self.preparedMigrations().flatMap { migrations in
-            return EventLoopFutureQueue(eventLoop: self.database.eventLoop).append(each: migrations, self.revert)
-        }
+        return self.preparedMigrations().sequencedFlatMapEach(self.revert)
     }
 
     // MARK: Preview

Sources/FluentKit/Query/Builder/QueryBuilder.swift

@@ -234,14 +234,8 @@ public final class QueryBuilder<Model>
                     return self.database.eventLoop.makeSucceededFuture(())
                 }
                 // run eager loads
-                return EventLoopFutureQueue(eventLoop: self.database.eventLoop).append(each: self.eagerLoaders) { loader in
+                return self.eagerLoaders.sequencedFlatMapEach(on: self.database.eventLoop) { loader in
                     return loader.anyRun(models: all, on: self.database)
-                }.flatMapErrorThrowing { error in
-                    if case .previousError(let error) = error as? EventLoopFutureQueue.ContinueError {
-                        throw error
-                    } else {
-                        throw error
-                    }
                 }
             }
         } else {