Batch network packets in prepared statements

packages/pg/bench.js

@@ -47,21 +47,27 @@ const run = async () => {
   for (let i = 0; i < 4; i++) {
     let queries = await bench(client, params, seconds * 1000)
     console.log('')
-    console.log('little queries:', queries)
+    console.log('param queries:', queries)
     console.log('qps', queries / seconds)
-    console.log('on my laptop best so far seen 733 qps')
+    console.log('on my laptop best so far seen 987 qps')
+
+    queries = await bench(client, { ...params, name: 'params' }, seconds * 1000)
+    console.log('')
+    console.log('named queries:', queries)
+    console.log('qps', queries / seconds)
+    console.log('on my laptop best so far seen 937 qps')
 
     console.log('')
     queries = await bench(client, seq, seconds * 1000)
     console.log('sequence queries:', queries)
     console.log('qps', queries / seconds)
-    console.log('on my laptop best so far seen 1309 qps')
+    console.log('on my laptop best so far seen 2725 qps')
 
     console.log('')
     queries = await bench(client, insert, seconds * 1000)
     console.log('insert queries:', queries)
     console.log('qps', queries / seconds)
-    console.log('on my laptop best so far seen 6445 qps')
+    console.log('on my laptop best so far seen 27383 qps')
 
     console.log('')
     console.log('Warming up bytea test')
@@ -75,7 +81,7 @@ const run = async () => {
     const time = Date.now() - start
     console.log('bytea time:', time, 'ms')
     console.log('bytea length:', results.rows[0].data.byteLength, 'bytes')
-    console.log('on my laptop best so far seen 1107ms and 104857600 bytes')
+    console.log('on my laptop best so far seen 1407ms and 104857600 bytes')
     await new Promise((resolve) => setTimeout(resolve, 250))
   }
 

packages/pg/lib/query.js

@@ -161,7 +161,21 @@ class Query extends EventEmitter {
       return new Error('Query values must be an array')
     }
     if (this.requiresPreparation()) {
-      this.prepare(connection)
+      // If we're using the extended query protocol we fire off several separate commands
+      // to the backend. On some versions of node & some operating system versions
+      // the network stack writes each message separately instead of buffering them together
+      // causing the client & network to send more slowly. Corking & uncorking the stream
+      // allows node to buffer up the messages internally before sending them all off at once.
+      // note: we're checking for existence of cork/uncork because some versions of streams
+      // might not have this (cloudflare?)
+      connection.stream.cork && connection.stream.cork()
+      try {
+        this.prepare(connection)
+      } finally {
+        // while unlikely for this.prepare to throw, if it does & we don't uncork this stream
+        // this client becomes unresponsive, so put in finally block "just in case"
+        connection.stream.uncork && connection.stream.uncork()
+      }
     } else {
       connection.query(this.text)
     }