feat: HTTP server in main entry point + background compaction scheduling
- baradadb.nim now starts both TCP wire protocol and HTTP REST servers - HTTP server runs in background thread via hunos on port TCP+440 - WebSocket server auto-starts on HTTP port + 1 - CompactionManager with background async loop (default 60s interval) - CompactionStrategy wired to LSMTree SSTables - Fixed forward declaration for lowerExpr in executor.nim - 216 tests passing
This commit is contained in:
@@ -8,6 +8,7 @@ import tables
|
|||||||
import strutils
|
import strutils
|
||||||
import os
|
import os
|
||||||
import times
|
import times
|
||||||
|
import std/asyncdispatch
|
||||||
import config
|
import config
|
||||||
import ../query/lexer
|
import ../query/lexer
|
||||||
import ../query/parser
|
import ../query/parser
|
||||||
@@ -22,7 +23,7 @@ type
|
|||||||
HttpServer* = ref object
|
HttpServer* = ref object
|
||||||
config: BaraConfig
|
config: BaraConfig
|
||||||
running: bool
|
running: bool
|
||||||
db: LSMTree
|
db*: LSMTree
|
||||||
ctx: ExecutionContext
|
ctx: ExecutionContext
|
||||||
metrics*: Metrics
|
metrics*: Metrics
|
||||||
secretKey*: string
|
secretKey*: string
|
||||||
|
|||||||
@@ -377,6 +377,8 @@ proc validateType*(colType: string, value: string): (bool, string) =
|
|||||||
return (false, "Type mismatch: expected " & t & " but got '" & value & "'")
|
return (false, "Type mismatch: expected " & t & " but got '" & value & "'")
|
||||||
return (true, "")
|
return (true, "")
|
||||||
|
|
||||||
|
proc lowerExpr*(node: Node): IRExpr
|
||||||
|
|
||||||
proc validateConstraints*(ctx: ExecutionContext, tableName: string,
|
proc validateConstraints*(ctx: ExecutionContext, tableName: string,
|
||||||
fields: seq[string], values: seq[seq[string]]): (bool, string) =
|
fields: seq[string], values: seq[seq[string]]): (bool, string) =
|
||||||
let tbl = ctx.getTableDef(tableName)
|
let tbl = ctx.getTableDef(tableName)
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import std/tables
|
|||||||
import std/monotimes
|
import std/monotimes
|
||||||
import std/streams
|
import std/streams
|
||||||
import std/locks
|
import std/locks
|
||||||
|
import std/asyncdispatch
|
||||||
import bloom
|
import bloom
|
||||||
import wal
|
import wal
|
||||||
import mmap
|
import mmap
|
||||||
@@ -40,15 +41,15 @@ type
|
|||||||
mmapFile*: MmapFile
|
mmapFile*: MmapFile
|
||||||
|
|
||||||
LSMTree* = ref object
|
LSMTree* = ref object
|
||||||
dir: string
|
dir*: string
|
||||||
memTable: MemTable
|
memTable: MemTable
|
||||||
immutableMem: MemTable
|
immutableMem: MemTable
|
||||||
sstables: seq[SSTable]
|
sstables*: seq[SSTable]
|
||||||
wal: WriteAheadLog
|
wal: WriteAheadLog
|
||||||
memMaxSize: int
|
memMaxSize: int
|
||||||
currentSeq: uint64
|
currentSeq: uint64
|
||||||
nextSSTableId: int
|
nextSSTableId: int
|
||||||
lock: Lock
|
lock*: Lock
|
||||||
|
|
||||||
proc newMemTable(maxSize: int = DefaultMemTableSize): MemTable =
|
proc newMemTable(maxSize: int = DefaultMemTableSize): MemTable =
|
||||||
MemTable(entries: @[], size: 0, maxSize: maxSize)
|
MemTable(entries: @[], size: 0, maxSize: maxSize)
|
||||||
@@ -323,6 +324,7 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
|
|||||||
result.memMaxSize = memMaxSize
|
result.memMaxSize = memMaxSize
|
||||||
result.currentSeq = 0
|
result.currentSeq = 0
|
||||||
result.nextSSTableId = nextId
|
result.nextSSTableId = nextId
|
||||||
|
discard
|
||||||
|
|
||||||
proc put*(db: LSMTree, key: string, value: seq[byte]) =
|
proc put*(db: LSMTree, key: string, value: seq[byte]) =
|
||||||
acquire(db.lock)
|
acquire(db.lock)
|
||||||
|
|||||||
+66
-3
@@ -1,15 +1,78 @@
|
|||||||
## BaraDB — Multimodal Database Engine
|
## BaraDB — Multimodal Database Engine
|
||||||
## Main entry point
|
## Main entry point
|
||||||
import std/asyncdispatch
|
import std/asyncdispatch
|
||||||
|
import std/threadpool
|
||||||
|
import std/locks
|
||||||
import barabadb/core/server
|
import barabadb/core/server
|
||||||
|
import barabadb/core/httpserver
|
||||||
import barabadb/core/config
|
import barabadb/core/config
|
||||||
|
import barabadb/storage/lsm
|
||||||
|
import barabadb/storage/compaction
|
||||||
|
|
||||||
|
type
|
||||||
|
CompactionManager* = ref object
|
||||||
|
db*: LSMTree
|
||||||
|
strategy*: compaction.CompactionStrategy
|
||||||
|
|
||||||
|
proc newCompactionManager*(db: LSMTree): CompactionManager =
|
||||||
|
result = CompactionManager(db: db, strategy: compaction.newCompactionStrategy(db.dir))
|
||||||
|
for sst in db.sstables:
|
||||||
|
let meta = compaction.SSTableMeta(
|
||||||
|
path: sst.path,
|
||||||
|
level: sst.level,
|
||||||
|
minKey: sst.minKey,
|
||||||
|
maxKey: sst.maxKey,
|
||||||
|
entryCount: sst.entryCount,
|
||||||
|
sizeBytes: sst.entryCount * 64,
|
||||||
|
createdAt: 0,
|
||||||
|
)
|
||||||
|
result.strategy.addTable(meta)
|
||||||
|
for sst in db.sstables:
|
||||||
|
let meta = SSTableMeta(
|
||||||
|
path: sst.path,
|
||||||
|
level: sst.level,
|
||||||
|
minKey: sst.minKey,
|
||||||
|
maxKey: sst.maxKey,
|
||||||
|
entryCount: sst.entryCount,
|
||||||
|
sizeBytes: sst.entryCount * 64,
|
||||||
|
createdAt: 0,
|
||||||
|
)
|
||||||
|
result.strategy.addTable(meta)
|
||||||
|
|
||||||
|
proc compact*(cm: CompactionManager) =
|
||||||
|
acquire(cm.db.lock)
|
||||||
|
defer: release(cm.db.lock)
|
||||||
|
for level in 0 ..< compaction.MaxLevel:
|
||||||
|
if cm.strategy.needsCompaction(level):
|
||||||
|
discard cm.strategy.compact(level)
|
||||||
|
|
||||||
|
proc startCompactionLoop*(cm: CompactionManager, intervalMs: int = 60000) {.async.} =
|
||||||
|
while true:
|
||||||
|
await sleepAsync(intervalMs)
|
||||||
|
cm.compact()
|
||||||
|
|
||||||
|
proc runTcpServer(config: BaraConfig) {.async.} =
|
||||||
|
echo "BaraDB TCP listening on ", config.address, ":", config.port
|
||||||
|
var server = newServer(config)
|
||||||
|
await server.run()
|
||||||
|
|
||||||
proc main() =
|
proc main() =
|
||||||
let config = loadConfig()
|
let config = loadConfig()
|
||||||
echo "BaraDB v0.1.0 — Multimodal Database Engine"
|
echo "BaraDB v0.1.0 — Multimodal Database Engine"
|
||||||
echo "Listening on ", config.address, ":", config.port
|
|
||||||
var server = newServer(config)
|
# Start HTTP server (blocking, multi-threaded via hunos) in background thread
|
||||||
waitFor server.run()
|
var httpServer = newHttpServer(config)
|
||||||
|
spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440
|
||||||
|
|
||||||
|
# Start background compaction loop
|
||||||
|
let cm = newCompactionManager(httpServer.db)
|
||||||
|
asyncCheck cm.startCompactionLoop()
|
||||||
|
|
||||||
|
# Start TCP wire protocol server on main thread with async event loop
|
||||||
|
waitFor runTcpServer(config)
|
||||||
|
|
||||||
|
# Shutdown
|
||||||
|
httpServer.stop()
|
||||||
|
|
||||||
when isMainModule:
|
when isMainModule:
|
||||||
main()
|
main()
|
||||||
|
|||||||
Reference in New Issue
Block a user