set parallelism

README.md

@@ -31,4 +31,9 @@ GPU is occupied by which job(s)
 **DisableSchedule**
 ```
 ?action=debug_disable
+```
+
+**UpdateMaxParallelism**
+```
+?action=debug_update_parallelism&parallelism=5
 ```

src/main.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"os"
 	"time"
+	"strconv"
 )
 
 var addr = flag.String("addr", "0.0.0.0:8080", "http service address")
@@ -180,6 +181,14 @@ func serverAPI(w http.ResponseWriter, r *http.Request) {
 		w.Write(js)
 		break
 
+	case "debug_update_parallelism":
+		log.Debug("update_parallelism")
+		parallelism, _ := strconv.Atoi(r.URL.Query().Get("parallelism"))
+		js, _ := json.Marshal(scheduler.UpdateParallelism(parallelism))
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(js)
+		break
+
 	default:
 		http.Error(w, "Not Found", http.StatusNotFound)
 		break

src/scheduler.go

@@ -32,4 +32,6 @@ type Scheduler interface {
 	Enable() bool
 
 	Disable() bool
+
+	UpdateParallelism(parallelism int) bool
 }

src/scheduler_FCFS.go

@@ -13,8 +13,9 @@ type SchedulerFCFS struct {
 	mu         sync.Mutex
 	scheduling sync.Mutex
 
-	jobs    map[string]*JobManager
-	enabled bool
+	jobs        map[string]*JobManager
+	enabled     bool
+	parallelism int
 }
 
 func (scheduler *SchedulerFCFS) Start() {
@@ -254,3 +255,9 @@ func (scheduler *SchedulerFCFS) Disable() bool {
 	scheduler.enabled = false
 	return true
 }
+
+func (scheduler *SchedulerFCFS) UpdateParallelism(parallelism int) bool {
+	scheduler.parallelism = parallelism
+	log.Info("parallelism is updated to", parallelism)
+	return true
+}

src/scheduler_fair.go

@@ -26,6 +26,7 @@ type SchedulerFair struct {
 	resourceAllocations map[string]*ResourceCount
 	enabled             bool
 	latestPoolIndex     int
+	parallelism         int
 }
 
 type FairJobSorter []Job
@@ -50,6 +51,8 @@ func (scheduler *SchedulerFair) Start() {
 	scheduler.enabled = true
 	scheduler.schedulingJobsCnt = 0
 
+	scheduler.parallelism = 1
+
 	go func() {
 		for {
 			log.Debug("Scheduling")
@@ -58,7 +61,7 @@ func (scheduler *SchedulerFair) Start() {
 				continue
 			}
 			scheduler.schedulingMu.Lock()
-			if scheduler.schedulingJobsCnt >= pool.poolsCount/10 {
+			if scheduler.schedulingJobsCnt >= scheduler.parallelism {
 				scheduler.schedulingMu.Unlock()
 				time.Sleep(time.Millisecond * 100)
 				continue
@@ -434,3 +437,9 @@ func (scheduler *SchedulerFair) Disable() bool {
 	log.Info("scheduler is disabled", time.Now())
 	return true
 }
+
+func (scheduler *SchedulerFair) UpdateParallelism(parallelism int) bool {
+	scheduler.parallelism = parallelism
+	log.Info("parallelism is updated to", parallelism)
+	return true
+}

src/scheduler_priority.go

@@ -13,8 +13,9 @@ type SchedulerPriority struct {
 	mu         sync.Mutex
 	scheduling sync.Mutex
 
-	jobs    map[string]*JobManager
-	enabled bool
+	jobs        map[string]*JobManager
+	enabled     bool
+	parallelism int
 }
 
 func (scheduler *SchedulerPriority) Start() {
@@ -278,3 +279,9 @@ func (scheduler *SchedulerPriority) Disable() bool {
 	scheduler.enabled = false
 	return true
 }
+
+func (scheduler *SchedulerPriority) UpdateParallelism(parallelism int) bool {
+	scheduler.parallelism = parallelism
+	log.Info("parallelism is updated to", parallelism)
+	return true
+}