Merge branch 'master' into fs2

Author: williamstein

src/packages/conat/files/read.ts

@@ -41,6 +41,10 @@ for await (const chunk of await a.readFile({project_id:'00847397-d6a8-4cb0-96a8-
 import { conat } from "@cocalc/conat/client";
 import { projectSubject } from "@cocalc/conat/names";
 import { type Subscription } from "@cocalc/conat/core/client";
+import { delay } from "awaiting";
+import { getLogger } from "@cocalc/conat/client";
+
+const logger = getLogger("conat:files:read");
 
 let subs: { [name: string]: Subscription } = {};
 export async function close({ project_id, compute_server_id, name = "" }) {
@@ -72,6 +76,7 @@ export async function createServer({
     compute_server_id,
     name,
   });
+  logger.debug("createServer", { subject });
   const cn = await conat();
   const sub = await cn.subscribe(subject);
   subs[subject] = sub;
@@ -89,36 +94,65 @@ async function listen({ sub, createReadStream }) {
 }
 
 async function handleMessage(mesg, createReadStream) {
+  logger.debug("handleMessage", mesg.subject);
   try {
     await sendData(mesg, createReadStream);
     await mesg.respond(null, { headers: { done: true } });
   } catch (err) {
-    // console.log("sending ERROR", err);
+    logger.debug("handleMessage: ERROR", err);
     mesg.respondSync(null, { headers: { error: `${err}` } });
   }
 }
 
-const MAX_CHUNK_SIZE = 16384 * 16 * 3;
+// 4MB -- chunks may be slightly bigger
+const CHUNK_SIZE = 4194304;
+const CHUNK_INTERVAL = 250;
 
 function getSeqHeader(seq) {
   return { headers: { seq } };
 }
 
 async function sendData(mesg, createReadStream) {
   const { path } = mesg.data;
+  logger.debug("sendData: starting", { path });
   let seq = 0;
+  const chunks: Buffer[] = [];
+  let size = 0;
+  const sendChunks = async () => {
+    // Not only is waiting for the response useful to make sure somebody is listening,
+    // we also use await here partly to space out the messages to avoid saturing
+    // the websocket connection, since doing so would break everything
+    // (heartbeats, etc.) and disconnect us, when transfering a large file.
+    seq += 1;
+    logger.debug("sendData: sending", { path, seq });
+    const data = Buffer.concat(chunks);
+    const { count } = await mesg.respond(data, getSeqHeader(seq));
+    if (count == 0) {
+      logger.debug("sendData: nobody is listening");
+      // nobody is listening so don't waste effort sending...
+      throw Error("receiver is gone");
+    }
+    size = 0;
+    chunks.length = 0;
+    // Delay a little just to give other messages a chance, so we don't get disconnected
+    // e.g., due to lack of heartbeats. Also, this reduces the load on conat-router.
+    await delay(CHUNK_INTERVAL);
+  };
+
   for await (let chunk of createReadStream(path, {
-    highWaterMark: 16384 * 16 * 3,
+    highWaterMark: CHUNK_SIZE,
   })) {
-    // console.log("sending ", { seq, bytes: chunk.length });
-    // We must break the chunk into smaller messages or it will
-    // get bounced by conat...
-    while (chunk.length > 0) {
-      seq += 1;
-      mesg.respondSync(chunk.slice(0, MAX_CHUNK_SIZE), getSeqHeader(seq));
-      chunk = chunk.slice(MAX_CHUNK_SIZE);
+    chunks.push(chunk);
+    size += chunk.length;
+    if (size >= CHUNK_SIZE) {
+      // send it
+      await sendChunks();
     }
   }
+  if (size > 0) {
+    await sendChunks();
+  }
+  logger.debug("sendData: done", { path }, "successfully sent ", seq, "chunks");
 }
 
 export interface ReadFileOptions {
@@ -136,6 +170,7 @@ export async function* readFile({
   name = "",
   maxWait = 1000 * 60 * 10, // 10 minutes
 }: ReadFileOptions) {
+  logger.debug("readFile", { project_id, compute_server_id, path });
   const cn = await conat();
   const subject = getSubject({
     project_id,

src/packages/conat/files/write.ts

@@ -73,6 +73,8 @@ import {
 import { projectSubject } from "@cocalc/conat/names";
 import { type Subscription } from "@cocalc/conat/core/client";
 import { type Readable } from "node:stream";
+import { getLogger } from "@cocalc/conat/client";
+const logger = getLogger("conat:files:write");
 
 function getWriteSubject({ project_id, compute_server_id }) {
   return projectSubject({
@@ -106,6 +108,7 @@ export async function createServer({
   createWriteStream: (path: string) => any;
 }) {
   const subject = getWriteSubject({ project_id, compute_server_id });
+  logger.debug("createServer", { subject });
   let sub = subs[subject];
   if (sub != null) {
     return;
@@ -141,6 +144,7 @@ async function handleMessage({
   let writeStream: null | Awaited<ReturnType<typeof createWriteStream>> = null;
   try {
     const { path, name, maxWait } = mesg.data;
+    logger.debug("handleMessage", { path, name, maxWait });
     writeStream = await createWriteStream(path);
     // console.log("created writeStream");
     writeStream.on("error", (err) => {
@@ -166,12 +170,15 @@ async function handleMessage({
       writeStream.write(chunk);
       chunks += 1;
       bytes += chunk.length;
+      logger.debug("handleMessage -- wrote", { path, name, bytes });
       // console.log("wrote ", bytes);
     }
     writeStream.end();
     writeStream.emit("rename");
     mesg.respondSync({ status: "success", bytes, chunks });
+    logger.debug("handleMessage -- SUCCESS", { path, name });
   } catch (err) {
+    logger.debug("handleMessage: ERROR", err);
     if (!error) {
       mesg.respondSync({ error: `${err}`, status: "error" });
       writeStream?.emit("remove");
@@ -194,6 +201,7 @@ export async function writeFile({
   stream,
   maxWait = 1000 * 60 * 10, // 10 minutes
 }): Promise<{ bytes: number; chunks: number }> {
+  logger.debug("writeFile", { project_id, compute_server_id, path, maxWait });
   const name = randomId();
   try {
     function createReadStream() {

src/packages/jupyter/kernel/kernel.ts

@@ -725,10 +725,7 @@ export class JupyterKernel
       dbg("queue is empty");
       return;
     }
-    dbg(
-      `queue has ${n} items; ensure kernel running`,
-      this._execute_code_queue,
-    );
+    dbg(`queue has ${n} items; ensure kernel running`);
     try {
       await this.ensure_running();
       await this._execute_code_queue[0].go();

src/packages/project/conat/connection.ts

@@ -70,14 +70,9 @@ async function callHub({
   timeout?: number;
 }) {
   const subject = `hub.project.${project_id}.${service}`;
-  try {
-    const data = { name, args };
-    const resp = await client.request(subject, data, { timeout });
-    return resp.data;
-  } catch (err) {
-    err.message = `${err.message} - callHub: subject='${subject}', name='${name}', `;
-    throw err;
-  }
+  const data = { name, args };
+  const resp = await client.request(subject, data, { timeout });
+  return resp.data;
 }
 
 async function versionCheckLoop(client) {