Add 'sdk/java-v2/' from commit '55f103e336ca9fb8bf1720d2ef4ee8dd4e221118'

[arvados.git] / lib / dispatchcloud / scheduler / scheduler.go

diff --git a/lib/dispatchcloud/scheduler/scheduler.go b/lib/dispatchcloud/scheduler/scheduler.go
index 0be8edb7b6420e00a07192ebcaed7eb5fa8e4186..eb82c488390e3751fd7d3383acb1c1cf72af5e37 100644 (file)
--- a/lib/dispatchcloud/scheduler/scheduler.go
+++ b/lib/dispatchcloud/scheduler/scheduler.go
@@ -7,10 +7,12 @@
 package scheduler
 
 import (
+       "context"
        "sync"
        "time"
 
-       "github.com/Sirupsen/logrus"
+       "git.curoverse.com/arvados.git/sdk/go/ctxlog"
+       "github.com/sirupsen/logrus"
 )
 
 // A Scheduler maps queued containers onto unallocated workers in
@@ -32,22 +34,30 @@ type Scheduler struct {
        staleLockTimeout    time.Duration
        queueUpdateInterval time.Duration
 
+       uuidOp map[string]string // operation in progress: "lock", "cancel", ...
+       mtx    sync.Mutex
+       wakeup *time.Timer
+
        runOnce sync.Once
        stop    chan struct{}
+       stopped chan struct{}
 }
 
 // New returns a new unstarted Scheduler.
 //
 // Any given queue and pool should not be used by more than one
 // scheduler at a time.
-func New(logger logrus.FieldLogger, queue ContainerQueue, pool WorkerPool, staleLockTimeout, queueUpdateInterval time.Duration) *Scheduler {
+func New(ctx context.Context, queue ContainerQueue, pool WorkerPool, staleLockTimeout, queueUpdateInterval time.Duration) *Scheduler {
        return &Scheduler{
-               logger:              logger,
+               logger:              ctxlog.FromContext(ctx),
                queue:               queue,
                pool:                pool,
                staleLockTimeout:    staleLockTimeout,
                queueUpdateInterval: queueUpdateInterval,
+               wakeup:              time.NewTimer(time.Second),
                stop:                make(chan struct{}),
+               stopped:             make(chan struct{}),
+               uuidOp:              map[string]string{},
        }
 }
 
@@ -60,13 +70,19 @@ func (sch *Scheduler) Start() {
 // Stop.
 func (sch *Scheduler) Stop() {
        close(sch.stop)
+       <-sch.stopped
 }
 
 func (sch *Scheduler) run() {
+       defer close(sch.stopped)
+
        // Ensure the queue is fetched once before attempting anything.
        for err := sch.queue.Update(); err != nil; err = sch.queue.Update() {
                sch.logger.Errorf("error updating queue: %s", err)
-               d := sch.queueUpdateInterval / 60
+               d := sch.queueUpdateInterval / 10
+               if d < time.Second {
+                       d = time.Second
+               }
                sch.logger.Infof("waiting %s before retry", d)
                time.Sleep(d)
        }
@@ -102,6 +118,7 @@ func (sch *Scheduler) run() {
                        return
                case <-queueNotify:
                case <-poolNotify:
+               case <-sch.wakeup.C:
                }
        }
 }