apf: use a list instead of slice for queueset

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/fifo_list.go

@@ -0,0 +1,102 @@
+/*
+Copyright 2021 The Kubernetes 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 queueset
+
+import (
+	"container/list"
+)
+
+// removeFromFIFOFunc removes a designated element from the list.
+// The complexity of the runtime cost is O(1)
+// It returns the request removed from the list.
+type removeFromFIFOFunc func() *request
+
+// walkFunc is called for each request in the list in the
+// oldest -> newest order.
+// ok: if walkFunc returns false then the iteration stops immediately.
+type walkFunc func(*request) (ok bool)
+
+// Internal interface to abstract out the implementation details
+// of the underlying list used to maintain the requests.
+//
+// Note that the FIFO list is not safe for concurrent use by multiple
+// goroutines without additional locking or coordination. It rests with
+// the user to ensure that the FIFO list is used with proper locking.
+type fifo interface {
+	// Enqueue enqueues the specified request into the list and
+	// returns a removeFromFIFOFunc function that can be used to remove the
+	// request from the list
+	Enqueue(*request) removeFromFIFOFunc
+
+	// Dequeue pulls out the oldest request from the list.
+	Dequeue() (*request, bool)
+
+	// Length returns the number of requests in the list.
+	Length() int
+
+	// Walk iterates through the list in order of oldest -> newest
+	// and executes the specified walkFunc for each request in that order.
+	//
+	// if the specified walkFunc returns false the Walk function
+	// stops the walk an returns immediately.
+	Walk(walkFunc)
+}
+
+// the FIFO list implementation is not safe for concurrent use by multiple
+// goroutines without additional locking or coordination.
+type requestFIFO struct {
+	*list.List
+}
+
+func newRequestFIFO() fifo {
+	return &requestFIFO{
+		List: list.New(),
+	}
+}
+
+func (l *requestFIFO) Length() int {
+	return l.Len()
+}
+
+func (l *requestFIFO) Enqueue(req *request) removeFromFIFOFunc {
+	e := l.PushBack(req)
+	return func() *request {
+		l.Remove(e)
+		return req
+	}
+}
+
+func (l *requestFIFO) Dequeue() (*request, bool) {
+	e := l.Front()
+	if e == nil {
+		return nil, false
+	}
+	defer l.Remove(e)
+
+	request, ok := e.Value.(*request)
+	return request, ok
+}
+
+func (l *requestFIFO) Walk(f walkFunc) {
+	for current := l.Front(); current != nil; current = current.Next() {
+		if r, ok := current.Value.(*request); ok {
+			if !f(r) {
+				return
+			}
+		}
+	}
+}

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/queueset/fifo_list_test.go

@@ -0,0 +1,183 @@
+/*
+Copyright 2021 The Kubernetes 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 queueset
+
+import (
+	"math/rand"
+	"testing"
+	"time"
+)
+
+func TestFIFOWithEnqueueDequeueSingleRequest(t *testing.T) {
+	req := &request{}
+
+	list := newRequestFIFO()
+	list.Enqueue(req)
+
+	reqGot, ok := list.Dequeue()
+	if !ok {
+		t.Errorf("Expected true, but got: %t", ok)
+	}
+	if req != reqGot {
+		t.Errorf("Expected dequued request: (%p), but got: (%p)", req, reqGot)
+	}
+	if list.Length() != 0 {
+		t.Errorf("Expected length: %d, but got: %d)", 0, list.Length())
+	}
+}
+
+func TestFIFOWithEnqueueDequeueMultipleRequests(t *testing.T) {
+	arrival := []*request{{}, {}, {}, {}, {}, {}}
+
+	list := newRequestFIFO()
+	for i := range arrival {
+		list.Enqueue(arrival[i])
+	}
+
+	dequeued := make([]*request, 0)
+	for list.Length() > 0 {
+		req, _ := list.Dequeue()
+		dequeued = append(dequeued, req)
+	}
+
+	verifyOrder(t, arrival, dequeued)
+}
+
+func TestFIFOWithEnqueueDequeueSomeRequestsRemainInQueue(t *testing.T) {
+	list := newRequestFIFO()
+
+	arrival := []*request{{}, {}, {}, {}, {}, {}}
+	half := len(arrival) / 2
+	for i := range arrival {
+		list.Enqueue(arrival[i])
+	}
+
+	dequeued := make([]*request, 0)
+	for i := 0; i < half; i++ {
+		req, _ := list.Dequeue()
+		dequeued = append(dequeued, req)
+	}
+
+	verifyOrder(t, arrival[:half], dequeued)
+}
+
+func TestFIFOWithRemoveMultipleRequestsInArrivalOrder(t *testing.T) {
+	list := newRequestFIFO()
+
+	arrival := []*request{{}, {}, {}, {}, {}, {}}
+	removeFn := make([]removeFromFIFOFunc, 0)
+	for i := range arrival {
+		removeFn = append(removeFn, list.Enqueue(arrival[i]))
+	}
+
+	dequeued := make([]*request, 0)
+	for _, f := range removeFn {
+		dequeued = append(dequeued, f())
+	}
+
+	if list.Length() != 0 {
+		t.Errorf("Expected length: %d, but got: %d)", 0, list.Length())
+	}
+
+	verifyOrder(t, arrival, dequeued)
+}
+
+func TestFIFOWithRemoveMultipleRequestsInRandomOrder(t *testing.T) {
+	list := newRequestFIFO()
+
+	arrival := []*request{{}, {}, {}, {}, {}, {}}
+	removeFn := make([]removeFromFIFOFunc, 0)
+	for i := range arrival {
+		removeFn = append(removeFn, list.Enqueue(arrival[i]))
+	}
+
+	dequeued := make([]*request, 0)
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	randomIndices := r.Perm(len(removeFn))
+	t.Logf("Random remove order: %v", randomIndices)
+	for i := range randomIndices {
+		dequeued = append(dequeued, removeFn[i]())
+	}
+
+	if list.Length() != 0 {
+		t.Errorf("Expected length: %d, but got: %d)", 0, list.Length())
+	}
+
+	verifyOrder(t, arrival, dequeued)
+}
+
+func TestFIFOWithRemoveIsIdempotent(t *testing.T) {
+	list := newRequestFIFO()
+
+	arrival := []*request{{}, {}, {}, {}}
+	removeFn := make([]removeFromFIFOFunc, 0)
+	for i := range arrival {
+		removeFn = append(removeFn, list.Enqueue(arrival[i]))
+	}
+
+	// pick one request to be removed at random
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	randomIndex := r.Intn(len(removeFn))
+	t.Logf("Random remove index: %d", randomIndex)
+
+	// remove the request from the fifo twice, we expect it to be idempotent
+	removeFn[randomIndex]()
+	removeFn[randomIndex]()
+
+	lengthExpected := len(arrival) - 1
+	if lengthExpected != list.Length() {
+		t.Errorf("Expected length: %d, but got: %d)", lengthExpected, list.Length())
+	}
+
+	orderExpected := append(arrival[0:randomIndex], arrival[randomIndex+1:]...)
+	remainingRequests := walkAll(list)
+	verifyOrder(t, orderExpected, remainingRequests)
+}
+
+func TestFIFOWithWalk(t *testing.T) {
+	list := newRequestFIFO()
+
+	arrival := []*request{{}, {}, {}, {}, {}, {}}
+	for i := range arrival {
+		list.Enqueue(arrival[i])
+	}
+
+	visited := walkAll(list)
+
+	verifyOrder(t, arrival, visited)
+}
+
+func verifyOrder(t *testing.T, expected, actual []*request) {
+	if len(expected) != len(actual) {
+		t.Fatalf("Expected slice length: %d, but got: %d", len(expected), len(actual))
+	}
+	for i := range expected {
+		if expected[i] != actual[i] {
+			t.Errorf("Dequeue order mismatch, expected request: (%p), but got: (%p)", expected[i], actual[i])
+		}
+	}
+}
+
+func walkAll(l fifo) []*request {
+	visited := make([]*request, 0)
+	l.Walk(func(req *request) bool {
+		visited = append(visited, req)
+		return true
+	})
+
+	return visited
+}