diff --git a/docs/bg/architecture.md b/docs/bg/architecture.md index 86a9100..04af52b 100644 --- a/docs/bg/architecture.md +++ b/docs/bg/architecture.md @@ -34,10 +34,31 @@ BaraDB е **мултимодален database engine**, написан на Nim, - **WebSocket** (`core/websocket.nim`): Full-duplex стрийминг - **Embedded** (`storage/lsm.nim`): Директен in-process достъп +### Сървърна Архитектура + +TCP и HTTP сървърите споделят един LSMTree инстанс за гарантиране на консистентност на данните: + +``` +┌─────────────────────────────────────────────────────────┐ +│ СПОДЕЛЕНО СЪХРАНЕНИЕ │ +│ LSMTree Инстанс │ +├────────────────────────┬────────────────────────────────┤ +│ TCP Сървър │ HTTP Сървър │ +│ (Бинарен Протокол) │ (REST API) │ +│ Порт: 9472 │ Порт: 9912 │ +│ TCP_NODELAY: ВКЛ │ Multi-threaded │ +└────────────────────────┴────────────────────────────────┘ +``` + +**Ключови оптимизации:** +- **Споделен LSMTree** — И двата сървъра работят върху един и същ database инстанс, елиминирайки несъответствие в данните +- **TCP_NODELAY** — Включен както на слушащия, така и на клиентските сокети за по-ниска латентност при малки съобщения +- **Безопасна конверсия на байтове** — Правилни `bytesToString`/`stringToBytes` функции вместо unsafe `cast` операции + ### Управление на Връзки - **Connection Pool** (`protocol/pool.nim`): Min/max лимити на връзки с idle timeout -- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket глобални и per-client лимити +- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket глобални и per-client лимити, интегрирани в TCP и HTTP хендърите - **Автентикация** (`protocol/auth.nim`): JWT с HMAC-SHA256 и достъп на база роли - **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 с автоматично генерирани сертификати @@ -95,7 +116,7 @@ BaraQL конвейрът: - **Raft Консенсус** (`core/raft.nim`): Leader election, log репликация - **Шардиране** (`core/sharding.nim`): Hash, range и consistent hashing - **Репликация** (`core/replication.nim`): Sync, async, semi-sync режими -- **Gossip Протокол** (`core/gossip.nim`): SWIM-подобно управление на членство +- **Gossip Протокол** (`core/gossip.nim`): SWIM-подобно управление на членство с експоненциален backoff при грешки - **Разпределени Транзакции** (`core/disttxn.nim`): Two-phase commit ## Ключови Дизайнерски Решения diff --git a/docs/bg/changelog.md b/docs/bg/changelog.md index 4696943..035134e 100644 --- a/docs/bg/changelog.md +++ b/docs/bg/changelog.md @@ -2,6 +2,21 @@ Всички забележителни промени в BaraDB са документирани в този файл. +## [Unreleased] — Подобрения на Сървърната Архитектура + +### Поправки + +- **Споделен Storage Инстанс** — TCP и HTTP сървърите вече споделят един LSMTree инстанс вместо да създават отделни, елиминирайки несъответствие в данните между протоколите +- **Unsafe Cast Операции** — Заменени са всички `cast[string]` и `cast[seq[byte]]` с безопасните `bytesToString`/`stringToBytes` помощни функции, които правилно копират данните за съвместимост с ARC/ORC управлението на паметта + +### Промени + +- **TCP_NODELAY Включен** — Както слушащите, така и клиентските сокети вече имат TCP_NODELAY, намалявайки латентността за малки съобщения (заявки, acks) чрез деактивиране на алгоритъма на Nagle +- **Rate Limiter Интегриран** — Token-bucket rate limiter вече е активен в TCP и HTTP сървър хендърите: + - TCP: Per-client rate limiting на query съобщения (код за грешка 429) + - HTTP: Rate limiting на `/query` и `/batch` endpoints с поддръжка на `X-Forwarded-For` +- **Gossip Обработка на Грешки** — Подобрено възстановяване при UDP socket грешки с експоненциален backoff (100ms-5s) и презаписване на сокета само след 10 последователни грешки + ## [Unreleased] — SQL:2023 Стабилизация ### Поправки diff --git a/docs/bg/configuration.md b/docs/bg/configuration.md index 57dd68f..31f6808 100644 --- a/docs/bg/configuration.md +++ b/docs/bg/configuration.md @@ -45,8 +45,10 @@ BaraDB може да се конфигурира чрез **променливи |------------|----------|----------| | `BARADB_AUTH_ENABLED` | `false` | Включване на автентикация | | `BARADB_JWT_SECRET` | — | JWT подписващ secret | -| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Глобални заявки в секунда | -| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Заявки в секунда за клиент | +| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Глобални заявки в минута (token-bucket) | +| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Заявки в минута за клиент (token-bucket) | + +> **Забележка:** Rate limiting се прилага на ниво протокол и в двата сървъра (TCP и HTTP). TCP използва ID на връзката за per-client лимити. HTTP използва `X-Forwarded-For` хедър (за прокси среди) или глобален ключ. ### Логване diff --git a/docs/bg/protocol.md b/docs/bg/protocol.md index cf15981..66057e6 100644 --- a/docs/bg/protocol.md +++ b/docs/bg/protocol.md @@ -11,6 +11,11 @@ BaraDB поддържа множество протоколи за клиент Бинарният протокол използва big-endian кодиране за всички многобайтови стойности. +### Оптимизации на Връзката + +- **TCP_NODELAY** — Включен както на сървърните, така и на клиентските сокети за минимизиране на латентността при малки съобщения (заявки, acks). Това деактивира алгоритъма на Nagle, гарантирайки че пакетите се изпращат незабавно без буфериране. +- **Rate Limiting** — Token-bucket rate limiter е интегриран на ниво протокол. Заявки надвишаващи лимита получават код за грешка 429. + ### Жизнен Цикъл на Връзката ``` @@ -107,6 +112,18 @@ BaraDB поддържа множество протоколи за клиент Базов URL: `http://localhost:9470/api/v1` +### Rate Limiting + +Всички query endpoints (`/query`, `/batch`) са защитени от token-bucket rate limiter: +- **Глобален лимит** — Конфигурира се чрез `BARADB_RATE_LIMIT_GLOBAL` (по подразбиране: 10,000 заявки/минута) +- **Per-client лимит** — Конфигурира се чрез `BARADB_RATE_LIMIT_PER_CLIENT` (по подразбиране: 1,000 заявки/минута) +- **Идентификация на клиент** — Използва `X-Forwarded-For` хедър когато е зад прокси, иначе използва глобален ключ + +При rate limiting, сървърът връща HTTP 429 с: +```json +{"error": "Rate limit exceeded"} +``` + ### Endpoints #### Health diff --git a/docs/en/architecture.md b/docs/en/architecture.md index 7039363..24b4165 100644 --- a/docs/en/architecture.md +++ b/docs/en/architecture.md @@ -34,10 +34,31 @@ Multiple communication protocols: - **WebSocket** (`core/websocket.nim`): Full-duplex streaming - **Embedded** (`storage/lsm.nim`): Direct in-process access +### Server Architecture + +The TCP and HTTP servers share a single LSMTree instance to ensure data consistency: + +``` +┌─────────────────────────────────────────────────────────┐ +│ SHARED STORAGE │ +│ LSMTree Instance │ +├────────────────────────┬────────────────────────────────┤ +│ TCP Server │ HTTP Server │ +│ (Binary Protocol) │ (REST API) │ +│ Port: 9472 │ Port: 9912 │ +│ TCP_NODELAY: ON │ Multi-threaded │ +└────────────────────────┴────────────────────────────────┘ +``` + +**Key optimizations:** +- **Shared LSMTree** — Both servers operate on the same database instance, eliminating data inconsistency +- **TCP_NODELAY** — Enabled on both listening and client sockets for lower latency on small messages +- **Safe byte conversion** — Proper `bytesToString`/`stringToBytes` functions instead of unsafe `cast` operations + ### Connection Management - **Connection Pool** (`protocol/pool.nim`): Min/max connection limits with idle timeout -- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket global and per-client limits +- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket global and per-client limits, integrated in both TCP and HTTP handlers - **Authentication** (`protocol/auth.nim`): JWT with HMAC-SHA256 and role-based access - **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 with auto-generated certificates @@ -121,7 +142,7 @@ Client → Protocol → Auth → Parser → AST → IR → Codegen - **Raft Consensus** (`core/raft.nim`): Leader election, log replication - **Sharding** (`core/sharding.nim`): Hash, range, and consistent hashing - **Replication** (`core/replication.nim`): Sync, async, semi-sync modes -- **Gossip Protocol** (`core/gossip.nim`): SWIM-like membership management +- **Gossip Protocol** (`core/gossip.nim`): SWIM-like membership management with exponential backoff error recovery - **Distributed Transactions** (`core/disttxn.nim`): Two-phase commit ## Key Design Decisions diff --git a/docs/en/changelog.md b/docs/en/changelog.md index 2e44ebd..9baaf70 100644 --- a/docs/en/changelog.md +++ b/docs/en/changelog.md @@ -2,6 +2,21 @@ All notable changes to BaraDB are documented in this file. +## [Unreleased] — Server Architecture Improvements + +### Fixed + +- **Shared Storage Instance** — TCP and HTTP servers now share a single LSMTree instance instead of creating separate ones, eliminating data inconsistency between protocols +- **Unsafe Cast Operations** — Replaced all `cast[string]` and `cast[seq[byte]]` with safe `bytesToString`/`stringToBytes` helper functions that properly copy data for ARC/ORC memory management compatibility + +### Changed + +- **TCP_NODELAY Enabled** — Both listening and client sockets now have TCP_NODELAY set, reducing latency for small messages (queries, acks) by disabling Nagle's algorithm +- **Rate Limiter Integrated** — Token-bucket rate limiter is now active in both TCP and HTTP server handlers: + - TCP: Per-client rate limiting on query messages (error code 429) + - HTTP: Rate limiting on `/query` and `/batch` endpoints with `X-Forwarded-For` support +- **Gossip Error Handling** — Improved UDP socket error recovery with exponential backoff (100ms-5s) and socket recreation only after 10 consecutive failures + ## [Unreleased] — AI-Native Platform ### Added diff --git a/docs/en/configuration.md b/docs/en/configuration.md index e343171..e0ae705 100644 --- a/docs/en/configuration.md +++ b/docs/en/configuration.md @@ -45,8 +45,10 @@ BaraDB can be configured via **environment variables**, a **config file**, or ** |----------|---------|-------------| | `BARADB_AUTH_ENABLED` | `false` | Enable authentication | | `BARADB_JWT_SECRET` | — | JWT signing secret | -| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Global requests per second | -| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Per-client requests per second | +| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Global requests per minute (token-bucket) | +| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Per-client requests per minute (token-bucket) | + +> **Note:** Rate limiting is enforced at the protocol level in both TCP and HTTP servers. TCP uses client connection ID for per-client limits. HTTP uses `X-Forwarded-For` header (for proxy setups) or a global key. ### Logging diff --git a/docs/en/protocol.md b/docs/en/protocol.md index a4c3698..6579d61 100644 --- a/docs/en/protocol.md +++ b/docs/en/protocol.md @@ -11,6 +11,11 @@ BaraDB supports multiple protocols for client communication: The binary protocol uses big-endian encoding for all multi-byte values. +### Connection Optimizations + +- **TCP_NODELAY** — Enabled on both server and client sockets to minimize latency for small messages (queries, acks). This disables Nagle's algorithm, ensuring packets are sent immediately without buffering. +- **Rate Limiting** — Token-bucket rate limiter is integrated at the protocol level. Queries exceeding the rate limit receive error code 429. + ### Connection Lifecycle ``` @@ -142,6 +147,18 @@ printf '\x00\x00\x00\x12\x01\x02\x00\x00\x00\x00\x08SELECT 1' > /dev/tcp/localho Base URL: `http://localhost:9470/api/v1` +### Rate Limiting + +All query endpoints (`/query`, `/batch`) are protected by a token-bucket rate limiter: +- **Global rate** — Configured via `BARADB_RATE_LIMIT_GLOBAL` (default: 10,000 requests/minute) +- **Per-client rate** — Configured via `BARADB_RATE_LIMIT_PER_CLIENT` (default: 1,000 requests/minute) +- **Client identification** — Uses `X-Forwarded-For` header when behind a proxy, otherwise uses a global key + +When rate-limited, the server returns HTTP 429 with: +```json +{"error": "Rate limit exceeded"} +``` + ### Endpoints #### Health diff --git a/src/barabadb/core/gossip.nim b/src/barabadb/core/gossip.nim index d82bcfa..582bb0e 100644 --- a/src/barabadb/core/gossip.nim +++ b/src/barabadb/core/gossip.nim @@ -320,21 +320,36 @@ proc startGossipListener*(gp: GossipProtocol) {.async.} = gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.bindAddr(Port(gp.gossipPort)) gp.running = true + var consecutiveErrors = 0 + const maxConsecutiveErrors = 10 + const baseRetryDelayMs = 100 + while gp.running: try: let (data, senderAddr, _) = await gp.sock.recvFrom(65535) + consecutiveErrors = 0 # Reset on success 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 = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) - gp.sock.setSockOpt(OptReuseAddr, true) - gp.sock.bindAddr(Port(gp.gossipPort)) - except: + except CatchableError as e: + if not gp.running: break + inc consecutiveErrors + if consecutiveErrors >= maxConsecutiveErrors: + # Recreate socket after too many errors + try: + gp.sock.close() + except: + discard + try: + gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + gp.sock.setSockOpt(OptReuseAddr, true) + gp.sock.bindAddr(Port(gp.gossipPort)) + consecutiveErrors = 0 + except: + break + # Exponential backoff with cap + let delayMs = min(baseRetryDelayMs * (1 shl min(consecutiveErrors, 6)), 5000) + await sleepAsync(delayMs) proc startGossip*(gp: GossipProtocol, healthIntervalMs: int = 1000, gossipIntervalMs: int = 2000) = diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index d97649c..c3c4f77 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -19,6 +19,7 @@ import ../protocol/wire import ../core/websocket import jwt as jwtlib import ../protocol/auth +import ../protocol/ratelimit type HttpServer* = ref object @@ -29,6 +30,7 @@ type metrics*: Metrics secretKey*: string authManager*: AuthManager + rateLimiter*: RateLimiter ws*: WsServer Metrics* = ref object @@ -38,13 +40,12 @@ type selectCount*: int activeConnections*: int -proc newHttpServer*(config: BaraConfig): HttpServer = - let dataDir = config.dataDir / "server" - let db = newLSMTree(dataDir) +proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer = let ctx = newExecutionContext(db) ctx.txnManager = newTxnManager() let secret = config.getEffectiveJwtSecret() let ws = newWsServer(config, secret) + let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient) ctx.onChange = proc(ev: ChangeEvent) = let msg = $ev.kind & " " & ev.table asyncCheck ws.broadcastToTable(ev.table, msg) @@ -52,8 +53,14 @@ proc newHttpServer*(config: BaraConfig): HttpServer = HttpServer(config: config, running: false, db: db, ctx: ctx, secretKey: secret, authManager: am, + rateLimiter: rl, metrics: Metrics(), ws: ws) +proc newHttpServer*(config: BaraConfig): HttpServer = + let dataDir = config.dataDir / "server" + let db = newLSMTree(dataDir) + return newHttpServerWithDb(config, db) + # ---------------------------------------------------------------------- # JWT helpers # ---------------------------------------------------------------------- @@ -124,6 +131,13 @@ proc queryHandler(server: HttpServer): RequestHandler = userId = uid role = r + # Rate limiting + let clientKey = request.headers["X-Forwarded-For"] + let rateLimitKey = if clientKey.len > 0: clientKey else: "http-global" + if not server.rateLimiter.allowRequest(rateLimitKey): + ctx.json(%*{"error": "Rate limit exceeded"}, 429) + return + let body = parseJson(request.body) if body == nil or "query" notin body: ctx.json(%*{"error": "Missing 'query' in request body"}, 400) diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 444f600..e4fe9e1 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -8,6 +8,11 @@ import std/os import std/endians import std/monotimes import std/locks +import std/nativesockets +when defined(windows): + from std/winlean import TCP_NODELAY +else: + from std/posix import TCP_NODELAY import config import logging import ../protocol/wire @@ -22,6 +27,7 @@ import ../core/disttxn import ../core/replication import ../core/sharding import ../core/gossip +import ../protocol/ratelimit import jwt as jwtlib type @@ -37,12 +43,11 @@ type clusterMembership*: ClusterMembership gossipProtocol*: GossipProtocol tls*: TLSContext + rateLimiter*: RateLimiter activeConnections*: int activeConnectionsLock*: Lock -proc newServer*(config: BaraConfig): Server = - let dataDir = config.dataDir / "server" - let db = newLSMTree(dataDir) +proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server = let ctx = newExecutionContext(db) ctx.txnManager = newTxnManager() var tls: TLSContext = nil @@ -85,19 +90,40 @@ proc newServer*(config: BaraConfig): Server = gp.onSuspect = proc(nodeId: string) {.gcsafe.} = cm.onNodeSuspect(nodeId) + # Initialize rate limiter + let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient) + result = Server(config: config, running: false, db: db, ctx: ctx, txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), replicationManager: newReplicationManager(), shardRouter: shardRouter, clusterMembership: cm, gossipProtocol: gp, - tls: tls) + tls: tls, + rateLimiter: rl) initLock(result.activeConnectionsLock) +proc newServer*(config: BaraConfig): Server = + let dataDir = config.dataDir / "server" + let db = newLSMTree(dataDir) + return newServerWithDb(config, db) + # ---------------------------------------------------------------------- # Wire Protocol Helpers # ---------------------------------------------------------------------- +proc bytesToString(data: seq[byte]): string = + ## Safely convert seq[byte] to string (copies data) + result = newString(data.len) + for i in 0.. data.len: raise newException(ValueError, "readUint32BE: index out of bounds") @@ -381,7 +407,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} let key = data[0.. 0: - server.db.put(key, cast[seq[byte]](value)) + server.db.put(key, stringToBytes(value)) else: server.db.delete(key) await client.send("ACK " & $lsn & "\n") @@ -435,27 +461,35 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} if not authenticated and header.kind != mkAuth: let err = makeErrorMessage(header.requestId, 401, "Authentication required") - await client.send(cast[string](err)) + await client.send(bytesToString(err)) continue + # Rate limiting for query messages + if header.kind in {mkQuery, mkQueryParams}: + let clientKey = "client-" & $clientId + if not server.rateLimiter.allowRequest(clientKey): + let err = serializeError(429, "Rate limit exceeded", header.requestId) + await client.send(bytesToString(err)) + continue + case header.kind of mkAuth: - let tokenStr = parseAuthMessage(cast[seq[byte]](payload)) + let tokenStr = parseAuthMessage(stringToBytes(payload)) let (valid, userId, role) = verifyToken(secret, tokenStr) if valid: authenticated = true connCtx.currentUser = userId connCtx.currentRole = role let okMsg = makeAuthOkMessage(header.requestId) - await client.send(cast[string](okMsg)) + await client.send(bytesToString(okMsg)) info("Client " & $clientId & " authenticated as " & userId) else: let err = makeErrorMessage(header.requestId, 403, "Invalid token") - await client.send(cast[string](err)) + await client.send(bytesToString(err)) of mkQuery: var pos = 0 - let queryStr = readString(cast[seq[byte]](payload), pos) + let queryStr = readString(stringToBytes(payload), pos) info("[" & $clientId & "] Query: " & queryStr) # Shard-aware routing: check if this node should handle the write @@ -471,7 +505,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} if localShards.len == 0: shardCheck = false let err = serializeError(3, "Node not assigned to any shard", header.requestId) - await client.send(cast[string](err)) + await client.send(bytesToString(err)) break if shardCheck: @@ -484,15 +518,15 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} if success: let dataMsg = serializeResult(result, header.requestId) - await client.send(cast[string](dataMsg)) + await client.send(bytesToString(dataMsg)) let completeMsg = serializeComplete(result.affectedRows, header.requestId) - await client.send(cast[string](completeMsg)) + await client.send(bytesToString(completeMsg)) else: let errorMsg = serializeError(1, errorMsg, header.requestId) - await client.send(cast[string](errorMsg)) + await client.send(bytesToString(errorMsg)) of mkQueryParams: - let (queryStr, params) = readQueryParamsMessage(cast[seq[byte]](payload)) + let (queryStr, params) = readQueryParamsMessage(stringToBytes(payload)) info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)") let startTicks = getMonoTime().ticks() @@ -504,26 +538,26 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} if success: let dataMsg = serializeResult(result, header.requestId) - await client.send(cast[string](dataMsg)) + await client.send(bytesToString(dataMsg)) let completeMsg = serializeComplete(result.affectedRows, header.requestId) - await client.send(cast[string](completeMsg)) + await client.send(bytesToString(completeMsg)) else: let errorMsg = serializeError(1, errorMsg, header.requestId) - await client.send(cast[string](errorMsg)) + await client.send(bytesToString(errorMsg)) of mkPing: var pongMsg = WireMessage( header: MessageHeader(kind: mkPong, length: 0, requestId: header.requestId), payload: @[], ) - await client.send(cast[string](serializeMessage(pongMsg))) + await client.send(bytesToString(serializeMessage(pongMsg))) of mkClose: break else: let errorMsg = serializeError(2, "Unsupported message kind: " & $header.kind, header.requestId) - await client.send(cast[string](errorMsg)) + await client.send(bytesToString(errorMsg)) except Exception as e: errorMsg("Client " & $clientId & " error: " & e.msg) @@ -537,6 +571,11 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} info("Client " & $clientId & " disconnected") client.close() +proc setTcpNoDelay(sock: AsyncSocket) = + ## Enable TCP_NODELAY using the correct protocol level (IPPROTO_TCP). + ## asyncnet.setSockOpt uses SOL_SOCKET by default which fails for OptNoDelay. + setSockOptInt(sock.getFd, IPPROTO_TCP.cint, TCP_NODELAY, 1) + proc run*(server: Server) {.async.} = server.running = true var clientId = 0 @@ -548,6 +587,7 @@ proc run*(server: Server) {.async.} = let sock = newAsyncSocket() sock.setSockOpt(OptReuseAddr, true) + setTcpNoDelay(sock) sock.bindAddr(Port(server.config.port), server.config.address) sock.listen() if server.config.tlsEnabled: @@ -556,6 +596,7 @@ proc run*(server: Server) {.async.} = info("BaraDB listening on " & server.config.address & ":" & $server.config.port) while server.running: let client = await sock.accept() + setTcpNoDelay(client) acquire(server.activeConnectionsLock) var shouldAccept = true try: diff --git a/src/barabadb/core/websocket.nim b/src/barabadb/core/websocket.nim index 36a6b3d..3229fb3 100644 --- a/src/barabadb/core/websocket.nim +++ b/src/barabadb/core/websocket.nim @@ -5,6 +5,11 @@ import std/strutils import std/tables import std/base64 import std/sets +import std/nativesockets +when defined(windows): + from std/winlean import TCP_NODELAY +else: + from std/posix import TCP_NODELAY import config import jwt as jwtlib @@ -307,16 +312,22 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} = inc server.nextId asyncCheck server.handleWsClient(client, server.nextId) +proc setTcpNoDelay(sock: AsyncSocket) = + ## Enable TCP_NODELAY using the correct protocol level (IPPROTO_TCP). + setSockOptInt(sock.getFd, IPPROTO_TCP.cint, TCP_NODELAY, 1) + proc run*(server: WsServer, port: int = 9471) {.async.} = server.running = true let sock = newAsyncSocket() sock.setSockOpt(OptReuseAddr, true) + setTcpNoDelay(sock) sock.bindAddr(Port(port)) sock.listen() echo "BaraDB WebSocket listening on port ", port while server.running: let client = await sock.accept() + setTcpNoDelay(client) asyncCheck server.handleConnection(client) proc stop*(server: WsServer) = diff --git a/src/baradadb.nim b/src/baradadb.nim index d61a845..ac288d3 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -230,17 +230,21 @@ proc main() = warn("Failed to generate self-signed certificate. TLS disabled.") config.tlsEnabled = false + # Shared LSMTree instance for both TCP and HTTP servers + let dataDir = config.dataDir / "server" + let sharedDb = newLSMTree(dataDir) + # Start HTTP server (blocking, multi-threaded via hunos) in background thread - var httpServer = newHttpServer(config) + var httpServer = newHttpServerWithDb(config, sharedDb) spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440 # Start background compaction loop - let cm = newCompactionManager(httpServer.db) + let cm = newCompactionManager(sharedDb) asyncCheck cm.startCompactionLoop() # Create TCP server (initialization is synchronous, run is async) let localId {.used.} = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port - var tcpServer = newServer(config) + var tcpServer = newServerWithDb(config, sharedDb) # Start Raft cluster if enabled if config.raftEnabled: @@ -251,11 +255,9 @@ proc main() = if cmd == "put": let parts = cast[string](data).split("\x00") if parts.len >= 2: - httpServer.db.put(parts[0], cast[seq[byte]](parts[1])) - tcpServer.db.put(parts[0], cast[seq[byte]](parts[1])) + sharedDb.put(parts[0], cast[seq[byte]](parts[1])) elif cmd == "delete": - httpServer.db.delete(cast[string](data)) - tcpServer.db.delete(cast[string](data)) + sharedDb.delete(cast[string](data)) # Wire RAFT ↔ DistTxn wireRaftDistTxn(raftNode, tcpServer)