going over all this task stuff

1 files changed, 19 insertions, 11 deletions

diff --git a/app/relay/runner.js b/app/relay/runner.js
index 1a72025..94b7779 100644
--- a/app/relay/runner.js
+++ b/app/relay/runner.js
@@ -8,6 +8,8 @@ import uuidv1 from 'uuid/v1'
 import * as fs from 'fs'
 import * as path from 'path'
 
+import * as queue from 'queue'
+
 const idle_state = { status: 'IDLE', task: {} }
 
 export const state = {
@@ -182,14 +184,15 @@ export function run_script(task, cb) {
   }, cb)
 }
 
-export function run_task(task, preempt, watch){
+export function run_task(task, preempt=false, watch=false){
+  if (! task) return null
 	const module = modules[task.module]
 	if (! module) return { type: 'error', error: "No such module: " + task.module }
   const activity = module.activities[task.activity]
-  run_task_with_activity(task, module, activity, preempt, watch)
+  return run_task_with_activity(task, module, activity, preempt, watch)
 }
 
-export function run_task_with_activity(task, module, activity, preempt, watch) {
+export function run_task_with_activity(task, module, activity, preempt=false, watch=false) {
 	const { interpreter, params } = build_params(module, activity, task)
   if (! interpreter) return { type: 'error', error: "No such interpreter: " + activity.interpreter }
 
@@ -199,7 +202,7 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
       terminate(state.current_gpu_task)
     } else {
       console.log('already running GPU task :(', state.current_gpu_task.pid)
-      return { type: 'error', error: 'task already running on gpu' }
+      return { type: 'error', preempt: false, error: 'task already running on gpu' }
     }
   } else if (!interpreter.gpu && state.current_cpu_task.status !== 'IDLE') {
     if (preempt) {
@@ -207,7 +210,7 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
       terminate(state.current_cpu_task)
     } else {
       console.log('already running CPU task :(')
-      return { type: 'error', error: 'task already running on cpu' }
+      return { type: 'error', preempt: false, error: 'task already running on cpu' }
     }
   }
 
@@ -215,7 +218,9 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
   console.log(module.cwd)
 	console.log(interpreter.cmd, params)
 
+  task.uuid = task.uuid || uuidv1()
   task.started = new Date().toString()
+  task.processing = true
 	const subprocess = spawn(interpreter.cmd, params, {
     cwd: module.cwd,
   })
@@ -230,7 +235,6 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
     }
 	}
   
-  task.uuid = task.uuid || uuidv1()
   const processor = task.processor = interpreter.gpu ? 'gpu' : 'cpu'
   remote.emit('task_res', { type: 'task_begin', task })
   
@@ -248,6 +252,7 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
   subprocess.on('error', (err) => {
     if (finished) return
     finished = true
+    task.processing = false
 		console.log('task error', subprocess.exitCode, err)
     finish({
       type: 'task_error',
@@ -260,6 +265,8 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
     finished = true
 		console.log('task ended', subprocess.exitCode || '')
     set_connected(false)
+    task.processing = false
+    task.completed = true
     finish({
       type: 'task_finish',
       task,
@@ -277,14 +284,15 @@ export function run_task_with_activity(task, module, activity, preempt, watch) {
     }
     return run_next_task()
   }
+
+  return task
 }
 
 export function run_next_task(){
-  // get next task from the queue...
-  /*
-  const task = queue.get_next_task()
-  return run_task(task)
-  */
+  if (queue.is_active()) {
+    const task = queue.get_next_task()
+    return run_task(task)
+  }
 }
 
 export function stop_task(task){