better buffer utilization in pg

splitice · splitice · commit 44b4a8c32869 · 2025-06-04T19:38:30.000+10:00
diff --git a/packages/pg-protocol/src/buffer-reader.ts b/packages/pg-protocol/src/buffer-reader.ts
@@ -1,4 +1,4 @@
-const emptyBuffer = Buffer.allocUnsafe(0)
+const emptyBuffer = Buffer.allocUnsafeSlow(0)
 
 export class BufferReader {
   private buffer: Buffer = emptyBuffer
diff --git a/packages/pg-protocol/src/buffer-writer.ts b/packages/pg-protocol/src/buffer-writer.ts
@@ -5,7 +5,7 @@ export class Writer {
   private offset: number = 5
   private headerPosition: number = 0
   constructor(private size = 256) {
-    this.buffer = Buffer.allocUnsafe(size)
+    this.buffer = Buffer.allocUnsafeSlow(size)
   }
 
   private ensure(size: number): void {
@@ -15,7 +15,7 @@ export class Writer {
       // exponential growth factor of around ~ 1.5
       // https://stackoverflow.com/questions/2269063/buffer-growth-strategy
       var newSize = oldBuffer.length + (oldBuffer.length >> 1) + size
-      this.buffer = Buffer.allocUnsafe(newSize)
+      this.buffer = Buffer.allocUnsafeSlow(newSize)
       oldBuffer.copy(this.buffer)
     }
   }
@@ -79,7 +79,7 @@ export class Writer {
     var result = this.join(code)
     this.offset = 5
     this.headerPosition = 0
-    this.buffer = Buffer.allocUnsafe(this.size)
+    this.buffer = Buffer.allocUnsafeSlow(this.size)
     return result
   }
 }
diff --git a/packages/pg-protocol/src/parser.ts b/packages/pg-protocol/src/parser.ts
@@ -106,6 +106,9 @@ export class Parser {
         break
       }
     }
+    if(buffer === this.buffer) {
+      this.buffer = Buffer.from(this.buffer)
+    }
     if (offset === bufferFullLength) {
       // No more use for the buffer
       this.buffer = emptyBuffer
@@ -134,7 +137,7 @@ export class Parser {
           while (newLength >= newBufferLength) {
             newBufferLength *= 2
           }
-          newBuffer = Buffer.allocUnsafe(newBufferLength)
+          newBuffer = Buffer.allocUnsafeSlow(newBufferLength)
         }
         // Move the remaining buffer to the new one
         this.buffer.copy(newBuffer, 0, this.bufferOffset, this.bufferOffset + this.bufferLength)
diff --git a/packages/pg-protocol/src/serializer.ts b/packages/pg-protocol/src/serializer.ts
@@ -36,7 +36,7 @@ const startup = (opts: Record<string, string>): Buffer => {
 }
 
 const requestSsl = (): Buffer => {
-  const response = Buffer.allocUnsafe(8)
+  const response = Buffer.allocUnsafeSlow(8)
   response.writeInt32BE(8, 0)
   response.writeInt32BE(80877103, 4)
   return response
@@ -183,7 +183,7 @@ const execute = (config?: ExecOpts): Buffer => {
   const portalLength = Buffer.byteLength(portal)
   const len = 4 + portalLength + 1 + 4
   // one extra bit for code
-  const buff = Buffer.allocUnsafe(1 + len)
+  const buff = Buffer.allocUnsafeSlow(1 + len)
   buff[0] = code.execute
   buff.writeInt32BE(len, 1)
   buff.write(portal, 5, 'utf-8')
@@ -193,7 +193,7 @@ const execute = (config?: ExecOpts): Buffer => {
 }
 
 const cancel = (processID: number, secretKey: number): Buffer => {
-  const buffer = Buffer.allocUnsafe(16)
+  const buffer = Buffer.allocUnsafeSlow(16)
   buffer.writeInt32BE(16, 0)
   buffer.writeInt16BE(1234, 4)
   buffer.writeInt16BE(5678, 6)
@@ -211,7 +211,7 @@ const cstringMessage = (code: code, string: string): Buffer => {
   const stringLen = Buffer.byteLength(string)
   const len = 4 + stringLen + 1
   // one extra bit for code
-  const buffer = Buffer.allocUnsafe(1 + len)
+  const buffer = Buffer.allocUnsafeSlow(1 + len)
   buffer[0] = code
   buffer.writeInt32BE(len, 1)
   buffer.write(string, 5, 'utf-8')
diff --git a/packages/pg/lib/connection.js b/packages/pg/lib/connection.js
@@ -40,7 +40,7 @@ class Connection extends EventEmitter {
 
     this._connecting = true
     this.stream.setNoDelay(true)
-    this.stream.connect(port, host)
+    this.stream.connect({port, host})
 
     this.stream.once('connect', function () {
       if (self._keepAlive) {
diff --git a/packages/pg/lib/stream.js b/packages/pg/lib/stream.js
@@ -20,7 +20,8 @@ module.exports = {
 function getNodejsStreamFuncs() {
   function getStream(ssl) {
     const net = require('net')
-    return new net.Socket()
+    const socket = new net.Socket({onread:{buffer: Buffer.allocUnsafeSlow(2048), callback: (n, b)=>socket.emit('data', b.slice(0, n))}})
+    return socket
   }
 
   function getSecureStream(options) {

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-const emptyBuffer = Buffer.allocUnsafe(0)`
	`1`	`+const emptyBuffer = Buffer.allocUnsafeSlow(0)`
`2`	`2`
`3`	`3`	`export class BufferReader {`
`4`	`4`	`private buffer: Buffer = emptyBuffer`
Original file line number	Diff line number	Diff line change
`@@ -5,7 +5,7 @@ export class Writer {`
`5`	`5`	`private offset: number = 5`
`6`	`6`	`private headerPosition: number = 0`
`7`	`7`	`constructor(private size = 256) {`
`8`		`- this.buffer = Buffer.allocUnsafe(size)`
	`8`	`+ this.buffer = Buffer.allocUnsafeSlow(size)`
`9`	`9`	`}`
`10`	`10`
`11`	`11`	`private ensure(size: number): void {`
`@@ -15,7 +15,7 @@ export class Writer {`
`15`	`15`	`// exponential growth factor of around ~ 1.5`
`16`	`16`	`// https://stackoverflow.com/questions/2269063/buffer-growth-strategy`
`17`	`17`	`var newSize = oldBuffer.length + (oldBuffer.length >> 1) + size`
`18`		`- this.buffer = Buffer.allocUnsafe(newSize)`
	`18`	`+ this.buffer = Buffer.allocUnsafeSlow(newSize)`
`19`	`19`	`oldBuffer.copy(this.buffer)`
`20`	`20`	`}`
`21`	`21`	`}`
`@@ -79,7 +79,7 @@ export class Writer {`
`79`	`79`	`var result = this.join(code)`
`80`	`80`	`this.offset = 5`
`81`	`81`	`this.headerPosition = 0`
`82`		`- this.buffer = Buffer.allocUnsafe(this.size)`
	`82`	`+ this.buffer = Buffer.allocUnsafeSlow(this.size)`
`83`	`83`	`return result`
`84`	`84`	`}`
`85`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -106,6 +106,9 @@ export class Parser {`
`106`	`106`	`break`
`107`	`107`	`}`
`108`	`108`	`}`
	`109`	`+ if(buffer === this.buffer) {`
	`110`	`+ this.buffer = Buffer.from(this.buffer)`
	`111`	`+ }`
`109`	`112`	`if (offset === bufferFullLength) {`
`110`	`113`	`// No more use for the buffer`
`111`	`114`	`this.buffer = emptyBuffer`
`@@ -134,7 +137,7 @@ export class Parser {`
`134`	`137`	`while (newLength >= newBufferLength) {`
`135`	`138`	`newBufferLength *= 2`
`136`	`139`	`}`
`137`		`- newBuffer = Buffer.allocUnsafe(newBufferLength)`
	`140`	`+ newBuffer = Buffer.allocUnsafeSlow(newBufferLength)`
`138`	`141`	`}`
`139`	`142`	`// Move the remaining buffer to the new one`
`140`	`143`	`this.buffer.copy(newBuffer, 0, this.bufferOffset, this.bufferOffset + this.bufferLength)`
Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,8 @@ module.exports = {`
`20`	`20`	`function getNodejsStreamFuncs() {`
`21`	`21`	`function getStream(ssl) {`
`22`	`22`	`const net = require('net')`
`23`		`- return new net.Socket()`
	`23`	`+ const socket = new net.Socket({onread:{buffer: Buffer.allocUnsafeSlow(2048), callback: (n, b)=>socket.emit('data', b.slice(0, n))}})`
	`24`	`+ return socket`
`24`	`25`	`}`
`25`	`26`
`26`	`27`	`function getSecureStream(options) {`