Allow for gc during adds
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Dec 5, 2015 at 20:31 UTC
c49dcffce2233689ff1f6dfd003a5545338ebc5e
4 files changed
+66
-13
blocks/blockstore/blockstore.go
+14
-1
@@ -5,6 +5,7 @@ package blockstore
5
import (
6
"errors"
7
"sync"
8
+ "sync/atomic"
9
10
ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
11
dsns "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/namespace"
@@ -49,6 +50,10 @@ type GCBlockstore interface {
50
// at the same time, but no GC should not happen simulatenously.
51
// Reading during Pinning is safe, and requires no lock.
52
PinLock() func()
53
+
54
+ // GcRequested returns true if GCLock has been called and is waiting to
55
+ // take the lock
56
+ GCRequested() bool
57
}
58
59
func NewBlockstore(d ds.Batching) *blockstore {
@@ -63,7 +68,9 @@ func NewBlockstore(d ds.Batching) *blockstore {
68
type blockstore struct {
69
datastore ds.Batching
70
66
- lk sync.RWMutex
71
+ lk sync.RWMutex
72
+ gcreq int32
73
+ gcreqlk sync.Mutex
74
}
75
76
func (bs *blockstore) Get(k key.Key) (*blocks.Block, error) {
@@ -192,7 +199,9 @@ func (bs *blockstore) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
199
}
200
201
func (bs *blockstore) GCLock() func() {
202
+ atomic.AddInt32(&bs.gcreq, 1)
203
bs.lk.Lock()
204
+ atomic.AddInt32(&bs.gcreq, -1)
205
return bs.lk.Unlock
206
}
207
@@ -200,3 +209,7 @@ func (bs *blockstore) PinLock() func() {
209
bs.lk.RLock()
210
return bs.lk.RUnlock
211
}
212
+
213
+func (bs *blockstore) GCRequested() bool {
214
+ return atomic.LoadInt32(&bs.gcreq) > 0
215
+}
blocks/blockstore/write_cache.go
+4
@@ -66,3 +66,7 @@ func (w *writecache) GCLock() func() {
66
func (w *writecache) PinLock() func() {
67
return w.blockstore.(GCBlockstore).PinLock()
68
}
69
+
70
+func (w *writecache) GCRequested() bool {
71
+ return w.blockstore.(GCBlockstore).GCRequested()
72
+}
core/coreunix/add.go
+47
-11
@@ -12,6 +12,7 @@ import (
12
syncds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
13
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
14
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
15
+ key "github.com/ipfs/go-ipfs/blocks/key"
16
bserv "github.com/ipfs/go-ipfs/blockservice"
17
"github.com/ipfs/go-ipfs/exchange/offline"
18
importer "github.com/ipfs/go-ipfs/importer"
@@ -99,6 +100,8 @@ type Adder struct {
100
Chunker string
101
root *dag.Node
102
mr *mfs.Root
103
+ unlock func()
104
+ tempRoot key.Key
105
}
106
107
// Perform the actual add & pin locally, outputting results to reader
@@ -157,6 +160,14 @@ func (params *Adder) PinRoot() error {
160
return err
161
}
162
163
+ if params.tempRoot != "" {
164
+ err := params.node.Pinning.Unpin(params.ctx, params.tempRoot, true)
165
+ if err != nil {
166
+ return err
167
+ }
168
+ params.tempRoot = rnk
169
+ }
170
+
171
params.node.Pinning.PinWithMode(rnk, pin.Recursive)
172
return params.node.Pinning.Flush()
173
}
@@ -256,7 +267,7 @@ func AddR(n *core.IpfsNode, root string) (key string, err error) {
267
return "", err
268
}
269
259
- err = fileAdder.AddFile(f)
270
+ err = fileAdder.addFile(f)
271
if err != nil {
272
return "", err
273
}
@@ -289,7 +300,7 @@ func AddWrapped(n *core.IpfsNode, r io.Reader, filename string) (string, *dag.No
300
unlock := n.Blockstore.PinLock()
301
defer unlock()
302
292
- err = fileAdder.AddFile(file)
303
+ err = fileAdder.addFile(file)
304
if err != nil {
305
return "", nil, err
306
}
@@ -330,12 +341,24 @@ func (params *Adder) addNode(node *dag.Node, path string) error {
341
342
// Add the given file while respecting the params.
343
func (params *Adder) AddFile(file files.File) error {
344
+ params.unlock = params.node.Blockstore.PinLock()
345
+ defer params.unlock()
346
+
347
+ return params.addFile(file)
348
+}
349
+
350
+func (adder *Adder) addFile(file files.File) error {
351
+ err := adder.maybePauseForGC()
352
+ if err != nil {
353
+ return err
354
+ }
355
+
356
switch {
334
- case files.IsHidden(file) && !params.Hidden:
357
+ case files.IsHidden(file) && !adder.Hidden:
358
log.Debugf("%s is hidden, skipping", file.FileName())
359
return &hiddenFileError{file.FileName()}
360
case file.IsDirectory():
338
- return params.addDir(file)
361
+ return adder.addDir(file)
362
}
363
364
// case for symlink
@@ -346,29 +369,29 @@ func (params *Adder) AddFile(file files.File) error {
369
}
370
371
dagnode := &dag.Node{Data: sdata}
349
- _, err = params.node.DAG.Add(dagnode)
372
+ _, err = adder.node.DAG.Add(dagnode)
373
if err != nil {
374
return err
375
}
376
354
- return params.addNode(dagnode, s.FileName())
377
+ return adder.addNode(dagnode, s.FileName())
378
}
379
380
// case for regular file
381
// if the progress flag was specified, wrap the file so that we can send
382
// progress updates to the client (over the output channel)
383
var reader io.Reader = file
361
- if params.Progress {
362
- reader = &progressReader{file: file, out: params.out}
384
+ if adder.Progress {
385
+ reader = &progressReader{file: file, out: adder.out}
386
}
387
365
- dagnode, err := params.add(reader)
388
+ dagnode, err := adder.add(reader)
389
if err != nil {
390
return err
391
}
392
393
// patch it into the root
371
- return params.addNode(dagnode, file.FileName())
394
+ return adder.addNode(dagnode, file.FileName())
395
}
396
397
func (params *Adder) addDir(dir files.File) error {
@@ -388,7 +411,7 @@ func (params *Adder) addDir(dir files.File) error {
411
break
412
}
413
391
- err = params.AddFile(file)
414
+ err = params.addFile(file)
415
if _, ok := err.(*hiddenFileError); ok {
416
// hidden file error, skip file
417
continue
@@ -400,6 +423,19 @@ func (params *Adder) addDir(dir files.File) error {
423
return nil
424
}
425
426
+func (adder *Adder) maybePauseForGC() error {
427
+ if adder.node.Blockstore.GCRequested() {
428
+ err := adder.PinRoot()
429
+ if err != nil {
430
+ return err
431
+ }
432
+
433
+ adder.unlock()
434
+ adder.unlock = adder.node.Blockstore.PinLock()
435
+ }
436
+ return nil
437
+}
438
+
439
// outputDagnode sends dagnode info over the output channel
440
func outputDagnode(out chan interface{}, name string, dn *dag.Node) error {
441
if out == nil {
pin/gc/gc.go
+1
-1
@@ -24,7 +24,6 @@ var log = logging.Logger("gc")
24
// deletes any block that is not found in the marked set.
25
func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner) (<-chan key.Key, error) {
26
unlock := bs.GCLock()
27
- defer unlock()
27
28
bsrv := bserv.New(bs, offline.Exchange(bs))
29
ds := dag.NewDAGService(bsrv)
@@ -42,6 +41,7 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner) (<-chan key.
41
output := make(chan key.Key)
42
go func() {
43
defer close(output)
44
+ defer unlock()
45
for {
46
select {
47
case k, ok := <-keychan: