Hummingbird2 updates (#9043)

* Use one connection for mulitple queries

* Use prepared statements

* Reduce connection amount to 100

Author: adam-fowler

frameworks/Swift/hummingbird2/src-postgres/Sources/server/Controllers/FortunesController.swift

@@ -35,15 +35,27 @@ final class FortunesController: Sendable {
     /// is delivered to the client using a server-side HTML template. The message 
     /// text must be considered untrusted and properly escaped and the UTF-8 fortune messages must be rendered properly.
     @Sendable func fortunes(request: Request, context: Context) async throws -> HTML {
-        let rows = try await self.postgresClient.query("SELECT id, message FROM Fortune")
+        let rows = try await self.postgresClient.execute(SelectFortuneStatement())
         var fortunes: [Fortune] = []
-        for try await (id, message) in rows.decode((Int32, String).self, context: .default) {
-            fortunes.append(.init(id: id, message: message))
+        for try await fortune in rows {
+            fortunes.append(.init(id: fortune.0, message: fortune.1))
         }
 
         fortunes.append(.init(id: 0, message: "Additional fortune added at request time."))
         let sortedFortunes = fortunes.sorted { $0.message < $1.message }
         return HTML(html: self.template.render(sortedFortunes) )
 
     }
+
+    struct SelectFortuneStatement: PostgresPreparedStatement {
+        typealias Row = (Int32, String)
+
+        static var sql = "SELECT id, message FROM Fortune"
+
+        func makeBindings() throws -> PostgresNIO.PostgresBindings {
+            return .init()
+        }
+
+        func decodeRow(_ row: PostgresNIO.PostgresRow) throws -> Row { try row.decode(Row.self) }
+    }
 }

frameworks/Swift/hummingbird2/src-postgres/Sources/server/Controllers/WorldController.swift

@@ -16,11 +16,11 @@ struct WorldController {
     /// simple database table. That row is then serialized as a JSON response.
     @Sendable func single(request: Request, context: Context) async throws -> World {
         let id = Int32.random(in: 1...10_000)
-        let rows = try await self.postgresClient.query("SELECT id, randomnumber FROM World WHERE id = \(id)")
-        for try await (id, randomNumber) in rows.decode((Int32, Int32).self, context: .default) {
-            return World(id: id, randomNumber: randomNumber)
+        let rows = try await self.postgresClient.execute(SelectWorldStatement(id: id))
+        guard let row = try await rows.first(where: {_ in true }) else {
+            throw HTTPError(.notFound)
         }
-        throw HTTPError(.notFound)
+        return World(id: row.0, randomNumber: row.1)
     }
 
     /// In this test, each request is processed by fetching multiple rows from a 
@@ -29,21 +29,16 @@ struct WorldController {
     /// All tests are run at 512 concurrency.
     @Sendable func multiple(request: Request, context: Context) async throws -> [World] {
         let queries = (request.uri.queryParameters.get("queries", as: Int.self) ?? 1).bound(1, 500)
-        return try await withThrowingTaskGroup(of: World.self) { group in
+        return try await self.postgresClient.withConnection { conn in
+            var result: [World] = .init()
+            result.reserveCapacity(queries)
             for _ in 0..<queries {
-                group.addTask {
-                    let id = Int32.random(in: 1...10_000)
-                    let rows = try await self.postgresClient.query("SELECT id, randomnumber FROM World WHERE id = \(id)")
-                    for try await (id, randomNumber) in rows.decode((Int32, Int32).self, context: .default) {
-                        return World(id: id, randomNumber: randomNumber)
-                    }
+                let id = Int32.random(in: 1...10_000)
+                let rows = try await conn.execute(SelectWorldStatement(id: id), logger: context.logger)
+                guard let row = try await rows.first(where: {_ in true }) else {
                     throw HTTPError(.notFound)
                 }
-            }
-            var result: [World] = .init()
-            result.reserveCapacity(queries)
-            for try await world in group {
-                result.append(world)
+                result.append( World(id: row.0, randomNumber: row.1))
             }
             return result
         }
@@ -59,25 +54,54 @@ struct WorldController {
     /// query to fetch the object. All tests are run at 512 concurrency.
     @Sendable func updates(request: Request, context: Context) async throws -> [World] {
         let queries = (request.uri.queryParameters.get("queries", as: Int.self) ?? 1).bound(1, 500)
-        return try await withThrowingTaskGroup(of: World.self) { group in
+        return try await self.postgresClient.withConnection { conn in
+            var result: [World] = .init()
+            result.reserveCapacity(queries)
             for _ in 0..<queries {
-                group.addTask {
-                    let id = Int32.random(in: 1...10_000)
-                    let rows = try await self.postgresClient.query("SELECT id FROM World WHERE id = \(id)")
-                    for try await (id) in rows.decode((Int32).self, context: .default) {
-                        let randomNumber = Int32.random(in: 1...10_000)
-                        try await self.postgresClient.query("UPDATE World SET randomnumber = \(randomNumber) WHERE id = \(id)")
-                        return World(id: id, randomNumber: randomNumber)
-                    }
+                let id = Int32.random(in: 1...10_000)
+                let rows = try await conn.execute(SelectWorldStatement(id: id), logger: context.logger)
+                guard let row = try await rows.first(where: {_ in true }) else {
                     throw HTTPError(.notFound)
                 }
-            }
-            var result: [World] = .init()
-            result.reserveCapacity(queries)
-            for try await world in group {
-                result.append(world)
+                let randomNumber = Int32.random(in: 1...10_000)
+                _ = try await conn.execute(UpdateWorldStatement(id: id, randomNumber: randomNumber), logger: context.logger)
+                result.append(World(id: row.0, randomNumber: randomNumber))
             }
             return result
         }
     }
+
+    struct SelectWorldStatement: PostgresPreparedStatement {
+        typealias Row = (Int32, Int32)
+
+        let id: Int32
+
+        static var sql = "SELECT id, randomnumber FROM World WHERE id = $1"
+
+        func makeBindings() throws -> PostgresNIO.PostgresBindings {
+            var bindings = PostgresNIO.PostgresBindings(capacity: 1)
+            bindings.append(.init(int32: self.id))
+            return bindings
+        }
+
+        func decodeRow(_ row: PostgresNIO.PostgresRow) throws -> Row { try row.decode(Row.self) }
+    }
+
+    struct UpdateWorldStatement: PostgresPreparedStatement {
+        typealias Row = Int32
+
+        let id: Int32
+        let randomNumber: Int32
+
+        static var sql = "UPDATE World SET randomnumber = $2 WHERE id = $1"
+
+        func makeBindings() throws -> PostgresNIO.PostgresBindings {
+            var bindings = PostgresNIO.PostgresBindings(capacity: 2)
+            bindings.append(.init(int32: self.id))
+            bindings.append(.init(int32: self.randomNumber))
+            return bindings
+        }
+
+        func decodeRow(_ row: PostgresNIO.PostgresRow) throws -> Row { try row.decode(Row.self) }
+    }
 }

frameworks/Swift/hummingbird2/src-postgres/Sources/server/main.swift

@@ -39,7 +39,7 @@ func runApp() async throws {
         database: "hello_world", 
         tls: .disable
     )
-    postgresConfiguration.options.maximumConnections = 1900
+    postgresConfiguration.options.maximumConnections = 100
     let postgresClient = PostgresClient(
         configuration: postgresConfiguration, 
         eventLoopGroup: MultiThreadedEventLoopGroup.singleton