gitbase: close siva FS after use

When using siva filesystems the FS itself was closed by the garbage
collector. This caused to reach opened files limit when the tables did not
trigger garbage collection for some time. One of these cases is getting
remotes:

    select * from remotes;

Now when opening a siva FS a function is added to the repository to
close it.

Also added tests to check that the files are indeed closed. repository
and go-billy-siva are mocked to keep track of opened files.

Signed-off-by: Javi Fontan <[email protected]>

Author: jfontan

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
 }