Add locking interface to blockstore
The addition of a locking interface to the blockstore allows us to perform atomic operations on the underlying datastore without having to worry about different operations happening in the background, such as garbage collection. License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Jul 7, 2015 at 08:56 UTC
da0d48e6c56e8b40686950f44f22b2e939c0d3b6
3 files changed
+50
-29
blocks/blockstore/blockstore.go
+21
-1
@@ -4,6 +4,7 @@ package blockstore
4
5
import (
6
"errors"
7
+ "sync"
8
9
ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
10
dsns "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/namespace"
@@ -35,7 +36,14 @@ type Blockstore interface {
36
AllKeysChan(ctx context.Context) (<-chan key.Key, error)
37
}
38
38
-func NewBlockstore(d ds.ThreadSafeDatastore) Blockstore {
39
+type GCBlockstore interface {
40
+ Blockstore
41
+
42
+ Lock() func()
43
+ RLock() func()
44
+}
45
+
46
+func NewBlockstore(d ds.ThreadSafeDatastore) *blockstore {
47
dd := dsns.Wrap(d, BlockPrefix)
48
return &blockstore{
49
datastore: dd,
@@ -46,6 +54,8 @@ type blockstore struct {
54
datastore ds.Batching
55
// cant be ThreadSafeDatastore cause namespace.Datastore doesnt support it.
56
// we do check it on `NewBlockstore` though.
57
+
58
+ lk sync.RWMutex
59
}
60
61
func (bs *blockstore) Get(k key.Key) (*blocks.Block, error) {
@@ -172,3 +182,13 @@ func (bs *blockstore) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
182
183
return output, nil
184
}
185
+
186
+func (bs *blockstore) Lock() func() {
187
+ bs.lk.Lock()
188
+ return bs.lk.Unlock
189
+}
190
+
191
+func (bs *blockstore) RLock() func() {
192
+ bs.lk.RLock()
193
+ return bs.lk.RUnlock
194
+}
blocks/blockstore/write_cache.go
+9
-1
@@ -8,7 +8,7 @@ import (
8
)
9
10
// WriteCached returns a blockstore that caches up to |size| unique writes (bs.Put).
11
-func WriteCached(bs Blockstore, size int) (Blockstore, error) {
11
+func WriteCached(bs Blockstore, size int) (*writecache, error) {
12
c, err := lru.New(size)
13
if err != nil {
14
return nil, err
@@ -58,3 +58,11 @@ func (w *writecache) PutMany(bs []*blocks.Block) error {
58
func (w *writecache) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
59
return w.blockstore.AllKeysChan(ctx)
60
}
61
+
62
+func (w *writecache) Lock() func() {
63
+ return w.blockstore.(GCBlockstore).Lock()
64
+}
65
+
66
+func (w *writecache) RLock() func() {
67
+ return w.blockstore.(GCBlockstore).RLock()
68
+}
blocks/key/key_set.go
+20
-27
@@ -1,46 +1,39 @@
1
package key
2
3
-import (
4
- "sync"
5
-)
6
-
3
type KeySet interface {
4
Add(Key)
5
+ Has(Key) bool
6
Remove(Key)
7
Keys() []Key
8
}
9
13
-type ks struct {
14
- lock sync.RWMutex
15
- data map[Key]struct{}
10
+type keySet struct {
11
+ keys map[Key]struct{}
12
}
13
14
func NewKeySet() KeySet {
19
- return &ks{
20
- data: make(map[Key]struct{}),
21
- }
15
+ return &keySet{make(map[Key]struct{})}
16
}
17
24
-func (wl *ks) Add(k Key) {
25
- wl.lock.Lock()
26
- defer wl.lock.Unlock()
27
-
28
- wl.data[k] = struct{}{}
18
+func (gcs *keySet) Add(k Key) {
19
+ gcs.keys[k] = struct{}{}
20
}
21
31
-func (wl *ks) Remove(k Key) {
32
- wl.lock.Lock()
33
- defer wl.lock.Unlock()
34
-
35
- delete(wl.data, k)
22
+func (gcs *keySet) Has(k Key) bool {
23
+ _, has := gcs.keys[k]
24
+ return has
25
}
26
38
-func (wl *ks) Keys() []Key {
39
- wl.lock.RLock()
40
- defer wl.lock.RUnlock()
41
- keys := make([]Key, 0)
42
- for k := range wl.data {
43
- keys = append(keys, k)
27
+func (ks *keySet) Keys() []Key {
28
+ var out []Key
29
+ for k, _ := range ks.keys {
30
+ out = append(out, k)
31
}
45
- return keys
32
+ return out
33
}
34
+
35
+func (ks *keySet) Remove(k Key) {
36
+ delete(ks.keys, k)
37
+}
38
+
39
+// TODO: implement disk-backed keyset for working with massive DAGs