spike(perf): pipeline reference-returning calls via client-allocated object refs

packages/@jsii/kernel/src/api.ts

@@ -147,6 +147,14 @@ export interface CreateRequest {
    * Declarations of method overrides that should trigger callbacks
    */
   readonly overrides?: Override[];
+
+  /**
+   * A client-allocated object id to register the new instance under, instead of
+   * the kernel allocating one. Enables the host to pipeline `create` calls
+   * (use the reference before the kernel confirms it). Must be unique; the host
+   * is responsible for avoiding collisions with kernel-allocated ids.
+   */
+  readonly objid?: string;
 }
 
 export type CreateResponse = ObjRef;
@@ -161,11 +169,15 @@ export interface DelResponse {}
 export interface GetRequest {
   readonly objref: ObjRef;
   readonly property: string;
+  /** Client-allocated id to alias a reference-typed property value under. */
+  readonly objid?: string;
 }
 
 export interface StaticGetRequest {
   readonly fqn: string;
   readonly property: string;
+  /** Client-allocated id to alias a reference-typed property value under. */
+  readonly objid?: string;
 }
 
 export interface GetResponse {
@@ -191,12 +203,16 @@ export interface StaticInvokeRequest {
   readonly fqn: string;
   readonly method: string;
   readonly args?: any[];
+  /** Client-allocated id to alias the (reference-typed) result under. */
+  readonly objid?: string;
 }
 
 export interface InvokeRequest {
   readonly objref: ObjRef;
   readonly method: string;
   readonly args?: any[];
+  /** Client-allocated id to alias the (reference-typed) result under. */
+  readonly objid?: string;
 }
 
 export interface InvokeResponse {

packages/@jsii/kernel/src/kernel.ts

@@ -262,6 +262,9 @@ export class Kernel {
 
     this.#debug('value:', value);
     const ret = this.#fromSandbox(value, ti, `of static property ${symbol}`);
+    if (req.objid != null) {
+      this.#objects.aliasObject(req.objid, value);
+    }
     this.#debug('ret', ret);
     return { value: ret };
   }
@@ -320,6 +323,9 @@ export class Kernel {
       ti,
       () => `of property ${fqn}.${property}`,
     );
+    if (req.objid != null) {
+      this.#objects.aliasObject(req.objid, value);
+    }
     this.#debug('ret:', ret);
     return { value: ret };
   }
@@ -384,6 +390,9 @@ export class Kernel {
       ti.returns ?? 'void',
       () => `returned by method ${fqn ? `${fqn}#` : ''}${method}`,
     );
+    if (req.objid != null) {
+      this.#objects.aliasObject(req.objid, ret);
+    }
     this.#debug('invoke result', result);
 
     return { result };
@@ -421,13 +430,15 @@ export class Kernel {
     });
 
     this.#debug('method returned:', ret);
-    return {
-      result: this.#fromSandbox(
-        ret,
-        ti.returns ?? 'void',
-        `returned by static method ${fqn}.${method}`,
-      ),
-    };
+    const result = this.#fromSandbox(
+      ret,
+      ti.returns ?? 'void',
+      `returned by static method ${fqn}.${method}`,
+    );
+    if (req.objid != null) {
+      this.#objects.aliasObject(req.objid, ret);
+    }
+    return { result };
   }
 
   public begin(req: api.BeginRequest): api.BeginResponse {
@@ -694,7 +705,12 @@ export class Kernel {
         ctorResult.parameters,
       ),
     );
-    const objref = this.#objects.registerObject(obj, fqn, req.interfaces ?? []);
+    const objref = this.#objects.registerObject(
+      obj,
+      fqn,
+      req.interfaces ?? [],
+      req.objid,
+    );
 
     // overrides: for each one of the override method names, installs a
     // method on the newly created object which represents the remote "reverse proxy".

packages/@jsii/kernel/src/objects.ts

@@ -163,6 +163,7 @@ export class ObjectTable {
     obj: unknown,
     fqn: spec.FQN,
     interfaces?: spec.FQN[],
+    providedId?: string,
   ): api.ObjRef {
     if (fqn === undefined) {
       throw new JsiiFault('FQN cannot be undefined');
@@ -196,13 +197,31 @@ export class ObjectTable {
 
     interfaces = this.#removeRedundant(interfaces, fqn);
 
-    const objid = this.#makeId(fqn);
+    const objid = providedId ?? this.#makeId(fqn);
     this.#objects.set(objid, { instance: obj, fqn, interfaces });
     tagObject(obj, objid, interfaces);
 
     return { [api.TOKEN_REF]: objid, [api.TOKEN_INTERFACES]: interfaces };
   }
 
+  /**
+   * Registers an additional (client-allocated) id pointing at the same object
+   * an existing reference denotes. Used for pipelined invoke results: the host
+   * mints a fresh id before the call returns, and the kernel aliases it to
+   * whatever object the call actually produced (fresh or pre-existing). A no-op
+   * if the value is not an object reference.
+   */
+  public aliasObject(clientId: string, obj: unknown): void {
+    const ref = objectReference(obj);
+    if (!ref) {
+      return;
+    }
+    const entry = this.#objects.get(ref[api.TOKEN_REF]);
+    if (entry) {
+      this.#objects.set(clientId, entry);
+    }
+  }
+
   /**
    * Find the object and registered type for the given ObjRef
    */

packages/@jsii/python-runtime/src/jsii/_kernel/providers/process.py

@@ -150,8 +150,74 @@ def jdefault(obj):
     raise TypeError("Don't know how to convert object to JSON: %r" % obj)
 
 
+import sys as _sys
+import typing as _typing
+
+# POC (protocol pipelining): code-object id -> jsii fqn of a generated binding's
+# declared reference return type (or None). Resolved once, then cached.
+_RET_FQN_CACHE: dict = {}
+
+
+def _caller_jsii_return_fqn():
+    """Return the jsii fqn of the calling generated binding's declared return
+    type, but only for true reference types (classes / interface proxies). Used
+    to decide whether a call's result can be pipelined (returned as a handle the
+    host can use before the kernel confirms it). Returns None for value returns,
+    structs, enums, Optionals, unions, and anything unresolved -- those stay
+    synchronous. POC shortcut: walks frames + reads annotations; production would
+    thread the return fqn through pacmak codegen instead.
+    """
+    frame = _sys._getframe(1)
+    while frame is not None:
+        mod = frame.f_globals.get("__name__", "")
+        if mod != "jsii" and not mod.startswith("jsii."):
+            break
+        frame = frame.f_back
+    if frame is None:
+        return None
+    code = frame.f_code
+    key = id(code)
+    if key in _RET_FQN_CACHE:
+        return _RET_FQN_CACHE[key]
+    fqn = None
+    try:
+        name = code.co_name
+        holder = frame.f_locals.get("self", None)
+        if holder is None:
+            holder = frame.f_locals.get("cls", None)
+        if holder is not None:
+            klass = holder if isinstance(holder, type) else type(holder)
+            func = None
+            for k in getattr(klass, "__mro__", [klass]):
+                if name in vars(k):
+                    raw = vars(k)[name]
+                    func = (
+                        raw.fget
+                        if isinstance(raw, property)
+                        else getattr(raw, "__func__", raw)
+                    )
+                    break
+            if func is not None:
+                ret = _typing.get_type_hints(func).get("return")
+                cand = getattr(ret, "__jsii_type__", None)
+                if cand is not None:
+                    from ... import _reference_map as _rm
+
+                    if cand in _rm._types:
+                        fqn = cand
+    except Exception:
+        fqn = None
+    _RET_FQN_CACHE[key] = fqn
+    return fqn
+
+
 class _NodeProcess:
     def __init__(self):
+        # POC (pipelining) state: fire-and-forget reference-returning calls
+        # without waiting; drain acks at sync points; cap to avoid deadlock.
+        self._pending = 0
+        self._objid_seq = 0
+        self._PIPELINE_CAP = 64
         self._serializer = cattr.Converter()
         self._serializer.register_unstructure_hook(enum.Enum, _unstructure_enum)
         self._serializer.register_unstructure_hook(
@@ -318,9 +384,7 @@ def handshake(self) -> None:
             or resp.hello == f"@jsii/runtime@0.0.0"
         ), f"Invalid JSII Runtime Version: {resp.hello!r}"
 
-    def send(
-        self, request: KernelRequest, response_type: Type[KernelResponse]
-    ) -> KernelResponse:
+    def _write(self, request) -> None:
         req_dict = self._serializer.unstructure(request)
 
         if _stack_traces_enabled():
@@ -330,11 +394,51 @@ def send(
 
         data = json.dumps(req_dict, default=jdefault).encode("utf8")
 
-        # Send our data, ensure that it is framed with a trailing \n
         assert self._process.stdin is not None
         self._process.stdin.write(b"%b\n" % (data,))
         self._process.stdin.flush()
 
+    def _drain(self) -> None:
+        # Read and discard acks for fire-and-forget calls. POC shortcut: errors
+        # are ignored; callbacks during the pipelined window are unsupported.
+        while self._pending > 0:
+            resp = self._serializer.structure(self._next_message(), _ProcessResponse)
+            if isinstance(resp, _CallbackResponse):
+                raise RuntimeError("pipelining POC: callback during drain")
+            self._pending -= 1
+
+    def _fire(self, request, fqn: str) -> str:
+        # Mint a client object id, fire the request without waiting, and return
+        # the id. The kernel registers/aliases this id to whatever the call
+        # produces (in order), so later references resolve correctly.
+        self._objid_seq += 1
+        objid = f"{fqn}@{1_000_000_000 + self._objid_seq}"
+        self._write(attr.evolve(request, objid=objid))
+        self._pending += 1
+        if self._pending >= self._PIPELINE_CAP:
+            self._drain()
+        return objid
+
+    def create_async(self, request: "CreateRequest") -> "CreateResponse":
+        objid = self._fire(request, request.fqn)
+        return CreateResponse(ref=objid, interfaces=request.interfaces)
+
+    def invoke_async(self, request, return_fqn: str) -> "InvokeResponse":
+        return InvokeResponse(result=ObjRef(ref=self._fire(request, return_fqn)))
+
+    def get_async(self, request, return_fqn: str) -> "GetResponse":
+        return GetResponse(value=ObjRef(ref=self._fire(request, return_fqn)))
+
+    def send(
+        self, request: KernelRequest, response_type: Type[KernelResponse]
+    ) -> KernelResponse:
+        # Any synchronous call is a barrier: the kernel processes requests in
+        # order, so its response comes after all pending fire-and-forget acks.
+        if self._pending:
+            self._drain()
+
+        self._write(request)
+
         resp: _ProcessResponse = self._serializer.structure(
             self._next_message(),
             _ProcessResponse,  # pyright: ignore[reportArgumentType]
@@ -370,24 +474,36 @@ def invokeBinScript(self, request: InvokeScriptRequest) -> InvokeScriptResponse:
         return self._process.send(request, InvokeScriptResponse)
 
     def create(self, request: CreateRequest) -> CreateResponse:
-        return self._process.send(request, CreateResponse)
+        return self._process.create_async(request)
 
     def get(self, request: GetRequest) -> GetResponse:
+        fqn = _caller_jsii_return_fqn()
+        if fqn is not None:
+            return self._process.get_async(request, fqn)
         return self._process.send(request, GetResponse)
 
     def set(self, request: SetRequest) -> SetResponse:
         return self._process.send(request, SetResponse)
 
     def sget(self, request: StaticGetRequest) -> GetResponse:
+        fqn = _caller_jsii_return_fqn()
+        if fqn is not None:
+            return self._process.get_async(request, fqn)
         return self._process.send(request, GetResponse)
 
     def sset(self, request: StaticSetRequest) -> SetResponse:
         return self._process.send(request, SetResponse)
 
     def invoke(self, request: InvokeRequest) -> Union[InvokeResponse, Callback]:
+        fqn = _caller_jsii_return_fqn()
+        if fqn is not None:
+            return self._process.invoke_async(request, fqn)
         return self._process.send(request, InvokeResponse)
 
     def sinvoke(self, request: StaticInvokeRequest) -> InvokeResponse:
+        fqn = _caller_jsii_return_fqn()
+        if fqn is not None:
+            return self._process.invoke_async(request, fqn)
         return self._process.send(request, InvokeResponse)
 
     def delete(self, request: DeleteRequest) -> DeleteResponse:

packages/@jsii/python-runtime/src/jsii/_kernel/types.py

@@ -71,6 +71,10 @@ class CreateRequest:
     args: List[Any] = attr.Factory(list)
     overrides: List[Override] = attr.Factory(list)
     interfaces: Optional[List[str]] = None
+    # POC (pipelining): client-allocated object id. When set, the kernel
+    # registers the new instance under this id instead of allocating one,
+    # letting the host use the reference without waiting for the response.
+    objid: Optional[str] = None
 
 
 @attr.s(auto_attribs=True, frozen=True, slots=True)
@@ -90,12 +94,14 @@ class DeleteResponse: ...
 class GetRequest:
     objref: ObjRef
     property: str
+    objid: Optional[str] = None  # POC (pipelining): alias reference-typed value
 
 
 @attr.s(auto_attribs=True, frozen=True, slots=True)
 class StaticGetRequest:
     fqn: str
     property: str
+    objid: Optional[str] = None  # POC (pipelining): alias reference-typed value
 
 
 @attr.s(auto_attribs=True, frozen=True, slots=True)
@@ -126,13 +132,15 @@ class StaticInvokeRequest:
     fqn: str
     method: str
     args: Optional[List[Any]] = attr.Factory(list)
+    objid: Optional[str] = None  # POC (pipelining): alias reference-typed result
 
 
 @attr.s(auto_attribs=True, frozen=True, slots=True)
 class InvokeRequest:
     objref: ObjRef
     method: str
     args: Optional[List[Any]] = attr.Factory(list)
+    objid: Optional[str] = None  # POC (pipelining): alias reference-typed result
 
 
 @attr.s(auto_attribs=True, frozen=True, slots=True)