Change queue to be timestamp based
Michael Avila committed
Jul 2, 2019 at 09:49 UTC
b10e3891357956021516eec0b4463d4c351af595
2 files changed
+37
-129
provider/queue/queue.go
+29
-93
@@ -3,8 +3,7 @@ package queue
3
import (
4
"context"
5
"fmt"
6
- "strconv"
7
- "strings"
6
+ "time"
7
8
cid "github.com/ipfs/go-cid"
9
datastore "github.com/ipfs/go-datastore"
@@ -25,8 +24,6 @@ type Queue struct {
24
// e.g. provider vs reprovider
25
name string
26
ctx context.Context
28
- tail uint64
29
- head uint64
27
ds datastore.Datastore // Must be threadsafe
28
dequeue chan cid.Cid
29
enqueue chan cid.Cid
@@ -37,16 +34,10 @@ type Queue struct {
34
// NewQueue creates a queue for cids
35
func NewQueue(ctx context.Context, name string, ds datastore.Datastore) (*Queue, error) {
36
namespaced := namespace.Wrap(ds, datastore.NewKey("/"+name+"/queue/"))
40
- head, tail, err := getQueueHeadTail(ctx, namespaced)
41
- if err != nil {
42
- return nil, err
43
- }
37
cancelCtx, cancel := context.WithCancel(ctx)
38
q := &Queue{
39
name: name,
40
ctx: cancelCtx,
48
- head: head,
49
- tail: tail,
41
ds: namespaced,
42
dequeue: make(chan cid.Cid),
43
enqueue: make(chan cid.Cid),
@@ -77,41 +68,6 @@ func (q *Queue) Dequeue() <-chan cid.Cid {
68
return q.dequeue
69
}
70
80
-// Look for next Cid in the queue and return it. Skip over gaps and mangled data
81
-func (q *Queue) nextEntry() (datastore.Key, cid.Cid) {
82
- for {
83
- if q.head >= q.tail {
84
- return datastore.Key{}, cid.Undef
85
- }
86
-
87
- key := q.queueKey(q.head)
88
- value, err := q.ds.Get(key)
89
-
90
- if err != nil {
91
- if err == datastore.ErrNotFound {
92
- log.Warningf("Error missing entry in queue: %s", key)
93
- } else {
94
- log.Errorf("Error fetching from queue: %s", err)
95
- }
96
- q.head++ // move on
97
- continue
98
- }
99
-
100
- c, err := cid.Parse(value)
101
- if err != nil {
102
- log.Warningf("Error marshalling Cid from queue: ", err)
103
- q.head++
104
- err = q.ds.Delete(key)
105
- if err != nil {
106
- log.Warningf("Provider queue failed to delete: %s", key)
107
- }
108
- continue
109
- }
110
-
111
- return key, c
112
- }
113
-}
114
-
71
// Run dequeues and enqueues when available.
72
func (q *Queue) work() {
73
go func() {
@@ -124,7 +80,26 @@ func (q *Queue) work() {
80
81
for {
82
if c == cid.Undef {
127
- k, c = q.nextEntry()
83
+ head, e := q.getQueueHead()
84
+
85
+ if e != nil {
86
+ log.Errorf("error querying for head of queue: %s, stopping provider", e)
87
+ return
88
+ } else if head != nil {
89
+ k = datastore.NewKey(head.Key)
90
+ c, e = cid.Parse(head.Value)
91
+ if e != nil {
92
+ log.Warningf("error parsing queue entry cid with key (%s), removing it from queue: %s", head.Key, e)
93
+ err := q.ds.Delete(k)
94
+ if err != nil {
95
+ log.Errorf("error deleting queue entry with key (%s), due to error (%s), stopping provider", head.Key, err)
96
+ return
97
+ }
98
+ continue
99
+ }
100
+ } else {
101
+ c = cid.Undef
102
+ }
103
}
104
105
// If c != cid.Undef set dequeue and attempt write, otherwise wait for enqueue
@@ -135,14 +110,12 @@ func (q *Queue) work() {
110
111
select {
112
case toQueue := <-q.enqueue:
138
- nextKey := q.queueKey(q.tail)
113
+ nextKey := datastore.NewKey(fmt.Sprintf("%d", time.Now().UnixNano()))
114
115
if err := q.ds.Put(nextKey, toQueue.Bytes()); err != nil {
116
log.Errorf("Failed to enqueue cid: %s", err)
117
continue
118
}
144
-
145
- q.tail++
119
case dequeue <- c:
120
err := q.ds.Delete(k)
121
@@ -151,7 +124,6 @@ func (q *Queue) work() {
124
continue
125
}
126
c = cid.Undef
154
- q.head++
127
case <-q.ctx.Done():
128
return
129
}
@@ -159,53 +131,17 @@ func (q *Queue) work() {
131
}()
132
}
133
162
-func (q *Queue) queueKey(id uint64) datastore.Key {
163
- s := fmt.Sprintf("%016X", id)
164
- return datastore.NewKey(s)
165
-}
166
-
167
-func getQueueHeadTail(ctx context.Context, datastore datastore.Datastore) (uint64, uint64, error) {
168
- head, err := getQueueHead(datastore)
169
- if err != nil {
170
- return 0, 0, err
171
- }
172
- tail, err := getQueueTail(datastore)
173
- if err != nil {
174
- return 0, 0, err
175
- }
176
- return head, tail, nil
177
-}
178
-
179
-func getQueueHead(ds datastore.Datastore) (uint64, error) {
180
- return getFirstIDByOrder(ds, query.OrderByKey{})
181
-}
182
-
183
-func getQueueTail(ds datastore.Datastore) (uint64, error) {
184
- tail, err := getFirstIDByOrder(ds, query.OrderByKeyDescending{})
134
+func (q *Queue) getQueueHead() (*query.Result, error) {
135
+ qry := query.Query{Orders: []query.Order{query.OrderByKey{}}}
136
+ results, err := q.ds.Query(qry)
137
if err != nil {
186
- return 0, err
187
- }
188
- if tail > 0 {
189
- tail++
190
- }
191
- return tail, nil
192
-}
193
-
194
-func getFirstIDByOrder(ds datastore.Datastore, order query.Order) (uint64, error) {
195
- q := query.Query{Orders: []query.Order{order}}
196
- results, err := ds.Query(q)
197
- if err != nil {
198
- return 0, err
138
+ return nil, err
139
}
140
defer results.Close()
141
r, ok := results.NextSync()
142
if !ok {
203
- return 0, nil
204
- }
205
- trimmed := strings.TrimPrefix(r.Key, "/")
206
- id, err := strconv.ParseUint(trimmed, 16, 64)
207
- if err != nil {
208
- return 0, err
143
+ return nil, nil
144
}
210
- return id, nil
145
+
146
+ return &r, nil
147
}
provider/queue/queue_test.go
+8
-36
@@ -5,9 +5,9 @@ import (
5
"testing"
6
"time"
7
8
- cid "github.com/ipfs/go-cid"
9
- datastore "github.com/ipfs/go-datastore"
10
- sync "github.com/ipfs/go-datastore/sync"
8
+ "github.com/ipfs/go-cid"
9
+ "github.com/ipfs/go-datastore"
10
+ "github.com/ipfs/go-datastore/sync"
11
"github.com/ipfs/go-ipfs-blocksutil"
12
)
13
@@ -55,36 +55,6 @@ func TestBasicOperation(t *testing.T) {
55
assertOrdered(cids, queue, t)
56
}
57
58
-func TestSparseDatastore(t *testing.T) {
59
- ctx := context.Background()
60
- defer ctx.Done()
61
-
62
- ds := sync.MutexWrap(datastore.NewMapDatastore())
63
- queue, err := NewQueue(ctx, "test", ds)
64
- if err != nil {
65
- t.Fatal(err)
66
- }
67
-
68
- cids := makeCids(10)
69
- for _, c := range cids {
70
- queue.Enqueue(c)
71
- }
72
-
73
- // remove entries in the middle
74
- err = queue.ds.Delete(queue.queueKey(5))
75
- if err != nil {
76
- t.Fatal(err)
77
- }
78
-
79
- err = queue.ds.Delete(queue.queueKey(6))
80
- if err != nil {
81
- t.Fatal(err)
82
- }
83
-
84
- expected := append(cids[:5], cids[7:]...)
85
- assertOrdered(expected, queue, t)
86
-}
87
-
58
func TestMangledData(t *testing.T) {
59
ctx := context.Background()
60
defer ctx.Done()
@@ -100,13 +70,15 @@ func TestMangledData(t *testing.T) {
70
queue.Enqueue(c)
71
}
72
103
- // remove entries in the middle
104
- err = queue.ds.Put(queue.queueKey(5), []byte("borked"))
73
+ // put bad data in the queue
74
+ queueKey := datastore.NewKey("/test/0")
75
+ err = queue.ds.Put(queueKey, []byte("borked"))
76
if err != nil {
77
t.Fatal(err)
78
}
79
109
- expected := append(cids[:5], cids[6:]...)
80
+ // expect to only see the valid cids we entered
81
+ expected := cids
82
assertOrdered(expected, queue, t)
83
}
84