feat(c): non-scalar CBOR-free fast path

Author: gmelodie

CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to this project are documented in this file.
 ## [Unreleased]
 
 ### Changed
+- **`abi = c` non-scalar procs no longer marshal through CBOR.** The foreign
+  surface is unchanged — the generated headers and exported symbols are
+  byte-identical — but the hop between the caller and the FFI thread now carries
+  the packed `_CWire` struct itself instead of CBOR-encoding it and decoding it
+  back. A request is packed into a `malloc`'d owned copy on the calling thread
+  and unpacked (then freed) on the FFI thread; an object reply rides back as its
+  `_CWire` image and a `string` reply as raw UTF-8, so the round trip through
+  `cborEncodeShared`/`cborDecodePtr` is gone from both directions. Only the
+  scalar fast path was CBOR-free before
+  ([#131](https://github.com/logos-messaging/nim-ffi/issues/131)).
+- `_CWire` seq/Option payload buffers are allocated with libc `malloc`
+  (`cwireAllocBuf`) rather than `allocShared`, so a wire packed on the calling
+  thread can be freed on the FFI thread — the cross-thread ownership the
+  CBOR-free request path relies on, and consistent with the libc-backed request
+  envelope.
 - User event callbacks now run on a dedicated event thread fed by a
   bounded SPSC queue (default capacity 1024), so a slow listener can no
   longer block the FFI thread or concurrent `add_event_listener` /

ffi/codegen/c.nim

@@ -1231,7 +1231,7 @@ proc emitAbiMethod(
 
 func abiScalarOkLines(m: FFIProcMeta, fnType: string): seq[string] =
   ## Trampoline RET_OK branch. A string return rides as its own UTF-8; every
-  ## other scalar is the 8-byte image `ffiScalarRetBytes` packs (ints
+  ## other scalar is the 8-byte image `ffiRawRetBytes` packs (ints
   ## sign-extended, floats widened to double, bool as 0/1).
   let rt = m.returnTypeName.strip()
   if rt == "string" or rt == "cstring":

ffi/ffi_thread_request.nim

@@ -17,11 +17,13 @@ type FFIThreadRequest* = object
   callback*: FFICallBack
   userData*: pointer
   reqId*: cstring ## Req type name used to look up the handler.
-  data*: ptr UncheckedArray[byte] ## Owned CBOR-encoded request payload.
+  data*: ptr UncheckedArray[byte]
+    ## Owned request payload: CBOR-encoded, or a packed `_CWire` struct on the
+    ## `abi = c` path. Nil on the scalar fast path.
   dataLen*: int
-  isScalar*: bool
-    ## Scalar fast path: args rode inline in `scalarArgs`, so a 0-length return
-    ## is a real empty string, not a CBOR "no value".
+  rawReply*: bool
+    ## CBOR-free request (scalar fast path or `abi = c`): the reply is raw bytes,
+    ## so a 0-length one is a real empty string, not a CBOR "no value".
   scalarArgs*: array[MaxScalarArgs, uint64]
     ## Inlined scalar args (no per-call c_malloc); a plain array keeps
     ## `deleteRequest` unaliased.
@@ -63,7 +65,7 @@ proc allocBaseRequest(
   ret[].reqId = reqId.alloc()
   ret[].data = nil
   ret[].dataLen = 0
-  ret[].isScalar = false
+  ret[].rawReply = false
   ret[].next = nil
   ret[].responded = false
   return ret
@@ -121,11 +123,14 @@ proc initFromOwnedShared*(
     reqId: cstring,
     data: ptr UncheckedArray[byte],
     dataLen: int,
+    rawReply: bool = false,
 ): ptr type T =
   ## Adopts an already-c_malloc'd buffer (no copy); `deleteRequest` c_frees it.
-  ## Pass `(nil, 0)` for an empty payload.
+  ## Pass `(nil, 0)` for an empty payload. Set `rawReply` when the handler answers
+  ## with raw (non-CBOR) bytes, so an empty reply isn't read as "no value".
   var ret = allocBaseRequest(callback, userData, reqId)
   adoptOwnedSharedPayload(ret, data, dataLen)
+  ret[].rawReply = rawReply
   return ret
 
 proc initScalar*(
@@ -140,14 +145,14 @@ proc initScalar*(
     "initScalar: " & $args.len & " scalar args exceed MaxScalarArgs (" & $MaxScalarArgs &
       ")"
   var ret = allocBaseRequest(callback, userData, reqId)
-  ret[].isScalar = true
+  ret[].rawReply = true
   for i in 0 ..< args.len:
     ret[].scalarArgs[i] = args[i]
   ret
 
-func ffiScalarRetBytes*[T](x: T): seq[byte] =
-  ## Scalar handler result as raw bytes, no CBOR: string/cstring ride as UTF-8,
-  ## other scalars as the 8-byte native image of `ffiPackScalar(x)`.
+func ffiRawRetBytes*[T](x: T): seq[byte] =
+  ## CBOR-free handler result as raw bytes: string/cstring ride as UTF-8, other
+  ## scalars as the 8-byte native image of `ffiPackScalar(x)`.
   when T is string:
     var b = newSeq[byte](x.len)
     if x.len > 0:
@@ -195,8 +200,8 @@ proc fireCallback*(res: Result[seq[byte], string], request: ptr FFIThreadRequest
         cast[csize_t](bytes.len),
         request[].userData,
       )
-    elif request[].isScalar:
-      # Scalar 0-byte return is a real empty string, not CBOR "no value".
+    elif request[].rawReply:
+      # A CBOR-free 0-byte return is a real empty string, not CBOR "no value".
       var empty: byte
       request[].callback(
         RET_OK, cast[ptr cchar](addr empty), 0.csize_t, request[].userData

ffi/internal/c_macro_helpers.nim

@@ -22,7 +22,7 @@ proc markCWireEmitted(typeName: string) {.compileTime.} =
   if not isCWireEmitted(typeName):
     emittedCWireTypes.add(typeName)
 
-proc cwireTypeName(userTypeName: string): string =
+proc cwireTypeName*(userTypeName: string): string =
   userTypeName & "_CWire"
 
 proc seqItemsField(obj, field: NimNode): NimNode =
@@ -31,7 +31,7 @@ proc seqItemsField(obj, field: NimNode): NimNode =
 proc seqLenField(obj, field: NimNode): NimNode =
   newDotExpr(obj, ident($field & cwireLenSuffix))
 
-proc isStringType(t: NimNode): bool =
+proc isStringType*(t: NimNode): bool =
   t.kind == nnkIdent and ($t == "string" or $t == "cstring")
 
 proc isBracketOf(t: NimNode, heads: openArray[string]): bool =
@@ -294,12 +294,12 @@ proc emitSeqPack(dstObj, srcAccess, fieldNameIdent, userType: NimNode): NimNode
       `items` = nil
       `count` = 0
     else:
-      `items` = cast[`bufType`](allocShared(sizeof(`wireElem`) * `srcAccess`.len()))
+      `items` = cast[`bufType`](cwireAllocBuf(sizeof(`wireElem`) * `srcAccess`.len()))
       `forLoop`
       `count` = `srcAccess`.len()
 
 proc emitOptionPack(dstAccess, srcAccess, userType: NimNode): NimNode =
-  ## Pack an Option into a `ptr`: some → `allocShared` box, none → nil. Payload
+  ## Pack an Option into a `ptr`: some → `cwireAllocBuf` box, none → nil. Payload
   ## read into a local once so a composite inner type isn't re-`get()` per element.
   let innerType = userType[1]
   let wireInner = wireValueType(innerType)
@@ -308,7 +308,7 @@ proc emitOptionPack(dstAccess, srcAccess, userType: NimNode): NimNode =
   let elemPack = emitElemPack(nnkBracketExpr.newTree(dstAccess), innerVal, innerType)
   quote:
     if `srcAccess`.isSome():
-      `dstAccess` = cast[`bufType`](allocShared(sizeof(`wireInner`)))
+      `dstAccess` = cast[`bufType`](cwireAllocBuf(sizeof(`wireInner`)))
       let `innerVal` = `srcAccess`.get()
       `elemPack`
     else:
@@ -379,7 +379,7 @@ proc emitSeqFree(dstObj, fieldNameIdent, userType: NimNode): NimNode =
   quote:
     if not `items`.isNil():
       `freeLoop`
-      deallocShared(`items`)
+      cwireFreeBuf(`items`)
       `items` = nil
       `count` = 0
 
@@ -390,7 +390,7 @@ proc emitOptionFree(dstAccess, userType: NimNode): NimNode =
   quote:
     if not `dstAccess`.isNil():
       `freeInner`
-      deallocShared(`dstAccess`)
+      cwireFreeBuf(`dstAccess`)
       `dstAccess` = nil
 
 proc emitFreeStmt(dstObj, fieldNameIdent, userType: NimNode): seq[NimNode] =
@@ -539,6 +539,9 @@ type
     paramNames: seq[string] ## envelope field names (the extra params)
     paramTypes: seq[NimNode] ## envelope field types
     respType: NimNode ## method result T; empty for a ctor
+    handler: NimNode
+      ## FFI-thread handler, deferred here so it lands after the `_CWire`
+      ## companions it packs/unpacks through.
 
 var cAbiSpecs {.compileTime.}: seq[CAbiSpec]
 
@@ -553,7 +556,7 @@ proc registerCAbiMethod*(
     libType, envelope: NimNode,
     paramNames: seq[string],
     paramTypes: seq[NimNode],
-    respType: NimNode,
+    respType, handler: NimNode,
 ) {.compileTime.} =
   ## Record an `abi = c` method for `flushCAbiDispatch`. Nodes are `copyNimTree`
   ## frozen: reusing the Req section's originals (bound to `nnkSym`) would ICE.
@@ -566,6 +569,7 @@ proc registerCAbiMethod*(
       paramNames: paramNames,
       paramTypes: copyTypes(paramTypes),
       respType: respType.copyNimTree(),
+      handler: handler.copyNimTree(),
     )
   )
 
@@ -574,6 +578,7 @@ proc registerCAbiCtor*(
     libType, envelope: NimNode,
     paramNames: seq[string],
     paramTypes: seq[NimNode],
+    handler: NimNode,
 ) {.compileTime.} =
   ## Record an `abi = c` ctor for `flushCAbiDispatch`; see `registerCAbiMethod`
   ## for why nodes are `copyNimTree` frozen.
@@ -586,6 +591,7 @@ proc registerCAbiCtor*(
       paramNames: paramNames,
       paramTypes: copyTypes(paramTypes),
       respType: newEmptyNode(),
+      handler: handler.copyNimTree(),
     )
   )
 
@@ -630,15 +636,16 @@ proc replyTrampProc(trampName, body: NimNode): NimNode =
     pragmas = cdeclReplyPragma(),
   )
 
-proc objectTrampBody(boxName, respType, respWire: NimNode): NimNode =
-  ## Reply trampoline for an object return: decode, `cwirePack` into `_CWire`,
-  ## hand a pointer to the caller, release. `reply` is nil only on error.
+proc objectTrampBody(boxName, respWire: NimNode): NimNode =
+  ## Reply trampoline for an object return: the payload is already the packed
+  ## `_CWire` image, so hand its address straight to the caller and release the
+  ## buffers it owns. `reply` is nil only on error.
   quote:
     let box = cast[ptr `boxName`](ud)
     if box.isNil():
       return
     if ret == RET_STALE_WARN:
-      # Non-terminal progress signal: keep the box, don't decode.
+      # Non-terminal progress signal: keep the box, don't read the payload.
       return
     defer:
       freeBox(box)
@@ -651,53 +658,46 @@ proc objectTrampBody(boxName, respType, respWire: NimNode): NimNode =
           copyMem(addr em[0], msg, int(len))
         box.fn(ret, nil, em.cstring, box.ud)
         return
-      let decoded =
-        cborDecodePtr(cast[ptr UncheckedArray[byte]](msg), int(len), `respType`)
-      if decoded.isErr():
-        box.fn(RET_ERR, nil, decoded.error.cstring, box.ud)
-      else:
-        var wire: `respWire`
-        cwirePack(wire, decoded.get())
-        box.fn(RET_OK, addr wire, "".cstring, box.ud)
-        cwireFree(wire)
+      if msg.isNil() or int(len) != sizeof(`respWire`):
+        box.fn(RET_ERR, nil, "abi = c reply: unexpected payload size".cstring, box.ud)
+        return
+      var wire = cast[ptr `respWire`](msg)[]
+      box.fn(RET_OK, addr wire, "".cstring, box.ud)
+      cwireFree(wire)
     except CatchableError as e:
       box.fn(RET_ERR, nil, e.msg.cstring, box.ud)
 
 proc stringTrampBody(boxName: NimNode): NimNode =
-  ## Reply trampoline for a `string` return (and the ctor's address string):
-  ## decode and hand the caller a NUL-terminated `cstring`. Reply/error strings
-  ## are always non-nil empty on the paths they don't apply to (no nil deref).
+  ## Reply trampoline for a `string` return (and the ctor's address string): the
+  ## payload is raw UTF-8, copied into a NUL-terminated `cstring` since the wire
+  ## bytes aren't terminated. Reply/error strings are always non-nil empty on the
+  ## paths they don't apply to (no nil deref).
   quote:
     let box = cast[ptr `boxName`](ud)
     if box.isNil():
       return
     if ret == RET_STALE_WARN:
-      # Non-terminal progress signal: keep the box, don't decode.
+      # Non-terminal progress signal: keep the box, don't read the payload.
       return
     defer:
       freeBox(box)
     if box.fn.isNil():
       return
     try:
+      var payload = newString(int(len))
+      if int(len) > 0 and not msg.isNil():
+        copyMem(addr payload[0], msg, int(len))
       if ret != RET_OK:
-        var em = newString(int(len))
-        if int(len) > 0:
-          copyMem(addr em[0], msg, int(len))
-        box.fn(ret, "".cstring, em.cstring, box.ud)
+        box.fn(ret, "".cstring, payload.cstring, box.ud)
         return
-      let decoded = cborDecodePtr(cast[ptr UncheckedArray[byte]](msg), int(len), string)
-      if decoded.isErr():
-        box.fn(RET_ERR, "".cstring, decoded.error.cstring, box.ud)
-      else:
-        let replyStr = decoded.get()
-        box.fn(RET_OK, replyStr.cstring, "".cstring, box.ud)
+      box.fn(RET_OK, payload.cstring, "".cstring, box.ud)
     except CatchableError as e:
       box.fn(RET_ERR, "".cstring, e.msg.cstring, box.ud)
 
 proc exportedMethodProc(
     spec: CAbiSpec, boxName, envWire, trampName, poolIdent, cbType: NimNode
 ): NimNode =
-  # No `foreignThreadGc`: `cwireUnpack`/`cborEncodeShared` alloc on the calling thread (already GC-registered); wrapping would free its live ORC heap.
+  # No `foreignThreadGc`: `cwireUnpack`/`cwirePack` alloc on the calling thread (already GC-registered); wrapping would free its live ORC heap.
   let envName = spec.envelope
   let libFFICtx =
     nnkPtrTy.newTree(nnkBracketExpr.newTree(ident("FFIContext"), spec.libType))
@@ -713,23 +713,31 @@ proc exportedMethodProc(
     if not `poolIdent`.isValidCtx(cast[pointer](ctx)):
       onReply(RET_ERR, `emptyReply`, "ctx is not a valid FFI context".cstring, userData)
       return RET_ERR
-    var reqObj: `envName` = cwireUnpack(req[])
-    let enc = cborEncodeShared(reqObj)
-    let reqBuf = enc.data
-    let reqBufLen = enc.len
+    var ownedWire: `envWire`
+    cwirePack(ownedWire, cwireUnpack(req[]))
+    let ownedCopy = cwireOwnedCopy(ownedWire)
+    if ownedCopy.isNil():
+      cwireFree(ownedWire)
+      onReply(RET_ERR, `emptyReply`, "out of memory".cstring, userData)
+      return RET_ERR
+    let reqBuf = cast[ptr UncheckedArray[byte]](ownedCopy)
     let box = cast[ptr `boxName`](allocBox(sizeof(`boxName`)))
     box.fn = onReply
     box.ud = userData
     let typeStr = $`envName`
     let reqPtr = FFIThreadRequest.initFromOwnedShared(
-      `trampName`, box, typeStr.cstring, reqBuf, reqBufLen
+      `trampName`, box, typeStr.cstring, reqBuf, sizeof(`envWire`), rawReply = true
     )
     let sendRes =
       try:
         ffi_context.sendRequestToFFIThread(ctx, reqPtr)
       except Exception as e:
         Result[void, string].err("sendRequestToFFIThread exception: " & e.msg)
     if sendRes.isErr():
+      # A rejected send already `deleteRequest`ed the struct copy, which frees only
+      # the struct itself; `ownedWire` still aliases its field buffers, so free them
+      # here — on success the FFI thread's unpack does it instead.
+      cwireFree(ownedWire)
       onReply(RET_ERR, `emptyReply`, sendRes.error.cstring, userData)
       return RET_ERR
     return RET_OK
@@ -774,23 +782,31 @@ proc exportedCtorProc(
         )
       return nil
     let ctx = ctxRes.get()
-    var reqObj: `envName` = cwireUnpack(req[])
-    let enc = cborEncodeShared(reqObj)
-    let reqBuf = enc.data
-    let reqBufLen = enc.len
+    var ownedWire: `envWire`
+    cwirePack(ownedWire, cwireUnpack(req[]))
+    let ownedCopy = cwireOwnedCopy(ownedWire)
+    if ownedCopy.isNil():
+      cwireFree(ownedWire)
+      if not onCreated.isNil():
+        onCreated(RET_ERR, "".cstring, "out of memory".cstring, userData)
+      return nil
+    let reqBuf = cast[ptr UncheckedArray[byte]](ownedCopy)
     let box = cast[ptr `boxName`](allocBox(sizeof(`boxName`)))
     box.fn = onCreated
     box.ud = userData
     let typeStr = $`envName`
     let reqPtr = FFIThreadRequest.initFromOwnedShared(
-      `trampName`, box, typeStr.cstring, reqBuf, reqBufLen
+      `trampName`, box, typeStr.cstring, reqBuf, sizeof(`envWire`), rawReply = true
     )
     let sendRes =
       try:
         ctx.sendRequestToFFIThread(reqPtr)
       except Exception as e:
         Result[void, string].err("sendRequestToFFIThread exception: " & e.msg)
     if sendRes.isErr():
+      # See exportedMethodProc: the rejected send freed the struct copy, not the
+      # field buffers `ownedWire` still aliases.
+      cwireFree(ownedWire)
       if not onCreated.isNil():
         onCreated(RET_ERR, "".cstring, sendRes.error.cstring, userData)
       return nil
@@ -838,6 +854,7 @@ proc flushCAbiDispatch*(): NimNode {.compileTime.} =
   for spec in cAbiSpecs:
     let envName = spec.envelope
     ensureCWireForFields(sink, $envName, spec.paramNames, spec.paramTypes)
+    sink.add(spec.handler)
     let envWire = ident(cwireTypeName($envName))
     let boxName = ident($envName & "CBox")
     let trampName = ident($envName & "CReply")
@@ -861,7 +878,7 @@ proc flushCAbiDispatch*(): NimNode {.compileTime.} =
         let respWire = ident(cwireTypeName($rt))
         let cbType = cAbiCbType(nnkPtrTy.newTree(respWire))
         sink.add(boxTypeDef(boxName, cbType))
-        sink.add(replyTrampProc(trampName, objectTrampBody(boxName, rt, respWire)))
+        sink.add(replyTrampProc(trampName, objectTrampBody(boxName, respWire)))
         sink.add(
           exportedMethodProc(spec, boxName, envWire, trampName, poolIdent, cbType)
         )