@cryptotaxi247 / kubo / commits / 1595253b7

Remove locking entirely

License: MIT Signed-off-by: Erik Ingenito <erik@carbonfive.com>

Erik Ingenito committed Mar 15, 2019 at 14:19 UTC 1595253b7b6f73b2b1f121677f3f32e7b2a6cb2d
4 files changed +71 -142
provider/provider.go
+9 -13
@@ -11,13 +11,9 @@ import (
11 routing "github.com/libp2p/go-libp2p-routing"
12 )
13
14 -var (
15 - log = logging.Logger("provider")
16 -)
14 +var log = logging.Logger("provider")
15
18 -const (
19 - provideOutgoingWorkerLimit = 8
20 -)
16 +const provideOutgoingWorkerLimit = 8
17
18 // Provider announces blocks to the network
19 type Provider interface {
@@ -44,13 +40,13 @@ func NewProvider(ctx context.Context, queue *Queue, contentRouting routing.Conte
40
41 // Start workers to handle provide requests.
42 func (p *provider) Run() {
47 - p.queue.Run()
43 p.handleAnnouncements()
44 }
45
46 // Provide the given cid using specified strategy.
47 func (p *provider) Provide(root cid.Cid) error {
53 - return p.queue.Enqueue(root)
48 + p.queue.Enqueue(root)
49 + return nil
50 }
51
52 // Handle all outgoing cids by providing (announcing) them
@@ -61,12 +57,12 @@ func (p *provider) handleAnnouncements() {
57 select {
58 case <-p.ctx.Done():
59 return
64 - case entry := <-p.queue.Dequeue():
65 - log.Info("announce - start - ", entry.cid)
66 - if err := p.contentRouting.Provide(p.ctx, entry.cid, true); err != nil {
67 - log.Warningf("Unable to provide entry: %s, %s", entry.cid, err)
60 + case c := <-p.queue.Dequeue():
61 + log.Info("announce - start - ", c)
62 + if err := p.contentRouting.Provide(p.ctx, c, true); err != nil {
63 + log.Warningf("Unable to provide entry: %s, %s", c, err)
64 }
69 - log.Info("announce - end - ", entry.cid)
65 + log.Info("announce - end - ", c)
66 }
67 }
68 }()
provider/provider_test.go
+2 -2
@@ -42,7 +42,7 @@ func TestAnnouncement(t *testing.T) {
42
43 cids := cid.NewSet()
44
45 - for i := 0; i < 100; i++ {
45 + for i := 0; i < 1000; i++ {
46 c := blockGenerator.Next().Cid()
47 cids.Add(c)
48 }
@@ -63,7 +63,7 @@ func TestAnnouncement(t *testing.T) {
63 t.Fatal("Wrong CID provided")
64 }
65 cids.Remove(cp)
66 - case <-time.After(time.Second * 1):
66 + case <-time.After(time.Second * 5):
67 t.Fatal("Timeout waiting for cids to be provided.")
68 }
69 }
provider/queue.go
+55 -113
@@ -2,27 +2,15 @@ package provider
2
3 import (
4 "context"
5 - "errors"
5 + "github.com/ipfs/go-cid"
6 + "github.com/ipfs/go-datastore"
7 + "github.com/ipfs/go-datastore/namespace"
8 + "github.com/ipfs/go-datastore/query"
9 "math"
10 "strconv"
11 "strings"
9 - "sync"
10 -
11 - cid "github.com/ipfs/go-cid"
12 - datastore "github.com/ipfs/go-datastore"
13 - namespace "github.com/ipfs/go-datastore/namespace"
14 - query "github.com/ipfs/go-datastore/query"
12 )
13
17 -// Entry allows for the durability in the queue. When a cid is dequeued it is
18 -// not removed from the datastore until you call Complete() on the entry you
19 -// receive.
20 -type Entry struct {
21 - cid cid.Cid
22 - key datastore.Key
23 - queue *Queue
24 -}
25 -
14 // Queue provides a durable, FIFO interface to the datastore for storing cids
15 //
16 // Durability just means that cids in the process of being provided when a
@@ -32,17 +20,15 @@ type Queue struct {
20 // used to differentiate queues in datastore
21 // e.g. provider vs reprovider
22 name string
35 -
23 ctx context.Context
24
25 tail uint64
26 head uint64
27
41 - enqueueLock sync.Mutex
28 ds datastore.Datastore // Must be threadsafe
29
44 - dequeue chan *Entry
45 - added chan struct{}
30 + dequeue chan cid.Cid
31 + enqueue chan cid.Cid
32 }
33
34 // NewQueue creates a queue for cids
@@ -57,124 +43,85 @@ func NewQueue(ctx context.Context, name string, ds datastore.Datastore) (*Queue,
43 ctx: ctx,
44 head: head,
45 tail: tail,
60 - enqueueLock: sync.Mutex{},
46 ds: namespaced,
62 - dequeue: make(chan *Entry),
63 - added: make(chan struct{}),
47 + dequeue: make(chan cid.Cid),
48 + enqueue: make(chan cid.Cid),
49 }
50 + q.work()
51 return q, nil
52 }
53
54 // Enqueue puts a cid in the queue
69 -func (q *Queue) Enqueue(cid cid.Cid) error {
70 - q.enqueueLock.Lock()
71 - defer q.enqueueLock.Unlock()
72 -
73 - nextKey := q.queueKey(q.tail)
74 -
75 - if err := q.ds.Put(nextKey, cid.Bytes()); err != nil {
76 - return err
77 - }
78 -
79 - q.tail++
80 -
55 +func (q *Queue) Enqueue(cid cid.Cid) {
56 select {
82 - case q.added <- struct{}{}:
57 + case q.enqueue <- cid:
58 case <-q.ctx.Done():
84 - default:
59 }
86 -
87 - return nil
60 }
61
62 // Dequeue returns a channel that if listened to will remove entries from the queue
91 -func (q *Queue) Dequeue() <-chan *Entry {
63 +func (q *Queue) Dequeue() <-chan cid.Cid {
64 return q.dequeue
65 }
66
95 -// IsEmpty returns whether or not the queue has any items
96 -func (q *Queue) IsEmpty() bool {
97 - return (q.tail - q.head) == 0
98 -}
99 -
100 -// Run dequeues items when the dequeue channel is available to
101 -// be written to.
102 -func (q *Queue) Run() {
67 +// Run dequeues and enqueues when available.
68 +func (q *Queue) work() {
69 go func() {
70 for {
105 - if q.IsEmpty() {
106 - select {
107 - case <-q.ctx.Done():
108 - return
109 - case <-q.added:
71 + var c cid.Cid = cid.Undef
72 + var key datastore.Key
73 + var dequeue chan cid.Cid
74 +
75 + // If we're not empty dequeue a cid and ship it
76 + if q.head < q.tail {
77 + key = q.queueKey(q.head)
78 + value, err := q.ds.Get(key)
79 +
80 + if err == datastore.ErrNotFound {
81 + log.Warningf("Missing entry in queue: %s", err)
82 + q.head++
83 + continue
84 + } else if err != nil {
85 + log.Warningf("Error fetching from queue: %s", err)
86 + continue
87 + }
88 +
89 + c, err = cid.Parse(value)
90 + if err != nil {
91 + log.Warningf("Error marshalling Cid from queue: ", err)
92 + q.head++
93 + err = q.ds.Delete(key)
94 + continue
95 }
96 }
97
113 - entry, err := q.next()
114 - if err != nil {
115 - log.Warningf("Error Dequeue()-ing: %s, %s", entry, err)
116 - continue
98 + if c != cid.Undef {
99 + dequeue = q.dequeue
100 }
101
102 select {
103 + case toQueue := <-q.enqueue:
104 + nextKey := q.queueKey(q.tail)
105 +
106 + if err := q.ds.Put(nextKey, toQueue.Bytes()); err != nil {
107 + log.Errorf("Failed to enqueue cid: %s", err)
108 + }
109 +
110 + q.tail++
111 + case dequeue <- c:
112 + q.head++
113 + err := q.ds.Delete(key)
114 +
115 + if err != nil {
116 + log.Errorf("Failed to delete queued cid: %s", err)
117 + }
118 case <-q.ctx.Done():
119 return
122 - case q.dequeue <- entry:
123 - q.head++
124 - err = q.ds.Delete(entry.key)
120 }
121 }
122 }()
123 }
124
130 -// Find the next item in the queue, crawl forward if an entry is not
131 -// found in the next spot.
132 -func (q *Queue) next() (*Entry, error) {
133 - var key datastore.Key
134 - var value []byte
135 - var err error
136 - for {
137 - if q.head >= q.tail {
138 - return nil, errors.New("next: no more entries in queue returning")
139 - }
140 - select {
141 - case <-q.ctx.Done():
142 - return nil, nil
143 - default:
144 - }
145 - key = q.queueKey(q.head)
146 -
147 - value, err = q.ds.Get(key)
148 -
149 - value, err = q.ds.Get(key)
150 - if err == datastore.ErrNotFound {
151 - q.head++
152 - continue
153 - } else if err != nil {
154 - return nil, err
155 - } else {
156 - break
157 - }
158 - }
159 -
160 - id, err := cid.Parse(value)
161 - if err != nil {
162 - return nil, err
163 - }
164 -
165 - entry := &Entry{
166 - cid: id,
167 - key: key,
168 - queue: q,
169 - }
170 -
171 - if err != nil {
172 - return nil, err
173 - }
174 -
175 - return entry, nil
176 -}
177 -
125 func (q *Queue) queueKey(id uint64) datastore.Key {
126 return datastore.NewKey(strconv.FormatUint(id, 10))
127 }
@@ -190,11 +137,6 @@ func getQueueHeadTail(ctx context.Context, name string, datastore datastore.Data
137 var tail uint64
138 var head uint64 = math.MaxUint64
139 for entry := range results.Next() {
193 - select {
194 - case <-ctx.Done():
195 - return 0, 0, nil
196 - default:
197 - }
140 trimmed := strings.TrimPrefix(entry.Key, "/")
141 id, err := strconv.ParseUint(trimmed, 10, 64)
142 if err != nil {
provider/queue_test.go
+5 -14
@@ -11,7 +11,7 @@ import (
11 )
12
13 func makeCids(n int) []cid.Cid {
14 - cids := make([]cid.Cid, 0, 10)
14 + cids := make([]cid.Cid, 0, n)
15 for i := 0; i < 10; i++ {
16 c := blockGenerator.Next().Cid()
17 cids = append(cids, c)
@@ -23,8 +23,8 @@ func assertOrdered(cids []cid.Cid, q *Queue, t *testing.T) {
23 for _, c := range cids {
24 select {
25 case dequeued := <- q.dequeue:
26 - if c != dequeued.cid {
27 - t.Fatalf("Error in ordering of CIDs retrieved from queue. Expected: %s, got: %s", c, dequeued.cid)
26 + if c != dequeued {
27 + t.Fatalf("Error in ordering of CIDs retrieved from queue. Expected: %s, got: %s", c, dequeued)
28 }
29
30 case <-time.After(time.Second * 1):
@@ -42,15 +42,11 @@ func TestBasicOperation(t *testing.T) {
42 if err != nil {
43 t.Fatal(err)
44 }
45 - queue.Run()
45
46 cids := makeCids(10)
47
48 for _, c := range cids {
50 - err = queue.Enqueue(c)
51 - if err != nil {
52 - t.Fatal("Failed to enqueue CID")
53 - }
49 + queue.Enqueue(c)
50 }
51
52 assertOrdered(cids, queue, t)
@@ -65,15 +61,11 @@ func TestInitialization(t *testing.T) {
61 if err != nil {
62 t.Fatal(err)
63 }
68 - queue.Run()
64
65 cids := makeCids(10)
66
67 for _, c := range cids {
73 - err = queue.Enqueue(c)
74 - if err != nil {
75 - t.Fatal("Failed to enqueue CID")
76 - }
68 + queue.Enqueue(c)
69 }
70
71 assertOrdered(cids[:5], queue, t)
@@ -83,7 +75,6 @@ func TestInitialization(t *testing.T) {
75 if err != nil {
76 t.Fatal(err)
77 }
86 - queue.Run()
78
79 assertOrdered(cids[5:], queue, t)
80 }