Go: [BREAKING] explicit destruction for futures and transactions

Stop relying on Go's GC finalizers and instead use explicit Close() calls
to release resources, which is more Go-idiomatic.

NOTE: this would be a breaking change for users because without any code breakage
they would now have huge memory leaks on their FoundationDB clients; a release management
solution must be devised for this problem.

Author: gm42

bindings/go/src/fdb/database.go

@@ -28,7 +28,6 @@ import "C"
 
 import (
 	"errors"
-	"runtime"
 )
 
 // ErrMultiVersionClientUnavailable is returned when the multi-version client API is unavailable.
@@ -92,6 +91,7 @@ func (opt DatabaseOptions) setOpt(code int, param []byte) error {
 // preferable to use the (Database).Transact method, which handles
 // automatically creating and committing a transaction with appropriate retry
 // behavior.
+// Close() must be called on the returned transaction to avoid a memory leak.
 func (d Database) CreateTransaction() (Transaction, error) {
 	var outt *C.FDBTransaction
 
@@ -100,10 +100,6 @@ func (d Database) CreateTransaction() (Transaction, error) {
 	}
 
 	t := &transaction{outt, d}
-	// transactions cannot be destroyed explicitly if any future is still potentially used
-	// thus the GC is used to figure out when all Go wrapper objects for futures have gone out of scope,
-	// making the transaction ready to be garbage-collected.
-	runtime.SetFinalizer(t, (*transaction).destroy)
 
 	return Transaction{t}, nil
 }
@@ -114,18 +110,21 @@ func (d Database) CreateTransaction() (Transaction, error) {
 // if the data directory is writeable by creating a validation file. The address must be a
 // process address is the form of IP:Port pair.
 func (d Database) RebootWorker(address string, checkFile bool, suspendDuration int) error {
-	t := &futureInt64{
-		future: newFutureWithDb(
-			d.database,
-			nil,
-			C.fdb_database_reboot_worker(
-				d.ptr,
-				byteSliceToPtr([]byte(address)),
-				C.int(len(address)),
-				C.fdb_bool_t(boolToInt(checkFile)),
-				C.int(suspendDuration),
-			),
+	f := newFutureWithDb(
+		d.database,
+		nil,
+		C.fdb_database_reboot_worker(
+			d.ptr,
+			byteSliceToPtr([]byte(address)),
+			C.int(len(address)),
+			C.fdb_bool_t(boolToInt(checkFile)),
+			C.int(suspendDuration),
 		),
+	)
+	defer f.Close()
+
+	t := &futureInt64{
+		future: f,
 	}
 
 	dbVersion, err := t.Get()
@@ -145,8 +144,11 @@ func (d Database) GetClientStatus() ([]byte, error) {
 		return nil, errAPIVersionUnset
 	}
 
+	f := newFutureWithDb(d.database, nil, C.fdb_database_get_client_status(d.ptr))
+	defer f.Close()
+
 	st := &futureByteSlice{
-		future: newFutureWithDb(d.database, nil, C.fdb_database_get_client_status(d.ptr)),
+		future: f,
 	}
 
 	b, err := st.Get()
@@ -173,7 +175,10 @@ func retryable(wrapped func() (interface{}, error), onError func(Error) FutureNi
 		// fdb.Error
 		var ep Error
 		if errors.As(err, &ep) {
-			processedErr := onError(ep).Get()
+			f := onError(ep)
+			defer f.Close()
+
+			processedErr := f.Get()
 			var newEp Error
 			if !errors.As(processedErr, &newEp) || newEp.Code != ep.Code {
 				// override original error only if not an Error or code changed
@@ -214,18 +219,22 @@ func retryable(wrapped func() (interface{}, error), onError func(Error) FutureNi
 // Transaction and Database objects.
 func (d Database) Transact(f func(Transaction) (interface{}, error)) (interface{}, error) {
 	tr, err := d.CreateTransaction()
-	// Any error here is non-retryable
 	if err != nil {
+		// Any error here is non-retryable
 		return nil, err
 	}
+	defer tr.Close()
 
 	wrapped := func() (ret interface{}, err error) {
 		defer panicToError(&err)
 
 		ret, err = f(tr)
 
 		if err == nil {
-			err = tr.Commit().Get()
+			f := tr.Commit()
+			defer f.Close()
+
+			err = f.Get()
 		}
 
 		return
@@ -263,14 +272,14 @@ func (d Database) ReadTransact(f func(ReadTransaction) (interface{}, error)) (in
 		// Any error here is non-retryable
 		return nil, err
 	}
+	defer tr.Close()
 
 	wrapped := func() (ret interface{}, err error) {
 		defer panicToError(&err)
 
 		ret, err = f(tr)
 
-		// read-only transactions are not committed and will be destroyed automatically via GC,
-		// once all the futures go out of scope
+		// read-only transactions are not committed and will be destroyed automatically by the deferred Close()
 
 		return
 	}
@@ -300,6 +309,7 @@ func (d Database) LocalityGetBoundaryKeys(er ExactRange, limit int, readVersion
 	if err != nil {
 		return nil, err
 	}
+	defer tr.Close()
 
 	if readVersion != 0 {
 		tr.SetReadVersion(readVersion)

bindings/go/src/fdb/directory/allocator.go

@@ -95,6 +95,7 @@ func (hca highContentionAllocator) allocate(tr fdb.Transaction, s subspace.Subsp
 			// Increment the allocation count for the current window
 			tr.Add(hca.counters.Sub(start), oneBytes)
 			countFuture := tr.Snapshot().Get(hca.counters.Sub(start))
+			defer countFuture.Close()
 
 			allocatorMutex.Unlock()
 

bindings/go/src/fdb/directory/directory_layer.go

@@ -87,8 +87,12 @@ func (dl directoryLayer) createOrOpen(rtr fdb.ReadTransaction, tr *fdb.Transacti
 		return nil, errors.New("the root directory cannot be opened")
 	}
 
-	existingNode := dl.find(rtr, path).prefetchMetadata(rtr)
-	if existingNode.exists() {
+	existingNode, exists := dl.find(rtr, path).prefetchMetadata(rtr)
+	if exists {
+		defer existingNode._layer.Close()
+	}
+
+	if exists {
 		if existingNode.isInPartition(nil, false) {
 			subpath := existingNode.getPartitionSubpath()
 			enc, err := existingNode.getContents(dl, nil)
@@ -225,9 +229,11 @@ func (dl directoryLayer) Exists(rt fdb.ReadTransactor, path []string) (bool, err
 			return false, err
 		}
 
-		node := dl.find(rtr, path).prefetchMetadata(rtr)
-		if !node.exists() {
+		node, exists := dl.find(rtr, path).prefetchMetadata(rtr)
+		if !exists {
 			return false, nil
+		} else {
+			defer node._layer.Close()
 		}
 
 		if node.isInPartition(nil, false) {
@@ -252,9 +258,11 @@ func (dl directoryLayer) List(rt fdb.ReadTransactor, path []string) ([]string, e
 			return nil, err
 		}
 
-		node := dl.find(rtr, path).prefetchMetadata(rtr)
-		if !node.exists() {
+		node, exists := dl.find(rtr, path).prefetchMetadata(rtr)
+		if !exists {
 			return nil, ErrDirNotExists
+		} else {
+			defer node._layer.Close()
 		}
 
 		if node.isInPartition(nil, true) {
@@ -291,8 +299,14 @@ func (dl directoryLayer) Move(t fdb.Transactor, oldPath []string, newPath []stri
 			return nil, errors.New("the destination directory cannot be a subdirectory of the source directory")
 		}
 
-		oldNode := dl.find(tr, oldPath).prefetchMetadata(tr)
-		newNode := dl.find(tr, newPath).prefetchMetadata(tr)
+		oldNode, oldNodeExists := dl.find(tr, oldPath).prefetchMetadata(tr)
+		if oldNodeExists {
+			defer oldNode._layer.Close()
+		}
+		newNode, newNodeExists := dl.find(tr, newPath).prefetchMetadata(tr)
+		if newNodeExists {
+			defer newNode._layer.Close()
+		}
 
 		if !oldNode.exists() {
 			return nil, errors.New("the source directory does not exist")
@@ -349,10 +363,11 @@ func (dl directoryLayer) Remove(t fdb.Transactor, path []string) (bool, error) {
 			return false, errors.New("the root directory cannot be removed")
 		}
 
-		node := dl.find(tr, path).prefetchMetadata(tr)
-
-		if !node.exists() {
+		node, exists := dl.find(tr, path).prefetchMetadata(tr)
+		if !exists {
 			return false, nil
+		} else {
+			defer node._layer.Close()
 		}
 
 		if node.isInPartition(nil, false) {
@@ -416,16 +431,10 @@ func (dl directoryLayer) subdirNames(rtr fdb.ReadTransaction, node subspace.Subs
 	sd := node.Sub(_SUBDIRS)
 
 	rr := rtr.GetRange(sd, fdb.RangeOptions{})
-	ri := rr.Iterator()
 
 	var ret []string
-
-	for ri.Advance() {
-		kv, err := ri.Get()
-		if err != nil {
-			return nil, err
-		}
-
+	kvs := rr.GetSliceOrPanic()
+	for _, kv := range kvs {
 		p, err := sd.Unpack(kv.Key)
 		if err != nil {
 			return nil, err
@@ -441,13 +450,10 @@ func (dl directoryLayer) subdirNodes(tr fdb.Transaction, node subspace.Subspace)
 	sd := node.Sub(_SUBDIRS)
 
 	rr := tr.GetRange(sd, fdb.RangeOptions{})
-	ri := rr.Iterator()
 
 	var ret []subspace.Subspace
-
-	for ri.Advance() {
-		kv := ri.MustGet()
-
+	kvs := rr.GetSliceOrPanic()
+	for _, kv := range kvs {
 		ret = append(ret, dl.nodeWithPrefix(kv.Value))
 	}
 
@@ -507,7 +513,10 @@ func (dl directoryLayer) isPrefixFree(rtr fdb.ReadTransaction, prefix []byte) (b
 }
 
 func (dl directoryLayer) checkVersion(rtr fdb.ReadTransaction, tr *fdb.Transaction) error {
-	version, err := rtr.Get(dl.rootNode.Sub([]byte("version"))).Get()
+	f := rtr.Get(dl.rootNode.Sub([]byte("version")))
+	defer f.Close()
+
+	version, err := f.Get()
 	if err != nil {
 		return err
 	}
@@ -590,8 +599,18 @@ func (dl directoryLayer) nodeWithPrefix(prefix []byte) subspace.Subspace {
 
 func (dl directoryLayer) find(rtr fdb.ReadTransaction, path []string) *node {
 	n := &node{dl.rootNode, []string{}, path, nil}
+	futures := make([]fdb.FutureByteSlice, len(path))
 	for i := range path {
-		n = &node{dl.nodeWithPrefix(rtr.Get(n.subspace.Sub(_SUBDIRS, path[i])).MustGet()), path[:i+1], path, nil}
+		futures[i] = rtr.Get(n.subspace.Sub(_SUBDIRS, path[i]))
+	}
+	defer func() {
+		for _, f := range futures {
+			f.Close()
+		}
+	}()
+
+	for i, f := range futures {
+		n = &node{dl.nodeWithPrefix(f.MustGet()), path[:i+1], path, nil}
 		if !n.exists() || bytes.Compare(n.layer(rtr).MustGet(), []byte("partition")) == 0 {
 			return n
 		}

bindings/go/src/fdb/directory/node.go

@@ -43,11 +43,12 @@ func (n *node) exists() bool {
 	return true
 }
 
-func (n *node) prefetchMetadata(rtr fdb.ReadTransaction) *node {
+func (n *node) prefetchMetadata(rtr fdb.ReadTransaction) (*node, bool) {
 	if n.exists() {
 		n.layer(rtr)
+		return n, true
 	}
-	return n
+	return n, false
 }
 
 func (n *node) layer(rtr fdb.ReadTransaction) fdb.FutureByteSlice {