Provider queue updates to address deadlocks
License: MIT Signed-off-by: Erik Ingenito <erik@carbonfive.com>
Erik Ingenito committed
Mar 14, 2019 at 16:48 UTC
8337035466942d787ffce694c8f7d18368bb95e1
2 files changed
+18
-25
provider/provider.go
+3
-2
@@ -5,9 +5,10 @@ package provider
5
6
import (
7
"context"
8
- "github.com/ipfs/go-cid"
8
+
9
+ cid "github.com/ipfs/go-cid"
10
logging "github.com/ipfs/go-log"
10
- "github.com/libp2p/go-libp2p-routing"
11
+ routing "github.com/libp2p/go-libp2p-routing"
12
)
13
14
var (
provider/queue.go
+15
-23
@@ -24,6 +24,8 @@ type Entry struct {
24
25
// Complete the entry by removing it from the queue
26
func (e *Entry) Complete() error {
27
+ e.queue.lock.Lock()
28
+ defer e.queue.lock.Unlock()
29
return e.queue.remove(e.key)
30
}
31
@@ -46,9 +48,7 @@ type Queue struct {
48
datastore ds.Datastore
49
50
dequeue chan *Entry
49
- notEmpty chan struct{}
50
-
51
- isRunning bool
51
+ added chan struct{}
52
}
53
54
// NewQueue creates a queue for cids
@@ -66,8 +66,7 @@ func NewQueue(ctx context.Context, name string, datastore ds.Datastore) (*Queue,
66
lock: sync.Mutex{},
67
datastore: namespaced,
68
dequeue: make(chan *Entry),
69
- notEmpty: make(chan struct{}),
70
- isRunning: false,
69
+ added: make(chan struct{}),
70
}
71
return q, nil
72
}
@@ -77,8 +76,6 @@ func (q *Queue) Enqueue(cid cid.Cid) error {
76
q.lock.Lock()
77
defer q.lock.Unlock()
78
80
- wasEmpty := q.IsEmpty()
81
-
79
nextKey := q.queueKey(q.tail)
80
81
if err := q.datastore.Put(nextKey, cid.Bytes()); err != nil {
@@ -87,11 +84,10 @@ func (q *Queue) Enqueue(cid cid.Cid) error {
84
85
q.tail++
86
90
- if q.isRunning && wasEmpty {
91
- select {
92
- case q.notEmpty <- struct{}{}:
87
+ select {
88
+ case q.added <- struct{}{}:
89
case <-q.ctx.Done():
94
- }
90
+ default:
91
}
92
93
return nil
@@ -110,20 +106,13 @@ func (q *Queue) IsEmpty() bool {
106
// Run dequeues items when the dequeue channel is available to
107
// be written to.
108
func (q *Queue) Run() {
113
- q.isRunning = true
109
go func() {
110
for {
116
- select {
117
- case <-q.ctx.Done():
118
- return
119
- default:
120
- }
111
if q.IsEmpty() {
112
select {
113
case <-q.ctx.Done():
114
return
125
- // wait for a notEmpty message
126
- case <-q.notEmpty:
115
+ case <-q.added:
116
}
117
}
118
@@ -138,6 +127,7 @@ func (q *Queue) Run() {
127
return
128
case q.dequeue <- entry:
129
}
130
+
131
}
132
}()
133
}
@@ -146,14 +136,16 @@ func (q *Queue) Run() {
136
// found in the next spot.
137
func (q *Queue) next() (*Entry, error) {
138
q.lock.Lock()
149
- defer q.lock.Unlock()
139
+ defer func() {
140
+ q.lock.Unlock()
141
+ }()
142
143
var nextKey ds.Key
144
var value []byte
145
var err error
146
for {
147
if q.head >= q.tail {
156
- return nil, errors.New("no more entries in queue")
148
+ return nil, errors.New("next: no more entries in queue returning")
149
}
150
select {
151
case <-q.ctx.Done():
@@ -194,8 +186,8 @@ func (q *Queue) queueKey(id uint64) ds.Key {
186
187
// crawl over the queue entries to find the head and tail
188
func getQueueHeadTail(ctx context.Context, name string, datastore ds.Datastore) (uint64, uint64, error) {
197
- query := query.Query{}
198
- results, err := datastore.Query(query)
189
+ q := query.Query{}
190
+ results, err := datastore.Query(q)
191
if err != nil {
192
return 0, 0, err
193
}