merge: interactive hunk-by-hunk port stash-ipc-work → stable-ipc

Merged via hunk_merger — 47 hunks reviewed across 2 file(s):
  14 taken  ·  25 kept ours  ·  8 edited  ·  0 skipped

Per-file decisions:
  src/omnipkg/dispatcher.c: +1 taken, 13 kept
  src/omnipkg/isolation/worker_daemon.py: +13 taken, 12 kept, 8 edited

Staged: 2 files changed, 66 insertions(+), 43 deletions(-)

Source branch : stash-ipc-work
Target branch : stable-ipc
Session date  : 2026-05-30 02:12:11

[gitship-hunk-merger]

Author: 1minds3t

src/omnipkg/dispatcher.c

@@ -629,10 +629,28 @@ static int send_json_msg(int sock, const char *json_str) {
     uint64_t len = (uint64_t)strlen(json_str);
     uint8_t be_len[8];
     for (int i = 0; i < 8; i++)
-        be_len[7 - i] = (uint8_t)((len >> (i * 8)) & 0xFF);
-    if (!write_all(sock, be_len, 8)) return 0;
-    if (!write_all(sock, json_str, (size_t)len)) return 0;
-    return 1;
+     * This eliminates the double kernel wake-up that was costing ~110us:
+     * Python would wake on the 8-byte header, block waiting for the payload,
+     * OS context-switched back to C, C sent payload, switched back to Python.
+     * A single write lets the kernel deliver everything in one shot. */
+    if (len < 8192) {
+        /* Hot path: stack-allocated, zero heap pressure */
+        uint8_t buf[8192 + 8];
+        for (int i = 0; i < 8; i++)
+            buf[7 - i] = (uint8_t)((len >> (i * 8)) & 0xFF);
+        memcpy(buf + 8, json_str, (size_t)len);
+        return write_all(sock, buf, (size_t)(len + 8));
+    }
+
+    /* Cold path: large payload (rare — run_cli argv is almost always < 8KB) */
+    uint8_t *buf = (uint8_t *)malloc((size_t)(len + 8));
+    if (!buf) return 0;
+    for (int i = 0; i < 8; i++)
+        buf[7 - i] = (uint8_t)((len >> (i * 8)) & 0xFF);
+    memcpy(buf + 8, json_str, (size_t)len);
+    int ok = write_all(sock, buf, (size_t)(len + 8));
+    free(buf);
+    return ok;
 }
 
 static char* recv_json_msg(int sock) {

src/omnipkg/isolation/worker_daemon.py

@@ -4050,10 +4050,16 @@ def _run_cli(self, req, conn):
                     pass
 
             _t_ready = _rt.perf_counter()
-            safe_print(f"[RUN-CLI] ready_line={repr(ready_line)} ({(_t_ready-_t_got_worker)*1000:.2f}ms)", file=sys.stderr)
+            if os.environ.get('OMNIPKG_DEBUG') == '1':
+                _elapsed_us = int((_t_ready - locals().get('_t_post_flush', _t_ready)) * 1000000)
+                safe_print(f"[RUN-CLI-T] queue.get READY={_elapsed_us}us drained={_drain_count} line={repr(ready_line)}", file=sys.stderr)
+
+            _t_first_frame = None
+            _frame_count = 0
 
             if ready_parsed is None:
-                safe_print(f"[RUN-CLI] worker READY timeout - subprocess fallback", file=sys.stderr)
+                if os.environ.get('OMNIPKG_DEBUG') == '1':
+                    safe_print(f"[RUN-CLI] worker READY timeout - subprocess fallback", file=sys.stderr)
                 worker.force_shutdown()
                 try:
                     _fb_python = req.get("python_exe") or sys.executable
@@ -4079,13 +4085,16 @@ def _run_cli(self, req, conn):
                     send_json(conn, {"status": "ERROR", "error": str(_fb_err), "exit_code": 1})
                 return
 
+            # ── Zero-frame relay loop ─────────────────────────────────────────
+            # Worker now captures stdout/stderr into StringIO and embeds them
+            # in the single COMPLETED frame. No stream frames are emitted.
+            # This eliminates ~750us of queue-wake overhead (was 5x ~150us).
+            #
+            # Protocol: worker sends exactly two frames: READY then COMPLETED.
+            # NEEDS_INPUT still works: worker sends NEEDS_INPUT mid-execution
+            # (only for interactive commands), we relay it, then loop back.
             while True:
                 try:
-                    # Use a short poll interval so we remain responsive.
-                    # safe_input() in the worker blocks until it gets a
-                    # stdin_line reply, so there's no busy-spin here -
-                    # the worker simply won't produce output until we
-                    # send the reply.
                     line = worker.stdout_queue.get(timeout=86400.0)
                 except queue.Empty:
                     break
@@ -4096,49 +4105,44 @@ def _run_cli(self, req, conn):
                 msg = json.loads(line)
 
                 #  stdout / stderr stream - forward to C dispatcher 
-                if msg.get("stream"):
+                if '"status": "COMPLETED"' in line or '"status": "SHM_COMPLETED"' in line or '"status": "ERROR"' in line:
+                    _t_completed = _rt.perf_counter()
+                    _completed_cleanly = ('"status": "COMPLETED"' in line) or ('"status": "SHM_COMPLETED"' in line)
                     try:
-                        send_json(conn, msg)
+                        worker._last_result = json.loads(line)
+                    except Exception as _parse_err:
+                        if os.environ.get('OMNIPKG_DEBUG') == '1':
+                            safe_print(f"[RUN-CLI] terminal frame parse failed: {_parse_err!r} line={line[:120]!r}", file=sys.stderr)
+                    
+                    raw = line.encode("utf-8")
+                    try:
+                        conn.sendall(len(raw).to_bytes(8, "big") + raw)
                     except Exception:
-                        break
+                        pass
+                    
+                    if os.environ.get('OMNIPKG_DEBUG') == '1':
+                        safe_print(f"[RUN-CLI] COMPLETED in {(_t_completed-_t0)*1000:.2f}ms total frames={_frame_count}", file=sys.stderr)
+                    break
 
-                #  NEEDS_INPUT - relay prompt to C dispatcher, wait for
-                #    the user's reply, then send it back to the worker 
-                elif msg.get("status") == "NEEDS_INPUT":
+                elif '"status": "NEEDS_INPUT"' in line:
+                    # Interactive command — relay prompt, read reply, loop
                     try:
-                        # 1. Forward the NEEDS_INPUT frame to the C dispatcher.
-                        #    It will print the prompt on the real terminal and
-                        #    send back {"type":"stdin_line","data":"<text>"}.
-                        send_json(conn, msg)
-
-                        # 2. Wait for the stdin_line reply from the C dispatcher.
-                        #    We read directly from conn (the socket) here.
-                        #    recv_json() must be your existing framed-read helper.
-                        reply = recv_json(conn)   # blocks until dispatcher replies
-
+                        msg_ni = json.loads(line)
+                        send_json(conn, msg_ni)
+                        reply = recv_json(conn)
                         if reply and reply.get("type") == "stdin_line":
-                            # 3. Forward the user's input to the worker over its stdin.
                             worker.process.stdin.write(json.dumps(reply) + "\n")
-                            worker.process.stdin.flush()
                         else:
-                            # Unexpected reply or closed conn - send empty line
-                            # so the worker doesn't hang forever.
-                            worker.process.stdin.write(
-                                json.dumps({"type": "stdin_line", "data": ""}) + "\n"
-                            )
-                            worker.process.stdin.flush()
+                            worker.process.stdin.write('{"type": "stdin_line", "data": ""}\\n')
+                        worker.process.stdin.flush()
                     except Exception as _ni_err:
-                        safe_print(f"[RUN-CLI] NEEDS_INPUT relay failed: {_ni_err}", file=sys.stderr)
-                        # Send empty line - worker returns default and continues.
+                        if os.environ.get('OMNIPKG_DEBUG') == '1':
+                            safe_print(f"[RUN-CLI] NEEDS_INPUT relay failed: {_ni_err}", file=sys.stderr)
                         try:
-                            worker.process.stdin.write(
-                                json.dumps({"type": "stdin_line", "data": ""}) + "\n"
-                            )
+                            worker.process.stdin.write('{"type": "stdin_line", "data": ""}\\n')
                             worker.process.stdin.flush()
                         except Exception:
                             break
-                    # Continue the loop - worker resumes and will send more output.
-
                 #  COMPLETED / ERROR - forward and finish 
                 elif msg.get("status") in ("COMPLETED", "ERROR"):
                     _t_completed = _rt.perf_counter()
@@ -4721,7 +4725,7 @@ def _execute_code(
                 with self.pool_lock:
                     self.workers[worker_key] = {
                         "worker": worker,
-			"pinned": pin,
+                        "pinned": pin,
                         "created": time.time(),
                         "last_used": time.time(),
                         "request_count": 0,
@@ -7517,6 +7521,7 @@ def cli_idle_config(python_version: str = None, count: int = None):
 # 
 
 if __name__ == "__main__":
+    sys.setswitchinterval(0.0001)  # 100us GIL slice: 39 threads, was 5ms default
     #  CRITICAL: Check if we're a daemon child on Windows FIRST
     if IS_WINDOWS and os.environ.get("OMNIPKG_DAEMON_CHILD") == "1":
         try: