summaryrefslogtreecommitdiff
path: root/app/relay
diff options
context:
space:
mode:
Diffstat (limited to 'app/relay')
-rw-r--r--app/relay/queue.js62
-rw-r--r--app/relay/remote.js8
-rw-r--r--app/relay/runner.js27
3 files changed, 67 insertions, 30 deletions
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){