Merge pull request #343 from engineer-man/process-janitor

rework process janitor
This commit is contained in:
Thomas Hobson 2021-10-02 00:08:58 +13:00 committed by GitHub
commit 9e81d2ff67
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 47 additions and 27 deletions

View File

@ -146,26 +146,31 @@ class Job {
const kill_timeout = set_timeout( const kill_timeout = set_timeout(
_ => proc.kill('SIGKILL'), async _ => {
logger.info(`Timeout exceeded timeout=${timeout} uuid=${this.uuid}`)
process.kill(proc.pid, 'SIGKILL')
},
timeout timeout
); );
proc.stderr.on('data', data => { proc.stderr.on('data', async data => {
if(eventBus !== null) { if(eventBus !== null) {
eventBus.emit("stderr", data); eventBus.emit("stderr", data);
} else if (stderr.length > config.output_max_size) { } else if (stderr.length > config.output_max_size) {
proc.kill('SIGKILL'); logger.info(`stderr length exceeded uuid=${this.uuid}`)
process.kill(proc.pid, 'SIGKILL')
} else { } else {
stderr += data; stderr += data;
output += data; output += data;
} }
}); });
proc.stdout.on('data', data => { proc.stdout.on('data', async data => {
if(eventBus !== null){ if(eventBus !== null){
eventBus.emit("stdout", data); eventBus.emit("stdout", data);
} else if (stdout.length > config.output_max_size) { } else if (stdout.length > config.output_max_size) {
proc.kill('SIGKILL'); logger.info(`stdout length exceeded uuid=${this.uuid}`)
process.kill(proc.pid, 'SIGKILL')
} else { } else {
stdout += data; stdout += data;
output += data; output += data;
@ -179,6 +184,7 @@ class Job {
proc.stdout.destroy(); proc.stdout.destroy();
await this.cleanup_processes() await this.cleanup_processes()
logger.debug(`Finished exit cleanup uuid=${this.uuid}`)
}; };
proc.on('exit', async (code, signal) => { proc.on('exit', async (code, signal) => {
@ -284,36 +290,47 @@ class Job {
this.state = job_states.EXECUTED; this.state = job_states.EXECUTED;
} }
async cleanup_processes() { async cleanup_processes(dont_wait = []) {
let processes = [1]; let processes = [1];
logger.debug(`Cleaning up processes uuid=${this.uuid}`)
while (processes.length > 0) { while (processes.length > 0) {
processes = await new Promise((resolve, reject) => processes = []
cp.execFile('ps', ['awwxo', 'pid,ruid'], (err, stdout) => {
if (err === null) {
const lines = stdout.split('\n').slice(1); //Remove header with slice
const procs = lines.map(line => {
const [pid, ruid] = line
.trim()
.split(/\s+/)
.map(n => parseInt(n));
return { pid, ruid };
});
resolve(procs); const proc_ids = await fs.readdir("/proc");
} else {
reject(error);
processes = await Promise.all(proc_ids.map(async (proc_id) => {
if(isNaN(proc_id)) return -1;
try{
const proc_status = await fs.read_file(path.join("/proc",proc_id,"status"));
const proc_lines = proc_status.to_string().split("\n")
const uid_line = proc_lines.find(line=>line.starts_with("Uid:"))
const [_, ruid, euid, suid, fuid] = uid_line.split(/\s+/);
if(ruid == this.uid || euid == this.uid)
return parse_int(proc_id)
}catch{
return -1
} }
})
);
processes = processes.filter(proc => proc.ruid === this.uid); return -1
}))
processes = processes.filter(p => p > 0)
if(processes.length > 0)
logger.debug(`Got processes to kill: ${processes} uuid=${this.uuid}`)
for (const proc of processes) { for (const proc of processes) {
// First stop the processes, but keep their resources allocated so they cant re-fork // First stop the processes, but keep their resources allocated so they cant re-fork
try { try {
process.kill(proc.pid, 'SIGSTOP'); process.kill(proc, 'SIGSTOP');
} catch { } catch {
// Could already be dead // Could already be dead
} }
@ -322,14 +339,17 @@ class Job {
for (const proc of processes) { for (const proc of processes) {
// Then clear them out of the process tree // Then clear them out of the process tree
try { try {
process.kill(proc.pid, 'SIGKILL'); process.kill(proc, 'SIGKILL');
} catch { } catch {
// Could already be dead and just needs to be waited on // Could already be dead and just needs to be waited on
} }
wait_pid(proc.pid); if(!dont_wait.includes(proc))
wait_pid(proc);
} }
} }
logger.debug(`Cleaned up processes uuid=${this.uuid}`)
} }
async cleanup_filesystem() { async cleanup_filesystem() {