715ef8267e
* add functionality to wait for specialization by keeping track of incoming requests * format executor package * fix required capacity to specialise new pod condition * move handling concurrency logic into pool cache from executor * remove unused methods and structs * implement queue in to store the svc wait * create a queue struct and its methods to handle concurrent inputs * use newly created queue to store waiting for svc requests * add waiting requests in queue and use them when a svc is ready * set function to request in queue if the context is still alive * remove concurrency approach to set svc for waiting requests * update the active requests whenever requests from pool are assigned a svc * add doc to define why the conditions exist * remove unwanted params in strcut and clean up code * set error while getting svc value if sum of specialization in progress and specialized is only more than concurrency limit * remove duplicate functions and unnecessary values in struct * close svc channel on set value and create constants for default concurrency and rpp * get next value in queue in case context is timed out for fetched value * remove specializationInProgress counter from pool cache * return in case the queue is empty wihle setting func to svc * test getSvcVaue and setSvcValue in poolcache * add unit tests for GetConcurrent and GetRequestsPerPod methods * reorder imports * add fuzzy testing for getSVCValue and setSVCValue in poolcache * restructure go mod file and update pool cache test cases * Add tests and bug fixes * refactor code and add test cases * add svcWaiting check while setting svc value --------- Signed-off-by: Sanket Sudake <sanketsudake@gmail.com> Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>
46 lines
617 B
Go
46 lines
617 B
Go
package fscache
|
|
|
|
import (
|
|
"container/list"
|
|
"sync"
|
|
)
|
|
|
|
type Queue struct {
|
|
items *list.List
|
|
mutex sync.Mutex
|
|
}
|
|
|
|
func NewQueue() *Queue {
|
|
return &Queue{
|
|
items: list.New(),
|
|
}
|
|
}
|
|
|
|
func (q *Queue) Push(item *svcWait) {
|
|
q.mutex.Lock()
|
|
defer q.mutex.Unlock()
|
|
q.items.PushBack(item)
|
|
}
|
|
|
|
func (q *Queue) Pop() *svcWait {
|
|
q.mutex.Lock()
|
|
defer q.mutex.Unlock()
|
|
|
|
item := q.items.Front()
|
|
if item == nil {
|
|
return nil
|
|
}
|
|
q.items.Remove(item)
|
|
svcWait, ok := item.Value.(*svcWait)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
return svcWait
|
|
}
|
|
|
|
func (q *Queue) Len() int {
|
|
q.mutex.Lock()
|
|
defer q.mutex.Unlock()
|
|
return q.items.Len()
|
|
}
|