import debug from "debug" ; const L = debug ( "project:usage-info:server" ) ; import { EventEmitter } from "events" ; import { delay } from "awaiting" ; import { ProjectInfoServer , get_ProjectInfoServer } from "../project-info" ; import { ProjectInfo , Process } from "../project-info/types" ; import { UsageInfo } from "./types" ; function is_diff ( prev , next , key ) { const a = prev [ key ] ?? 0 ; const b = next [ key ] ?? 0 ; if ( a === 0 && b === 0 ) return false ; return Math . abs ( b - a ) / Math . min ( a , b ) > 0.05 ; } export class UsageInfoServer extends EventEmitter { private readonly dbg : Function ; private running = false ; private readonly testing : boolean ; private readonly project_info : ProjectInfoServer ; private readonly path : string ; private info ? : ProjectInfo ; private usage ? : UsageInfo ; private last ? : UsageInfo ; constructor ( path , testing = false ) { super ( ) ; this . testing = testing ; this . path = path ; this . dbg = L ; this . project_info = get_ProjectInfoServer ( ) ; this . dbg ( "starting" ) ; } private async init ( ) { this . project_info . start ( ) ; this . project_info . on ( "info" , ( info ) => { this . info = info ; this . update ( ) ; } ) ; } private path_process ( ) { if ( this . info ?. processes == null ) return ; for ( const p of Object . values ( this . info . processes ) ) { const cocalc = p . cocalc ; if ( cocalc == null || cocalc . type != "jupyter" ) continue ; if ( cocalc . path == this . path ) return p ; } } private proces_tree_stats ( ppid , stats ) { const procs = this . info ?. processes ; if ( procs == null ) return ; for ( const proc of Object . values ( procs ) ) { if ( proc . ppid != ppid ) continue ; this . proces_tree_stats ( proc . pid , stats ) ; stats . mem += proc . stat . mem . rss ; stats . cpu += proc . cpu . pct ; } } private usage_children ( pid ) { const stats = { mem : 0 , cpu : 0 } ; this . proces_tree_stats ( pid , stats ) ; return stats ; } private path_usage_info ( ) { const proc = this . path_process ( ) ; if ( proc == null ) { return { cpu : 0 , mem : 0 , cpu_chld : 0 , mem_chld : 0 } ; } else { const children = this . usage_children ( proc . pid ) ; return { cpu : Math . round ( proc . cpu . pct ) , cpu_chld : Math . round ( children . cpu ) , mem : Math . round ( proc . stat . mem . rss ) , mem_chld : Math . round ( children . mem ) , } ; } } private update ( ) { if ( this . info == null ) { L ( "was told to update, but there is no ProjectInfo" ) ; return ; } const cg = this . info . cgroup ; const du = this . info . disk_usage ; if ( cg == null || du == null ) { this . dbg ( "info incomplete, can't send usage data" ) ; return ; } const mem_rss = cg . mem_stat . total_rss + ( du . tmp ?. usage ?? 0 ) ; const mem_tot = cg . mem_stat . hierarchical_memory_limit ; const usage = { time : Date . now ( ) , ... this . path_usage_info ( ) , mem_limit : mem_tot , cpu_limit : cg . cpu_cores_limit , mem_free : Math . max ( 0 , mem_tot - mem_rss ) , } ; if ( this . should_update ( usage ) ) { this . usage = usage ; this . emit ( "usage" , this . usage ) ; this . last = this . usage ; } } private should_update ( usage ) { if ( this . last == null ) return true ; if ( usage == null ) return false ; const keys = [ "cpu" , "mem" , "cpu_chld" , "mem_chld" ] ; for ( const key of keys ) { if ( ( this . last [ key ] ?? 0 ) >= 1 && ( usage [ key ] ?? 0 ) < 1 ) return true ; if ( is_diff ( usage , this . last , key ) ) return true ; } if ( is_diff ( usage , this . last , "mem_free" ) ) return true ; return false ; } private async get_usage ( ) { this . update ( ) ; return this . usage ; } public stop ( ) { this . running = false ; } public async start ( ) { if ( this . running ) { this . dbg ( "UsageInfoServer already running, cannot be started twice" ) ; } else { await this . _start ( ) ; } } private async _start ( ) { this . dbg ( "start" ) ; if ( this . running ) { throw Error ( "Cannot start UsageInfoServer twice" ) ; } this . running = true ; await this . init ( ) ; const usage = await this . get_usage ( ) ; this . emit ( "usage" , usage ) ; while ( this . testing ) { await delay ( 5000 ) ; const usage = await this . get_usage ( ) ; this . emit ( "usage" , usage ) ; } } } if ( require . main === module ) { const uis = new UsageInfoServer ( "testing.ipynb" , true ) ; uis . start ( ) ; let cnt = 0 ; uis . on ( "usage" , ( usage ) => { console . log ( JSON . stringify ( usage , null , 2 ) ) ; cnt += 1 ; if ( cnt >= 2 ) process . exit ( ) ; } ) ; }