Add fixed period repo GC + test
License: MIT Signed-off-by: rht <rhtbot@gmail.com>
rht committed
Oct 20, 2015 at 10:56 UTC
48a33ffb673fb3eaf6d20cd7ca6bed04426a2abb
12 files changed
+214
-15
cmd/ipfs/daemon.go
+30
-2
@@ -19,6 +19,7 @@ import (
19
"github.com/ipfs/go-ipfs/core"
20
commands "github.com/ipfs/go-ipfs/core/commands"
21
corehttp "github.com/ipfs/go-ipfs/core/corehttp"
22
+ corerepo "github.com/ipfs/go-ipfs/core/corerepo"
23
"github.com/ipfs/go-ipfs/core/corerouting"
24
conn "github.com/ipfs/go-ipfs/p2p/net/conn"
25
peer "github.com/ipfs/go-ipfs/p2p/peer"
@@ -36,6 +37,7 @@ const (
37
ipnsMountKwd = "mount-ipns"
38
unrestrictedApiAccessKwd = "unrestricted-api"
39
unencryptTransportKwd = "disable-transport-encryption"
40
+ enableGCKwd = "enable-gc"
41
// apiAddrKwd = "address-api"
42
// swarmAddrKwd = "address-swarm"
43
)
@@ -114,6 +116,7 @@ future version, along with this notice. Please move to setting the HTTP Headers.
116
cmds.StringOption(ipnsMountKwd, "Path to the mountpoint for IPNS (if using --mount)"),
117
cmds.BoolOption(unrestrictedApiAccessKwd, "Allow API access to unlisted hashes"),
118
cmds.BoolOption(unencryptTransportKwd, "Disable transport encryption (for debugging protocols)"),
119
+ cmds.BoolOption(enableGCKwd, "Enable automatic periodic repo garbage collection"),
120
121
// TODO: add way to override addresses. tricky part: updating the config if also --init.
122
// cmds.StringOption(apiAddrKwd, "Address for the daemon rpc API (overrides config)"),
@@ -277,15 +280,23 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
280
}
281
}
282
283
+ // repo blockstore GC - if --enable-gc flag is present
284
+ err, gcErrc := maybeRunGC(req, node)
285
+ if err != nil {
286
+ res.SetError(err, cmds.ErrNormal)
287
+ return
288
+ }
289
+
290
fmt.Printf("Daemon is ready\n")
291
// collect long-running errors and block for shutdown
292
// TODO(cryptix): our fuse currently doesnt follow this pattern for graceful shutdown
283
- for err := range merge(apiErrc, gwErrc) {
293
+ for err := range merge(apiErrc, gwErrc, gcErrc) {
294
if err != nil {
295
+ log.Error(err)
296
res.SetError(err, cmds.ErrNormal)
286
- return
297
}
298
}
299
+ return
300
}
301
302
// serveHTTPApi collects options, creates listener, prints status message and starts serving requests
@@ -478,6 +489,23 @@ func mountFuse(req cmds.Request) error {
489
return nil
490
}
491
492
+func maybeRunGC(req cmds.Request, node *core.IpfsNode) (error, <-chan error) {
493
+ enableGC, _, err := req.Option(enableGCKwd).Bool()
494
+ if err != nil {
495
+ return err, nil
496
+ }
497
+ if !enableGC {
498
+ return nil, nil
499
+ }
500
+
501
+ errc := make(chan error)
502
+ go func() {
503
+ errc <- corerepo.PeriodicGC(req.Context(), node)
504
+ close(errc)
505
+ }()
506
+ return nil, errc
507
+}
508
+
509
// merge does fan-in of multiple read-only error channels
510
// taken from http://blog.golang.org/pipelines
511
func merge(cs ...<-chan error) <-chan error {
commands/request.go
+1
-1
@@ -43,7 +43,7 @@ func (c *Context) GetConfig() (*config.Config, error) {
43
}
44
45
// GetNode returns the node of the current Command exection
46
-// context. It may construct it with the providied function.
46
+// context. It may construct it with the provided function.
47
func (c *Context) GetNode() (*core.IpfsNode, error) {
48
var err error
49
if c.node == nil {
core/commands/add.go
+8
@@ -89,6 +89,7 @@ remains to be implemented.
89
// see comment above
90
return nil
91
}
92
+
93
log.Debugf("Total size of file being added: %v\n", size)
94
req.Values()["size"] = size
95
@@ -100,6 +101,13 @@ remains to be implemented.
101
res.SetError(err, cmds.ErrNormal)
102
return
103
}
104
+ // check if repo will exceed storage limit if added
105
+ // TODO: this doesn't handle the case if the hashed file is already in blocks (deduplicated)
106
+ // TODO: conditional GC is disabled due to it is somehow not possible to pass the size to the daemon
107
+ //if err := corerepo.ConditionalGC(req.Context(), n, uint64(size)); err != nil {
108
+ // res.SetError(err, cmds.ErrNormal)
109
+ // return
110
+ //}
111
112
progress, _, _ := req.Option(progressOptionName).Bool()
113
trickle, _, _ := req.Option(trickleOptionName).Bool()
core/commands/cat.go
+5
@@ -5,6 +5,7 @@ import (
5
6
cmds "github.com/ipfs/go-ipfs/commands"
7
core "github.com/ipfs/go-ipfs/core"
8
+ "github.com/ipfs/go-ipfs/core/corerepo"
9
coreunix "github.com/ipfs/go-ipfs/core/coreunix"
10
11
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
@@ -44,6 +45,10 @@ it contains.
45
return
46
}
47
48
+ if err := corerepo.ConditionalGC(req.Context(), node, length); err != nil {
49
+ res.SetError(err, cmds.ErrNormal)
50
+ return
51
+ }
52
res.SetLength(length)
53
54
reader := io.MultiReader(readers...)
core/corerepo/gc.go
+138
-4
@@ -1,21 +1,77 @@
1
package corerepo
2
3
import (
4
+ "errors"
5
+ "time"
6
+
7
+ humanize "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/dustin/go-humanize"
8
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
9
key "github.com/ipfs/go-ipfs/blocks/key"
10
"github.com/ipfs/go-ipfs/core"
7
-
11
+ repo "github.com/ipfs/go-ipfs/repo"
12
logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
13
)
14
15
var log = logging.Logger("corerepo")
16
17
+var ErrMaxStorageExceeded = errors.New("Maximum storage limit exceeded. Maybe unpin some files?")
18
+
19
type KeyRemoved struct {
20
Key key.Key
21
}
22
23
+type GC struct {
24
+ Node *core.IpfsNode
25
+ Repo repo.Repo
26
+ StorageMax uint64
27
+ StorageGC uint64
28
+ SlackGB uint64
29
+ Storage uint64
30
+}
31
+
32
+func NewGC(n *core.IpfsNode) (*GC, error) {
33
+ r := n.Repo
34
+ cfg, err := r.Config()
35
+ if err != nil {
36
+ return nil, err
37
+ }
38
+
39
+ // check if cfg has these fields initialized
40
+ // TODO: there should be a general check for all of the cfg fields
41
+ // maybe distinguish between user config file and default struct?
42
+ if cfg.Datastore.StorageMax == "" {
43
+ r.SetConfigKey("Datastore.StorageMax", "10GB")
44
+ cfg.Datastore.StorageMax = "10GB"
45
+ }
46
+ if cfg.Datastore.StorageGCWatermark == 0 {
47
+ r.SetConfigKey("Datastore.StorageGCWatermark", 90)
48
+ cfg.Datastore.StorageGCWatermark = 90
49
+ }
50
+
51
+ storageMax, err := humanize.ParseBytes(cfg.Datastore.StorageMax)
52
+ if err != nil {
53
+ return nil, err
54
+ }
55
+ storageGC := storageMax * uint64(cfg.Datastore.StorageGCWatermark) / 100
56
+
57
+ // calculate the slack space between StorageMax and StorageGCWatermark
58
+ // used to limit GC duration
59
+ slackGB := (storageMax - storageGC) / 10e9
60
+ if slackGB < 1 {
61
+ slackGB = 1
62
+ }
63
+
64
+ return &GC{
65
+ Node: n,
66
+ Repo: r,
67
+ StorageMax: storageMax,
68
+ StorageGC: storageGC,
69
+ SlackGB: slackGB,
70
+ }, nil
71
+}
72
+
73
func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
18
- ctx, cancel := context.WithCancel(context.Background())
74
+ ctx, cancel := context.WithCancel(ctx)
75
defer cancel() // in case error occurs during operation
76
keychan, err := n.Blockstore.AllKeysChan(ctx)
77
if err != nil {
@@ -23,8 +79,7 @@ func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
79
}
80
for k := range keychan { // rely on AllKeysChan to close chan
81
if !n.Pinning.IsPinned(k) {
26
- err := n.Blockstore.DeleteBlock(k)
27
- if err != nil {
82
+ if err := n.Blockstore.DeleteBlock(k); err != nil {
83
return err
84
}
85
}
@@ -66,3 +121,82 @@ func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemo
121
}()
122
return output, nil
123
}
124
+
125
+func PeriodicGC(ctx context.Context, node *core.IpfsNode) error {
126
+ cfg, err := node.Repo.Config()
127
+ if err != nil {
128
+ return err
129
+ }
130
+
131
+ if cfg.Datastore.GCPeriod == "" {
132
+ node.Repo.SetConfigKey("Datastore.GCPeriod", "1h")
133
+ cfg.Datastore.GCPeriod = "1h"
134
+ }
135
+
136
+ period, err := time.ParseDuration(cfg.Datastore.GCPeriod)
137
+ if err != nil {
138
+ return err
139
+ }
140
+ if int64(period) == 0 {
141
+ // if duration is 0, it means GC is disabled.
142
+ return nil
143
+ }
144
+
145
+ gc, err := NewGC(node)
146
+ if err != nil {
147
+ return err
148
+ }
149
+
150
+ for {
151
+ select {
152
+ case <-ctx.Done():
153
+ return nil
154
+ case <-time.After(period):
155
+ // the private func maybeGC doesn't compute storageMax, storageGC, slackGC so that they are not re-computed for every cycle
156
+ if err := gc.maybeGC(ctx, 0); err != nil {
157
+ return err
158
+ }
159
+ }
160
+ }
161
+}
162
+
163
+func ConditionalGC(ctx context.Context, node *core.IpfsNode, offset uint64) error {
164
+ gc, err := NewGC(node)
165
+ if err != nil {
166
+ return err
167
+ }
168
+ return gc.maybeGC(ctx, offset)
169
+}
170
+
171
+func (gc *GC) maybeGC(ctx context.Context, offset uint64) error {
172
+ storage, err := gc.Repo.GetStorageUsage()
173
+ if err != nil {
174
+ return err
175
+ }
176
+
177
+ if storage+offset > gc.StorageMax {
178
+ err := ErrMaxStorageExceeded
179
+ log.Error(err)
180
+ return err
181
+ }
182
+
183
+ if storage+offset > gc.StorageGC {
184
+ // Do GC here
185
+ log.Info("Starting repo GC...")
186
+ defer log.EventBegin(ctx, "repoGC").Done()
187
+ // 1 minute is sufficient for ~1GB unlink() blocks each of 100kb in SSD
188
+ _ctx, cancel := context.WithTimeout(ctx, time.Duration(gc.SlackGB)*time.Minute)
189
+ defer cancel()
190
+
191
+ if err := GarbageCollect(gc.Node, _ctx); err != nil {
192
+ return err
193
+ }
194
+ newStorage, err := gc.Repo.GetStorageUsage()
195
+ if err != nil {
196
+ return err
197
+ }
198
+ log.Infof("Repo GC done. Released %s\n", humanize.Bytes(uint64(storage-newStorage)))
199
+ return nil
200
+ }
201
+ return nil
202
+}
repo/config/datastore.go
+5
-2
@@ -5,8 +5,11 @@ const DefaultDataStoreDirectory = "datastore"
5
6
// Datastore tracks the configuration of the datastore.
7
type Datastore struct {
8
- Type string
9
- Path string
8
+ Type string
9
+ Path string
10
+ StorageMax string // in B, kB, kiB, MB, ...
11
+ StorageGCWatermark int64 // in percentage to multiply on StorageMax
12
+ GCPeriod string // in ns, us, ms, s, m, h
13
}
14
15
// DataStorePath returns the default data store path given a configuration root
repo/config/init.go
+5
-2
@@ -87,8 +87,11 @@ func datastoreConfig() (*Datastore, error) {
87
return nil, err
88
}
89
return &Datastore{
90
- Path: dspath,
91
- Type: "leveldb",
90
+ Path: dspath,
91
+ Type: "leveldb",
92
+ StorageMax: "10GB",
93
+ StorageGCWatermark: 90, // 90%
94
+ GCPeriod: "1h",
95
}, nil
96
}
97
repo/fsrepo/fsrepo.go
+16
-2
@@ -23,7 +23,6 @@ import (
23
mfsr "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
24
serialize "github.com/ipfs/go-ipfs/repo/fsrepo/serialize"
25
dir "github.com/ipfs/go-ipfs/thirdparty/dir"
26
- u "github.com/ipfs/go-ipfs/util"
26
util "github.com/ipfs/go-ipfs/util"
27
ds2 "github.com/ipfs/go-ipfs/util/datastore2"
28
logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
@@ -166,7 +165,7 @@ func open(repoPath string) (repo.Repo, error) {
165
}
166
167
func newFSRepo(rpath string) (*FSRepo, error) {
169
- expPath, err := u.TildeExpansion(filepath.Clean(rpath))
168
+ expPath, err := util.TildeExpansion(filepath.Clean(rpath))
169
if err != nil {
170
return nil, err
171
}
@@ -587,6 +586,21 @@ func (r *FSRepo) Datastore() ds.ThreadSafeDatastore {
586
return d
587
}
588
589
+// GetStorageUsage computes the storage space taken by the repo in bytes
590
+func (r *FSRepo) GetStorageUsage() (uint64, error) {
591
+ pth, err := config.PathRoot()
592
+ if err != nil {
593
+ return 0, err
594
+ }
595
+
596
+ var du uint64
597
+ err = filepath.Walk(pth, func(p string, f os.FileInfo, err error) error {
598
+ du += uint64(f.Size())
599
+ return nil
600
+ })
601
+ return du, err
602
+}
603
+
604
var _ io.Closer = &FSRepo{}
605
var _ repo.Repo = &FSRepo{}
606
repo/mock.go
+2
@@ -34,6 +34,8 @@ func (m *Mock) GetConfigKey(key string) (interface{}, error) {
34
35
func (m *Mock) Datastore() ds.ThreadSafeDatastore { return m.D }
36
37
+func (m *Mock) GetStorageUsage() (uint64, error) { return 0, nil }
38
+
39
func (m *Mock) Close() error { return errTODO }
40
41
func (m *Mock) SetAPIAddr(addr string) error { return errTODO }
repo/repo.go
+1
@@ -21,6 +21,7 @@ type Repo interface {
21
GetConfigKey(key string) (interface{}, error)
22
23
Datastore() datastore.ThreadSafeDatastore
24
+ GetStorageUsage() (uint64, error)
25
26
// SetAPIAddr sets the API address in the repo.
27
SetAPIAddr(addr string) error
test/sharness/lib/test-lib.sh
+2
-2
@@ -324,8 +324,8 @@ disk_usage() {
324
FreeBSD)
325
DU="du -s -A -B 1"
326
;;
327
- Darwin | DragonFly)
328
- DU="du"
327
+ Darwin | DragonFly | *)
328
+ DU="du -s"
329
;;
330
esac
331
$DU "$1" | awk "{print \$1}"
test/sharness/t0080-repo.sh
+1
@@ -55,6 +55,7 @@ test_expect_success "'ipfs pin rm' output looks good" '
55
'
56
57
test_expect_failure "ipfs repo gc fully reverse ipfs add" '
58
+ ipfs repo gc &&
59
random 100000 41 >gcfile &&
60
disk_usage "$IPFS_PATH/blocks" >expected &&
61
hash=`ipfs add -q gcfile` &&