Support stdin via SharedArrayBuffer (#217)

Author: ianthomas23

packages/xeus/src/coincident.worker.ts

@@ -4,6 +4,8 @@
 
 import coincident from 'coincident';
 
+import type { KernelMessage } from '@jupyterlab/services';
+
 import {
   ContentsAPI,
   DriveFS,
@@ -67,7 +69,10 @@ export class XeusCoincidentKernel extends XeusRemoteKernel {
   }
 
   protected _initializeStdin(baseUrl: string, browsingContextId: string): void {
-    // TODO: SharedArrayBuffer implementation
+    globalThis.get_stdin = (
+      inputRequest: KernelMessage.IInputRequestMsg
+    ): KernelMessage.IInputReplyMsg =>
+      workerAPI.processStdinRequest(inputRequest);
   }
 }
 

packages/xeus/src/interfaces.ts

@@ -5,6 +5,8 @@
  * Definitions for the Xeus kernel.
  */
 
+import type { KernelMessage } from '@jupyterlab/services';
+
 import {
   TDriveMethod,
   TDriveRequest,
@@ -36,6 +38,15 @@ export interface IXeusWorkerKernel extends IWorkerKernel {
    */
   processMessage(msg: any): void;
 
+  /**
+   * Process stdin request, blocking until the reply is received.
+   * This is sync for the web worker, async for the UI thread.
+   * @param inputRequest
+   */
+  processStdinRequest(
+    inputRequest: KernelMessage.IInputRequestMsg
+  ): KernelMessage.IInputReplyMsg;
+
   /**
    * Process worker message
    * @param msg

packages/xeus/src/web_worker_kernel.ts

@@ -100,6 +100,17 @@ export class WebWorkerKernel implements IKernel {
 
         return await this._contentsProcessor.processDriveRequest(data);
       };
+
+      // Stdin request is synchronous from the web worker's point of view, blocking
+      // until the reply is received. From the UI thread's point of view it is async.
+      (remote.processStdinRequest as any) = async (
+        inputRequest: KernelMessage.IInputRequestMsg
+      ): Promise<KernelMessage.IInputReplyMsg> => {
+        this._processCoincidentWorkerMessage({ data: inputRequest });
+        this._inputDelegate =
+          new PromiseDelegate<KernelMessage.IInputReplyMsg>();
+        return await this._inputDelegate.promise;
+      };
     } else {
       this._worker.onmessage = e => {
         this._processComlinkWorkerMessage(e.data);
@@ -128,12 +139,11 @@ export class WebWorkerKernel implements IKernel {
   }
 
   private async _sendMessageToWorker(msg: any): Promise<void> {
-    if (msg.header.msg_type !== 'input_reply') {
+    if (msg.header.msg_type === 'input_reply') {
+      this._inputDelegate.resolve(msg);
+    } else {
       this._executeDelegate = new PromiseDelegate<void>();
-    }
-
-    await this._remoteKernel.processMessage({ msg, parent: this.parent });
-    if (msg.header.msg_type !== 'input_reply') {
+      await this._remoteKernel.processMessage({ msg, parent: this.parent });
       return await this._executeDelegate.promise;
     }
   }
@@ -304,6 +314,7 @@ export class WebWorkerKernel implements IKernel {
   private _worker: Worker;
   private _sendMessage: IKernel.SendMessage;
   private _executeDelegate = new PromiseDelegate<void>();
+  private _inputDelegate = new PromiseDelegate<KernelMessage.IInputReplyMsg>();
   private _parentHeader:
     | KernelMessage.IHeader<KernelMessage.MessageType>
     | undefined = undefined;

packages/xeus/src/worker.ts

@@ -106,7 +106,8 @@ export abstract class XeusRemoteKernel {
     }
 
     if (msg_type === 'input_reply') {
-      // Should never be called as input_reply messages are returned via service worker
+      // Should never be called as input_reply messages are handled by get_stdin
+      // via SharedArrayBuffer or service worker.
     } else {
       rawXServer.notify_listener(event.msg);
     }