Merge pull request #497 from jfontan/fix/close-siva-files

gitbase: close siva FS after use

Author: ajnavarro

blobs.go

@@ -396,6 +396,15 @@ func (i *blobsKeyValueIter) Close() error {
 	if i.blobs != nil {
 		i.blobs.Close()
 	}
+
+	if i.idx != nil {
+		i.idx.Close()
+	}
+
+	if i.repo != nil {
+		i.repo.Close()
+	}
+
 	return nil
 }
 

blobs_test.go

@@ -236,3 +236,11 @@ func TestBlobsIndex(t *testing.T) {
 		)},
 	)
 }
+
+func TestBlobsIndexIterClosed(t *testing.T) {
+	testTableIndexIterClosed(t, new(blobsTable))
+}
+
+func TestBlobsIterClosed(t *testing.T) {
+	testTableIterClosed(t, new(blobsTable))
+}

commit_blobs_test.go

@@ -1,12 +1,14 @@
 package gitbase
 
 import (
+	"io"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"gopkg.in/src-d/go-git.v4/plumbing"
 	"gopkg.in/src-d/go-mysql-server.v0/sql"
 	"gopkg.in/src-d/go-mysql-server.v0/sql/expression"
+	"gopkg.in/src-d/go-mysql-server.v0/sql/plan"
 )
 
 func TestCommitBlobsTableRowIter(t *testing.T) {
@@ -202,3 +204,29 @@ func TestCommitBlobsRowKeyMapper(t *testing.T) {
 
 	require.Equal(row, row2)
 }
+
+func TestCommitBlobsIndexIterClosed(t *testing.T) {
+	testTableIndexIterClosed(t, new(commitBlobsTable))
+}
+
+// This one is not using testTableIterClosed as it takes too much time
+// to go through all the rows. Here we limit it to the first 100.
+func TestCommitBlobsIterClosed(t *testing.T) {
+	require := require.New(t)
+	ctx, closed := setupSivaCloseRepos(t, "_testdata")
+
+	table := new(commitBlobsTable)
+	iter, err := plan.NewResolvedTable(table).RowIter(ctx)
+	require.NoError(err)
+
+	for i := 0; i < 100; i++ {
+		_, err = iter.Next()
+		if err != nil {
+			require.Equal(io.EOF, err)
+			break
+		}
+	}
+
+	iter.Close()
+	require.True(closed.Check())
+}

commit_files.go

@@ -301,6 +301,10 @@ func (i *commitFilesRowIter) Close() error {
 		i.files.Close()
 	}
 
+	if i.repo != nil {
+		i.repo.Close()
+	}
+
 	if i.index != nil {
 		return i.index.Close()
 	}
@@ -488,6 +492,14 @@ func (i *commitFilesKeyValueIter) Close() error {
 		i.files.Close()
 	}
 
+	if i.idx != nil {
+		i.idx.Close()
+	}
+
+	if i.repo != nil {
+		i.repo.Close()
+	}
+
 	return nil
 }
 

commit_files_test.go

@@ -98,3 +98,11 @@ func TestEncodeCommitFileIndexKey(t *testing.T) {
 
 	require.Equal(k, k2)
 }
+
+func TestCommitFilesIndexIterClosed(t *testing.T) {
+	testTableIndexIterClosed(t, new(commitFilesTable))
+}
+
+func TestCommitFilesIterClosed(t *testing.T) {
+	testTableIterClosed(t, new(commitFilesTable))
+}

commit_trees_test.go

@@ -168,3 +168,11 @@ func TestCommitTreesRowKeyMapper(t *testing.T) {
 
 	require.Equal(row, row2)
 }
+
+func TestCommitTreesIndexIterClosed(t *testing.T) {
+	testTableIndexIterClosed(t, new(commitTreesTable))
+}
+
+func TestCommitTreesIterClosed(t *testing.T) {
+	testTableIterClosed(t, new(commitTreesTable))
+}

commits.go

@@ -190,7 +190,9 @@ func (i *commitIter) Close() error {
 		i.iter.Close()
 	}
 
-	i.repo.Close()
+	if i.repo != nil {
+		i.repo.Close()
+	}
 
 	return nil
 }
@@ -275,6 +277,10 @@ func (i *commitsByHashIter) Close() {
 	if i.commitIter != nil {
 		i.commitIter.Close()
 	}
+
+	if i.repo != nil {
+		i.repo.Close()
+	}
 }
 
 func (i *commitsByHashIter) nextScan() (*object.Commit, error) {
@@ -374,6 +380,15 @@ func (i *commitsKeyValueIter) Close() error {
 	if i.commits != nil {
 		i.commits.Close()
 	}
+
+	if i.idx != nil {
+		i.idx.Close()
+	}
+
+	if i.repo != nil {
+		i.repo.Close()
+	}
+
 	return nil
 }
 

commits_test.go

@@ -298,3 +298,11 @@ func TestCommitsIndex(t *testing.T) {
 		)},
 	)
 }
+
+func TestCommitsIndexIterClosed(t *testing.T) {
+	testTableIndexIterClosed(t, new(commitsTable))
+}
+
+func TestCommitsIterClosed(t *testing.T) {
+	testTableIterClosed(t, new(commitsTable))
+}

common_test.go

@@ -2,11 +2,22 @@ package gitbase
 
 import (
 	"context"
+	"io"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"gopkg.in/src-d/go-billy-siva.v4"
+	billy "gopkg.in/src-d/go-billy.v4"
+	"gopkg.in/src-d/go-billy.v4/osfs"
 	fixtures "gopkg.in/src-d/go-git-fixtures.v3"
+	git "gopkg.in/src-d/go-git.v4"
 	"gopkg.in/src-d/go-git.v4/plumbing/cache"
+	"gopkg.in/src-d/go-git.v4/storage/filesystem"
 	"gopkg.in/src-d/go-mysql-server.v0/sql"
 	"gopkg.in/src-d/go-mysql-server.v0/sql/plan"
 )
@@ -65,3 +76,180 @@ func buildSession(t *testing.T, repos fixtures.Fixtures,
 func tableToRows(ctx *sql.Context, t sql.Table) ([]sql.Row, error) {
 	return sql.NodeToRows(ctx, plan.NewResolvedTable(t))
 }
+
+/*
+
+The following code adds utilities to test that siva files are properly closed.
+Instead of using normal setup you can use setupSivaCloseRepos, it returns
+a context with a pool with all the sivas in "_testdata" directory. It also
+tracks all siva filesystems opened. Its closed state can be checked with
+closedSiva.Check().
+
+*/
+
+type closedSiva struct {
+	closed []bool
+	m      sync.Mutex
+}
+
+func (c *closedSiva) NewFS(path string) (billy.Filesystem, error) {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	localfs := osfs.New(filepath.Dir(path))
+
+	tmpDir, err := ioutil.TempDir(os.TempDir(), "gitbase-siva")
+	if err != nil {
+		return nil, err
+	}
+
+	tmpfs := osfs.New(tmpDir)
+
+	fs, err := sivafs.NewFilesystem(localfs, filepath.Base(path), tmpfs)
+	if err != nil {
+		return nil, err
+	}
+
+	pos := len(c.closed)
+	c.closed = append(c.closed, false)
+
+	fun := func() {
+		c.m.Lock()
+		defer c.m.Unlock()
+		c.closed[pos] = true
+	}
+
+	return &closedSivaFilesystem{fs, fun}, nil
+}
+
+func (c *closedSiva) Check() bool {
+	for _, f := range c.closed {
+		if !f {
+			return false
+		}
+	}
+
+	return true
+}
+
+type closedSivaFilesystem struct {
+	sivafs.SivaFS
+	closeFunc func()
+}
+
+func (c *closedSivaFilesystem) Sync() error {
+	if c.closeFunc != nil {
+		c.closeFunc()
+	}
+
+	return c.SivaFS.Sync()
+}
+
+var _ repository = new(closedSivaRepository)
+
+type closedSivaRepository struct {
+	path  string
+	siva  *closedSiva
+	cache cache.Object
+}
+
+func (c *closedSivaRepository) ID() string {
+	return c.path
+}
+
+func (c *closedSivaRepository) Repo() (*Repository, error) {
+	fs, err := c.FS()
+	if err != nil {
+		return nil, err
+	}
+
+	s := fs.(*closedSivaFilesystem)
+	closeFunc := func() { s.Sync() }
+
+	sto := filesystem.NewStorageWithOptions(fs, c.Cache(), gitStorerOptions)
+	repo, err := git.Open(sto, nil)
+	if err != nil {
+		return nil, err
+
+	}
+
+	return NewRepository(c.path, repo, closeFunc), nil
+}
+
+func (c *closedSivaRepository) FS() (billy.Filesystem, error) {
+	return c.siva.NewFS(c.path)
+}
+
+func (c *closedSivaRepository) Path() string {
+	return c.path
+}
+
+func (c *closedSivaRepository) Cache() cache.Object {
+	if c.cache == nil {
+		c.cache = cache.NewObjectLRUDefault()
+	}
+
+	return c.cache
+}
+
+// setupSivaCloseRepos creates a pool with siva files that can be checked
+// if they've been closed.
+func setupSivaCloseRepos(t *testing.T, dir string) (*sql.Context, *closedSiva) {
+	require := require.New(t)
+
+	t.Helper()
+
+	cs := new(closedSiva)
+	pool := NewRepositoryPool(cache.DefaultMaxSize)
+
+	filepath.Walk(dir,
+		func(path string, info os.FileInfo, err error) error {
+			if strings.HasSuffix(path, ".siva") {
+				repo := &closedSivaRepository{path: path, siva: cs}
+				err := pool.Add(repo)
+				require.NoError(err)
+			}
+
+			return nil
+		},
+	)
+
+	session := NewSession(pool, WithSkipGitErrors(true))
+	ctx := sql.NewContext(context.TODO(), sql.WithSession(session))
+
+	return ctx, cs
+}
+
+func testTableIndexIterClosed(t *testing.T, table sql.IndexableTable) {
+	t.Helper()
+
+	require := require.New(t)
+	ctx, closed := setupSivaCloseRepos(t, "_testdata")
+
+	iter, err := table.IndexKeyValues(ctx, nil)
+	require.NoError(err)
+
+	for {
+		_, i, err := iter.Next()
+		if err != nil {
+			require.Equal(io.EOF, err)
+			break
+		}
+
+		i.Close()
+	}
+
+	iter.Close()
+	require.True(closed.Check())
+}
+
+func testTableIterClosed(t *testing.T, table sql.IndexableTable) {
+	t.Helper()
+
+	require := require.New(t)
+	ctx, closed := setupSivaCloseRepos(t, "_testdata")
+	_, err := tableToRows(ctx, table)
+	require.NoError(err)
+
+	require.True(closed.Check())
+}

files.go

@@ -492,6 +492,14 @@ func (i *filesKeyValueIter) Close() error {
 		i.files.Close()
 	}
 
+	if i.idx != nil {
+		i.idx.Close()
+	}
+
+	if i.repo != nil {
+		i.repo.Close()
+	}
+
 	return nil
 }