Use Conductor-based streams for StdIO (#107)

* Add StdIO streams + Bump conductor version

* Abstract away error and output displays

* Formatting changes

* Ensure the write functions are called asynchronously and cache readers and writers

* More formatting changes

* Add input function

* Pass the ErrorValue creation process to the displayError function

* Remove console.log from streams file

* Fix controller close + use allSettled instead of Promise.all

---------

Co-authored-by: Martin Henz <henz@comp.nus.edu.sg>

Author: AaravMalani

package.json

@@ -55,7 +55,7 @@
     "typescript-eslint": "^8.56.1"
   },
   "dependencies": {
-    "@sourceacademy/conductor": "^0.2.3",
+    "@sourceacademy/conductor": "^0.3.0",
     "@sourceacademy/wasm-util": "^1.0.4",
     "fast-levenshtein": "^3.0.0",
     "mathjs": "^14.9.1",

src/conductor/PyEvaluator.ts

@@ -2,10 +2,17 @@
 // https://github.com/source-academy/conductor
 // Original author(s): Source Academy Team
 
+import { ErrorType } from "@sourceacademy/conductor/common";
 import { BasicEvaluator, IRunnerPlugin } from "@sourceacademy/conductor/runner";
 import { Context } from "../cse-machine/context";
+import {
+  createErrorStream,
+  createInputStream,
+  createOutputStream,
+  destroyStreams,
+  displayError,
+} from "../cse-machine/streams";
 import { IOptions, runInContext } from "../runner/pyRunner";
-import { Finished } from "../types";
 
 const defaultContext = new Context();
 const defaultOptions: IOptions = {
@@ -26,16 +33,25 @@ export default class PyEvaluator extends BasicEvaluator {
 
   async evaluateChunk(chunk: string): Promise<void> {
     try {
-      const result = await runInContext(
+      this.context.streams = {
+        initialised: true,
+        stdout: createOutputStream(this.conductor),
+        stderr: createErrorStream(this.conductor),
+        stdin: createInputStream(this.conductor),
+      };
+      await runInContext(
         chunk, // Code
         this.context,
         this.options,
       );
-      this.conductor.sendOutput(
-        `${(result as Finished).representation.toString((result as Finished).value)}`,
-      );
     } catch (error) {
-      this.conductor.sendOutput(`Error: ${error instanceof Error ? error.message : error}`);
+      if (error instanceof SyntaxError) {
+        await displayError(this.context, error, ErrorType.EVALUATOR_SYNTAX);
+        return;
+      }
+      await displayError(this.context, error, ErrorType.INTERNAL);
+    } finally {
+      await destroyStreams(this.context);
     }
   }
 }

src/cse-machine/context.ts

@@ -1,16 +1,28 @@
+import { ConductorError } from "@sourceacademy/conductor/common";
 import { StmtNS } from "../ast-types";
 import { ModuleContext, NativeStorage } from "../types";
 import { Control } from "./control";
 import { Environment } from "./environment";
 import { CseError } from "./error";
 import { Heap } from "./heap";
 import { Stash, Value } from "./stash";
+import { ReadableContext, WritableContext } from "./streams";
 import { Node } from "./types";
 
 export class Context {
   public control: Control;
   public stash: Stash;
-  public output: string = "";
+
+  public streams:
+    | {
+        initialised: false;
+      }
+    | {
+        initialised: true;
+        stdout: WritableContext<string>;
+        stderr: WritableContext<ConductorError>;
+        stdin: ReadableContext<string>;
+      };
   //public environment: Environment;
   public errors: CseError[] = [];
   public moduleContexts: { [name: string]: ModuleContext };
@@ -46,6 +58,7 @@ export class Context {
       this.runtime.environments.push(globalEnvironment);
       this.runtime.environmentTree.insert(globalEnvironment);
     }
+    this.streams = this.createEmptyStreams();
     this.nativeStorage = {
       builtins: new Map<string, Value>(),
       previousProgramsIdentifiers: new Set<string>(),
@@ -82,6 +95,10 @@ export class Context {
     changepointSteps: [],
   });
 
+  createEmptyStreams = (): { initialised: false } => ({
+    initialised: false,
+  });
+
   public reset(program?: StmtNS.Stmt): void {
     this.control = new Control(program);
     this.stash = new Stash();

src/cse-machine/interpreter.ts

@@ -6,6 +6,7 @@
 
 /* tslint:disable:max-classes-per-file */
 
+import { ErrorType } from "@sourceacademy/conductor/common";
 import { ExprNS, StmtNS } from "../ast-types";
 import * as error from "../errors/errors";
 import { BuiltinReassignmentError } from "../errors/errors";
@@ -31,6 +32,7 @@ import {
   isFalsy,
 } from "./operators";
 import { Stash, Value } from "./stash";
+import { displayError } from "./streams";
 import {
   AppInstr,
   AssmtInstr,
@@ -53,12 +55,7 @@ type CmdEvaluator = (
   control: Control,
   stash: Stash,
   isPrelude: boolean,
-) => void;
-
-let cseFinalPrint = "";
-export function addPrint(str: string) {
-  cseFinalPrint = cseFinalPrint + str + "\n";
-}
+) => void | Promise<void>;
 
 /**
  * Function that returns the appropriate Promise<Result> given the output of CSE machine evaluating, depending
@@ -73,7 +70,7 @@ export function CSEResultPromise(context: Context, value: Value): Promise<Result
       resolve({ status: "suspended-cse-eval", context });
     } else if (value.type === "error") {
       const msg = value.message;
-      const representation = new Representation(cseFinalPrint + msg);
+      const representation = new Representation(msg);
       resolve({ status: "finished", context, value, representation });
     } else {
       const representation = new Representation(toPythonString(value));
@@ -93,26 +90,26 @@ let source = "";
  * @param options Evaluation options.
  * @returns The result of running the CSE machine.
  */
-export function evaluate(
+export async function evaluate(
   code: string,
   program: StmtNS.Stmt,
   context: Context,
   options: RecursivePartial<IOptions> = {},
-): Value {
+): Promise<Value> {
   source = code;
   try {
     // TODO: is undefined variables check necessary for Python?
     // checkProgramForUndefinedVariables(program, context)
   } catch (error) {
-    return { type: "error", message: error instanceof Error ? error.message : String(error) };
+    return displayError(context, error, ErrorType.EVALUATOR_RUNTIME);
   }
 
   try {
     context.runtime.isRunning = true;
     context.control = new Control(program);
     context.stash = new Stash();
     // Adaptation for new feature
-    const result = runCSEMachine(
+    const result = await runCSEMachine(
       code,
       context,
       context.control,
@@ -121,9 +118,9 @@ export function evaluate(
       options.stepLimit!,
       options.isPrelude,
     );
-    return context.output ? { type: "string", value: context.output } : result;
+    return result;
   } catch (error) {
-    return { type: "error", message: error instanceof Error ? error.message : String(error) };
+    return await displayError(context, error, ErrorType.EVALUATOR_RUNTIME);
   } finally {
     context.runtime.isRunning = false;
   }
@@ -179,15 +176,15 @@ export function evaluate(
  * @param isPrelude Whether the program is the prelude.
  * @returns The top value of the stash after execution.
  */
-export function runCSEMachine(
+export async function runCSEMachine(
   code: string,
   context: Context,
   control: Control,
   stash: Stash,
   envSteps: number,
   stepLimit: number,
   isPrelude: boolean = false,
-): Value {
+): Promise<Value> {
   const eceState = generateCSEMachineStateStream(
     code,
     context,
@@ -200,7 +197,7 @@ export function runCSEMachine(
 
   // Execute the generator until it completes
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  for (const _value of eceState) {
+  for await (const _value of eceState) {
   }
 
   // Return the value at the top of the storage as the result
@@ -219,7 +216,7 @@ export function runCSEMachine(
  * @param isPrelude Whether the program is the prelude.
  * @yields The current state of the stash, control stack, and step count.
  */
-export function* generateCSEMachineStateStream(
+export async function* generateCSEMachineStateStream(
   code: string,
   context: Context,
   control: Control,
@@ -265,7 +262,7 @@ export function* generateCSEMachineStateStream(
       context.runtime.nodes.shift();
       context.runtime.nodes.unshift(command);
 
-      cmdEvaluators[nodeType](code, command, context, control, stash, isPrelude);
+      await cmdEvaluators[nodeType](code, command, context, control, stash, isPrelude);
 
       if (context.runtime.break && context.runtime.debuggerOn) {
         // TODO
@@ -278,7 +275,7 @@ export function* generateCSEMachineStateStream(
     } else {
       // Command is an instruction
       const instr = command as Instr;
-      cmdEvaluators[instr.instrType](code, command, context, control, stash, isPrelude);
+      await cmdEvaluators[instr.instrType](code, command, context, control, stash, isPrelude);
     }
 
     command = control.peek();
@@ -760,7 +757,7 @@ const cmdEvaluators: { [type: string]: CmdEvaluator } = {
     stash.pop();
   },
 
-  [InstrType.APPLICATION]: function (
+  [InstrType.APPLICATION]: async function (
     code: string,
     command: ControlItem,
     context: Context,
@@ -802,7 +799,7 @@ const cmdEvaluators: { [type: string]: CmdEvaluator } = {
       }
     } else {
       if (callable && callable.type === "builtin") {
-        const result = callable.func(args, code, instr.srcNode, context);
+        const result = await callable.func(args, code, instr.srcNode, context);
         stash.push(result);
       }
     }

src/cse-machine/stash.ts

@@ -81,7 +81,9 @@ export interface NoneValue {
 export interface BuiltinValue {
   type: "builtin";
   name: string;
-  func: (args: Value[], code: string, command: ControlItem, context: Context) => Value;
+  func:
+    | ((args: Value[], code: string, command: ControlItem, context: Context) => Value)
+    | ((args: Value[], code: string, command: ControlItem, context: Context) => Promise<Value>);
 }
 
 export class Stash extends Stack<Value> {

src/cse-machine/streams.ts

@@ -0,0 +1,97 @@
+import { ConductorError, ErrorType } from "@sourceacademy/conductor/common";
+import { IRunnerPlugin } from "@sourceacademy/conductor/runner";
+import { Context } from "./context";
+import { ErrorValue } from "./stash";
+
+export type WritableContext<T> = {
+  stream: WritableStream<T>;
+  writer: WritableStreamDefaultWriter<T>;
+};
+export type ReadableContext<T> = {
+  stream: ReadableStream<T>;
+  reader: ReadableStreamDefaultReader<T>;
+};
+
+export function createOutputStream(conductor: IRunnerPlugin): WritableContext<string> {
+  const stream = new WritableStream<string>({
+    write: chunk => {
+      conductor.sendOutput(chunk);
+    },
+  });
+  const writer = stream.getWriter();
+  return { stream, writer };
+}
+
+export function createErrorStream(conductor: IRunnerPlugin): WritableContext<ConductorError> {
+  const stream = new WritableStream<ConductorError>({
+    write: chunk => {
+      conductor.sendError(chunk);
+    },
+  });
+
+  const writer = stream.getWriter();
+  return { stream, writer };
+}
+
+export const createInputStream = (conductor: IRunnerPlugin): ReadableContext<string> => {
+  const stream = new ReadableStream<string>({
+    async pull(controller) {
+      const input = await conductor.requestInput();
+      controller.enqueue(input);
+    },
+  });
+  const reader = stream.getReader();
+  return { stream, reader };
+};
+
+export const displayError = async (
+  context: Context,
+  error: unknown,
+  type: ErrorType,
+): Promise<ErrorValue> => {
+  const name =
+    typeof error === "object" && error !== null && "name" in error && typeof error.name === "string"
+      ? error.name
+      : "Error";
+  const message =
+    typeof error === "object" &&
+    error !== null &&
+    "message" in error &&
+    typeof error.message === "string"
+      ? error.message
+      : String(error);
+  if (context.streams.initialised) {
+    await context.streams.stderr.writer.write({ name, message, errorType: type });
+  }
+  return { type: "error", message: message };
+};
+
+export const displayOutput = async (context: Context, output: string) => {
+  if (context.streams.initialised) {
+    await context.streams.stdout.writer.write(output);
+  }
+};
+
+export const receiveInput = async (context: Context): Promise<string> => {
+  if (context.streams.initialised) {
+    const reader = context.streams.stdin.reader;
+    const { value } = await reader.read();
+    return value ?? "";
+  }
+  return "";
+};
+
+export const destroyStreams = async (context: Context) => {
+  if (context.streams.initialised) {
+    context.streams.stdout.writer.releaseLock();
+    context.streams.stderr.writer.releaseLock();
+    context.streams.stdin.reader.releaseLock();
+
+    await Promise.allSettled([
+      context.streams.stdout.stream.close(),
+      context.streams.stderr.stream.close(),
+      context.streams.stdin.stream.cancel(),
+    ]);
+  }
+  context.streams = { initialised: false };
+};

src/runner/pyRunner.ts

@@ -1,9 +1,9 @@
+import { StmtNS } from "../ast-types";
 import { Context } from "../cse-machine/context";
 import { CSEResultPromise, evaluate } from "../cse-machine/interpreter";
-import { RecursivePartial, Result } from "../types";
 import { parse } from "../parser/parser-adapter";
 import { analyze } from "../resolver/analysis";
-import { StmtNS } from "../ast-types";
+import { RecursivePartial, Result } from "../types";
 
 type Stmt = StmtNS.Stmt;
 
@@ -33,12 +33,12 @@ export async function runInContext(
   return result;
 }
 
-export function runCSEMachine(
+export async function runCSEMachine(
   code: string,
   program: Stmt,
   context: Context,
   options: RecursivePartial<IOptions> = {},
 ): Promise<Result> {
-  const result = evaluate(code, program, context, options as IOptions);
+  const result = await evaluate(code, program, context, options as IOptions);
   return CSEResultPromise(context, result);
 }