stand-alone remote runtime with useMQ

Author: lihebi

cpkernel/docker-compose.yml

@@ -0,0 +1,18 @@
+# Use postgres/example user/password credentials
+version: "3.5"
+
+services:
+  rabbitmq:
+    image: rabbitmq:3-management-alpine
+    restart: always
+    volumes:
+      - ./rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins
+    ports:
+      # The AMQP port
+      # - 5672:5672
+      # the management UI
+      - 15672:15672
+      # STOMP-over-WebSockets
+      - 15674:15674
+      # 61613 STOMP clients, internally used by kernel.js
+      - 61613:61613

cpkernel/rabbitmq/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_management,rabbitmq_prometheus,rabbitmq_web_stomp].

cpkernel/src/kernel.ts

@@ -12,6 +12,8 @@ import { v4 as uuidv4 } from "uuid";
 import crypto from "crypto";
 import Docker from "dockerode";
 
+import Stomp from "stompjs";
+
 import path from "path";
 
 import os from "os";
@@ -558,16 +560,16 @@ async function createContainer(image, name, network) {
   });
 }
 
-// var mq_client = Stomp.overTCP("rabbitmq", 61613);
-
 class MyMqSocket {
   queue;
-  constructor(queue) {
+  mq_client;
+  constructor(queue, mq_client) {
     this.queue = queue;
+    this.mq_client = mq_client
   }
   send(obj) {
     // FIXME need to make sure it is connected
-    // mq_client.send(this.queue, {}, obj);
+    this.mq_client.send(this.queue, {}, obj);
   }
 }
 
@@ -718,6 +720,7 @@ function handleIOPub_stream({ msgs, socket }) {
   }
 }
 
+let mq_client = null;
 export class CodePodKernel {
   lang;
   startupFile;
@@ -727,6 +730,8 @@ export class CodePodKernel {
   sessionId;
   wire;
   socket;
+  mq_socket;
+  useMQ;
   mapEval({ code, namespace }) { }
   mapAddImport({
     from,
@@ -749,9 +754,10 @@ export class CodePodKernel {
     name: string;
   }) { }
   constructor() { }
-  async init({ sessionId, socket }) {
+  async init({ sessionId, socket, useMQ }) {
     console.log("=== INIT!!");
     this.sessionId = sessionId;
+    this.useMQ = useMQ;
     let network = process.env["KERNEL_NETWORK"] || "codepod";
     let name = `cpkernel_${network}_${sessionId}_${this.lang}`;
     // await removeContainer(name);
@@ -774,7 +780,26 @@ export class CodePodKernel {
     console.log("connecting to zmq ..");
     // this.wire = new ZmqWire(JSON.parse(readFileSync(this.fname)), ip);
     this.wire = new ZmqWire(spec, "127.0.0.1");
-    // this.mq_socket = new MyMqSocket(sessionId);
+
+    if (useMQ) {
+      if (!mq_client) {
+        mq_client = Stomp.overTCP("localhost", 61613);
+        mq_client.connect(
+          "guest",
+          "guest",
+          function () {
+            console.log("connected");
+          },
+          function () {
+            console.log("error");
+            throw new Error("Cannot connect to RabbitMQ server");
+          },
+          "/"
+        );
+      }
+      this.mq_socket = new MyMqSocket(sessionId, mq_client);
+    }
+    
     if (socket) {
       // listen to IOPub here
       this.addSocket(socket);
@@ -809,7 +834,9 @@ export class CodePodKernel {
     }
     this.socket = socket;
     // DEBUG
-    // socket = this.mq_socket;
+    if (this.useMQ) {
+      socket = this.mq_socket;
+    }
     this.wire.setOnIOPub((topic, msgs) => {
       // console.log("-----", topic, msgs);
       // iracket's topic seems to be an ID. I should use msg type instead
@@ -846,7 +873,7 @@ export class CodePodKernel {
     this.wire.setOnShell((msgs) => {
       // DEBUG
       // socket = this.mq_socket;
-      socket = this.socket;
+      // socket = this.socket;
       switch (msgs.header.msg_type) {
         case "execute_reply":
           {
@@ -1145,22 +1172,24 @@ export async function createKernel({
   lang,
   sessionId,
   socket,
+  useMQ
 }: {
   lang: string;
   sessionId: string;
   socket: any;
+  useMQ: boolean;
 }) {
   let kernels = detectKernels()
   console.log("===", "createKernel", lang);
   switch (lang) {
     case "julia":
-      return await new JuliaKernel({ kernelJson: kernels["julia"] }).init({ sessionId, socket });
+      return await new JuliaKernel({ kernelJson: kernels["julia"] }).init({ sessionId, socket, useMQ });
     case "javascript":
-      return await new JavascriptKernel({ kernelJson: kernels["javascript"] }).init({ sessionId, socket });
+      return await new JavascriptKernel({ kernelJson: kernels["javascript"] }).init({ sessionId, socket, useMQ });
     case "racket":
-      return await new RacketKernel({ kernelJson: kernels["racket"] }).init({ sessionId, socket });
+      return await new RacketKernel({ kernelJson: kernels["racket"] }).init({ sessionId, socket, useMQ });
     case "python":
-      return await new PythonKernel({ kernelJson: kernels["python"] }).init({ sessionId, socket });
+      return await new PythonKernel({ kernelJson: kernels["python"] }).init({ sessionId, socket, useMQ });
     default:
       console.log("ERROR: language not implemented", lang);
     // throw new Error(`Language not valid: ${lang}`);

cpkernel/src/server.js

@@ -26,7 +26,8 @@ export function startSocketServer() {
     // listenOnKernelManagement(socket);
     // listenOnSessionManagement(socket);
     // listenOnRunCode(socket);
-    listenOnMessage(socket);
+    // useMQ=true when creating a stand-alone runtime server
+    listenOnMessage(socket, true);
   });
 
   http_server.listen({ port: 14321 }, () => {

cpkernel/src/socket.js

@@ -152,7 +152,7 @@ async function killKernel({ sessionId, lang }) {
   }
 }
 
-async function getSessionKernel({ sessionId, lang, socket }) {
+async function getSessionKernel({ sessionId, lang, socket, useMQ }) {
   if (!sessionId || !lang) {
     console.log("sesisonId or lang is undefined", sessionId, lang);
     return null;
@@ -173,18 +173,18 @@ async function getSessionKernel({ sessionId, lang, socket }) {
   // FIXME what if the process never finish?
   session[lang] = "spawning";
   console.log("spawning kernel ..");
-  let kernel = await createKernel({ lang, sessionId, socket });
+  let kernel = await createKernel({ lang, sessionId, socket, useMQ });
   console.log("returning the newly spawned kernel");
   session[lang] = kernel;
   return kernel;
 }
 
-export function listenOnMessage(socket) {
+export function listenOnMessage(socket, useMQ = false) {
   socket.on("message", async (msg) => {
     let { type, payload } = JSON.parse(msg.toString());
     if (type === "ping") return;
     let { sessionId, lang } = payload;
-    let kernel = await getSessionKernel({ sessionId, lang, socket });
+    let kernel = await getSessionKernel({ sessionId, lang, socket, useMQ });
     if (!kernel) {
       console.log("ERROR: kernel error");
       return;