Make queue operation more clear
License: MIT Signed-off-by: Erik Ingenito <erik@carbonfive.com>
Erik Ingenito committed
Mar 15, 2019 at 20:45 UTC
6c1eca959eeda71868bb7bcc0523e093b248a75c
2 files changed
+45
-29
provider/provider_test.go
+1
-1
@@ -42,7 +42,7 @@ func TestAnnouncement(t *testing.T) {
42
43
cids := cid.NewSet()
44
45
- for i := 0; i < 1000; i++ {
45
+ for i := 0; i < 100; i++ {
46
c := blockGenerator.Next().Cid()
47
cids.Add(c)
48
}
provider/queue.go
+44
-28
@@ -61,37 +61,50 @@ func (q *Queue) Dequeue() <-chan cid.Cid {
61
return q.dequeue
62
}
63
64
+type entry struct {
65
+ cid cid.Cid
66
+ key datastore.Key
67
+}
68
+
69
+// Look for next Cid in the queue and return it. Skip over gaps and mangled data
70
+func (q *Queue) nextEntry() (datastore.Key, cid.Cid) {
71
+ for {
72
+ if q.head >= q.tail {
73
+ return datastore.Key{}, cid.Undef
74
+ }
75
+
76
+ key := q.queueKey(q.head)
77
+ value, err := q.ds.Get(key)
78
+
79
+ if err == datastore.ErrNotFound {
80
+ log.Warningf("Error missing entry in queue: %s", key)
81
+ q.head++ // move on
82
+ continue
83
+ } else if err != nil {
84
+ log.Warningf("Error fetching from queue: %s", err)
85
+ continue
86
+ }
87
+
88
+ c, err := cid.Parse(value)
89
+ if err != nil {
90
+ log.Warningf("Error marshalling Cid from queue: ", err)
91
+ q.head++
92
+ err = q.ds.Delete(key)
93
+ continue
94
+ }
95
+
96
+ return key, c
97
+ }
98
+}
99
+
100
// Run dequeues and enqueues when available.
101
func (q *Queue) work() {
102
go func() {
103
+
104
for {
68
- var c cid.Cid = cid.Undef
69
- var key datastore.Key
105
+ k, c := q.nextEntry()
106
var dequeue chan cid.Cid
107
72
- // If we're not empty dequeue a cid and ship it
73
- if q.head < q.tail {
74
- key = q.queueKey(q.head)
75
- value, err := q.ds.Get(key)
76
-
77
- if err == datastore.ErrNotFound {
78
- log.Warningf("Missing entry in queue: %s", err)
79
- q.head++
80
- continue
81
- } else if err != nil {
82
- log.Warningf("Error fetching from queue: %s", err)
83
- continue
84
- }
85
-
86
- c, err = cid.Parse(value)
87
- if err != nil {
88
- log.Warningf("Error marshalling Cid from queue: ", err)
89
- q.head++
90
- err = q.ds.Delete(key)
91
- continue
92
- }
93
- }
94
-
108
if c != cid.Undef {
109
dequeue = q.dequeue
110
}
@@ -102,16 +115,19 @@ func (q *Queue) work() {
115
116
if err := q.ds.Put(nextKey, toQueue.Bytes()); err != nil {
117
log.Errorf("Failed to enqueue cid: %s", err)
118
+ continue
119
}
120
121
q.tail++
122
case dequeue <- c:
109
- q.head++
110
- err := q.ds.Delete(key)
123
+ err := q.ds.Delete(k)
124
125
if err != nil {
113
- log.Errorf("Failed to delete queued cid: %s", err)
126
+ log.Errorf("Failed to delete queued cid %s with key %s: %s", c, k, err)
127
+ continue
128
}
129
+
130
+ q.head++
131
case <-q.ctx.Done():
132
return
133
}