import os from 'os' ; import { spawn , ChildProcess , exec } from 'child_process' ; import fs , { WriteStream } from 'fs' ; import path from 'path' ; import { deepCopy } from '../utils/DeepCopy' ; import { DeepPartial } from '../utils/DeepPartial' ; import { deepMerge } from '../utils/DeepMerge' ; import { FatalError , MatchError , NotSupportedError , } from '../DimensionError' ; import { Design } from '../Design' ; import { Logger } from '../Logger' ; import { Agent } from '../Agent' ; import { Match } from '../Match' ; import Dockerode from 'dockerode' ; import { isChildProcess } from '../utils/TypeGuards' ; import { noop } from '../utils' ; import { removeDirectorySync } from '../utils/System' ; ignore type EngineOptions = MatchEngine . EngineOptions ; /**
 * The Match Engine that takes a  Design and its specified  EngineOptions to form the backend
 * for running matches with agents.
  export class MatchEngine { /** The design the engine runs on */ private design : Design ; /** Engine options */ private engineOptions : EngineOptions = deepCopy ( DefaultMatchEngineOptions ) ; /** Override options */ private overrideOptions : Design . OverrideOptions ; /** Logger */ private log = new Logger ( ) ; /**
     * A coordination signal to ensure that all processes are indeed killed due to asynchronous initialization of agents
     * There is a race condition when a tournament/match is being destroyed and while every match is being destroyed, some
     * matches are in the initialization stage where they call the engine's initialize function. As a result, when we
     * send a match destroy signal, we spawn some processes and haven't spawned some others for the agents. As a result,
     * all processes eventually get spawned but not all are cleaned up and killed.
     */ killOffSignal = false ; /** approx extra buffer time given to agents due to engine processing for timeout mechanism */ static timeoutBuffer = 25 ; private docker : Dockerode ; /**
     * Single memory watch interval so that all memory checks are made together
     */ private memoryWatchInterval = null ; param design param loggingLevel constructor ( design , loggingLevel ) { this . design = design ; this . engineOptions = deepMerge ( this . engineOptions , deepCopy ( this . design . getDesignOptions ( ) . engineOptions ) ) ; this . overrideOptions = deepCopy ( this . design . getDesignOptions ( ) . override ) ; this . log . identifier = `Engine` ; this . setLogLevel ( loggingLevel ) ; this . docker = new Dockerode ( { socketPath : '/var/run/docker.sock' } ) ; } /** Set log level */ setLogLevel ( loggingLevel ) { this . log . level = loggingLevel ; } /** Get the engine options */ getEngineOptions ( ) { return this . engineOptions ; } /** Set the engine options */ setEngineOptions ( newOptions = { } ) { this . engineOptions = deepMerge ( this . engineOptions , newOptions ) ; } param agents param match returns async initialize ( agents , match ) { this . log . systembar ( ) ; const agentSetupPromises = [ ] ; match . agents . forEach ( ( agent ) => { agentSetupPromises . push ( this . initializeAgent ( agent , match ) ) ; } , this ) ; await Promise . all ( agentSetupPromises ) ; this . log . system ( 'FINISHED INITIALIZATION OF PROCESSES\n' ) ; return ; } /**
     * Initializes a single agent, called by  initialize param agent param match private async initializeAgent ( agent , match ) { this . log . system ( 'Setting up and spawning ' + agent . name + ` | Command: ${ agent . cmd } ${ agent . src }` ) ; if ( match . configs . secureMode ) { const name = `${ match . id }_agent_${ agent . id }` ; await agent . setupContainer ( name , this . docker , this . engineOptions ) ; } if ( this . engineOptions . memory . active ) { if ( agent . options . secureMode ) { } } const errorLogFilepath = path . join ( match . getMatchErrorLogDirectory ( ) , agent . getAgentErrorLogFilename ( ) ) ; let errorLogWriteStream = null ; if ( match . configs . storeErrorLogs ) { errorLogWriteStream = fs . createWriteStream ( errorLogFilepath ) ; agent . errorLogWriteStream = errorLogWriteStream ; errorLogWriteStream . write ( '=== Agent Install Log ===\n' ) ; } await agent . _install ( errorLogWriteStream , errorLogWriteStream , this . engineOptions ) ; this . log . system ( 'Succesfully ran install step for agent ' + agent . id ) ; if ( match . configs . storeErrorLogs ) { errorLogWriteStream . write ( '=== Agent Compile Log ===\n' ) ; } await agent . _compile ( errorLogWriteStream , errorLogWriteStream , this . engineOptions ) ; this . log . system ( 'Succesfully ran compile step for agent ' + agent . id ) ; let p = null ; p = await agent . _spawn ( ) ; this . log . system ( 'Spawned agent ' + agent . id ) ; if ( isChildProcess ( p ) ) { agent . _storeProcess ( p ) ; agent . streams . in = p . stdin ; agent . streams . out = p . stdout ; agent . streams . err = p . stderr ; p . on ( 'close' , ( code ) => { agent . emit ( Agent . AGENT_EVENTS . CLOSE , code ) ; } ) ; agent . streams . in . on ( 'error' , noop ) ; } else { agent . streams . in = p . in ; agent . streams . out = p . out ; agent . streams . err = p . err ; const containerExec = p . exec ; p . stream . on ( 'end' , async ( ) => { const endRes = await containerExec . inspect ( ) ; agent . emit ( Agent . AGENT_EVENTS . CLOSE , endRes . ExitCode ) ; } ) ; } agent . on ( Agent . AGENT_EVENTS . EXCEED_MEMORY_LIMIT , ( ) => { this . engineOptions . memory . memoryCallback ( agent , match , this . engineOptions ) ; } ) ; agent . on ( Agent . AGENT_EVENTS . TIMEOUT , ( ) => { this . engineOptions . timeout . timeoutCallback ( agent , match , this . engineOptions ) ; } ) ; match . idToAgentsMap . set ( agent . id , agent ) ; agent . status = Agent . Status . RUNNING ; agent . streams . out . on ( 'readable' , ( ) => { let data ; while ( ( data = agent . streams . out . read ( ) ) ) { const strs = `${ data }` . split ( /\r?\n/ ) ; const endsWithNewline = strs [ strs . length - 1 ] === '' ; const agentOutputContainsNewline = strs . length > 1 ; if ( agentOutputContainsNewline ) { strs [ 0 ] = agent . _buffer . join ( '' ) . concat ( strs [ 0 ] ) ; agent . _buffer = [ ] ; } for ( let i = 0 ; i < strs . length - 1 ; i ++ ) { if ( strs [ i ] === '' ) continue ; if ( agent . isAllowedToSendCommands ( ) ) { this . handleCommand ( agent , strs [ i ] ) ; } } if ( this . engineOptions . commandLines . waitForNewline && strs . length >= 1 && ! endsWithNewline ) { agent . _buffer . push ( strs [ strs . length - 1 ] ) ; } } } ) ; if ( ! this . engineOptions . noStdErr ) { agent . streams . err . on ( 'data' , ( data ) => { const strs = `${ data }` . split ( /\r?\n/ ) ; for ( const str of strs ) { if ( str === '' ) continue ; this . log . custom ( `[Agent ${ agent . id } Log]` . cyan , Logger . LEVEL . WARN , str ) ; } } ) ; } if ( match . configs . storeErrorLogs ) { errorLogWriteStream . write ( '=== Agent Error Log ===\n' ) ; agent . streams . err . on ( 'data' , ( data ) => { if ( agent . _logsize < agent . options . logLimit ) { agent . _logsize += data . length ; errorLogWriteStream . write ( `${ data }` ) ; } else { if ( agent . _trimmed === false ) { agent . _trimmed = true ; this . log . warn ( `agent ${ agent . id }'s logs were trimmed` ) ; errorLogWriteStream . write ( `\nend of logs as logs were trimmed\n` ) ; } } } ) ; } agent . on ( Agent . AGENT_EVENTS . CLOSE , ( code ) => { if ( ! agent . isTerminated ( ) ) { if ( agent . options . secureMode && code === 137 ) { this . kill ( agent , 'agent closed but agent not terminated yet, likely exceeded memory' ) ; this . engineOptions . memory . memoryCallback ( agent , match , this . engineOptions ) ; } else { this . kill ( agent , 'agent closed but agent not terminated yet' ) ; } } this . log . system ( `${ agent . name } | id: ${ agent . id } - exited with code ${ code }` ) ; } ) ; if ( this . engineOptions . memory . active ) { if ( ! agent . options . secureMode ) { agent . _setupMemoryWatcher ( this . engineOptions ) ; } } if ( this . killOffSignal ) { this . kill ( agent , 'engine has killOffSignal on, killing agent during initialization' ) ; } } param agent param str private async handleCommand ( agent , str ) { if ( this . engineOptions . commandStreamType === MatchEngine . COMMAND_STREAM_TYPE . SEQUENTIAL ) { switch ( this . engineOptions . commandFinishPolicy ) { case MatchEngine . COMMAND_FINISH_POLICIES . FINISH_SYMBOL : if ( `${ str }` === this . engineOptions . commandFinishSymbol ) { await agent . _finishMove ( ) ; } else { agent . currentMoveCommands . push ( str ) ; } break ; case MatchEngine . COMMAND_FINISH_POLICIES . LINE_COUNT : if ( `${ str }` === this . engineOptions . commandFinishSymbol ) { await agent . _finishMove ( ) ; } else if ( agent . currentMoveCommands . length < this . engineOptions . commandLines . max - 1 ) { agent . currentMoveCommands . push ( str ) ; } else if ( agent . currentMoveCommands . length == this . engineOptions . commandLines . max - 1 ) { await agent . _finishMove ( ) ; agent . currentMoveCommands . push ( str ) ; } break ; case MatchEngine . COMMAND_FINISH_POLICIES . CUSTOM : throw new NotSupportedError ( 'Custom command finish policies are not allowed yet' ) ; break ; } } } param match public async stop ( match ) { const stopPromises = [ ] ; match . agents . forEach ( ( agent ) => { stopPromises . push ( agent . stop ( ) ) ; } ) ; await Promise . all ( stopPromises ) ; this . log . system ( 'Stopped all agents' ) ; } param match public async resume ( match ) { const resumePromises = [ ] ; match . agents . forEach ( ( agent ) => { resumePromises . push ( agent . resume ( ) ) ; } ) ; await Promise . all ( resumePromises ) ; this . log . system ( 'Resumed all agents' ) ; } /**
     * Kills all intervals and agents and processes from a match and cleans up. Kills any game processes as well. Shouldn't be used
     * for custom design based matches. Called by  Match param match public async killAndClean ( match ) { this . killOffSignal = true ; clearInterval ( this . memoryWatchInterval ) ; const cleanUpPromises = [ ] ; if ( match . agents ) { match . agents . forEach ( ( agent ) => { cleanUpPromises . push ( this . kill ( agent , 'cleanup' ) ) ; } ) ; } await Promise . all ( cleanUpPromises ) ; } param agent public async kill ( agent , reason = 'unspecified' ) { try { await agent . _terminate ( ) ; this . log . system ( `Killed off agent ${ agent . id } - ${ agent . name }` , `Reason: ${ reason }` ) ; } catch ( err ) { this . log . system ( 'This should not happen when terminating agents.' , err ) ; } agent . _currentMoveResolve ( ) ; } param match returns a promise that resolves with an array of  MatchEngine . Command elements, holding the command and id
     * of the agent that sent it
      public getCommands ( match ) { return new Promise ( ( resolve ) => { const commands = [ ] ; const nonTerminatedAgents = match . agents . filter ( ( agent ) => { return ! agent . isTerminated ( ) ; } ) ; const allAgentMovePromises = nonTerminatedAgents . map ( ( agent ) => { return agent . _currentMovePromise ; } ) ; Promise . all ( allAgentMovePromises ) . then ( ( ) => { this . log . system ( `All move promises resolved` ) ; match . agents . forEach ( ( agent ) => { agent . currentMoveCommands . forEach ( ( commandString ) => { commandString . split ( this . engineOptions . commandDelimiter ) . forEach ( ( c ) => { if ( c !== '' ) { commands . push ( { command : c , agentID : agent . id } ) ; } } ) ; } ) ; } ) ; this . log . systemIO ( `Agent commands at end of time step ${ match . timeStep } to be sent to match on time step ${ match . timeStep + 1 } ` ) ; this . log . systemIO ( commands . length ? JSON . stringify ( commands ) : 'No commands' ) ; resolve ( commands ) ; } ) ; } ) ; } param match param message param agentID public send ( match , message , agentID ) { return new Promise ( ( resolve , reject ) => { const agent = match . idToAgentsMap . get ( agentID ) ; if ( agent . options . detached || ( ! agent . inputDestroyed ( ) && ! agent . isTerminated ( ) ) ) { const written = agent . options . detached ? `${ message }` : `${ message }\n` ; const bufferReachedHighWaterMark = agent . write ( written , ( error ) => { if ( error ) reject ( error ) ; resolve ( true ) ; } ) ; if ( ! bufferReachedHighWaterMark ) { this . log . system ( `Agent ${ agentID } buffer reached high watermark` ) ; } } else { this . log . error ( `Agent ${ agentID } - ${ agent . name } - has been killed off already, can't send messages now` ) ; resolve ( false ) ; } } ) ; } param match async initializeCustom ( ) { return true ; } /**
     * Run a custom match. A custom match much print to stdout all relevant data to be used by the engine and
     * Dimensions framework. All output after the conclude command from  Design . OverrideOptions is outputted
     * is stored as a list of new line delimited strings and returned as the match results. The match must exit with
     * exit code 0 to be marked as succesfully complete and the processing of results stops and this function resolves
     *  param match public runCustom ( match ) { return new Promise ( ( resolve , reject ) => { if ( this . overrideOptions . active == false ) { reject ( new FatalError ( 'Override was not set active! Make sure to set the overide.active field to true' ) ) ; } const cmd = this . overrideOptions . command ; const parsed = this . parseCustomArguments ( match , this . overrideOptions . arguments ) ; let matchProcessTimer ; match . matchProcess = spawn ( cmd , parsed ) . on ( 'error' , ( err ) => { if ( err ) throw err ; } ) ; this . log . system ( `${ match . name } | id: ${ match . id } - spawned: ${ cmd } ${ parsed . join ( ' ' ) }` ) ; const errorLogFilepath = path . join ( match . getMatchErrorLogDirectory ( ) , `match_error.log` ) ; let errorLogWriteStream = null ; if ( match . configs . storeErrorLogs ) { errorLogWriteStream = fs . createWriteStream ( errorLogFilepath ) ; } if ( match . configs . storeErrorLogs ) { errorLogWriteStream . write ( '=== Custom Match Error Log ===\n' ) ; match . matchProcess . stderr . pipe ( errorLogWriteStream ) ; } let matchTimedOut = false ; if ( this . overrideOptions . timeout !== null ) { matchProcessTimer = setTimeout ( ( ) => { this . log . system ( `${ match . name } | id: ${ match . id } - Timed out` ) ; match . matchProcess . kill ( 'SIGKILL' ) ; matchTimedOut = true ; } , this . overrideOptions . timeout ) ; } let processingStage = false ; match . matchProcess . stdout . on ( 'readable' , ( ) => { let data ; while ( ( data = match . matchProcess . stdout . read ( ) ) ) { const strs = `${ data }` . split ( /\r?\n/ ) ; for ( let i = 0 ; i < strs . length ; i ++ ) { const str = strs [ i ] ; if ( str === '' ) continue ; if ( str === this . overrideOptions . conclude_command ) { processingStage = true ; } else if ( ! processingStage ) { match . state . matchOutput . push ( str ) ; } else { match . results . push ( str ) ; } } } } ) ; match . matchProcess . stdout . on ( 'close' , ( code ) => { this . log . system ( `${ match . name } | id: ${ match . id } - exited with code ${ code }` ) ; if ( matchTimedOut ) { reject ( new MatchError ( 'Match timed out' ) ) ; } else { clearTimeout ( matchProcessTimer ) ; resolve ( match . results ) ; } match . agents . forEach ( ( agent ) => { if ( agent . options . secureMode ) { const tmpdir = os . tmpdir ( ) ; if ( agent . cwd . slice ( 0 , tmpdir . length ) === tmpdir ) { removeDirectorySync ( agent . cwd ) ; } else { this . log . error ( "couldn't remove agent files while in secure mode" ) ; } } } ) ; } ) ; } ) ; } /**
     * Attempts to stop a  Match based on a custom  Design param match public async stopCustom ( match ) { match . matchProcess . kill ( 'SIGSTOP' ) ; } /**
     * Attempts to resume a  Match based on a custom  Design param match public async resumeCustom ( match ) { match . matchProcess . kill ( 'SIGCONT' ) ; } param match public async killAndCleanCustom ( match ) { if ( match . matchProcess ) match . matchProcess . kill ( 'SIGKILL' ) ; } param match param args private parseCustomArguments ( match , args ) { if ( match . matchStatus === Match . Status . UNINITIALIZED ) { throw new FatalError ( `Match ${ match . id } - ${ match . name } is not initialized yet` ) ; } const parsed = [ ] ; for ( let i = 0 ; i < args . length ; i ++ ) { switch ( args [ i ] ) { case MatchEngine . DynamicDataStrings . D_FILES : match . agents . forEach ( ( agent ) => { parsed . push ( agent . file ) ; } ) ; break ; case MatchEngine . DynamicDataStrings . D_TOURNAMENT_IDS : match . agents . forEach ( ( agent ) => { parsed . push ( agent . tournamentID . id ? agent . tournamentID : '0' ) ; } ) ; break ; case MatchEngine . DynamicDataStrings . D_AGENT_IDS : match . agents . forEach ( ( agent ) => { parsed . push ( agent . id ) ; } ) ; break ; case MatchEngine . DynamicDataStrings . D_MATCH_ID : parsed . push ( match . id ) ; break ; case MatchEngine . DynamicDataStrings . D_MATCH_NAME : parsed . push ( match . name ) ; break ; case MatchEngine . DynamicDataStrings . D_NAMES : match . agents . forEach ( ( agent ) => { let parsedName = agent . name ; parsedName = parsedName . replace ( /\//g , '-' ) ; parsedName = parsedName . replace ( / /g , '_' ) ; parsed . push ( parsedName ) ; } ) ; break ; default : parsed . push ( args [ i ] ) ; break ; } } return parsed ; } /**
     * Returns the logger for this match engine
     */ public getLogger ( ) { return this . log ; } } export namespace MatchEngine { /**
     * Various policies available that describe the requirements before an agent is marked as done with sending commands
     * at some time step
     */ export enum COMMAND_FINISH_POLICIES { /**
         * Agent's finish their commands by sending a finish symbol, namely  EngineOptions . commandFinishSymbol FINISH_SYMBOL = 'finish_symbol' , /**
         * Agent's finish their commands by either sending a finish symmbol or after they send
         *  EngineOptions . commandLines . max lines
          LINE_COUNT = 'line_count' , /**
         * Custom finishing policy provided by user. Not allowed at the moment
         */ CUSTOM = 'custom' } /**
     * Engine Options that specify how the  MatchEngine should operate on a  Match export interface EngineOptions { /** The command streaming type */ commandStreamType : MatchEngine . COMMAND_STREAM_TYPE ; /**
         * Delimiter for seperating commands from agents in their stdout and then sending these delimited commands to
         *  Design . update . If an agent sent `move a b 3,run 24 d,t 3` and the delimiter is `','` then the
         *  Design . update function will receive commands `'move a b 3'` and `'run 24 d'` and `'t 3'`
         *  default commandDelimiter : string ; default commandFinishSymbol : string ; default commandFinishPolicy : MatchEngine . COMMAND_FINISH_POLICIES ; /**
         * Options for the  COMMAND_FINISH_POLICIES . LINE_COUNT finishing policy. Used only if this policy is active
          commandLines : { default max : number ; default waitForNewline : boolean ; } ; default noStdErr : boolean ; /**
         * Options for timeouts of agents
         */ timeout : { default active : boolean ; default max : number ; /**
             * the callback called when an agent times out.
             * Default is kill the agent with  Match . kill .
              timeoutCallback : ( agent : Agent , match : Match , engineOptions : EngineOptions ) => void ; } ; /**
         * Options related to the memory usage of agents. The memoryCallback is called when the limit is reached
         */ memory : { default active : boolean ; default limit : number ; /**
             * The callback called when an agent raeches the memory limit
             * Default is kill the agent with  Match . kill memoryCallback : ( agent : Agent , match : Match , engineOptions : EngineOptions ) => void ; default checkRate : number ; default usePs : boolean ; } ; } /** Standard ways for commands from agents to be streamed to the MatchEngine for the  Design to handle  export enum COMMAND_STREAM_TYPE { /** First come first serve for commands run. Not implemented */ PARALLEL = 'parallel' , /** Each agent's set of commands is run before the next agent */ SEQUENTIAL = 'sequential' } /**
     * A command delimited by the delimiter of the match engine from all commands sent by agent specified by agentID
     */ export interface Command { /**
         * The string command received
         */ command : string ; /**
         * The id of the agent that sent this command
         */ agentID : Agent . ID ; } /**
     * Dynammic Data strings are strings in the  OverrideOptions arguments array that are automatically replaced
     * with dynamic data as defined in the documentation of these enums
      export enum DynamicDataStrings { example D_FILES = 'D_FILES' , example D_AGENT_IDS = 'D_AGENT_IDS' , example D_TOURNAMENT_IDS = 'D_TOURNAMENT_IDS' , example D_MATCH_ID = 'D_MATCH_ID' , example D_MATCH_NAME = 'D_MATCH_NAME' , example D_NAMES = 'D_NAMES' } } export const DefaultMatchEngineOptions = { commandStreamType : MatchEngine . COMMAND_STREAM_TYPE . SEQUENTIAL , commandDelimiter : ',' , commandFinishSymbol : 'D_FINISH' , commandFinishPolicy : MatchEngine . COMMAND_FINISH_POLICIES . FINISH_SYMBOL , commandLines : { max : 1 , waitForNewline : true , } , noStdErr : true , timeout : { max : 1000 , active : true , timeoutCallback : ( agent , match , engineOptions ) : void => { match . kill ( agent . id , 'timed out' ) ; match . log . error ( `agent ${ agent . id } - '${ agent . name }' timed out after ${ engineOptions . timeout . max } ms` ) ; } , } , memory : { limit : 1000000000 , active : true , usePs : true , memoryCallback : ( agent , match , engineOptions ) : void => { match . kill ( agent . id , 'exceed memory limit' ) ; match . log . error ( `agent ${ agent . id } - '${ agent . name }' reached the memory limit of ${ engineOptions . memory . limit / 1000000 } MB` ) ; } , checkRate : 100 , } , } ;