MAKE-117: remove wait from queue interface

mongodb · Dec 7, 2016 · 3c6a620 · 3c6a620
1 parent fb04937
commit 3c6a620
Show file tree

Hide file tree

Showing 21 changed files with 115 additions and 111 deletions.
diff --git a/doc.go b/doc.go
@@ -43,7 +43,7 @@ Consider the following example:
       // handle error case
    }
 
-   queue.Wait() // waits for all tasks to finish.
+   Wait(queue) // waits for all tasks to finish.
    queue.Close() // waits for all tasks to finish and releases
 		 // all resources.
 */

diff --git a/interface.go b/interface.go
@@ -73,7 +73,7 @@ type Queue interface {
 	Get(string) (Job, bool)
 
 	// Returns the next job in the queue. These calls are
-	// non-blocking and return errors
+	// blocking, but may be interrupted with a canceled context.
 	Next(context.Context) Job
 
 	// Makes it possible to detect if a Queue has started
@@ -104,9 +104,6 @@ type Queue interface {
 	// Begins the execution of the job Queue, using the embedded
 	// Runner.
 	Start(context.Context) error
-
-	// Waits for all jobs to complete.
-	Wait()
 }
 
 // Runner describes a simple worker interface for executing jobs in

diff --git a/pool/group_test.go b/pool/group_test.go
@@ -160,8 +160,8 @@ func (s *UnorderedGroupSuite) TestPoolStartsAndProcessesJobs() {
 	s.True(queueTwo.Started())
 
 	s.True(s.pool.Started())
-	queueTwo.Wait()
-	queueOne.Wait()
+	amboy.Wait(queueTwo)
+	amboy.Wait(queueOne)
 
 	for _, job := range jobsOne {
 		s.True(job.Completed(), fmt.Sprintf("%T\n\t%+v", job, job))

diff --git a/pool/local_test.go b/pool/local_test.go
@@ -87,7 +87,7 @@ func (s *LocalWorkersSuite) TestPoolStartsAndProcessesJobs() {
 	s.True(s.pool.Started())
 	s.True(s.queue.Started())
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 
 	counter := 0
 	for j := range s.queue.Results() {

diff --git a/pool/mock_queue_test.go b/pool/mock_queue_test.go
@@ -5,7 +5,6 @@ import (
 
 	"github.com/mongodb/amboy"
 	"github.com/pkg/errors"
-	"github.com/tychoish/grip"
 	"golang.org/x/net/context"
 )
 
@@ -130,16 +129,6 @@ func (q *QueueTester) Results() <-chan amboy.Job {
 	return output
 }
 
-func (q *QueueTester) Wait() {
-	for {
-		grip.Debugf("test queue status: %+v\n", q.Stats())
-
-		if q.isComplete() {
-			return
-		}
-	}
-}
-
 func (q *QueueTester) isComplete() bool {
 	q.mutex.RLock()
 	defer q.mutex.RUnlock()

diff --git a/pool/single_test.go b/pool/single_test.go
@@ -73,7 +73,7 @@ func (s *SingleRunnerSuite) TestPoolStartsAndProcessesJobs() {
 	s.True(s.pool.Started())
 	s.True(s.queue.Started())
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 
 	for _, job := range jobs {
 		s.True(job.Completed())

diff --git a/queue/fixed.go b/queue/fixed.go
@@ -170,14 +170,3 @@ func (q *LocalLimitedSize) Start(ctx context.Context) error {
 	grip.Info("job server running")
 	return nil
 }
-
-// Wait blocks until all tasks are complete.
-func (q *LocalLimitedSize) Wait() {
-	for {
-		stats := q.Stats()
-		grip.Debugf("%d jobs complete of %d total", stats.Completed, stats.Total)
-		if stats.Total == stats.Completed {
-			break
-		}
-	}
-}
diff --git a/queue/fixed_test.go b/queue/fixed_test.go
@@ -106,7 +106,7 @@ func (s *LimitedSizeQueueSuite) TestGetMethodOnlyReturnsCompletedJobs() {
 		jobs[j.ID()] = j
 	}
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 	s.queue.Runner().Close()
 
 	for name, j := range jobs {

diff --git a/queue/ordered.go b/queue/ordered.go
@@ -27,7 +27,6 @@ import (
 	"github.com/mongodb/amboy/pool"
 	"github.com/pkg/errors"
 	"github.com/tychoish/grip"
-	"github.com/tychoish/grip/sometimes"
 	"golang.org/x/net/context"
 )
 
@@ -343,15 +342,3 @@ func (q *LocalOrdered) Complete(ctx context.Context, j amboy.Job) {
 	defer q.mutex.Unlock()
 	q.tasks.completed[j.ID()] = true
 }
-
-// Wait blocks until all pending jobs in the queue are complete.
-func (q *LocalOrdered) Wait() {
-	for {
-		stats := q.Stats()
-		grip.DebugWhenf(sometimes.Fifth(),
-			"waiting for %d pending jobs (total=%d)", stats.Pending, stats.Total)
-		if stats.Pending == 0 {
-			break
-		}
-	}
-}
diff --git a/queue/ordered_test.go b/queue/ordered_test.go
@@ -5,6 +5,7 @@ package queue
 import (
 	"testing"
 
+	"github.com/mongodb/amboy"
 	"github.com/mongodb/amboy/dependency"
 	"github.com/mongodb/amboy/job"
 	"github.com/mongodb/amboy/pool"
@@ -112,7 +113,7 @@ func (s *OrderedQueueSuite) TestResultsChannelProducesPointersToConsistentJobObj
 	s.NoError(s.queue.Put(job))
 	s.NoError(s.queue.Start(ctx))
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 
 	result, ok := <-s.queue.Results()
 	s.True(ok)
@@ -128,7 +129,7 @@ func (s *OrderedQueueSuite) TestQueueCanOnlyBeStartedOnce() {
 	s.NoError(s.queue.Start(ctx))
 	s.True(s.queue.Started())
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 	s.True(s.queue.Started())
 
 	// you can call start more than once until the queue has
@@ -210,7 +211,7 @@ func (s *OrderedQueueSuite) TestPassedIsCompletedButDoesNotRun() {
 	defer cancel()
 
 	s.NoError(s.queue.Start(ctx))
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 	s.False(j1.Completed())
 	s.True(j2.Completed())
 }
diff --git a/queue/priority.go b/queue/priority.go
@@ -8,7 +8,6 @@ import (
 	"github.com/mongodb/amboy/pool"
 	"github.com/mongodb/amboy/queue/driver"
 	"github.com/tychoish/grip"
-	"github.com/tychoish/grip/sometimes"
 	"golang.org/x/net/context"
 )
 
@@ -161,15 +160,3 @@ func (q *LocalPriorityQueue) Start(ctx context.Context) error {
 
 	return nil
 }
-
-// Wait blocks until there is no pending work remaining in the queue.
-func (q *LocalPriorityQueue) Wait() {
-	for {
-		stats := q.Stats()
-		grip.DebugWhenf(sometimes.Fifth(),
-			"%d jobs complete of %d total", stats.Completed, stats.Total)
-		if stats.Total == stats.Completed {
-			break
-		}
-	}
-}
diff --git a/queue/remote.go b/queue/remote.go
@@ -270,14 +270,3 @@ func (q *RemoteUnordered) Start(ctx context.Context) error {
 
 	return nil
 }
-
-// Wait blocks until there are no pending jobs in the queue.
-func (q *RemoteUnordered) Wait() {
-	for {
-		stats := q.Stats()
-		if stats.Total == stats.Completed {
-			break
-		}
-		time.Sleep(50 * time.Millisecond)
-	}
-}
diff --git a/queue/shuffled.go b/queue/shuffled.go
@@ -20,7 +20,6 @@ import (
 	"github.com/mongodb/amboy"
 	"github.com/pkg/errors"
 	"github.com/tychoish/grip"
-	"github.com/tychoish/grip/sometimes"
 	"golang.org/x/net/context"
 )
 
@@ -261,15 +260,3 @@ func (q *LocalShuffled) SetRunner(r amboy.Runner) error {
 func (q *LocalShuffled) Runner() amboy.Runner {
 	return q.runner
 }
-
-// Wait blocks until all submitted jobs are complete.
-func (q *LocalShuffled) Wait() {
-	for {
-		stats := q.Stats()
-		grip.DebugWhenf(sometimes.Fifth(),
-			"%d jobs complete of %d total", stats.Completed, stats.Total)
-		if stats.Total == stats.Completed {
-			break
-		}
-	}
-}
diff --git a/queue/shuffled_test.go b/queue/shuffled_test.go
@@ -124,7 +124,7 @@ func (s *ShuffledQueueSuite) TestGetMethodRetrieves() {
 	jReturn, ok = s.queue.Get(j.ID())
 	s.True(ok)
 	s.Exactly(jReturn, j)
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 
 	jReturn, ok = s.queue.Get(j.ID())
 	s.True(ok)

diff --git a/queue/smoke_test.go b/queue/smoke_test.go
@@ -56,7 +56,7 @@ func runUnorderedSmokeTest(ctx context.Context, q amboy.Queue, size int, assert
 	wg.Wait()
 
 	assert.Equal(numJobs, q.Stats().Total, fmt.Sprintf("with %d workers", size))
-	q.Wait()
+	amboy.Wait(q)
 
 	grip.Infof("workers complete for %d worker smoke test", size)
 	assert.Equal(numJobs, q.Stats().Completed, fmt.Sprintf("%+v", q.Stats()))
@@ -105,8 +105,8 @@ func runMultiQueueSingleBackEndSmokeTest(ctx context.Context, qOne, qTwo amboy.Q
 	grip.Infof("before wait statsTwo: %+v", statsTwo)
 
 	// wait for all jobs to complete.
-	qOne.Wait()
-	qTwo.Wait()
+	amboy.Wait(qOne)
+	amboy.Wait(qTwo)
 
 	grip.Infof("after wait statsOne: %+v", qOne.Stats())
 	grip.Infof("after wait statsTwo: %+v", qTwo.Stats())

diff --git a/queue/unordered.go b/queue/unordered.go
@@ -21,7 +21,6 @@ import (
 	"github.com/mongodb/amboy/pool"
 	"github.com/pkg/errors"
 	"github.com/tychoish/grip"
-	"github.com/tychoish/grip/sometimes"
 	"golang.org/x/net/context"
 )
 
@@ -211,15 +210,3 @@ func (q *LocalUnordered) Complete(ctx context.Context, j amboy.Job) {
 		q.numCompleted++
 	}()
 }
-
-// Wait blocks until all currently pending jobs jobs have completed.
-func (q *LocalUnordered) Wait() {
-	for {
-		stats := q.Stats()
-		grip.DebugWhenf(sometimes.Fifth(),
-			"waiting for %d pending jobs (total=%d)", stats.Pending, stats.Total)
-		if stats.Pending == 0 {
-			break
-		}
-	}
-}
diff --git a/queue/unordered_test.go b/queue/unordered_test.go
@@ -3,6 +3,7 @@ package queue
 import (
 	"testing"
 
+	"github.com/mongodb/amboy"
 	"github.com/mongodb/amboy/job"
 	"github.com/mongodb/amboy/pool"
 	"github.com/stretchr/testify/require"
@@ -93,7 +94,7 @@ func (s *LocalQueueSuite) TestResultsChannelProducesPointersToConsistentJobObjec
 	s.NoError(s.queue.Start(ctx))
 	s.NoError(s.queue.Put(job))
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 
 	result, ok := <-s.queue.Results()
 	s.True(ok)
@@ -116,7 +117,7 @@ func (s *LocalQueueSuite) TestJobsChannelProducesJobObjects() {
 		s.NoError(s.queue.Put(job))
 	}
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 
 	for j := range s.queue.Results() {
 		shellJob, ok := j.(*job.ShellJob)
@@ -156,7 +157,7 @@ func (s *LocalQueueSuite) TestQueueCanOnlyBeStartedOnce() {
 	s.NoError(s.queue.Start(ctx))
 	s.True(s.queue.Started())
 
-	s.queue.Wait()
+	amboy.Wait(s.queue)
 	s.True(s.queue.Started())
 
 	// you can call start more than once until the queue has

diff --git a/rest/client_test.go b/rest/client_test.go
@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mongodb/amboy"
 	"github.com/mongodb/amboy/job"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -324,7 +325,7 @@ func (s *ClientSuite) TestGetStatsHelperWithActualJob() {
 	j := job.NewShellJob("true", "")
 
 	s.NoError(s.service.queue.Put(j))
-	s.service.queue.Wait()
+	amboy.Wait(s.service.queue)
 
 	st, err := s.client.getStats(ctx)
 	s.NoError(err, fmt.Sprintf("%+v", st))
@@ -377,7 +378,7 @@ func (s *ClientSuite) TestJobStatusWithValidJob() {
 
 	j := job.NewShellJob("echo foo", "")
 	s.NoError(s.service.queue.Put(j))
-	s.service.queue.Wait()
+	amboy.Wait(s.service.queue)
 	st, err := s.client.jobStatus(ctx, j.ID())
 	s.NoError(err)
 	s.Equal(j.ID(), st.ID)
@@ -480,7 +481,7 @@ func (s *ClientSuite) TestPendingJobsWithCanceledContextReturnsError() {
 }
 
 func (s *ClientSuite) TestPendingJobsIsZeroAfterWaitingOnTheQueue() {
-	s.service.queue.Wait()
+	amboy.Wait(s.service.queue)
 	var err error
 	ctx := context.Background()
 	s.client, err = NewClient(s.info.host, s.info.port, "")
@@ -506,7 +507,7 @@ func (s *ClientSuite) TestJobCompleteWithCanceledContextReturnsError() {
 }
 
 func (s *ClientSuite) TestJobCompleteIsZeroAfterWaitingOnTheQueue() {
-	s.service.queue.Wait()
+	amboy.Wait(s.service.queue)
 	var err error
 	ctx := context.Background()
 	s.client, err = NewClient(s.info.host, s.info.port, "")
@@ -515,7 +516,7 @@ func (s *ClientSuite) TestJobCompleteIsZeroAfterWaitingOnTheQueue() {
 	j := job.NewShellJob("echo foo", "")
 	s.NoError(s.service.queue.Put(j))
 
-	s.service.queue.Wait()
+	amboy.Wait(s.service.queue)
 
 	isComplete, err := s.client.JobComplete(ctx, j.ID())
 	s.True(isComplete)

diff --git a/rest/job_test.go b/rest/job_test.go
@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/mongodb/amboy"
 	"github.com/mongodb/amboy/job"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -59,7 +60,7 @@ func (s *JobStatusSuite) TestIncorrectOrInvalidJobNamesReturnExpectedResults() {
 }
 
 func (s *JobStatusSuite) TestJobNameReturnsSuccessfulResponse() {
-	s.service.queue.Wait()
+	amboy.Wait(s.service.queue)
 
 	resp, err := s.service.getJobStatusResponse(s.jobName)
 	s.NoError(err)