Query for provider head/tail
License: MIT Signed-off-by: Michael Avila <davidmichaelavila@gmail.com>
Michael Avila committed
Mar 25, 2019 at 09:03 UTC
fca85a2bdd85dfadbf72aa2f56fd97e68117e6e9
4 files changed
+70
-36
go.mod
+2
-2
@@ -20,10 +20,10 @@ require (
20
github.com/ipfs/go-blockservice v0.0.3
21
github.com/ipfs/go-cid v0.0.1
22
github.com/ipfs/go-cidutil v0.0.1
23
- github.com/ipfs/go-datastore v0.0.1
23
+ github.com/ipfs/go-datastore v0.0.2
24
github.com/ipfs/go-detect-race v0.0.1
25
github.com/ipfs/go-ds-badger v0.0.2
26
- github.com/ipfs/go-ds-flatfs v0.0.1
26
+ github.com/ipfs/go-ds-flatfs v0.0.2
27
github.com/ipfs/go-ds-leveldb v0.0.1
28
github.com/ipfs/go-ds-measure v0.0.1
29
github.com/ipfs/go-fs-lock v0.0.1
go.sum
+4
-6
@@ -121,12 +121,14 @@ github.com/ipfs/go-cidutil v0.0.1 h1:UpDQI2LrihqOGY2mHaMhjrhh1DJ14N/58BQb7lKXvlQ
121
github.com/ipfs/go-cidutil v0.0.1/go.mod h1:/0H649ymJksNEZvBAkM18HIctk7tkONH9tspTeLok48=
122
github.com/ipfs/go-datastore v0.0.1 h1:AW/KZCScnBWlSb5JbnEnLKFWXL224LBEh/9KXXOrUms=
123
github.com/ipfs/go-datastore v0.0.1/go.mod h1:d4KVXhMt913cLBEI/PXAy6ko+W7e9AhyAKBGh803qeE=
124
+github.com/ipfs/go-datastore v0.0.2 h1:Blyjq95atbxmCHSaDt42phIhf9NGLflzwq/95FqsNs0=
125
+github.com/ipfs/go-datastore v0.0.2/go.mod h1:d4KVXhMt913cLBEI/PXAy6ko+W7e9AhyAKBGh803qeE=
126
github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
127
github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
128
github.com/ipfs/go-ds-badger v0.0.2 h1:7ToQt7QByBhOTuZF2USMv+PGlMcBC7FW7FdgQ4FCsoo=
129
github.com/ipfs/go-ds-badger v0.0.2/go.mod h1:Y3QpeSFWQf6MopLTiZD+VT6IC1yZqaGmjvRcKeSGij8=
128
-github.com/ipfs/go-ds-flatfs v0.0.1 h1:yqWwRYFOGNClUL7V2jvcx4KMMso1Jv+pgQzsv9/gWBs=
129
-github.com/ipfs/go-ds-flatfs v0.0.1/go.mod h1:YsMGWjUieue+smePAWeH/YhHtlmEMnEGhiwIn6K6rEM=
130
+github.com/ipfs/go-ds-flatfs v0.0.2 h1:1zujtU5bPBH6B8roE+TknKIbBCrpau865xUk0dH3x2A=
131
+github.com/ipfs/go-ds-flatfs v0.0.2/go.mod h1:YsMGWjUieue+smePAWeH/YhHtlmEMnEGhiwIn6K6rEM=
132
github.com/ipfs/go-ds-leveldb v0.0.1 h1:Z0lsTFciec9qYsyngAw1f/czhRU35qBLR2vhavPFgqA=
133
github.com/ipfs/go-ds-leveldb v0.0.1/go.mod h1:feO8V3kubwsEF22n0YRQCffeb79OOYIykR4L04tMOYc=
134
github.com/ipfs/go-ds-measure v0.0.1 h1:PrCueug+yZLkDCOthZTXKinuoCal/GvlAT7cNxzr03g=
@@ -143,8 +145,6 @@ github.com/ipfs/go-ipfs-chunker v0.0.1 h1:cHUUxKFQ99pozdahi+uSC/3Y6HeRpi9oTeUHbE
145
github.com/ipfs/go-ipfs-chunker v0.0.1/go.mod h1:tWewYK0we3+rMbOh7pPFGDyypCtvGcBFymgY4rSDLAw=
146
github.com/ipfs/go-ipfs-cmdkit v0.0.1 h1:X6YXEAjUljTzevE6DPUKXSqcgf+4FXzcn5B957F5MXo=
147
github.com/ipfs/go-ipfs-cmdkit v0.0.1/go.mod h1:9FtbMdUabcSqv/G4/8WCxSLxkZxn/aZEFrxxqnVcRbg=
146
-github.com/ipfs/go-ipfs-cmds v0.0.3 h1:QvNUE8lslNQghxXf6vzV1ZoMQCDDAtKG8f2oINiRew4=
147
-github.com/ipfs/go-ipfs-cmds v0.0.3/go.mod h1:1QVgxSgenZvOMGVC/XUTC7tJxRBGPLxYvpgPpCi3DUk=
148
github.com/ipfs/go-ipfs-cmds v0.0.4 h1:Iq4I8irWw5TmHe/4pjSyYJLbYkkdMOgHVe8ofJmPa4k=
149
github.com/ipfs/go-ipfs-cmds v0.0.4/go.mod h1:1QVgxSgenZvOMGVC/XUTC7tJxRBGPLxYvpgPpCi3DUk=
150
github.com/ipfs/go-ipfs-config v0.0.1 h1:6ED08emzI1imdsAjixFi2pEyZxTVD5ECKtCOxLBx+Uc=
@@ -220,8 +220,6 @@ github.com/jbenet/go-cienv v0.0.0-20150120210510-1bb1476777ec h1:DQqZhhDvrTrEQ3Q
220
github.com/jbenet/go-cienv v0.0.0-20150120210510-1bb1476777ec/go.mod h1:rGaEvXB4uRSZMmzKNLoXvTu1sfx+1kv/DojUlPrSZGs=
221
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
222
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
223
-github.com/jbenet/go-is-domain v0.0.0-20160119110217-ba9815c809e0 h1:qxMncUW0TzViA3REiN3/YgVOoekVtUtEY0O/j/Qlctg=
224
-github.com/jbenet/go-is-domain v0.0.0-20160119110217-ba9815c809e0/go.mod h1:I9DYFcJAixF5f9iOu/9oC451/bq+QDTaLGznkcJPWgg=
223
github.com/jbenet/go-is-domain v1.0.2 h1:11r5MSptcNFZyBoqubBQnVMUKRWLuRjL1banaIk+iYo=
224
github.com/jbenet/go-is-domain v1.0.2/go.mod h1:xbRLRb0S7FgzDBTJlguhDVwLYM/5yNtvktxj2Ttfy7Q=
225
github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c h1:uUx61FiAa1GI6ZmVd2wf2vULeQZIKG66eybjNXKYCz4=
provider/queue.go
+39
-27
@@ -2,7 +2,7 @@ package provider
2
3
import (
4
"context"
5
- "math"
5
+ "fmt"
6
"strconv"
7
"strings"
8
@@ -32,7 +32,7 @@ type Queue struct {
32
// NewQueue creates a queue for cids
33
func NewQueue(ctx context.Context, name string, ds datastore.Datastore) (*Queue, error) {
34
namespaced := namespace.Wrap(ds, datastore.NewKey("/"+name+"/queue/"))
35
- head, tail, err := getQueueHeadTail(ctx, name, namespaced)
35
+ head, tail, err := getQueueHeadTail(ctx, namespaced)
36
if err != nil {
37
return nil, err
38
}
@@ -142,40 +142,52 @@ func (q *Queue) work() {
142
}
143
144
func (q *Queue) queueKey(id uint64) datastore.Key {
145
- return datastore.NewKey(strconv.FormatUint(id, 10))
145
+ s := fmt.Sprintf("%016X", id)
146
+ return datastore.NewKey(s)
147
}
148
148
-// crawl over the queue entries to find the head and tail
149
-func getQueueHeadTail(ctx context.Context, name string, datastore datastore.Datastore) (uint64, uint64, error) {
150
- q := query.Query{}
151
- results, err := datastore.Query(q)
149
+func getQueueHeadTail(ctx context.Context, datastore datastore.Datastore) (uint64, uint64, error) {
150
+ head, err := getQueueHead(datastore)
151
if err != nil {
152
return 0, 0, err
153
}
154
+ tail, err := getQueueTail(datastore)
155
+ if err != nil {
156
+ return 0, 0, err
157
+ }
158
+ return head, tail, nil
159
+}
160
156
- var tail uint64
157
- var head uint64 = math.MaxUint64
158
- for entry := range results.Next() {
159
- trimmed := strings.TrimPrefix(entry.Key, "/")
160
- id, err := strconv.ParseUint(trimmed, 10, 64)
161
- if err != nil {
162
- return 0, 0, err
163
- }
161
+func getQueueHead(ds datastore.Datastore) (uint64, error) {
162
+ return getFirstIDByOrder(ds, query.OrderByKey{})
163
+}
164
165
- if id < head {
166
- head = id
167
- }
165
+func getQueueTail(ds datastore.Datastore) (uint64, error) {
166
+ tail, err := getFirstIDByOrder(ds, query.OrderByKeyDescending{})
167
+ if err != nil {
168
+ return 0, err
169
+ }
170
+ if tail > 0 {
171
+ tail++
172
+ }
173
+ return tail, nil
174
+}
175
169
- if (id + 1) > tail {
170
- tail = (id + 1)
171
- }
176
+func getFirstIDByOrder(ds datastore.Datastore, order query.Order) (uint64, error) {
177
+ q := query.Query{Orders: []query.Order{order}}
178
+ results, err := ds.Query(q)
179
+ if err != nil {
180
+ return 0, err
181
}
173
- if err := results.Close(); err != nil {
174
- return 0, 0, err
182
+ defer results.Close()
183
+ r, ok := results.NextSync()
184
+ if !ok {
185
+ return 0, nil
186
}
176
- if head == math.MaxUint64 {
177
- head = 0
187
+ trimmed := strings.TrimPrefix(r.Key, "/")
188
+ id, err := strconv.ParseUint(trimmed, 16, 64)
189
+ if err != nil {
190
+ return 0, err
191
}
179
-
180
- return head, tail, nil
192
+ return id, nil
193
}
provider/queue_test.go
+25
-1
@@ -12,7 +12,7 @@ import (
12
13
func makeCids(n int) []cid.Cid {
14
cids := make([]cid.Cid, 0, n)
15
- for i := 0; i < 10; i++ {
15
+ for i := 0; i < n; i++ {
16
c := blockGenerator.Next().Cid()
17
cids = append(cids, c)
18
}
@@ -129,3 +129,27 @@ func TestInitialization(t *testing.T) {
129
130
assertOrdered(cids[5:], queue, t)
131
}
132
+
133
+func TestInitializationWithManyCids(t *testing.T) {
134
+ ctx := context.Background()
135
+ defer ctx.Done()
136
+
137
+ ds := sync.MutexWrap(datastore.NewMapDatastore())
138
+ queue, err := NewQueue(ctx, "test", ds)
139
+ if err != nil {
140
+ t.Fatal(err)
141
+ }
142
+
143
+ cids := makeCids(25)
144
+ for _, c := range cids {
145
+ queue.Enqueue(c)
146
+ }
147
+
148
+ // make a new queue, same data
149
+ queue, err = NewQueue(ctx, "test", ds)
150
+ if err != nil {
151
+ t.Fatal(err)
152
+ }
153
+
154
+ assertOrdered(cids, queue, t)
155
+}