mirror of
https://github.com/logos-messaging/nim-sds.git
synced 2026-07-02 22:10:13 +00:00
feat: adopt nim-ffi v0.1.5 (macro-based snake_case FFI) (#81)
This commit is contained in:
Ivan FB
GitHub
parent
c88205dc82
commit
3aab2fbee0
24
CLAUDE.md
24
CLAUDE.md
@ -95,21 +95,23 @@ flake.nix / Makefile # Reproducible cross-platform build system
|
||||
|
||||
## FFI API (`library/libsds.nim`)
|
||||
|
||||
The C API wraps `ReliabilityManager` behind an opaque `SdsContext` handle:
|
||||
The C API (snake_case, generated by the nim-ffi `{.ffiCtor.}`/`{.ffi.}`/`{.ffiDtor.}`
|
||||
macros) wraps `ReliabilityManager` behind an opaque context handle. Requests and
|
||||
responses are marshalled as JSON; see `library/libsds.h`.
|
||||
|
||||
| Export | Maps to |
|
||||
|---|---|
|
||||
| `SdsNewReliabilityManager` | Create context |
|
||||
| `SdsWrapOutgoingMessage` | `wrapOutgoingMessage` |
|
||||
| `SdsUnwrapReceivedMessage` | `unwrapReceivedMessage` |
|
||||
| `SdsMarkDependenciesMet` | Notify buffered-message dependencies satisfied |
|
||||
| `SdsSetEventCallback` | Register event handler (JSON payloads) |
|
||||
| `SdsSetRetrievalHintProvider` | Register hint-provider callback |
|
||||
| `SdsStartPeriodicTasks` | Start periodic sync loop |
|
||||
| `SdsCleanupReliabilityManager` | Free context |
|
||||
| `SdsResetReliabilityManager` | Reset state without freeing |
|
||||
| `sds_create` | Create context (`{.ffiCtor.}`; `configJson` carries `participantId`) |
|
||||
| `sds_set_event_callback` | Register event handler (JSON payloads) |
|
||||
| `sds_set_retrieval_hint_provider` | Register hint-provider callback (hand-written) |
|
||||
| `sds_wrap_outgoing_message` | `wrapOutgoingMessage` |
|
||||
| `sds_unwrap_received_message` | `unwrapReceivedMessage` |
|
||||
| `sds_mark_dependencies_met` | Notify buffered-message dependencies satisfied |
|
||||
| `sds_start_periodic_tasks` | Start periodic sync loop |
|
||||
| `sds_reset` | Reset state without freeing |
|
||||
| `sds_destroy` | Recycle/free context (`{.ffiDtor.}`) |
|
||||
|
||||
Each `SdsContext` runs a dedicated Chronos async loop on a worker thread; application threads communicate with it via SPSC channels.
|
||||
Each context runs a dedicated Chronos async loop on a worker thread; application threads communicate with it via SPSC channels.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -1,8 +1,10 @@
|
||||
|
||||
// Generated manually and inspired by the one generated by the Nim Compiler.
|
||||
// In order to see the header file generated by Nim just run `make libsds`
|
||||
// from the root repo folder and the header should be created in
|
||||
// nimcache/release/libsds/libsds.h
|
||||
// C API for libsds, built on the nim-ffi framework.
|
||||
//
|
||||
// Parameters and results are marshalled as JSON: each request/response struct
|
||||
// in library/libsds.nim is a JSON object, passed in via the `*Json` cstring
|
||||
// argument and returned to the callback as a JSON string. Binary fields
|
||||
// (message bytes) are JSON arrays of byte values.
|
||||
#ifndef __libsds__
|
||||
#define __libsds__
|
||||
|
||||
@ -18,51 +20,52 @@
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
// Result/event callback. `msg` is the (JSON) payload of length `len`.
|
||||
// callerRet is one of the RET_* codes above.
|
||||
typedef void (*SdsCallBack) (int callerRet, const char* msg, size_t len, void* userData);
|
||||
|
||||
// Synchronous provider invoked by SDS-R to fetch a retrieval hint for a
|
||||
// message id. The implementation allocates `*hint` (and sets `*hintLen`); the
|
||||
// library takes ownership and frees it with deallocShared.
|
||||
typedef void (*SdsRetrievalHintProvider) (const char* messageId, char** hint, size_t* hintLen, void* userData);
|
||||
|
||||
|
||||
// --- Core API Functions ---
|
||||
|
||||
|
||||
void* SdsNewReliabilityManager(SdsCallBack callback, void* userData);
|
||||
// Create a context + ReliabilityManager. configJson: {"participantId":"..."}
|
||||
// (empty participantId disables SDS-R). Returns the context handle, or NULL on
|
||||
// failure. The callback also fires on async completion.
|
||||
void* sds_create(const char* configJson, SdsCallBack callback, void* userData);
|
||||
|
||||
void SdsSetEventCallback(void* ctx, SdsCallBack callback, void* userData);
|
||||
// Register the event callback (message_ready, message_sent,
|
||||
// missing_dependencies, periodic_sync, repair_ready). Payloads are JSON.
|
||||
void sds_set_event_callback(void* ctx, SdsCallBack callback, void* userData);
|
||||
|
||||
void SdsSetRetrievalHintProvider(void* ctx, SdsRetrievalHintProvider callback, void* userData);
|
||||
// Register the retrieval-hint provider used by SDS-R.
|
||||
int sds_set_retrieval_hint_provider(void* ctx, SdsRetrievalHintProvider callback, void* userData);
|
||||
|
||||
int SdsCleanupReliabilityManager(void* ctx, SdsCallBack callback, void* userData);
|
||||
// reqJson: {"message":[..bytes..],"messageId":"..","channelId":".."}
|
||||
// Result JSON: {"message":[..bytes..]}
|
||||
int sds_wrap_outgoing_message(void* ctx, SdsCallBack callback, void* userData, const char* reqJson);
|
||||
|
||||
int SdsResetReliabilityManager(void* ctx, SdsCallBack callback, void* userData);
|
||||
// reqJson: {"message":[..bytes..]}
|
||||
// Result JSON: {"message":[..],"channelId":"..","missingDeps":[{"messageId":"..","retrievalHint":"<base64>"}]}
|
||||
int sds_unwrap_received_message(void* ctx, SdsCallBack callback, void* userData, const char* reqJson);
|
||||
|
||||
int SdsWrapOutgoingMessage(void* ctx,
|
||||
void* message,
|
||||
size_t messageLen,
|
||||
const char* messageId,
|
||||
const char* channelId,
|
||||
SdsCallBack callback,
|
||||
void* userData);
|
||||
// reqJson: {"messageIds":["..",".."],"channelId":".."}
|
||||
int sds_mark_dependencies_met(void* ctx, SdsCallBack callback, void* userData, const char* reqJson);
|
||||
|
||||
int SdsUnwrapReceivedMessage(void* ctx,
|
||||
void* message,
|
||||
size_t messageLen,
|
||||
SdsCallBack callback,
|
||||
void* userData);
|
||||
int sds_reset(void* ctx, SdsCallBack callback, void* userData);
|
||||
|
||||
int SdsMarkDependenciesMet(void* ctx,
|
||||
char** messageIDs,
|
||||
size_t count,
|
||||
const char* channelId,
|
||||
SdsCallBack callback,
|
||||
void* userData);
|
||||
|
||||
int SdsStartPeriodicTasks(void* ctx, SdsCallBack callback, void* userData);
|
||||
int sds_start_periodic_tasks(void* ctx, SdsCallBack callback, void* userData);
|
||||
|
||||
// Tear down the context created by sds_create.
|
||||
int sds_destroy(void* ctx, SdsCallBack callback, void* userData);
|
||||
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
|
||||
#endif /* __libsds__ */
|
||||
#endif /* __libsds__ */
|
||||
|
||||
@ -1,4 +1,16 @@
|
||||
import std/[strutils, sequtils, json, base64, locks]
|
||||
## C-compatible FFI wrapper around the SDS ReliabilityManager.
|
||||
##
|
||||
## Built on the `nim-ffi` package's high-level macros: `declareLibrary` emits the
|
||||
## bootstrap + `sds_set_event_callback`; `{.ffiCtor.}`/`{.ffi.}`/`{.ffiDtor.}`
|
||||
## generate the C entry points, marshalling parameters and return values as JSON.
|
||||
## Exported C names are snake_case (`sds_wrap_outgoing_message`, …); see
|
||||
## `library/libsds.h`. The Go bindings (sds-go-bindings) must match this API.
|
||||
##
|
||||
## The one exception is `sds_set_retrieval_hint_provider`: it takes a C function
|
||||
## pointer, which has no sensible JSON representation, so it is hand-written and
|
||||
## dispatched to the worker thread to store the provider in a thread-local.
|
||||
|
||||
import std/[base64, json]
|
||||
import ffi
|
||||
import sds
|
||||
import ./events/[
|
||||
@ -6,467 +18,217 @@ import ./events/[
|
||||
json_periodic_sync_event, json_repair_ready_event,
|
||||
]
|
||||
|
||||
# Emit the library bootstrap: the {.exported.}/{.callback.} pragmas, the
|
||||
# `-fPIC`/soname linker flags, the `libsdsNimMain` import and the
|
||||
# `initializeLibrary()` proc the exported entry points call on every hop.
|
||||
declareLibraryBase("sds")
|
||||
|
||||
# C callback typedefs (mirrors libsds.h). `SdsCallBack` is structurally the
|
||||
# nim-ffi `FFICallBack`; the alias keeps the exported signatures readable.
|
||||
type SdsCallBack* = FFICallBack
|
||||
# Bootstrap (pragmas, linker flags, libsdsNimMain, initializeLibrary) plus the
|
||||
# `sds_set_event_callback(ctx, callback, userData)` C export. It also declares
|
||||
# the per-type `ReliabilityManagerFFIPool` used by the hand-written entry point
|
||||
# below (the ffiCtor/ffiDtor macros declare it too, guarded by `when not
|
||||
# declared`).
|
||||
declareLibrary("sds", ReliabilityManager)
|
||||
|
||||
type SdsRetrievalHintProvider* = proc(
|
||||
messageId: cstring, hint: ptr cstring, hintLen: ptr csize_t, userData: pointer
|
||||
) {.cdecl, gcsafe, raises: [].}
|
||||
|
||||
# One pool per library type; the macros that would normally declare it
|
||||
# (ffiCtor/ffiDtor) are not used here because we hand-write the entry points
|
||||
# to preserve the exact C ABI, so we declare it explicitly.
|
||||
var ReliabilityManagerFFIPool: FFIContextPool[ReliabilityManager]
|
||||
|
||||
# registerReqFFI inspects each request field's type via `$node`, which only
|
||||
# handles plain identifiers — a bracketed `SharedSeq[byte]` makes it choke. The
|
||||
# aliases give the generated request structs non-bracketed field types.
|
||||
type
|
||||
SdsSharedBytes = SharedSeq[byte]
|
||||
SdsSharedCstrs = SharedSeq[cstring]
|
||||
# The active retrieval-hint provider, stored per worker thread (one thread per
|
||||
# context). Set by sds_set_retrieval_hint_provider via a dispatched request so
|
||||
# the write lands on the worker thread, where the manager's hint closure reads
|
||||
# it during message processing.
|
||||
var sdsRetrievalHintCb {.threadvar.}: pointer
|
||||
var sdsRetrievalHintUserData {.threadvar.}: pointer
|
||||
|
||||
################################################################################
|
||||
### Retrieval-hint provider registry
|
||||
###
|
||||
### The retrieval-hint provider is a synchronous request/response callback
|
||||
### (the C side returns bytes inline), so it does not fit the fire-and-forget
|
||||
### event model. nim-ffi's FFIContext has no slot for it, so we keep a small
|
||||
### per-context registry here. A fixed array of plain (non-GC) records keeps
|
||||
### the lookup callable from the {.gcsafe.} hint closure running on the FFI
|
||||
### thread.
|
||||
### JSON-marshalled request/response types
|
||||
|
||||
type RetrievalHintSlot = object
|
||||
ctx: pointer
|
||||
cb: pointer
|
||||
userData: pointer
|
||||
type SdsConfig* {.ffi.} = object
|
||||
participantId: string ## empty disables SDS-R (see newReliabilityManager)
|
||||
|
||||
var retrievalHintSlots: array[MaxFFIContexts, RetrievalHintSlot]
|
||||
var retrievalHintsLock: Lock
|
||||
retrievalHintsLock.initLock()
|
||||
type SdsWrapRequest* {.ffi.} = object
|
||||
message: seq[byte]
|
||||
messageId: string
|
||||
channelId: string
|
||||
|
||||
proc setRetrievalHint(ctx: pointer, cb: pointer, userData: pointer) =
|
||||
withLock retrievalHintsLock:
|
||||
var free = -1
|
||||
for i in 0 ..< MaxFFIContexts:
|
||||
if retrievalHintSlots[i].ctx == ctx:
|
||||
retrievalHintSlots[i] = RetrievalHintSlot(ctx: ctx, cb: cb, userData: userData)
|
||||
return
|
||||
if free < 0 and retrievalHintSlots[i].ctx.isNil:
|
||||
free = i
|
||||
if free >= 0:
|
||||
retrievalHintSlots[free] = RetrievalHintSlot(ctx: ctx, cb: cb, userData: userData)
|
||||
type SdsWrapResponse* {.ffi.} = object
|
||||
message: seq[byte]
|
||||
|
||||
proc getRetrievalHint(ctx: pointer): tuple[cb: pointer, userData: pointer] {.gcsafe.} =
|
||||
withLock retrievalHintsLock:
|
||||
for i in 0 ..< MaxFFIContexts:
|
||||
if retrievalHintSlots[i].ctx == ctx:
|
||||
return (retrievalHintSlots[i].cb, retrievalHintSlots[i].userData)
|
||||
return (nil, nil)
|
||||
type SdsUnwrapRequest* {.ffi.} = object
|
||||
message: seq[byte]
|
||||
|
||||
proc clearRetrievalHint(ctx: pointer) =
|
||||
withLock retrievalHintsLock:
|
||||
for i in 0 ..< MaxFFIContexts:
|
||||
if retrievalHintSlots[i].ctx == ctx:
|
||||
retrievalHintSlots[i] = RetrievalHintSlot()
|
||||
return
|
||||
type SdsMarkDependenciesRequest* {.ffi.} = object
|
||||
messageIds: seq[string]
|
||||
channelId: string
|
||||
|
||||
################################################################################
|
||||
### Shared-memory copy helpers
|
||||
### Constructor — creates the FFI context and the ReliabilityManager.
|
||||
###
|
||||
### Request payloads carrying binary/pointer data must be deep-copied into
|
||||
### shared memory on the caller thread, because the FFI thread acks receipt
|
||||
### before it reads the payload — the caller may free its buffer in between.
|
||||
### cstring fields are deep-copied by the generated `ffiNewReq`; raw byte and
|
||||
### `char**` arrays are not, so we copy them here.
|
||||
### The event closures run on the worker thread and forward JSON payloads to the
|
||||
### C callback registered via sds_set_event_callback (dispatchFfiEvent reads the
|
||||
### per-thread callback state, so no context handle is needed here).
|
||||
|
||||
proc copyToSharedSeqByte(p: pointer, len: int): SharedSeq[byte] =
|
||||
if p.isNil or len <= 0:
|
||||
return (cast[ptr UncheckedArray[byte]](nil), 0)
|
||||
let data = allocShared(len)
|
||||
copyMem(data, p, len)
|
||||
return (cast[ptr UncheckedArray[byte]](data), len)
|
||||
proc sdsCreate*(
|
||||
config: SdsConfig
|
||||
): Future[Result[ReliabilityManager, string]] {.ffiCtor.} =
|
||||
# The ctor body runs on the (possibly recycled) worker thread. Drop any
|
||||
# retrieval-hint provider left over from a previous owner of this thread so a
|
||||
# stale C function pointer is never invoked.
|
||||
sdsRetrievalHintCb = nil
|
||||
sdsRetrievalHintUserData = nil
|
||||
|
||||
proc copyToSharedSeqCstr(p: pointer, count: int): SharedSeq[cstring] =
|
||||
if p.isNil or count <= 0:
|
||||
return (cast[ptr UncheckedArray[cstring]](nil), 0)
|
||||
let data = cast[ptr UncheckedArray[cstring]](allocShared(sizeof(cstring) * count))
|
||||
let src = cast[ptr UncheckedArray[cstring]](p)
|
||||
for i in 0 ..< count:
|
||||
data[i] = src[i].alloc()
|
||||
return (data, count)
|
||||
let rm = newReliabilityManager(participantId = config.participantId.SdsParticipantID).valueOr:
|
||||
error "Failed creating reliability manager", error = error
|
||||
return err("Failed creating reliability manager: " & $error)
|
||||
|
||||
proc freeSharedSeqCstr(s: var SharedSeq[cstring]) =
|
||||
if not s.data.isNil():
|
||||
for i in 0 ..< s.len:
|
||||
if not s.data[i].isNil:
|
||||
deallocShared(s.data[i])
|
||||
deallocShared(s.data)
|
||||
s.len = 0
|
||||
|
||||
################################################################################
|
||||
### Event callbacks
|
||||
###
|
||||
### These build the AppCallbacks closures handed to the ReliabilityManager.
|
||||
### They run on the FFI worker thread and forward JSON event payloads to the
|
||||
### C callback registered via SdsSetEventCallback (stored on the context).
|
||||
|
||||
proc onMessageReady(ctx: ptr FFIContext[ReliabilityManager]): MessageReadyCallback =
|
||||
return proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} =
|
||||
callEventCallback(ctx, "onMessageReady"):
|
||||
let messageReadyCb = proc(
|
||||
messageId: SdsMessageID, channelId: SdsChannelID
|
||||
) {.gcsafe.} =
|
||||
dispatchFfiEvent("message_ready"):
|
||||
$JsonMessageReadyEvent.new(messageId, channelId)
|
||||
|
||||
proc onMessageSent(ctx: ptr FFIContext[ReliabilityManager]): MessageSentCallback =
|
||||
return proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} =
|
||||
callEventCallback(ctx, "onMessageSent"):
|
||||
let messageSentCb = proc(
|
||||
messageId: SdsMessageID, channelId: SdsChannelID
|
||||
) {.gcsafe.} =
|
||||
dispatchFfiEvent("message_sent"):
|
||||
$JsonMessageSentEvent.new(messageId, channelId)
|
||||
|
||||
proc onMissingDependencies(
|
||||
ctx: ptr FFIContext[ReliabilityManager]
|
||||
): MissingDependenciesCallback =
|
||||
return proc(
|
||||
let missingDependenciesCb = proc(
|
||||
messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID
|
||||
) {.gcsafe.} =
|
||||
callEventCallback(ctx, "onMissingDependencies"):
|
||||
dispatchFfiEvent("missing_dependencies"):
|
||||
$JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId)
|
||||
|
||||
proc onPeriodicSync(ctx: ptr FFIContext[ReliabilityManager]): PeriodicSyncCallback =
|
||||
return proc() {.gcsafe.} =
|
||||
callEventCallback(ctx, "onPeriodicSync"):
|
||||
let periodicSyncCb = proc() {.gcsafe.} =
|
||||
dispatchFfiEvent("periodic_sync"):
|
||||
$JsonPeriodicSyncEvent.new()
|
||||
|
||||
proc onRepairReady(ctx: ptr FFIContext[ReliabilityManager]): RepairReadyCallback =
|
||||
return proc(message: seq[byte], channelId: SdsChannelID) {.gcsafe.} =
|
||||
callEventCallback(ctx, "onRepairReady"):
|
||||
let repairReadyCb = proc(message: seq[byte], channelId: SdsChannelID) {.gcsafe.} =
|
||||
dispatchFfiEvent("repair_ready"):
|
||||
$JsonRepairReadyEvent.new(message, channelId)
|
||||
|
||||
proc onRetrievalHint(ctx: ptr FFIContext[ReliabilityManager]): RetrievalHintProvider =
|
||||
return proc(messageId: SdsMessageID): seq[byte] {.gcsafe.} =
|
||||
let (cb, userData) = getRetrievalHint(cast[pointer](ctx))
|
||||
if cb.isNil():
|
||||
let retrievalHintProvider = proc(messageId: SdsMessageID): seq[byte] {.gcsafe.} =
|
||||
if sdsRetrievalHintCb.isNil():
|
||||
return @[]
|
||||
|
||||
var hint: cstring
|
||||
var hintLen: csize_t
|
||||
cast[SdsRetrievalHintProvider](cb)(
|
||||
messageId.cstring, addr hint, addr hintLen, userData
|
||||
cast[SdsRetrievalHintProvider](sdsRetrievalHintCb)(
|
||||
messageId.cstring, addr hint, addr hintLen, sdsRetrievalHintUserData
|
||||
)
|
||||
|
||||
if not hint.isNil() and hintLen > 0:
|
||||
var hintBytes = newSeq[byte](hintLen)
|
||||
copyMem(addr hintBytes[0], hint, hintLen)
|
||||
deallocShared(hint)
|
||||
return hintBytes
|
||||
|
||||
return @[]
|
||||
|
||||
################################################################################
|
||||
### Request handlers (executed on the FFI worker thread)
|
||||
await rm.setCallbacks(
|
||||
messageReadyCb, messageSentCb, missingDependenciesCb, periodicSyncCb,
|
||||
retrievalHintProvider, repairReadyCb,
|
||||
)
|
||||
|
||||
registerReqFFI(SdsCreateRmReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(): Future[Result[string, string]] {.async.} =
|
||||
# TODO: thread `participantId` through SdsNewReliabilityManager FFI input
|
||||
# and remove this hardcoded "". Empty id silently disables SDS-R; this is
|
||||
# acceptable as a temporary FFI-only fallback until sds-go-bindings and
|
||||
# logos-delivery's C-side caller are updated to supply the identity.
|
||||
let rm = newReliabilityManager(participantId = "".SdsParticipantID).valueOr:
|
||||
error "Failed creating reliability manager", error = error
|
||||
return err("Failed creating reliability manager: " & $error)
|
||||
|
||||
await rm.setCallbacks(
|
||||
onMessageReady(ctx), onMessageSent(ctx), onMissingDependencies(ctx),
|
||||
onPeriodicSync(ctx), onRetrievalHint(ctx), onRepairReady(ctx),
|
||||
)
|
||||
|
||||
# nim-ffi frees myLib on recycle, so (re)allocate it here.
|
||||
if ctx.myLib.isNil():
|
||||
ctx.myLib = createShared(ReliabilityManager)
|
||||
ctx.myLib[] = rm
|
||||
return ok("")
|
||||
|
||||
registerReqFFI(SdsResetRmReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(): Future[Result[string, string]] {.async.} =
|
||||
(await resetReliabilityManager(ctx.myLib[])).isOkOr:
|
||||
error "RESET_RELIABILITY_MANAGER failed", error = error
|
||||
return err("error processing RESET_RELIABILITY_MANAGER request: " & $error)
|
||||
return ok("")
|
||||
|
||||
registerReqFFI(SdsStartPeriodicTasksReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(): Future[Result[string, string]] {.async.} =
|
||||
ctx.myLib[].startPeriodicTasks()
|
||||
return ok("")
|
||||
|
||||
registerReqFFI(SdsWrapMessageReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(
|
||||
message: SdsSharedBytes, messageId: cstring, channelId: cstring
|
||||
): Future[Result[string, string]] {.async.} =
|
||||
var msg = message
|
||||
defer:
|
||||
deallocSharedSeq(msg)
|
||||
|
||||
let wrappedMessage = (
|
||||
await wrapOutgoingMessage(ctx.myLib[], message.toSeq(), $messageId, $channelId)
|
||||
).valueOr:
|
||||
error "WRAP_MESSAGE failed", error = error
|
||||
return err("error processing WRAP_MESSAGE request: " & $error)
|
||||
|
||||
# returns a comma-separated string of bytes
|
||||
return ok(wrappedMessage.mapIt($it).join(","))
|
||||
|
||||
registerReqFFI(SdsUnwrapMessageReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(message: SdsSharedBytes): Future[Result[string, string]] {.async.} =
|
||||
var msg = message
|
||||
defer:
|
||||
deallocSharedSeq(msg)
|
||||
|
||||
let (unwrappedMessage, missingDeps, extractedChannelId) = (
|
||||
await unwrapReceivedMessage(ctx.myLib[], message.toSeq())
|
||||
).valueOr:
|
||||
return err("error processing UNWRAP_MESSAGE request: " & $error)
|
||||
|
||||
# return the result as a json string
|
||||
var node = newJObject()
|
||||
node["message"] = %*unwrappedMessage
|
||||
node["channelId"] = %*extractedChannelId
|
||||
var missingDepsNode = newJArray()
|
||||
for dep in missingDeps:
|
||||
var depNode = newJObject()
|
||||
depNode["messageId"] = %*dep.messageId
|
||||
depNode["retrievalHint"] = %*encode(dep.retrievalHint)
|
||||
missingDepsNode.add(depNode)
|
||||
node["missingDeps"] = missingDepsNode
|
||||
return ok($node)
|
||||
|
||||
registerReqFFI(SdsMarkDepsReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(
|
||||
messageIds: SdsSharedCstrs, channelId: cstring
|
||||
): Future[Result[string, string]] {.async.} =
|
||||
var ids = messageIds
|
||||
defer:
|
||||
freeSharedSeqCstr(ids)
|
||||
|
||||
let messageIdSeq = ids.toSeq().mapIt($it)
|
||||
(await markDependenciesMet(ctx.myLib[], messageIdSeq, $channelId)).isOkOr:
|
||||
error "MARK_DEPENDENCIES_MET failed", error = error
|
||||
return err("error processing MARK_DEPENDENCIES_MET request: " & $error)
|
||||
return ok("")
|
||||
return ok(rm)
|
||||
|
||||
################################################################################
|
||||
### Dispatch helper
|
||||
###
|
||||
### Sends a request to the FFI worker thread and returns RET_OK/RET_ERR,
|
||||
### reporting any failure through the callback. The try/except keeps the
|
||||
### exported entry points `raises: []` (sendRequestToFFIThread can raise),
|
||||
### which `processReq` alone would not guarantee.
|
||||
### Async methods — each runs its body on the worker thread.
|
||||
|
||||
template dispatchReq(
|
||||
ctx: untyped, callback: FFICallBack, userData: pointer, reqExpr: untyped
|
||||
) =
|
||||
let sendRes =
|
||||
try:
|
||||
ffi_context.sendRequestToFFIThread(ctx, reqExpr)
|
||||
except Exception as exc:
|
||||
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
|
||||
if sendRes.isErr():
|
||||
let m = "libsds error: " & sendRes.error
|
||||
callback(RET_ERR, unsafeAddr m[0], cast[csize_t](m.len), userData)
|
||||
return RET_ERR
|
||||
return RET_OK
|
||||
proc sdsWrapOutgoingMessage*(
|
||||
rm: ReliabilityManager, req: SdsWrapRequest
|
||||
): Future[Result[SdsWrapResponse, string]] {.ffi.} =
|
||||
let wrapped = (
|
||||
await wrapOutgoingMessage(rm, req.message, req.messageId, req.channelId)
|
||||
).valueOr:
|
||||
error "WRAP_MESSAGE failed", error = error
|
||||
return err("error processing wrap request: " & $error)
|
||||
return ok(SdsWrapResponse(message: wrapped))
|
||||
|
||||
proc sdsUnwrapReceivedMessage*(
|
||||
rm: ReliabilityManager, req: SdsUnwrapRequest
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
# The response carries nested objects (missingDeps) which the framework's
|
||||
# object serializer cannot emit, so the JSON is built by hand and returned as
|
||||
# a string. Shape matches the legacy unwrap response.
|
||||
let (unwrapped, missingDeps, channelId) = (
|
||||
await unwrapReceivedMessage(rm, req.message)
|
||||
).valueOr:
|
||||
return err("error processing unwrap request: " & $error)
|
||||
|
||||
var node = newJObject()
|
||||
node["message"] = %*unwrapped
|
||||
node["channelId"] = %*channelId
|
||||
var missingDepsNode = newJArray()
|
||||
for dep in missingDeps:
|
||||
var depNode = newJObject()
|
||||
depNode["messageId"] = %*dep.messageId
|
||||
depNode["retrievalHint"] = %*encode(dep.retrievalHint)
|
||||
missingDepsNode.add(depNode)
|
||||
node["missingDeps"] = missingDepsNode
|
||||
return ok($node)
|
||||
|
||||
proc sdsMarkDependenciesMet*(
|
||||
rm: ReliabilityManager, req: SdsMarkDependenciesRequest
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(await markDependenciesMet(rm, req.messageIds, req.channelId)).isOkOr:
|
||||
error "MARK_DEPENDENCIES_MET failed", error = error
|
||||
return err("error processing mark-dependencies request: " & $error)
|
||||
return ok("")
|
||||
|
||||
proc sdsReset*(rm: ReliabilityManager): Future[Result[string, string]] {.ffi.} =
|
||||
(await resetReliabilityManager(rm)).isOkOr:
|
||||
error "RESET failed", error = error
|
||||
return err("error processing reset request: " & $error)
|
||||
return ok("")
|
||||
|
||||
proc sdsStartPeriodicTasks*(
|
||||
rm: ReliabilityManager
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
# The empty await forces the macro down its async path so the body runs on the
|
||||
# worker thread — startPeriodicTasks schedules futures on that thread's loop.
|
||||
await sleepAsync(chronos.milliseconds(0))
|
||||
rm.startPeriodicTasks()
|
||||
return ok("")
|
||||
|
||||
################################################################################
|
||||
### Exported C entry points (called from the application thread)
|
||||
###
|
||||
### Signatures must match library/libsds.h exactly. Each one validates the
|
||||
### context against the pool (rejecting nil/dangling pointers at the boundary),
|
||||
### checks the callback, deep-copies any pointer payloads into shared memory,
|
||||
### then dispatches a request to the FFI worker thread.
|
||||
### Destructor — runs library cleanup then tears down the FFI context.
|
||||
|
||||
proc SdsNewReliabilityManager(
|
||||
callback: FFICallBack, userData: pointer
|
||||
): pointer {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
proc sdsDestroy*(rm: ReliabilityManager) {.ffiDtor.} =
|
||||
discard
|
||||
|
||||
if isNil(callback):
|
||||
echo "error: missing callback in SdsNewReliabilityManager"
|
||||
return nil
|
||||
################################################################################
|
||||
### Retrieval-hint provider (hand-written: a C function pointer cannot be passed
|
||||
### as JSON). The setter dispatches a request so the provider is stored in the
|
||||
### worker thread's thread-local, where sdsCreate's hint closure reads it.
|
||||
|
||||
let ctx = ReliabilityManagerFFIPool.createFFIContext().valueOr:
|
||||
let msg = "Error creating SDS FFI context: " & $error
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return nil
|
||||
proc sdsNoopCallback(
|
||||
callerRet: cint, msg: ptr cchar, len: csize_t, userData: pointer
|
||||
) {.cdecl, gcsafe, raises: [].} =
|
||||
discard
|
||||
|
||||
let sendRes =
|
||||
try:
|
||||
ffi_context.sendRequestToFFIThread(ctx, SdsCreateRmReq.ffiNewReq(callback, userData))
|
||||
except Exception as exc:
|
||||
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
|
||||
if sendRes.isErr():
|
||||
let msg = "error creating reliability manager: " & sendRes.error
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
discard ReliabilityManagerFFIPool.destroyFFIContext(ctx)
|
||||
return nil
|
||||
registerReqFFI(SdsSetHintReq, ctx: ptr FFIContext[ReliabilityManager]):
|
||||
proc(cbPtr: pointer, udPtr: pointer): Future[Result[string, string]] {.async.} =
|
||||
sdsRetrievalHintCb = cbPtr
|
||||
sdsRetrievalHintUserData = udPtr
|
||||
return ok("")
|
||||
|
||||
return cast[pointer](ctx)
|
||||
|
||||
proc SdsSetEventCallback(
|
||||
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
|
||||
) {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
echo "error: invalid context in SdsSetEventCallback"
|
||||
return
|
||||
ctx[].callbackState.callback = cast[pointer](callback)
|
||||
ctx[].callbackState.userData = userData
|
||||
|
||||
proc SdsSetRetrievalHintProvider(
|
||||
proc sds_set_retrieval_hint_provider(
|
||||
ctx: ptr FFIContext[ReliabilityManager],
|
||||
callback: SdsRetrievalHintProvider,
|
||||
userData: pointer,
|
||||
) {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
echo "error: invalid context in SdsSetRetrievalHintProvider"
|
||||
return
|
||||
setRetrievalHint(cast[pointer](ctx), cast[pointer](callback), userData)
|
||||
|
||||
proc SdsCleanupReliabilityManager(
|
||||
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
|
||||
): cint {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
return RET_ERR
|
||||
if isNil(callback):
|
||||
return RET_MISSING_CALLBACK
|
||||
|
||||
clearRetrievalHint(cast[pointer](ctx))
|
||||
|
||||
# Recycle (not destroy) to reuse the worker + its fds. NON-BLOCKING: the FFI
|
||||
# thread fires `callback` once drained; the caller blocks on it, not the return.
|
||||
let res = releaseFFIContext(ctx, callback, userData)
|
||||
if res.isErr():
|
||||
let msg = "error cleaning up reliability manager: " & res.error
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
let sendRes =
|
||||
try:
|
||||
ffi_context.sendRequestToFFIThread(
|
||||
ctx,
|
||||
SdsSetHintReq.ffiNewReq(
|
||||
sdsNoopCallback, nil, cast[pointer](callback), userData
|
||||
),
|
||||
)
|
||||
except Exception as exc:
|
||||
Result[void, string].err("sendRequestToFFIThread exception: " & exc.msg)
|
||||
if sendRes.isErr():
|
||||
return RET_ERR
|
||||
return RET_OK
|
||||
|
||||
proc SdsResetReliabilityManager(
|
||||
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
|
||||
): cint {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
return RET_ERR
|
||||
if isNil(callback):
|
||||
return RET_MISSING_CALLBACK
|
||||
dispatchReq(ctx, callback, userData, SdsResetRmReq.ffiNewReq(callback, userData))
|
||||
|
||||
proc SdsWrapOutgoingMessage(
|
||||
ctx: ptr FFIContext[ReliabilityManager],
|
||||
message: pointer,
|
||||
messageLen: csize_t,
|
||||
messageId: cstring,
|
||||
channelId: cstring,
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
): cint {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
return RET_ERR
|
||||
if isNil(callback):
|
||||
return RET_MISSING_CALLBACK
|
||||
|
||||
if message == nil and messageLen > 0:
|
||||
let msg = "libsds error: message pointer is NULL but length > 0"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
if messageId == nil:
|
||||
let msg = "libsds error: message ID pointer is NULL"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
if channelId == nil:
|
||||
let msg = "libsds error: channel ID pointer is NULL"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
if $channelId == "":
|
||||
let msg = "libsds error: channel ID is empty string"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
let sharedMsg = copyToSharedSeqByte(message, messageLen.int)
|
||||
dispatchReq(
|
||||
ctx, callback, userData,
|
||||
SdsWrapMessageReq.ffiNewReq(callback, userData, sharedMsg, messageId, channelId),
|
||||
)
|
||||
|
||||
proc SdsUnwrapReceivedMessage(
|
||||
ctx: ptr FFIContext[ReliabilityManager],
|
||||
message: pointer,
|
||||
messageLen: csize_t,
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
): cint {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
return RET_ERR
|
||||
if isNil(callback):
|
||||
return RET_MISSING_CALLBACK
|
||||
|
||||
if message == nil and messageLen > 0:
|
||||
let msg = "libsds error: message pointer is NULL but length > 0"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
let sharedMsg = copyToSharedSeqByte(message, messageLen.int)
|
||||
dispatchReq(ctx, callback, userData, SdsUnwrapMessageReq.ffiNewReq(callback, userData, sharedMsg))
|
||||
|
||||
proc SdsMarkDependenciesMet(
|
||||
ctx: ptr FFIContext[ReliabilityManager],
|
||||
messageIds: pointer,
|
||||
count: csize_t,
|
||||
channelId: cstring,
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
): cint {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
return RET_ERR
|
||||
if isNil(callback):
|
||||
return RET_MISSING_CALLBACK
|
||||
|
||||
if messageIds == nil and count > 0:
|
||||
let msg = "libsds error: MessageIDs pointer is NULL but count > 0"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
if channelId == nil:
|
||||
let msg = "libsds error: channel ID pointer is NULL"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
if $channelId == "":
|
||||
let msg = "libsds error: channel ID is empty string"
|
||||
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](msg.len), userData)
|
||||
return RET_ERR
|
||||
|
||||
let sharedIds = copyToSharedSeqCstr(messageIds, count.int)
|
||||
dispatchReq(
|
||||
ctx, callback, userData,
|
||||
SdsMarkDepsReq.ffiNewReq(callback, userData, sharedIds, channelId),
|
||||
)
|
||||
|
||||
proc SdsStartPeriodicTasks(
|
||||
ctx: ptr FFIContext[ReliabilityManager], callback: FFICallBack, userData: pointer
|
||||
): cint {.dynlib, exportc, cdecl, raises: [].} =
|
||||
initializeLibrary()
|
||||
if not ReliabilityManagerFFIPool.isValidCtx(cast[pointer](ctx)):
|
||||
return RET_ERR
|
||||
if isNil(callback):
|
||||
return RET_MISSING_CALLBACK
|
||||
dispatchReq(ctx, callback, userData, SdsStartPeriodicTasksReq.ffiNewReq(callback, userData))
|
||||
# Emit binding metadata (no-op unless -d:ffiGenBindings). Must follow every
|
||||
# {.ffi.}/{.ffiCtor.}/{.ffiDtor.} annotation.
|
||||
genBindings()
|
||||
|
||||
@ -209,8 +209,8 @@
|
||||
}
|
||||
},
|
||||
"ffi": {
|
||||
"version": "#v0.1.5-rc.1",
|
||||
"vcsRevision": "e86b136e793adc9617e3c85ff4e8abe256a80100",
|
||||
"version": "0.1.5",
|
||||
"vcsRevision": "0a6bf6c49ff8d956899c2a3bb3a2b3ab8ed5281f",
|
||||
"url": "https://github.com/logos-messaging/nim-ffi",
|
||||
"downloadMethod": "git",
|
||||
"dependencies": [
|
||||
@ -219,7 +219,7 @@
|
||||
"taskpools"
|
||||
],
|
||||
"checksums": {
|
||||
"sha1": "301ae3a3a6889a1641a102a2c74798d1d82e6efb"
|
||||
"sha1": "eea561ccabe2f5fff6a586fbbf83d4e4dc41a7fb"
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
@ -117,8 +117,8 @@
|
||||
|
||||
ffi = pkgs.fetchgit {
|
||||
url = "https://github.com/logos-messaging/nim-ffi";
|
||||
rev = "d4c87c1f94c4678eea7d32a8f5f41c72420fadb6";
|
||||
sha256 = "14dm92l3wl8sc5a108612r1cgjvxksy2chzmn1asph6frl4lm641";
|
||||
rev = "0a6bf6c49ff8d956899c2a3bb3a2b3ab8ed5281f";
|
||||
sha256 = "0hif67h33fzn679pxq48v5y7al6ijgqdl5nf5gv0v4mkbhfbkffg";
|
||||
fetchSubmodules = true;
|
||||
};
|
||||
|
||||
|
||||
@ -16,7 +16,7 @@ requires "stew"
|
||||
requires "stint"
|
||||
requires "metrics"
|
||||
requires "results"
|
||||
requires "https://github.com/logos-messaging/nim-ffi#v0.1.5-rc.1"
|
||||
requires "https://github.com/logos-messaging/nim-ffi#v0.1.5"
|
||||
|
||||
proc buildLibrary(
|
||||
outLibNameAndExt: string,
|
||||
@ -188,10 +188,10 @@ proc buildMobileIOS(srcDir = ".", sdkPath = "") =
|
||||
exec "ar rcs " & aFileTmp & " " & objectFiles.join(" ")
|
||||
|
||||
# 4) Use libtool to localize all non-public symbols
|
||||
# Keep only Sds* functions as global, hide everything else to prevent conflicts
|
||||
# Keep only sds_* functions as global, hide everything else to prevent conflicts
|
||||
# with nim runtime symbols from libnim_status_client
|
||||
let keepSymbols =
|
||||
"_Sds*:_libsdsNimMain:_libsdsDatInit*:_libsdsInit*:_NimMainModule__libsds*"
|
||||
"_sds_*:_libsdsNimMain:_libsdsDatInit*:_libsdsInit*:_NimMainModule__libsds*"
|
||||
exec "xcrun libtool -static -o " & aFile & " " & aFileTmp &
|
||||
" -exported_symbols_list /dev/stdin <<< '" & keepSymbols & "' 2>/dev/null || cp " &
|
||||
aFileTmp & " " & aFile
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user