blob: 2c95e239b59f9c80d28805138ce5abe02e6d3534 [file] [log] [blame]
Kent Hagermana05f4d42020-04-01 15:11:22 -04001/*
2 * Copyright 2020-present Open Networking Foundation
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17package utils
18
19import (
20 "context"
21 "sync"
22)
23
24type request struct {
25 prev, next *request
26 notifyOnComplete chan<- struct{}
27}
28
29// RequestQueue represents a request processing queue where each request is processed to completion before another
30// request is given the green light to proceed.
31type RequestQueue struct {
32 mutex sync.Mutex
33
34 last, current *request
35 lastCompleteCh <-chan struct{}
36}
37
38// NewRequestQueue creates a new request queue
39func NewRequestQueue() *RequestQueue {
40 ch := make(chan struct{})
41 close(ch) // assume the "current" request is already complete
42 return &RequestQueue{lastCompleteCh: ch}
43}
44
45// WaitForGreenLight is invoked by a function processing a request to receive the green light before
46// proceeding. The caller can also provide a context with timeout. The timeout will be triggered if the wait is
47// too long (previous requests taking too long)
48func (rq *RequestQueue) WaitForGreenLight(ctx context.Context) error {
49 // add ourselves to the end of the queue
50 rq.mutex.Lock()
51 waitingOn := rq.lastCompleteCh
52
53 ch := make(chan struct{})
54 rq.lastCompleteCh = ch
55 r := &request{notifyOnComplete: ch}
56
57 if rq.last != nil {
58 rq.last.next, r.prev = r, rq.last
59 }
60 rq.last = r
61 rq.mutex.Unlock()
62
63 // wait for our turn
64 select {
65 case <-ctx.Done():
66 // canceled, so cleanup
67 rq.mutex.Lock()
68 defer rq.mutex.Unlock()
69
70 select {
71 case <-waitingOn:
72 // chan has been closed, so the lock has been acquired
73 // context is canceled, so just release the lock immediately
74 rq.current = r
75 rq.releaseWithoutLock()
76 default:
77 // on abort, skip our position in the queue
78 r.prev.notifyOnComplete = r.notifyOnComplete
79 // and remove ourselves from the queue
80 if r.next != nil { // if we are somewhere in the middle of the queue
81 r.prev.next = r.next
82 r.next.prev = r.prev
83 } else { // if we are at the end of the queue
84 rq.last = r.prev
85 r.prev.next = nil
86 }
87 }
88 return ctx.Err()
89
90 case <-waitingOn:
91 // lock is acquired
92 rq.current = r
93 return nil
94 }
95}
96
97// RequestComplete must be invoked by a process when it completes processing the request. That process must have
98// invoked WaitForGreenLight() before.
99func (rq *RequestQueue) RequestComplete() {
100 rq.mutex.Lock()
101 defer rq.mutex.Unlock()
102
103 rq.releaseWithoutLock()
104}
105
106func (rq *RequestQueue) releaseWithoutLock() {
107 // Notify the next waiting request. This will panic if the lock is released more than once.
108 close(rq.current.notifyOnComplete)
109
110 if rq.current.next != nil {
111 rq.current.next.prev = nil
112 }
113}