dan/errgroup
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

semaphore: add worker-pool example

Browse source 
I've commented several times in various forums that basically every
time I've seen the “worker goroutine” pattern in Go, there has turned
out to be a cleaner implementation using semaphores.

This change adds a simple such example. (For more complex usage, I
would generally pair the semaphore with an errgroup.Group.)

Change-Id: Ibf69ee761d14ba59c1acc6a2d595b4fcf0d8f6d6
Reviewed-on: https://go-review.googlesource.com/75170
Reviewed-by: Ross Light <light@google.com>
This commit is contained in:
Bryan C. Mills 2017-11-01 15:49:15 -04:00 • committed by Bryan Mills
commit fd80eb99c8
3 changed files with 98 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

19
semaphore/semaphore_test.go
View file

@ -2,7 +2,7 @@
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package semaphore
package semaphore_test
import (
"math/rand"
@ -13,11 +13,12 @@ import (
"golang.org/x/net/context"
"golang.org/x/sync/errgroup"
"golang.org/x/sync/semaphore"
)
const maxSleep = 1 * time.Millisecond
func HammerWeighted(sem *Weighted, n int64, loops int) {
func HammerWeighted(sem *semaphore.Weighted, n int64, loops int) {
for i := 0; i < loops; i++ {
sem.Acquire(context.Background(), n)
time.Sleep(time.Duration(rand.Int63n(int64(maxSleep/time.Nanosecond))) * time.Nanosecond)
@ -30,7 +31,7 @@ func TestWeighted(t *testing.T) {
n := runtime.GOMAXPROCS(0)
loops := 10000 / n
sem := NewWeighted(int64(n))
sem := semaphore.NewWeighted(int64(n))
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
@ -51,7 +52,7 @@ func TestWeightedPanic(t *testing.T) {
t.Fatal("release of an unacquired weighted semaphore did not panic")
}
}()
w := NewWeighted(1)
w := semaphore.NewWeighted(1)
w.Release(1)
}
@ -59,7 +60,7 @@ func TestWeightedTryAcquire(t *testing.T) {
t.Parallel()
ctx := context.Background()
sem := NewWeighted(2)
sem := semaphore.NewWeighted(2)
tries := []bool{}
sem.Acquire(ctx, 1)
tries = append(tries, sem.TryAcquire(1))
@ -83,7 +84,7 @@ func TestWeightedAcquire(t *testing.T) {
t.Parallel()
ctx := context.Background()
sem := NewWeighted(2)
sem := semaphore.NewWeighted(2)
tryAcquire := func(n int64) bool {
ctx, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
defer cancel()
@ -113,7 +114,7 @@ func TestWeightedDoesntBlockIfTooBig(t *testing.T) {
t.Parallel()
const n = 2
sem := NewWeighted(n)
sem := semaphore.NewWeighted(n)
{
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@ -132,7 +133,7 @@ func TestWeightedDoesntBlockIfTooBig(t *testing.T) {
})
}
if err := g.Wait(); err != nil {
t.Errorf("NewWeighted(%v) failed to AcquireCtx(_, 1) with AcquireCtx(_, %v) pending", n, n+1)
t.Errorf("semaphore.NewWeighted(%v) failed to AcquireCtx(_, 1) with AcquireCtx(_, %v) pending", n, n+1)
}
}
@ -143,7 +144,7 @@ func TestLargeAcquireDoesntStarve(t *testing.T) {
ctx := context.Background()
n := int64(runtime.GOMAXPROCS(0))
sem := NewWeighted(n)
sem := semaphore.NewWeighted(n)
running := true
var wg sync.WaitGroup