Repository: knative/pkg
Pull request #1512 (2020-07-17 03:32:41)

Implement Slowlane/FastLane queues
The suggested implementation is basically a Y joint where the data flows from 2 queues into the third queue, with the caveat that one of the paths has priority over the other.
This serves as a drop-in replacement in the controller.
Which will be in a separate pr.

/assign mattmoor

For #https://github.com/knative/pkg/issues/1505

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1512#" title="Author self-approved">vagababov</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:

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

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
---- |:------------:|:------------:|:-----:
[controller/two_lane_queue.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1512/pull-knative-pkg-go-coverage/1283968669172371456/artifacts/line-cov.html#file0) | Do not exist | 95.3% | 

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1512#" title="Author self-approved">vagababov</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **mattmoor**
You can assign the PR to them by writing `/assign @mattmoor` 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:

- ~~[controller/OWNERS](https://github.com/knative/pkg/blob/master/controller/OWNERS)~~ [vagababov]
- **[webhook/OWNERS](https://github.com/knative/pkg/blob/master/webhook/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":["mattmoor"]} -->
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
---- |:------------:|:------------:|:-----:
[controller/two_lane_queue.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1512/pull-knative-pkg-go-coverage/1284162193926918144/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[webhook/webhook.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1512/pull-knative-pkg-go-coverage/1284162193926918144/artifacts/line-cov.html#file2) | 87.0% | 86.8% | -0.2

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

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1512#" title="Author self-approved">vagababov</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:

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

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":[]} -->
mattmoor: /lgtm
/approve
/hold

It would be good to import this into serving / eventing and make sure things don't blow up.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/1512#issuecomment-660210611" title="Approved">mattmoor</a>*, *<a href="https://github.com/knative/pkg/pull/1512#" title="Author self-approved">vagababov</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:

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

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
---- |:------------:|:------------:|:-----:
[controller/two_lane_queue.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative_pkg/1512/pull-knative-pkg-go-coverage/1284165340904820736/artifacts/line-cov.html#file0) | Do not exist | 94.7% | 

vagababov: It's in serving: https://github.com/knative/serving/pull/8671
vagababov: Meh, the latest one I removed controller.
I am going to merge this, since it's just a library for now and stage another to import to serving.
mattmoor: /hold cancel

Patch:
diff --git a/controller/two_lane_queue.go b/controller/two_lane_queue.go
new file mode 100644
index 000000000..165eb2df3
--- /dev/null
+++ b/controller/two_lane_queue.go
@@ -0,0 +1,159 @@
+/*
+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
+
+    http://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 controller
+
+import (
+	"k8s.io/client-go/util/workqueue"
+)
+
+// twoLaneQueue is a rate limited queue that wraps around two queues
+// -- fast queue (anonymously aliased), whose contents are processed with priority.
+// -- slow queue (slowLane queue), whose contents are processed if fast queue has no items.
+// All the default methods operate on the fast queue, unless noted otherwise.
+type twoLaneQueue struct {
+	workqueue.RateLimitingInterface
+	slowLane workqueue.RateLimitingInterface
+	// consumerQueue is necessary to ensure that we're not reconciling
+	// the same object at the exact same time (e.g. if it had been enqueued
+	// in both fast and slow and is the only object there).
+	consumerQueue workqueue.Interface
+
+	name string
+
+	fastChan chan interface{}
+	slowChan chan interface{}
+}
+
+// Creates a new twoLaneQueue.
+func newTwoLaneWorkQueue(name string) *twoLaneQueue {
+	rl := workqueue.DefaultControllerRateLimiter()
+	tlq := &twoLaneQueue{
+		RateLimitingInterface: workqueue.NewNamedRateLimitingQueue(
+			rl,
+			name+"-fast",
+		),
+		slowLane: workqueue.NewNamedRateLimitingQueue(
+			rl,
+			name+"-slow",
+		),
+		consumerQueue: workqueue.NewNamed(name + "-consumer"),
+		name:          name,
+		fastChan:      make(chan interface{}),
+		slowChan:      make(chan interface{}),
+	}
+	// Run consumer thread.
+	go tlq.runConsumer()
+	// Run producer threads.
+	go process(tlq.RateLimitingInterface, tlq.fastChan)
+	go process(tlq.slowLane, tlq.slowChan)
+	return tlq
+}
+
+func process(q workqueue.Interface, ch chan interface{}) {
+	// Sender closes the channel
+	defer close(ch)
+	for {
+		i, d := q.Get()
+		// If the queue is empty and we're shutting down — stop the loop.
+		if d {
+			break
+		}
+		ch <- i
+		q.Done(i)
+	}
+}
+
+func (tlq *twoLaneQueue) runConsumer() {
+	// Shutdown flags.
+	fast, slow := true, true
+	// When both producer queues are shutdown stop the consumerQueue.
+	defer tlq.consumerQueue.ShutDown()
+	// While any of the queues is still running, try to read off of them.
+	for fast || slow {
+		// By default drain the fast lane.
+		// Channels in select are picked random, so first
+		// we have a select that only looks at the fast lane queue.
+		if fast {
+			select {
+			case item, ok := <-tlq.fastChan:
+				if !ok {
+					// This queue is shutdown and drained. Stop looking at it.
+					fast = false
+					continue
+				}
+				tlq.consumerQueue.Add(item)
+				continue
+			default:
+				// This immediately exits the wait if the fast chan is empty.
+			}
+		}
+
+		// If the fast lane queue had no items, we can select from both.
+		// Obviously if suddenly both are populated at the same time there's a
+		// 50% chance that the slow would be picked first, but this should be
+		// a rare occasion not to really worry about it.
+		select {
+		case item, ok := <-tlq.fastChan:
+			if !ok {
+				// This queue is shutdown and drained. Stop looking at it.
+				fast = false
+				continue
+			}
+			tlq.consumerQueue.Add(item)
+		case item, ok := <-tlq.slowChan:
+			if !ok {
+				// This queue is shutdown and drained. Stop looking at it.
+				slow = false
+				continue
+			}
+			tlq.consumerQueue.Add(item)
+		}
+	}
+}
+
+// Shutdown implements workqueue.Interace.
+// Shutdown shuts down both queues.
+func (tlq *twoLaneQueue) ShutDown() {
+	tlq.RateLimitingInterface.ShutDown()
+	tlq.slowLane.ShutDown()
+}
+
+// Done implements workqueue.Interface.
+// Done marks the item as completed in all the queues.
+// NB: this will just re-enqueue the object on the queue that
+// didn't originate the object.
+func (tlq *twoLaneQueue) Done(i interface{}) {
+	tlq.consumerQueue.Done(i)
+}
+
+// Get implements workqueue.Interface.
+// It gets the item from fast lane if it has anything, alternatively
+// the slow lane.
+func (tlq *twoLaneQueue) Get() (interface{}, bool) {
+	return tlq.consumerQueue.Get()
+}
+
+// Len returns the sum of lengths.
+// NB: actual _number_ of unique object might be less than this sum.
+func (tlq *twoLaneQueue) Len() int {
+	return tlq.RateLimitingInterface.Len() + tlq.slowLane.Len() + tlq.consumerQueue.Len()
+}
+
+// SlowLane gives direct access to the slow queue.
+func (tlq *twoLaneQueue) SlowLane() workqueue.RateLimitingInterface {
+	return tlq.slowLane
+}


Test patch:
diff --git a/controller/two_lane_queue_test.go b/controller/two_lane_queue_test.go
new file mode 100644
index 000000000..79c2fb34f
--- /dev/null
+++ b/controller/two_lane_queue_test.go
@@ -0,0 +1,133 @@
+/*
+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
+
+    http://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 controller
+
+import (
+	"strconv"
+	"testing"
+	"time"
+)
+
+func TestSlowQueue(t *testing.T) {
+	q := newTwoLaneWorkQueue("live-in-the-fast-lane")
+	q.SlowLane().Add("1")
+	if got, want := q.Len(), 1; got != want {
+		t.Errorf("Len = %d, want: 1", got)
+	}
+	k, done := q.Get()
+	if got, want := k.(string), "1"; got != want {
+		t.Errorf(`Got = %q, want: "1"`, got)
+	}
+	if done {
+		t.Error("The queue is unexpectedly shutdown")
+	}
+	q.Done(k)
+	q.ShutDown()
+	if !q.SlowLane().ShuttingDown() {
+		t.Error("ShutDown did not propagate to the slow queue")
+	}
+	if _, done := q.Get(); !done {
+		t.Error("Get did not return positive shutdown signal")
+	}
+}
+
+func TestDoubleKey(t *testing.T) {
+	// Verifies that we don't get double concurrent processing of the same key.
+	q := newTwoLaneWorkQueue("live-in-the-fast-lane")
+	q.Add("1")
+	q.SlowLane().Add("1")
+	t.Cleanup(q.ShutDown)
+
+	k, done := q.Get()
+	if got, want := k.(string), "1"; got != want {
+		t.Errorf(`Got = %q, want: "1"`, got)
+	}
+	if done {
+		t.Error("The queue is unexpectedly shutdown")
+	}
+
+	sentinel := make(chan struct{})
+	go func() {
+		defer close(sentinel)
+		k, done := q.Get()
+		if got, want := k.(string), "1"; got != want {
+			t.Errorf(`2nd time got = %q, want: "1"`, got)
+		}
+		if done {
+			t.Error("The queue is unexpectedly shutdown")
+		}
+		q.Done(k)
+	}()
+	select {
+	case <-sentinel:
+		t.Error("The sentinel should not have fired")
+	case <-time.After(200 * time.Millisecond):
+		// Expected.
+	}
+	// This should permit the re-reading of the same key.
+	q.Done(k)
+	select {
+	case <-sentinel:
+		// Expected.
+	case <-time.After(200 * time.Millisecond):
+		t.Error("The item was not processed as expected")
+	}
+}
+
+func TestOrder(t *testing.T) {
+	// Verifies that we read from the fast queue first.
+	q := newTwoLaneWorkQueue("live-in-the-fast-lane")
+	stop := make(chan struct{})
+	t.Cleanup(func() {
+		close(stop)
+		q.ShutDown()
+		// Drain the rest.
+		for q.Len() > 0 {
+			q.Get()
+		}
+	})
+
+	go func() {
+		for i := 1; ; i++ {
+			q.Add(strconv.Itoa(i))
+			// Get fewer of those, to ensure the first priority select wins.
+			if i%2 == 0 {
+				q.SlowLane().Add("slow" + strconv.Itoa(i))
+			}
+			select {
+			case <-stop:
+				return
+			default:
+			}
+		}
+	}()
+	done := time.After(300 * time.Millisecond)
+	for {
+		select {
+		case <-done:
+			return
+		default:
+		}
+		v, sd := q.Get()
+		if sd {
+			t.Error("Got shutdown signal")
+		} else if v.(string) == "slow" {
+			t.Error("Got item from the slow queue")
+		}
+		q.Done(v)
+	}
+}