const DEBUG = false ; export const VERSION = "5.3" ; import { EventEmitter } from "events" ; import { exists , unlink } from "./async-utils-node" ; import { createMainChannel } from "enchannel-zmq-backend" ; import { Channels , MessageType } from "@nteract/messaging" ; const { merge , copy , deep_copy , original_path , path_split , uuid , len , is_array , } = require ( "@cocalc/util/misc" ) ; import { SyncDB } from "@cocalc/sync/editor/db/sync" ; const { key_value_store } = require ( "@cocalc/util/key-value-store" ) ; import { get_blob_store } from "./jupyter-blobs-sqlite" ; import { JUPYTER_MIMETYPES } from "@cocalc/frontend/jupyter/util" ; import { is_likely_iframe , process as iframe_process , } from "@cocalc/frontend/jupyter/iframe" ; import { remove_redundant_reps } from "@cocalc/frontend/jupyter/import-from-ipynb" ; import { retry_until_success } from "@cocalc/util/async-utils" ; import { callback } from "awaiting" ; import { reuseInFlight } from "async-await-utils/hof" ; import { nbconvert } from "./convert" ; import { ExecOpts , KernelInfo , CodeExecutionEmitterInterface , } from "@cocalc/frontend/jupyter/project-interface" ; import { CodeExecutionEmitter } from "./execute-code" ; import { JupyterActions } from "@cocalc/frontend/jupyter/project-actions" ; import { JupyterStore } from "@cocalc/frontend/jupyter/store" ; import { JupyterKernelInterface } from "@cocalc/frontend/jupyter/project-interface" ; import { launch_jupyter_kernel , LaunchJupyterOpts , } from "./launch_jupyter_kernel" ; const SAGE_JUPYTER_ENV = merge ( copy ( process . env ) , { PYTHONUSERBASE : `${ process . env . HOME }/.local` , PYTHON_EGG_CACHE : `${ process . env . HOME }/.sage/.python-eggs` , R_MAKEVARS_USER : `${ process . env . HOME }/.sage/R/Makevars.user` , } ) ; export function jupyter_backend ( syncdb , client ) { const dbg = client . dbg ( "jupyter_backend" ) ; dbg ( ) ; const app_framework = require ( "@cocalc/frontend/app-framework" ) ; const project_id = client . client_id ( ) ; const path = original_path ( syncdb . get_path ( ) ) ; const redux_name = app_framework . redux_name ( project_id , path ) ; if ( app_framework . redux . getStore ( redux_name ) != null && app_framework . redux . getActions ( redux_name ) != null ) { return ; } const store = app_framework . redux . createStore ( redux_name , JupyterStore ) ; const actions = app_framework . redux . createActions ( redux_name , JupyterActions ) ; actions . _init ( project_id , path , syncdb , store , client ) ; syncdb . once ( "error" , ( err ) => dbg ( `syncdb ERROR -- ${ err }` ) ) ; syncdb . once ( "ready" , ( ) => dbg ( "syncdb ready" ) ) ; } export async function remove_jupyter_backend ( path , project_id ) { try { await get_existing_kernel ( path ) ?. close ( ) ; } catch ( _err ) { } const app_framework = require ( "@cocalc/frontend/app-framework" ) ; const redux_name = app_framework . redux_name ( project_id , path ) ; const actions = app_framework . redux . getActions ( redux_name ) ; if ( actions != null ) { try { await actions . close ( ) ; } catch ( _err ) { } } app_framework . redux . removeStore ( redux_name ) ; app_framework . redux . removeActions ( redux_name ) ; } class Client { client_id ( ) { return "123e4567-e89b-12d3-a456-426655440000" ; } is_project ( ) { return true ; } dbg ( f ) { return ( ... m ) => console . log ( new Date ( ) , `Client.${ f }: ` , ... m ) ; } } interface KernelParams { name : string ; client ? : Client ; verbose ? : boolean ; path : string ; actions ? : any ; } export function kernel ( opts ) { if ( opts . verbose === undefined ) { opts . verbose = true ; } if ( opts . client === undefined ) { opts . client = new Client ( ) ; } return new JupyterKernel ( opts . name , opts . verbose ? opts . client . dbg : undefined , opts . path , opts . actions ) ; } const _jupyter_kernels = { } ; export class JupyterKernel extends EventEmitter implements JupyterKernelInterface { public name : string ; public store : any ; public readonly identity : string = uuid ( ) ; private stderr : string = "" ; private _dbg : Function ; private _path : string ; private _actions : any ; private _state : string ; private _directory : string ; private _filename : string ; private _kernel : any ; private _kernel_info : KernelInfo ; _execute_code_queue : CodeExecutionEmitter [ ] = [ ] ; public channel ? : Channels ; private has_ensured_running : boolean = false ; constructor ( name , _dbg , _path , _actions ) { super ( ) ; this . spawn = reuseInFlight ( this . spawn . bind ( this ) ) ; this . kernel_info = reuseInFlight ( this . kernel_info . bind ( this ) ) ; this . nbconvert = reuseInFlight ( this . nbconvert . bind ( this ) ) ; this . ensure_running = reuseInFlight ( this . ensure_running . bind ( this ) ) ; this . close = this . close . bind ( this ) ; this . process_output = this . process_output . bind ( this ) ; this . name = name ; this . _dbg = _dbg ; this . _path = _path ; this . _actions = _actions ; this . store = key_value_store ( ) ; const { head , tail } = path_split ( this . _path ) ; this . _directory = head ; this . _filename = tail ; this . _set_state ( "off" ) ; this . _execute_code_queue = [ ] ; if ( _jupyter_kernels [ this . _path ] !== undefined ) { _jupyter_kernels [ this . _path ] . close ( ) ; } _jupyter_kernels [ this . _path ] = this ; const dbg = this . dbg ( "constructor" ) ; dbg ( ) ; process . on ( "exit" , this . close ) ; this . setMaxListeners ( 100 ) ; } public get_path ( ) { return this . _path ; } private _set_state ( state ) { if ( this . _state == state ) return ; this . _state = state ; this . emit ( "state" , this . _state ) ; this . emit ( this . _state ) ; } get_state ( ) { return this . _state ; } async spawn ( spawn_opts ? ) { if ( this . _state === "closed" ) { throw Error ( "closed" ) ; } if ( [ "running" , "starting" ] . includes ( this . _state ) ) { return ; } this . _set_state ( "spawning" ) ; const dbg = this . dbg ( "spawn" ) ; dbg ( "spawning kernel..." ) ; const opts = { detached : true , env : spawn_opts ?. env ?? { } , } ; if ( this . name . indexOf ( "sage" ) == 0 ) { dbg ( "setting special environment for sage.* kernels" ) ; opts . env = merge ( opts . env , SAGE_JUPYTER_ENV ) ; } opts . env . PLOTLY_RENDERER = "colab" ; opts . env . COCALC_JUPYTER_FILENAME = this . _path ; opts . env . COCALC_JUPYTER_KERNELNAME = this . name ; if ( this . _directory !== "" ) { opts . cwd = this . _directory ; } try { dbg ( "launching kernel interface..." ) ; this . _kernel = await launch_jupyter_kernel ( this . name , opts ) ; await this . finish_spawn ( ) ; } catch ( err ) { if ( this . _state === "closed" ) { throw Error ( "closed" ) ; } this . _set_state ( "off" ) ; throw err ; } } get_spawned_kernel ( ) { return this . _kernel ; } public get_connection_file ( ) { return this . _kernel ?. connection_file ; } private async finish_spawn ( ) { const dbg = this . dbg ( "finish_spawn" ) ; dbg ( "now finishing spawn of kernel..." ) ; this . _kernel . spawn . on ( "error" , ( err ) => { const error = `${ err }\n${ this . stderr }` ; dbg ( "kernel error" , error ) ; this . emit ( "kernel_error" , error ) ; } ) ; this . stderr = "" ; this . _kernel . spawn . stderr . on ( "data" , ( data ) => { const s = data . toString ( ) ; this . stderr += s ; if ( this . stderr . length > 5000 ) { this . stderr = this . stderr . slice ( this . stderr . length - 4000 ) ; } } ) ; this . _kernel . spawn . stdout . on ( "data" , ( _data ) => { } ) ; this . channel = await createMainChannel ( this . _kernel . config , "" , this . identity ) ; if ( DEBUG ) { this . low_level_dbg ( ) ; } this . channel ?. subscribe ( ( mesg ) => { switch ( mesg . channel ) { case "shell" : this . _set_state ( "running" ) ; this . emit ( "shell" , mesg ) ; break ; case "stdin" : this . emit ( "stdin" , mesg ) ; break ; case "iopub" : this . _set_state ( "running" ) ; if ( mesg . content != null && mesg . content . execution_state != null ) { this . emit ( "execution_state" , mesg . content . execution_state ) ; } if ( ( mesg . content != null ? mesg . content . comm_id : undefined ) !== undefined ) { this . process_comm_message_from_kernel ( mesg ) ; break ; } if ( this . _actions != null && this . _actions . capture_output_message ( mesg ) ) { break ; } this . emit ( "iopub" , mesg ) ; break ; } } ) ; this . _kernel . spawn . on ( "exit" , ( exit_code , signal ) => { this . dbg ( "kernel_exit" ) ( `spawned kernel terminated with exit code ${ exit_code } (signal=${ signal }); stderr=${ this . stderr }` ) ; const stderr = this . stderr ? `\n...\n${ this . stderr }` : "" ; if ( signal != null ) { this . emit ( "kernel_error" , `Kernel last terminated by signal ${ signal }.${ stderr }` ) ; } else if ( exit_code != null ) { this . emit ( "kernel_error" , `Kernel last exited with code ${ exit_code }.${ stderr }` ) ; } this . close ( ) ; } ) ; this . _set_state ( "starting" ) ; if ( this . _state === "closed" ) { throw Error ( "closed" ) ; } } async _get_kernel_info ( ) { const dbg = this . dbg ( "_get_kernel_info" ) ; const that = this ; async function f ( ) { if ( that . _state == "closed" ) return ; dbg ( "calling kernel_info_request..." , that . _state ) ; await that . call ( "kernel_info_request" ) ; dbg ( "called kernel_info_request" , that . _state ) ; if ( that . _state === "starting" ) { throw Error ( "still starting" ) ; } } dbg ( "getting kernel info to be certain kernel is fully usable..." ) ; await retry_until_success ( { start_delay : 500 , max_delay : 5000 , factor : 1.4 , max_time : 60000 , f : f , log : function ( ... args ) { dbg ( "retry_until_success" , ... args ) ; } , } ) ; if ( this . _state == "closed" ) { throw Error ( "closed" ) ; } dbg ( "successfully got kernel info" ) ; } signal ( signal ) { const dbg = this . dbg ( "signal" ) ; const spawn = this . _kernel != null ? this . _kernel . spawn : undefined ; const pid = spawn ?. pid ; dbg ( `pid=${ pid }, signal=${ signal }` ) ; if ( pid == null ) return ; try { this . clear_execute_code_queue ( ) ; process . kill ( - pid , signal ) ; } catch ( err ) { dbg ( `error: ${ err }` ) ; } } async close ( ) { this . dbg ( "close" ) ( ) ; if ( this . _state === "closed" ) { return ; } this . _set_state ( "closed" ) ; if ( this . store != null ) { this . store . close ( ) ; delete this . store ; } const kernel = _jupyter_kernels [ this . _path ] ; if ( kernel != null && kernel . identity === this . identity ) { delete _jupyter_kernels [ this . _path ] ; } this . removeAllListeners ( ) ; process . removeListener ( "exit" , this . close ) ; if ( this . _kernel != null ) { if ( this . _kernel . spawn != null ) { this . _kernel . spawn . removeAllListeners ( ) ; if ( this . _kernel . spawn . pid ) { try { process . kill ( - this . _kernel . spawn . pid , "SIGTERM" ) ; } catch ( err ) { } } this . _kernel . spawn . close ?. ( ) ; } if ( await exists ( this . _kernel . connection_file ) ) { try { await unlink ( this . _kernel . connection_file ) ; } catch { } } delete this . _kernel ; delete this . channel ; } if ( this . _execute_code_queue != null ) { for ( const code_snippet of this . _execute_code_queue ) { code_snippet . close ( ) ; } this . _execute_code_queue = [ ] ; } } public dbg ( f , trunc = 1000 ) { if ( ! this . _dbg ) { return function ( ) { } ; } else { return this . _dbg ( `jupyter.Kernel('${ this . name }',path='${ this . _path }').${ f }` , trunc ) ; } } low_level_dbg ( ) { const dbg = this . dbg ( "low_level_debug" , 10000 ) ; this . _kernel . spawn . all ?. on ( "data" , ( data ) => dbg ( "STDIO" , data . toString ( ) ) ) ; this . channel ?. subscribe ( ( mesg ) => { dbg ( JSON . stringify ( mesg ) ) ; } ) ; } private async ensure_running ( ) { const dbg = this . dbg ( "ensure_running" ) ; dbg ( this . _state ) ; if ( this . _state == "closed" ) { throw Error ( "closed so not possible to ensure running" ) ; } if ( this . _state == "running" ) { return ; } dbg ( "spawning" ) ; await this . spawn ( ) ; if ( ! this . has_ensured_running ) { dbg ( "waiting for kernel info" ) ; this . has_ensured_running = true ; await this . _get_kernel_info ( ) ; dbg ( "got kernel info" ) ; } } execute_code ( opts ) { if ( opts . halt_on_error === undefined ) { opts . halt_on_error = true ; } if ( this . _state === "closed" ) { throw Error ( "closed" ) ; } const code = new CodeExecutionEmitter ( this , opts ) ; this . _execute_code_queue . push ( code ) ; if ( this . _execute_code_queue . length === 1 ) { this . _process_execute_code_queue ( ) ; } return code ; } cancel_execute ( id ) { if ( this . _state === "closed" ) { return ; } const dbg = this . dbg ( `cancel_execute(id='${ id }')` ) ; if ( this . _execute_code_queue == null || this . _execute_code_queue . length === 0 ) { dbg ( "nothing to do" ) ; return ; } if ( this . _execute_code_queue . length > 1 ) { dbg ( "mutate this._execute_code_queue removing everything with the given id" ) ; for ( let i = this . _execute_code_queue . length - 1 ; i -- ; i >= 1 ) { const code = this . _execute_code_queue [ i ] ; if ( code . id === id ) { dbg ( `removing entry ${ i } from queue` ) ; this . _execute_code_queue . splice ( i , 1 ) ; code . cancel ( ) ; } } } if ( this . _execute_code_queue [ 0 ] . id === id ) { dbg ( "interrupting running computation" ) ; this . signal ( "SIGINT" ) ; } } async _process_execute_code_queue ( ) { const dbg = this . dbg ( "_process_execute_code_queue" ) ; dbg ( `state='${ this . _state }'` ) ; if ( this . _state === "closed" ) { dbg ( "closed" ) ; return ; } if ( this . _execute_code_queue == null ) { dbg ( "no queue" ) ; return ; } const n = this . _execute_code_queue . length ; if ( n === 0 ) { dbg ( "queue is empty" ) ; return ; } dbg ( `queue has ${ n } items; ensure kernel running` ) ; try { await this . ensure_running ( ) ; this . _execute_code_queue [ 0 ] . go ( ) ; } catch ( err ) { dbg ( `error running kernel -- ${ err }` ) ; for ( const code of this . _execute_code_queue ) { code . throw_error ( err ) ; } this . _execute_code_queue = [ ] ; } } public clear_execute_code_queue ( ) { const dbg = this . dbg ( "_clear_execute_code_queue" ) ; if ( this . _state === "closed" ) { dbg ( "no op since state is closed" ) ; return ; } if ( this . _execute_code_queue == null ) { dbg ( "nothing to do since queue is null" ) ; return ; } dbg ( `clearing queue of size ${ this . _execute_code_queue . length }` ) ; const mesg = { done : true } ; for ( const code_execution_emitter of this . _execute_code_queue . slice ( 1 ) ) { code_execution_emitter . emit_output ( mesg ) ; code_execution_emitter . close ( ) ; } this . _execute_code_queue = [ ] ; } async execute_code_now ( opts ) { this . dbg ( "execute_code_now" ) ( ) ; if ( this . _state === "closed" ) { throw Error ( "closed" ) ; } if ( opts . halt_on_error === undefined ) { opts . halt_on_error = true ; } await this . ensure_running ( ) ; return await new CodeExecutionEmitter ( this , opts ) . go ( ) ; } get_blob_store ( ) { return get_blob_store ( ) ; } process_output ( content ) { if ( this . _state === "closed" ) { return ; } const dbg = this . dbg ( "process_output" ) ; dbg ( JSON . stringify ( content ) ) ; if ( content . data == null ) { dbg ( "no data, so nothing to do" ) ; return ; } remove_redundant_reps ( content . data ) ; let type ; for ( type of JUPYTER_MIMETYPES ) { if ( content . data [ type ] != null ) { if ( type . split ( "/" ) [ 0 ] === "image" || type === "application/pdf" ) { content . data [ type ] = get_blob_store ( ) ?. save ( content . data [ type ] , type ) ; } else if ( type === "text/html" && is_likely_iframe ( content . data [ type ] ) ) { content . data [ "iframe" ] = iframe_process ( content . data [ type ] , get_blob_store ( ) ) ; delete content . data [ type ] ; } } } } async call ( msg_type , content ? ) { this . dbg ( "call" ) ( msg_type ) ; if ( ! this . has_ensured_running ) { await this . ensure_running ( ) ; } if ( this . channel == null || this . _state == "closed" ) { throw Error ( "not running, so can't call" ) ; } const message = { parent_header : { } , metadata : { } , channel : "shell" , content , header : { msg_id : uuid ( ) , username : "" , session : "" , msg_type : msg_type as MessageType , version : VERSION , date : new Date ( ) . toISOString ( ) , } , } ; this . channel ?. next ( message ) ; let the_mesg = undefined ; const wait_for_response = ( cb ) => { const f = ( mesg ) => { if ( mesg . parent_header . msg_id === message . header . msg_id ) { this . removeListener ( "shell" , f ) ; this . removeListener ( "closed" , g ) ; mesg = deep_copy ( mesg . content ) ; if ( len ( mesg . metadata ) === 0 ) { delete mesg . metadata ; } the_mesg = mesg ; cb ( ) ; } } ; const g = ( ) => { this . removeListener ( "shell" , f ) ; this . removeListener ( "closed" , g ) ; cb ( "closed" ) ; } ; this . on ( "shell" , f ) ; this . on ( "closed" , g ) ; } ; await callback ( wait_for_response ) ; return the_mesg ; } async complete ( opts ) { const dbg = this . dbg ( "complete" ) ; dbg ( `code='${ opts . code }', cursor_pos='${ opts . cursor_pos }'` ) ; return await this . call ( "complete_request" , opts ) ; } async introspect ( opts ) { const dbg = this . dbg ( "introspect" ) ; dbg ( `code='${ opts . code }', cursor_pos='${ opts . cursor_pos }', detail_level=${ opts . detail_level }` ) ; return await this . call ( "inspect_request" , opts ) ; } async kernel_info ( ) { if ( this . _kernel_info !== undefined ) { return this . _kernel_info ; } const info = await this . call ( "kernel_info_request" ) ; info . nodejs_version = process . version ; if ( this . _actions != null ) { info . start_time = this . _actions . store . get ( "start_time" ) ; } this . _kernel_info = info ; return info ; } async save_ipynb_file ( ) { if ( this . _actions != null ) { await this . _actions . save_ipynb_file ( ) ; } else { throw Error ( "save_ipynb_file -- ERROR: actions not known" ) ; } } more_output ( id ) { if ( id == null ) { throw new Error ( "must specify id" ) ; } if ( this . _actions == null ) { throw new Error ( "must have redux actions" ) ; } return this . _actions . store . get_more_output ( id ) || [ ] ; } async nbconvert ( args , timeout ? ) { if ( timeout === undefined ) { timeout = 30 ; } if ( ! is_array ( args ) ) { throw new Error ( "args must be an array" ) ; } args = copy ( args ) ; args . push ( "--" ) ; args . push ( this . _filename ) ; await nbconvert ( { args , timeout , directory : this . _directory , } ) ; } async load_attachment ( path ) { const dbg = this . dbg ( "load_attachment" ) ; dbg ( `path='${ path }'` ) ; if ( path [ 0 ] !== "/" ) { path = process . env . HOME + "/" + path ; } async function f ( ) { const bs = get_blob_store ( ) ; if ( bs == null ) throw new Error ( "BlobStore not available" ) ; return bs . readFile ( path , "base64" ) ; } try { return await retry_until_success ( { f : f , max_time : 30000 , } ) ; } catch ( err ) { unlink ( path ) ; throw err ; } } process_attachment ( base64 , mime ) { return get_blob_store ( ) ?. save ( base64 , mime ) ; } process_comm_message_from_kernel ( mesg ) { const dbg = this . dbg ( "process_comm_message_from_kernel" ) ; dbg ( mesg ) ; this . _actions . process_comm_message_from_kernel ( mesg ) ; } public send_comm_message_to_kernel ( msg_id , comm_id , data ) { const dbg = this . dbg ( "send_comm_message_to_kernel" ) ; const message = { parent_header : { } , metadata : { } , channel : "shell" , content : { comm_id , data } , header : { msg_id , username : "user" , session : "" , msg_type : "comm_msg" as MessageType , version : VERSION , date : new Date ( ) . toISOString ( ) , } , } ; dbg ( "sending " , JSON . stringify ( message ) ) ; this . channel ?. next ( message ) ; } } export function get_existing_kernel ( path ) { return _jupyter_kernels [ path ] ; } export function get_kernel_by_pid ( pid ) { for ( const kernel of Object . values ( _jupyter_kernels ) ) { if ( kernel . get_spawned_kernel ( ) ?. spawn . pid === pid ) { return kernel ; } } return ; }