-
Notifications
You must be signed in to change notification settings - Fork 0
/
queue.go
88 lines (78 loc) · 1.51 KB
/
queue.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
//Priority queue implementation, Heap based, Concurrency safe, FIFO
package PriorityQueue
import (
"container/heap"
"sync"
//"reflect"
"errors"
"math"
)
type Queue struct {
qh *QueueHeap
m sync.Mutex
count uint64
}
func Build() Queue {
qh := make(QueueHeap, 0)
heap.Init(&qh)
queue := Queue{qh: &qh}
return queue
}
func (q *Queue) Push(i interface{}, pr int) (bool, error) {
q.m.Lock()
defer q.m.Unlock()
if uint64(q.qh.Len()) == math.MaxUint64 {
return false, errors.New("queue is full")
}
if q.count == math.MaxUint64 {
q.count = q.qh.CollapseOrder()
}
q.count++
hi := HeapItem{
order: q.count,
priority: pr,
data: i,
}
heap.Push(q.qh, &hi)
return true, nil
}
func (q *Queue) Pull() (interface{}, error) {
if q.Len() == 0 {
return nil, errors.New("empty")
}
q.m.Lock()
defer q.m.Unlock()
item := heap.Pop(q.qh).(*HeapItem)
return item.data, nil
}
func (q *Queue) Len() int {
q.m.Lock()
defer q.m.Unlock()
return q.qh.Len()
}
//Receives channels in order of priority.(more is better) Returns output channel. (exp)
func Prioritize(ins ...chan interface{}) (chan interface{}, error) {
out := make(chan interface{})
q := Build()
c := sync.NewCond(&q.m)
for pr, ch := range ins {
go func(ch chan interface{}, pr int) {
for item := range ch {
q.Push(item, pr)
c.Signal()
}
}(ch, pr)
}
go func() {
for {
q.m.Lock()
if q.qh.Len() == 0 {
c.Wait()
}
item := heap.Pop(q.qh).(*HeapItem).data
q.m.Unlock()
out <- item
}
}()
return out, nil
}