GC now has concurrent mark phase

Introduce GC rework with concurrent mark phase, tricolor set and rework
that in future will allow to incremental and/or concurrent mark and
sweep phases.

Currently it is made a bit difficult by direct pins.

It will also allow for reuse of colorset for improved performance and no
set reallocation.

From observations pre-rewrok it should reduce GC lock time by half as
about 50% of the time is spent in mark phase.
This is consistent with observations made after the implementation.

License: MIT
Signed-off-by: Jakub Sztandera <[email protected]>

Author: Kubuxu

core/commands/repo.go

@@ -69,7 +69,11 @@ order to reclaim hard disk space.
 
 		streamErrors, _, _ := res.Request().Option("stream-errors").Bool()
 
-		gcOutChan := corerepo.GarbageCollectAsync(n, req.Context())
+		gcOutChan, err := corerepo.GarbageCollectAsync(req.Context(), n)
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
 
 		outChan := make(chan interface{}, cap(gcOutChan))
 		res.SetOutput((<-chan interface{})(outChan))

core/corerepo/gc.go

@@ -79,14 +79,13 @@ func BestEffortRoots(filesRoot *mfs.Root) ([]*cid.Cid, error) {
 	return []*cid.Cid{rootDag.Cid()}, nil
 }
 
-func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
+func GarbageCollect(ctx context.Context, n *core.IpfsNode) error {
 	ctx, cancel := context.WithCancel(ctx)
-	defer cancel() // in case error occurs during operation
-	roots, err := BestEffortRoots(n.FilesRoot)
+	defer cancel()
+	rmed, err := GarbageCollectAsync(ctx, n)
 	if err != nil {
 		return err
 	}
-	rmed := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
 
 	return CollectResult(ctx, rmed, nil)
 }
@@ -145,16 +144,22 @@ func (e *MultiError) Error() string {
 	return buf.String()
 }
 
-func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) <-chan gc.Result {
-	roots, err := BestEffortRoots(n.FilesRoot)
+func GarbageCollectAsync(ctx context.Context, n *core.IpfsNode) (<-chan gc.Result, error) {
+	g, err := gc.NewGC(n.Blockstore, n.DAG)
 	if err != nil {
-		out := make(chan gc.Result)
-		out <- gc.Result{Error: err}
-		close(out)
-		return out
+		return nil, err
+	}
+
+	err = g.AddPinSource(n.Pinning.PinSources()...)
+	if err != nil {
+		return nil, err
+	}
+	err = g.AddPinSource(*n.FilesRoot.PinSource())
+	if err != nil {
+		return nil, err
 	}
 
-	return gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
+	return g.Run(ctx), nil
 }
 
 func PeriodicGC(ctx context.Context, node *core.IpfsNode) error {
@@ -217,7 +222,7 @@ func (gc *GC) maybeGC(ctx context.Context, offset uint64) error {
 		log.Info("Watermark exceeded. Starting repo GC...")
 		defer log.EventBegin(ctx, "repoGC").Done()
 
-		if err := GarbageCollect(gc.Node, ctx); err != nil {
+		if err := GarbageCollect(ctx, gc.Node); err != nil {
 			return err
 		}
 		log.Infof("Repo GC done. See `ipfs repo stat` to see how much space got freed.\n")

core/coreunix/add_test.go

@@ -20,9 +20,9 @@ import (
 	"github.com/ipfs/go-ipfs/repo/config"
 	ds2 "github.com/ipfs/go-ipfs/thirdparty/datastore2"
 	pi "github.com/ipfs/go-ipfs/thirdparty/posinfo"
-	"gx/ipfs/QmSn9Td7xgxm9EV7iEjTckpUWmWApggzPxu7eFGWkkpwin/go-block-format"
 
 	cid "gx/ipfs/QmNp85zy9RLrQ5oQD4hPyS39ezrrXpcaa7R4Y9kxdWQLLQ/go-cid"
+	blocks "gx/ipfs/QmSn9Td7xgxm9EV7iEjTckpUWmWApggzPxu7eFGWkkpwin/go-block-format"
 )
 
 func TestAddRecursive(t *testing.T) {
@@ -46,6 +46,9 @@ func TestAddRecursive(t *testing.T) {
 }
 
 func TestAddGCLive(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	r := &repo.Mock{
 		C: config.Config{
 			Identity: config.Identity{
@@ -54,13 +57,13 @@ func TestAddGCLive(t *testing.T) {
 		},
 		D: ds2.ThreadSafeCloserMapDatastore(),
 	}
-	node, err := core.NewNode(context.Background(), &core.BuildCfg{Repo: r})
+	node, err := core.NewNode(ctx, &core.BuildCfg{Repo: r})
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	out := make(chan interface{})
-	adder, err := NewAdder(context.Background(), node.Pinning, node.Blockstore, node.DAG)
+	adder, err := NewAdder(ctx, node.Pinning, node.Blockstore, node.DAG)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -98,11 +101,18 @@ func TestAddGCLive(t *testing.T) {
 		t.Fatal("add shouldnt complete yet")
 	}
 
+	g, err := gc.NewGC(node.Blockstore, node.DAG)
+	if err != nil {
+		t.Fatal(err)
+	}
+	g.AddPinSource(node.Pinning.PinSources()...)
+
 	var gcout <-chan gc.Result
 	gcstarted := make(chan struct{})
+
 	go func() {
 		defer close(gcstarted)
-		gcout = gc.GC(context.Background(), node.Blockstore, node.DAG, node.Pinning, nil)
+		gcout = g.Run(ctx)
 	}()
 
 	// gc shouldnt start until we let the add finish its current file.
@@ -126,6 +136,7 @@ func TestAddGCLive(t *testing.T) {
 	<-gcstarted
 
 	for r := range gcout {
+		t.Logf("gc res: %v", r)
 		if r.Error != nil {
 			t.Fatal(err)
 		}
@@ -144,11 +155,11 @@ func TestAddGCLive(t *testing.T) {
 		last = c
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	ctx2, cancel := context.WithTimeout(ctx, time.Second*5)
 	defer cancel()
 
 	set := cid.NewSet()
-	err = dag.EnumerateChildren(ctx, node.DAG.GetLinks, last, set.Visit)
+	err = dag.EnumerateChildren(ctx2, node.DAG.GetLinks, last, set.Visit)
 	if err != nil {
 		t.Fatal(err)
 	}

mfs/system.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	dag "github.com/ipfs/go-ipfs/merkledag"
+	pin "github.com/ipfs/go-ipfs/pin"
 	ft "github.com/ipfs/go-ipfs/unixfs"
 
 	cid "gx/ipfs/QmNp85zy9RLrQ5oQD4hPyS39ezrrXpcaa7R4Y9kxdWQLLQ/go-cid"
@@ -123,6 +124,24 @@ func (kr *Root) Flush() error {
 	return nil
 }
 
+// PinSource returns information about pinning requirements.
+func (kr *Root) PinSource() *pin.Source {
+	return &pin.Source{
+		Get: func() ([]*cid.Cid, error) {
+			err := kr.Flush()
+			if err != nil {
+				return nil, err
+			}
+
+			nd, err := kr.GetValue().GetNode()
+			if err != nil {
+				return nil, err
+			}
+			return []*cid.Cid{nd.Cid()}, nil
+		},
+	}
+}
+
 // closeChild implements the childCloser interface, and signals to the publisher that
 // there are changes ready to be published
 func (kr *Root) closeChild(name string, nd node.Node, sync bool) error {