Repository: knative/pkg
Pull request #1341 (2020-05-16 19:48:21)

Move knative.dev/serving/pkg/pool to knative.dev/pkg/pool

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1341#" title="Author self-approved">tcnghia</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **vaikas**
You can assign the PR to them by writing `/assign @vaikas` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative/pkg/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["vaikas"]} -->
knative-prow-robot: The following users are mentioned in OWNERS file(s) but are not members of the knative org.

Once all users have been added as [members]() of the org, you can trigger verification by writing `/verify-owners` in a comment.
- api-reviewers
  - pool/OWNERS
- api-approvers
  - pool/OWNERS
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pool/pool.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1341/pull-knative-pkg-go-coverage/1261745437434449921/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 

knative-prow-robot: The following users are mentioned in OWNERS file(s) but are not members of the knative org.

Once all users have been added as [members]() of the org, you can trigger verification by writing `/verify-owners` in a comment.
- api-reviewers
  - pool/OWNERS
- api-approvers
  - pool/OWNERS
knative-prow-robot: The following users are mentioned in OWNERS file(s) but are not members of the knative org.

Once all users have been added as [members]() of the org, you can trigger verification by writing `/verify-owners` in a comment.
- apis-reviewers
  - pool/OWNERS
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1341#pullrequestreview-413104954" title="Approved">mattmoor</a>*, *<a href="https://github.com/knative/pkg/pull/1341#" title="Author self-approved">tcnghia</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative/pkg/blob/master/OWNERS)~~ [mattmoor]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-pkg-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pool/pool.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1341/pull-knative-pkg-go-coverage/1261749589812187138/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 

vagababov: /lgtm

Patch:
diff --git a/pool/OWNERS b/pool/OWNERS
new file mode 100644
index 000000000..219c8fd67
--- /dev/null
+++ b/pool/OWNERS
@@ -0,0 +1,7 @@
+# The OWNERS file is used by prow to automatically merge approved PRs.
+
+approvers:
+- apis-approvers
+
+labels:
+- area/API
diff --git a/pool/doc.go b/pool/doc.go
new file mode 100644
index 000000000..052b9ee93
--- /dev/null
+++ b/pool/doc.go
@@ -0,0 +1,21 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package pool contains a simple threadpool implementation that accepts
+// work in the form of `func() error` function.  The intent is for it to
+// support similar workloads to errgroup, but with a maximum number of
+// concurrent worker threads.
+package pool
diff --git a/pool/interface.go b/pool/interface.go
new file mode 100644
index 000000000..86cc8e48c
--- /dev/null
+++ b/pool/interface.go
@@ -0,0 +1,36 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pool
+
+import (
+	"golang.org/x/sync/errgroup"
+)
+
+// Interface defines an errgroup-compatible interface for interacting with
+// our threadpool.
+type Interface interface {
+	// Go queues a single unit of work for execution on this pool. All calls
+	// to Go must be finished before Wait is called.
+	Go(func() error)
+
+	// Wait blocks until all work is complete, returning the first
+	// error returned by any of the work.
+	Wait() error
+}
+
+// errgroup.Group implements Interface
+var _ Interface = (*errgroup.Group)(nil)
diff --git a/pool/pool.go b/pool/pool.go
new file mode 100644
index 000000000..9f298d41b
--- /dev/null
+++ b/pool/pool.go
@@ -0,0 +1,117 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pool
+
+import (
+	"context"
+	"sync"
+)
+
+type impl struct {
+	wg     sync.WaitGroup
+	workCh chan func() error
+
+	// Ensure that we Wait exactly once and memoize
+	// the result.
+	waitOnce sync.Once
+
+	cancel context.CancelFunc
+
+	// We're only interested in the first result so
+	// only set it once.
+	resultOnce sync.Once
+	result     error
+}
+
+// impl implements Interface
+var _ Interface = (*impl)(nil)
+
+// defaultCapacity is the number of work items or errors that we
+// can queue up before calls to Go will block, or work will
+// block until Wait is called.
+const defaultCapacity = 50
+
+// New creates a fresh worker pool with the specified size.
+func New(workers int) Interface {
+	return NewWithCapacity(workers, defaultCapacity)
+}
+
+// NewWithCapacity creates a fresh worker pool with the specified size.
+func NewWithCapacity(workers, capacity int) Interface {
+	i, _ := NewWithContext(context.Background(), workers, capacity)
+	return i
+}
+
+// NewWithContext creates a pool that is driven by a cancelable context.
+// Just like errgroup.Group on first error the context will be canceled as well.
+func NewWithContext(ctx context.Context, workers, capacity int) (Interface, context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	i := &impl{
+		cancel: cancel,
+		workCh: make(chan func() error, capacity),
+	}
+
+	// Start a go routine for each worker, which:
+	// 1. reads off of the work channel,
+	// 2. (optionally) sets the error as the result,
+	// 3. marks work as done in our sync.WaitGroup.
+	for idx := 0; idx < workers; idx++ {
+		go func() {
+			for work := range i.workCh {
+				i.exec(work)
+			}
+		}()
+	}
+	return i, ctx
+}
+
+func (i *impl) exec(w func() error) {
+	defer i.wg.Done()
+	if err := w(); err != nil {
+		i.resultOnce.Do(func() {
+			if i.cancel != nil {
+				i.cancel()
+			}
+			i.result = err
+		})
+	}
+}
+
+// Go implements Interface.
+func (i *impl) Go(w func() error) {
+	// Increment the amount of outstanding work we're waiting on.
+	i.wg.Add(1)
+	// Send the work along the queue.
+	i.workCh <- w
+}
+
+// Wait implements Interface.
+func (i *impl) Wait() error {
+	i.waitOnce.Do(func() {
+		// Wait for queued work to complete.
+		i.wg.Wait()
+		// Notify the context, that it's done now.
+		if i.cancel != nil {
+			i.cancel()
+		}
+
+		// Now we know there are definitely no new items arriving.
+		close(i.workCh)
+	})
+
+	return i.result
+}


Test patch:
diff --git a/pool/pool_test.go b/pool/pool_test.go
new file mode 100644
index 000000000..2cbab7a1b
--- /dev/null
+++ b/pool/pool_test.go
@@ -0,0 +1,246 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pool
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestParallelismNoErrors(t *testing.T) {
+	tests := []struct {
+		name string
+		size int
+		work int
+	}{{
+		name: "single threaded",
+		size: 1,
+		work: 3,
+	}, {
+		name: "three workers",
+		size: 3,
+		work: 10,
+	}, {
+		name: "ten workers",
+		size: 10,
+		work: 100,
+	}}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var (
+				// m guards max.
+				m      sync.Mutex
+				max    int32
+				active int32
+			)
+
+			// Use our own waitgroup to ensure that the work
+			// can all complete before we block on the error
+			// result.
+			wg := &sync.WaitGroup{}
+
+			worker := func() error {
+				defer wg.Done()
+				na := atomic.AddInt32(&active, 1)
+				defer atomic.AddInt32(&active, -1)
+
+				func() {
+					m.Lock()
+					defer m.Unlock()
+					if max < na {
+						max = na
+					}
+				}()
+
+				// Sleep a small amount to simulate work. This should be
+				// sufficient to saturate the threadpool before the first
+				// one wakes up.
+				time.Sleep(10 * time.Millisecond)
+				return nil
+			}
+
+			p := New(tc.size)
+			for idx := 0; idx < tc.work; idx++ {
+				wg.Add(1)
+				p.Go(worker)
+			}
+
+			// First wait for the waitgroup to finish, so that
+			// we are sure it isn't the Wait call that flushes
+			// remaining work.
+			wg.Wait()
+
+			if err := p.Wait(); err != nil {
+				t.Errorf("Wait() = %v", err)
+			}
+
+			if err := p.Wait(); err != nil {
+				t.Errorf("Wait() = %v", err)
+			}
+
+			if got, want := max, int32(tc.size); got != want {
+				t.Errorf("max active = %v, wanted %v", got, want)
+			}
+		})
+	}
+}
+
+func TestParallelismWithErrors(t *testing.T) {
+	tests := []struct {
+		name string
+		size int
+		work int
+	}{{
+		name: "single threaded",
+		size: 1,
+		work: 3,
+	}, {
+		name: "three workers",
+		size: 3,
+		work: 10,
+	}, {
+		name: "ten workers",
+		size: 10,
+		work: defaultCapacity + 10,
+	}}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var (
+				// m guards max.
+				m      sync.Mutex
+				max    int32
+				active int32
+			)
+
+			// Use our own waitgroup to ensure that the work
+			// can all complete before we block on the error
+			// result.
+			wg := &sync.WaitGroup{}
+
+			// The barrier holds all requests we don't expect to
+			// finish until we validate the condition.
+			barrier := make(chan struct{})
+
+			errExpected := errors.New("this is what I expect")
+			workerFactory := func(err error) func() error {
+				return func() error {
+					defer wg.Done()
+					na := atomic.AddInt32(&active, 1)
+					defer atomic.AddInt32(&active, -1)
+
+					func() {
+						m.Lock()
+						defer m.Unlock()
+						if max < na {
+							max = na
+						}
+					}()
+
+					// Sleep a small amount to simulate work. This should be
+					// sufficient to saturate the threadpool before the first
+					// one wakes up.
+					time.Sleep(10 * time.Millisecond)
+
+					// Make all unexpected errors wait.
+					if err != errExpected {
+						<-barrier
+					}
+					return err
+				}
+			}
+
+			p, ctx := NewWithContext(context.Background(), tc.size, defaultCapacity)
+
+			for i := 0; i < tc.work; i++ {
+				err := errors.New("this is not what I expect")
+				if i == tc.size-1 {
+					// Just fit in a request with the error we expect.
+					err = errExpected
+				}
+
+				wg.Add(1)
+				p.Go(workerFactory(err))
+			}
+
+			// Wait for the error to propagate.
+			<-ctx.Done()
+			// Allow all remaining requests to finish.
+			close(barrier)
+
+			// First wait for the waitgroup to finish, so that
+			// we are sure it isn't the Wait call that flushes
+			// remaining work.
+			wg.Wait()
+
+			if err := p.Wait(); err != errExpected {
+				t.Errorf("Wait() = %v, wanted %v", err, errExpected)
+			}
+
+			if got, want := max, int32(tc.size); got != want {
+				t.Errorf("max active = %v, wanted %v", got, want)
+			}
+		})
+	}
+}
+
+func TestWithContextWaitCancels(t *testing.T) {
+	pool, ctx := NewWithContext(context.Background(), 1 /*1 thread*/, 10 /*capacity*/)
+	for i := 0; i < 10; i++ {
+		pool.Go(func() error {
+			time.Sleep(10 * time.Millisecond)
+			return nil
+		})
+	}
+	if err := pool.Wait(); err != nil {
+		t.Fatal("pool.Wait =", err)
+	}
+	select {
+	case <-ctx.Done():
+	default:
+		t.Error("ctx is not canceled")
+	}
+}
+
+func TestErrorCancelsContext(t *testing.T) {
+	want := errors.New("i failed, sorry")
+	pool, ctx := NewWithContext(context.Background(), 1 /*1 thread*/, 10 /*capacity*/)
+	pool.Go(func() error {
+		return want
+	})
+	// Those don't matter, but generate load.
+	for i := 0; i < 10; i++ {
+		pool.Go(func() error {
+			time.Sleep(100 * time.Millisecond)
+			return nil
+		})
+	}
+	// This should be triggered basically immediately.
+	select {
+	case <-ctx.Done():
+	case <-time.After(100 * time.Millisecond):
+		t.Error("ctx is not canceled due to the first error")
+	}
+	if err := pool.Wait(); err != want {
+		t.Fatalf("pool.Wait() = %v, want: %v", err, want)
+	}
+}