new queue

5 files changed, 70 insertions, 33 deletions

diff --git a/app/client/modules/samplernn/samplernn.actions.js b/app/client/modules/samplernn/samplernn.actions.js
index 5324807..807a3d0 100644
--- a/app/client/modules/samplernn/samplernn.actions.js
+++ b/app/client/modules/samplernn/samplernn.actions.js
@@ -136,7 +136,7 @@ export const load_loss = () => dispatch => {
             return a
           }, {})
         )
-      // console.log(loss[name])
+      console.log(data, name, lossReport[name])
     })
     dispatch({
       type: types.samplernn.load_loss,
diff --git a/app/client/system/system.component.js b/app/client/system/system.component.js
index aaa10d6..2aa4cd4 100644
--- a/app/client/system/system.component.js
+++ b/app/client/system/system.component.js
@@ -109,8 +109,8 @@ class System extends Component {
                 <button onClick={() => actions.live.get_params()}>Get</button>
                 <button onClick={() => actions.live.set_param('fruit', choice(fruits))}>Set</button>
               </Param>
-              <Param title=''>
-                <button onClick={() => actions.system.enqueue_test_task(choice(fruits))}>+Add</button>
+              <Param title='Queue Tests'>
+                <button onClick={() => actions.system.enqueue_test_task(choice(fruits))}>+Add test task</button>
               </Param>
               <Param title=''>
                 <button onClick={() => actions.queue.start_task(wait_test_task, { preempt: true, watch: true })}>Wait and Buzz</button>
diff --git a/app/relay/queue.js b/app/relay/queue.js
index 9dd987a..d6e7ffe 100644
--- a/app/relay/queue.js
+++ b/app/relay/queue.js
@@ -1,25 +1,45 @@
 // get the processor that uses the task and give it its own queue!
 
-let queue = []
-let active = false
-let status = 'waiting'
-
-export const is_active = () => {
-  if (active && queue.length) {
-    return true
+export class Queue {
+  constructor(processor){
+    this.a = []
+    this.processor = processor
+    this.active = false
+    this.status = 'waiting'
+  }
+  is_active() {
+    if (this.active && this.a.length) {
+      return true
+    }
+    this.status = 'waiting'
+    this.active = false
+    return false
+  }
+  get_status() {
+    return this.status
+  }
+  activate() {
+    this.active = true
+    this.status = 'active'
+  }
+  deactivate(reason) {
+    this.active = false
+    this.status = reason || 'waiting'
+  }
+  add_task(task){
+    this.a.push(task)
+  }
+  remove_task(task){
+    this.a = this.a.filter(t => t.id !== task.id)
+  }
+  get_next_task(){
+    return this.a.shift()
+  }
+  list_tasks(){
+    return this.a
   }
-  return active = false
-}
-export const get_status = () => status
-export const activate = () => {
-  active = true
-  status = 'active'
-}
-export const deactivate = reason => {
-  active = false
-  status = reason
 }
-export const add_task = task => queue.push(task)
-export const remove_task = task => queue = queue.filter(t => t.id !== task.id)
-export const get_next_task = () => queue.shift()
-export const list_tasks = () => queue
+
+
+export const cpu = new Queue('cpu')
+export const gpu = new Queue('gpu')
\ No newline at end of file
diff --git a/app/relay/remote.js b/app/relay/remote.js
index c958eab..b4afab0 100644
--- a/app/relay/remote.js
+++ b/app/relay/remote.js
@@ -1,6 +1,6 @@
 import { rpc, get_connected } from './rpc'
 
-import * as queue from './queue'
+import * as q from './queue'
 
 const io = require('socket.io-client')
 const runner = require('./runner')
@@ -39,7 +39,7 @@ remote.on('cmd', (data) => {
 })
 
 remote.on('task', (data) => {
-  let response;
+  let queue, response;
   // console.log(data)
   console.log('task', data.type)
   switch(data.type) {
@@ -50,6 +50,7 @@ remote.on('task', (data) => {
       response = runner.stop_task(data.task, data.sigkill)
       break
     case 'add':
+      queue = runner.get_queue(data.task)
       queue.add_task(data.task)
       if (! queue.is_active()) {
         queue.activate()
@@ -57,6 +58,7 @@ remote.on('task', (data) => {
       }
       break
     case 'remove':
+      queue = q.get_processor(data_task)
       queue.remove_task(data.task)
       break
     case 'start_queue':
@@ -66,9 +68,11 @@ remote.on('task', (data) => {
       }
       break
     case 'stop_queue':
+      queue = q.get_processor(data_task)
       queue.deactivate('user')
       break
     case 'list':
+      queue = q.get_processor(data_task)
       response = { type: 'list', tasks: queue.list_tasks() }
       break
     case 'set_priority':
diff --git a/app/relay/runner.js b/app/relay/runner.js
index a449a6c..809de7a 100644
--- a/app/relay/runner.js
+++ b/app/relay/runner.js
@@ -8,7 +8,7 @@ import uuidv1 from 'uuid/v1'
 import * as fs from 'fs'
 import * as path from 'path'
 
-import * as queue from './queue'
+import * as q from './queue'
 
 const idle_state = { status: 'IDLE', task: {} }
 
@@ -212,6 +212,12 @@ export function get_processor(task){
   return interpreter.gpu ? 'gpu' : 'cpu'
 }
 
+export function get_queue(task){
+  const processor = get_processor(task)
+  if (!processor) return null
+  return (processor === 'cpu') ? q.cpu : q.gpu
+}
+
 export function run_task(task, preempt=false, watch=false){
   if (! task) return null
 	const module = modules[task.module]
@@ -338,12 +344,19 @@ export function start_queue(){
 }
 
 export function run_next_task(){
-  if (queue.is_active()) {
-    console.log(queue.list_tasks())
-    const task = queue.get_next_task()
-    return run_task(task, false, true)
-  }
-  return { status: 'not active' }
+  const status = [q.cpu, q.gpu].map(queue => {
+    const status = queue.processor === 'gpu' ? state.current_gpu_task.status : state.current_cpu_task.status
+    console.log(queue.is_active(), status)
+    if (queue.is_active() && status === 'IDLE') {
+      console.log(queue.processor, "is free")
+      const task = queue.get_next_task()
+      return run_task(task, false, true)
+    } else {
+      console.log(queue.processor, "is busy")
+      return { processor: queue.processor, status: 'busy' }
+    }
+  })
+  return status
 }
 
 export function stop_task(task, sigkill){