diff --git a/clients/javascript/baradb.js b/clients/javascript/baradb.js index c963a53..74cddef 100644 --- a/clients/javascript/baradb.js +++ b/clients/javascript/baradb.js @@ -244,6 +244,8 @@ class Client { this.requestId = 0; this._buffer = Buffer.alloc(0); this._pendingResolve = null; + this._requestQueue = []; + this._requestLock = false; } async connect() { @@ -478,70 +480,98 @@ class Client { throw new Error(`Unexpected auth response: 0x${header.kind.toString(16)}`); } + async _processQueue() { + if (this._requestLock || this._requestQueue.length === 0) return; + this._requestLock = true; + const { task, resolve, reject } = this._requestQueue.shift(); + try { + const result = await task(); + resolve(result); + } catch (err) { + reject(err); + } finally { + this._requestLock = false; + setImmediate(() => this._processQueue()); + } + } + + _enqueue(task) { + return new Promise((resolve, reject) => { + this._requestQueue.push({ task, resolve, reject }); + this._processQueue(); + }); + } + async ping() { if (!this.connected) throw new Error('Not connected'); - const msg = this._build(MsgKind.PING, Buffer.alloc(0)); - this.socket.write(msg); + return this._enqueue(async () => { + const msg = this._build(MsgKind.PING, Buffer.alloc(0)); + this.socket.write(msg); - const header = await this._readHeader(); - if (header.kind === MsgKind.PONG) return true; - if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); - return false; + const header = await this._readHeader(); + if (header.kind === MsgKind.PONG) return true; + if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); + return false; + }); } async query(sql) { if (!this.connected) throw new Error('Not connected'); - const queryBuf = Buffer.from(sql, 'utf-8'); - const payload = Buffer.alloc(4 + queryBuf.length + 1); - payload.writeUInt32BE(queryBuf.length, 0); - queryBuf.copy(payload, 4); - payload[4 + queryBuf.length] = ResultFormat.BINARY; + return this._enqueue(async () => { + const queryBuf = Buffer.from(sql, 'utf-8'); + const payload = Buffer.alloc(4 + queryBuf.length + 1); + payload.writeUInt32BE(queryBuf.length, 0); + queryBuf.copy(payload, 4); + payload[4 + queryBuf.length] = ResultFormat.BINARY; - const msg = this._build(MsgKind.QUERY, payload); - this.socket.write(msg); + const msg = this._build(MsgKind.QUERY, payload); + this.socket.write(msg); - const header = await this._readHeader(); - if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); - if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); - if (header.kind === MsgKind.COMPLETE) { - const data = await this._recvExact(header.length); - const result = new QueryResult(); - result.affectedRows = data.readUInt32BE(0); - return result; - } - return new QueryResult(); + const header = await this._readHeader(); + if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); + if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); + if (header.kind === MsgKind.COMPLETE) { + const data = await this._recvExact(header.length); + const result = new QueryResult(); + result.affectedRows = data.readUInt32BE(0); + return result; + } + return new QueryResult(); + }); } async queryParams(sql, params = []) { if (!this.connected) throw new Error('Not connected'); - const queryBuf = Buffer.from(sql, 'utf-8'); - const paramParts = []; - for (const p of params) { - paramParts.push(p.serialize()); - } - const paramsBuf = Buffer.concat(paramParts); + return this._enqueue(async () => { + const queryBuf = Buffer.from(sql, 'utf-8'); + const paramParts = []; + for (const p of params) { + paramParts.push(p.serialize()); + } + const paramsBuf = Buffer.concat(paramParts); - const payload = Buffer.alloc(4 + queryBuf.length + 1 + 4 + paramsBuf.length); - let pos = 0; - payload.writeUInt32BE(queryBuf.length, pos); pos += 4; - queryBuf.copy(payload, pos); pos += queryBuf.length; - payload[pos] = ResultFormat.BINARY; pos++; - payload.writeUInt32BE(params.length, pos); pos += 4; - paramsBuf.copy(payload, pos); + const payload = Buffer.alloc(4 + queryBuf.length + 1 + 4 + paramsBuf.length); + let pos = 0; + payload.writeUInt32BE(queryBuf.length, pos); pos += 4; + queryBuf.copy(payload, pos); pos += queryBuf.length; + payload[pos] = ResultFormat.BINARY; pos++; + payload.writeUInt32BE(params.length, pos); pos += 4; + paramsBuf.copy(payload, pos); - const msg = this._build(MsgKind.QUERY_PARAMS, payload); - this.socket.write(msg); + const msg = this._build(MsgKind.QUERY_PARAMS, payload); + this.socket.write(msg); - const header = await this._readHeader(); - if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); - if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); - if (header.kind === MsgKind.COMPLETE) { - const data = await this._recvExact(header.length); - const result = new QueryResult(); - result.affectedRows = data.readUInt32BE(0); - return result; - } - return new QueryResult(); + const header = await this._readHeader(); + if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); + if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); + if (header.kind === MsgKind.COMPLETE) { + const data = await this._recvExact(header.length); + const result = new QueryResult(); + result.affectedRows = data.readUInt32BE(0); + return result; + } + return new QueryResult(); + }); } async execute(sql) { diff --git a/src/barabadb/core/gossip.nim b/src/barabadb/core/gossip.nim index da6cbba..f2c37bf 100644 --- a/src/barabadb/core/gossip.nim +++ b/src/barabadb/core/gossip.nim @@ -4,6 +4,7 @@ import std/random import std/monotimes import std/asyncdispatch import std/net +import std/asyncnet import std/strutils import std/streams import std/os @@ -36,7 +37,7 @@ type fanout*: int # number of nodes to gossip to per round gossipPort*: int running*: bool - sock*: Socket + sock*: AsyncSocket onJoin*: proc(node: GossipNode) {.gcsafe.} onLeave*: proc(nodeId: string) {.gcsafe.} onSuspect*: proc(nodeId: string) {.gcsafe.} @@ -305,25 +306,21 @@ proc startGossipRound*(gp: GossipProtocol, intervalMs: int = 2000) {.async.} = gp.broadcastGossip() proc startGossipListener*(gp: GossipProtocol) {.async.} = - gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.bindAddr(Port(gp.gossipPort)) gp.running = true while gp.running: try: - var data = newString(65535) - var senderAddr = "" - var senderPort: Port - let bytesRead = gp.sock.recvFrom(data, 65535, senderAddr, senderPort) - if bytesRead > 0: - data.setLen(bytesRead) + let (data, senderAddr, senderPort) = await gp.sock.recvFrom(65535) + if data.len > 0: gp.handleIncomingGossip(data, senderAddr) except: # Small sleep on error to avoid spin gp.sock.close() # Recreate socket for next iteration try: - gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.bindAddr(Port(gp.gossipPort)) except: diff --git a/src/barabadb/protocol/wire.nim b/src/barabadb/protocol/wire.nim index 88963c7..259a132 100644 --- a/src/barabadb/protocol/wire.nim +++ b/src/barabadb/protocol/wire.nim @@ -169,12 +169,14 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) = of fkFloat32: var fl = val.float32Val var bytes: array[4, byte] - copyMem(addr bytes, unsafeAddr fl, 4) + var i32 = cast[int32](fl) + bigEndian32(addr bytes, unsafeAddr i32) buf.add(bytes) of fkFloat64: var fl = val.float64Val var bytes: array[8, byte] - copyMem(addr bytes, unsafeAddr fl, 8) + var i64 = cast[int64](fl) + bigEndian64(addr bytes, unsafeAddr i64) buf.add(bytes) of fkString: buf.writeString(val.strVal) of fkBytes: buf.writeBytes(val.bytesVal) @@ -192,7 +194,8 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) = for f in val.vecVal: var fl = f var bytes: array[4, byte] - copyMem(addr bytes, unsafeAddr fl, 4) + var i32 = cast[int32](fl) + bigEndian32(addr bytes, unsafeAddr i32) buf.add(bytes) of fkJson: buf.writeString(val.jsonVal) @@ -227,14 +230,18 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire var fl: float32 var bytes: array[4, byte] for i in 0..3: bytes[i] = buf[pos + i] - copyMem(addr fl, unsafeAddr bytes, 4) + var i32: int32 + bigEndian32(addr i32, unsafeAddr bytes) + fl = cast[float32](i32) pos += 4 result = WireValue(kind: fkFloat32, float32Val: fl) of fkFloat64: var fl: float64 var bytes: array[8, byte] for i in 0..7: bytes[i] = buf[pos + i] - copyMem(addr fl, unsafeAddr bytes, 8) + var i64: int64 + bigEndian64(addr i64, unsafeAddr bytes) + fl = cast[float64](i64) pos += 8 result = WireValue(kind: fkFloat64, float64Val: fl) of fkString: @@ -270,7 +277,9 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire var fl: float32 var bytes: array[4, byte] for j in 0..3: bytes[j] = buf[pos + j] - copyMem(addr fl, unsafeAddr bytes, 4) + var i32: int32 + bigEndian32(addr i32, unsafeAddr bytes) + fl = cast[float32](i32) pos += 4 vec.add(fl) result = WireValue(kind: fkVector, vecVal: vec)