Recsplit: cancelable build (#1073)

.github/workflows/ci.yml

@@ -49,7 +49,7 @@ jobs:
         if: matrix.os == 'ubuntu-20.04'
         uses: golangci/golangci-lint-action@v3
         with:
-          version: v1.53
+          version: v1.54
 
       - name: Test win
         if: matrix.os == 'windows-2022'

Makefile

@@ -80,7 +80,7 @@ lintci-deps-clean: golangci-lint-clean
 
 # download and build golangci-lint (https://golangci-lint.run)
 $(GOBINREL)/golangci-lint: | $(GOBINREL)
-	curl -sSfL https://github.com/raw/golangci/golangci-lint/master/install.sh | sh -s -- -b "$(GOBIN)" v1.53.3
+	curl -sSfL https://github.com/raw/golangci/golangci-lint/master/install.sh | sh -s -- -b "$(GOBIN)" v1.54.0
 
 golangci-lint-clean:
 	rm -f "$(GOBIN)/golangci-lint"

recsplit/index_test.go

@@ -18,6 +18,7 @@ package recsplit
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -47,7 +48,7 @@ func TestReWriteIndex(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
-	if err := rs.Build(); err != nil {
+	if err := rs.Build(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 	idx := MustOpen(indexFile)

recsplit/recsplit.go

@@ -18,6 +18,7 @@ package recsplit
 
 import (
 	"bufio"
+	"context"
 	"crypto/rand"
 	"encoding/binary"
 	"fmt"
@@ -534,8 +535,7 @@ func (rs *RecSplit) loadFuncOffset(k, _ []byte, _ etl.CurrentTableReader, _ etl.
 
 // Build has to be called after all the keys have been added, and it initiates the process
 // of building the perfect hash function and writing index into a file
-func (rs *RecSplit) Build() error {
-
+func (rs *RecSplit) Build(ctx context.Context) error {
 	if rs.built {
 		return fmt.Errorf("already built")
 	}
@@ -570,7 +570,7 @@ func (rs *RecSplit) Build() error {
 	if rs.lvl < log.LvlTrace {
 		log.Log(rs.lvl, "[index] calculating", "file", rs.indexFileName)
 	}
-	if err := rs.bucketCollector.Load(nil, "", rs.loadFuncBucket, etl.TransformArgs{}); err != nil {
+	if err := rs.bucketCollector.Load(nil, "", rs.loadFuncBucket, etl.TransformArgs{Quit: ctx.Done()}); err != nil {
 		return err
 	}
 	if len(rs.currentBucket) > 0 {

recsplit/recsplit_fuzz_test.go

@@ -18,6 +18,7 @@ limitations under the License.
 package recsplit
 
 import (
+	"context"
 	"path/filepath"
 	"testing"
 
@@ -73,7 +74,7 @@ func FuzzRecSplit(f *testing.F) {
 		if err := rs.AddKey(in[i:], off); err != nil {
 			t.Fatal(err)
 		}
-		if err = rs.Build(); err != nil {
+		if err = rs.Build(context.Background()); err != nil {
 			t.Fatal(err)
 		}
 		// Check that there is a bijection

recsplit/recsplit_test.go

@@ -17,6 +17,7 @@
 package recsplit
 
 import (
+	"context"
 	"fmt"
 	"path/filepath"
 	"testing"
@@ -41,16 +42,16 @@ func TestRecSplit2(t *testing.T) {
 	if err = rs.AddKey([]byte("first_key"), 0); err != nil {
 		t.Error(err)
 	}
-	if err = rs.Build(); err == nil {
+	if err = rs.Build(context.Background()); err == nil {
 		t.Errorf("test is expected to fail, too few keys added")
 	}
 	if err = rs.AddKey([]byte("second_key"), 0); err != nil {
 		t.Error(err)
 	}
-	if err = rs.Build(); err != nil {
+	if err = rs.Build(context.Background()); err != nil {
 		t.Error(err)
 	}
-	if err = rs.Build(); err == nil {
+	if err = rs.Build(context.Background()); err == nil {
 		t.Errorf("test is expected to fail, hash gunction was built already")
 	}
 	if err = rs.AddKey([]byte("key_to_fail"), 0); err == nil {
@@ -78,7 +79,7 @@ func TestRecSplitDuplicate(t *testing.T) {
 	if err := rs.AddKey([]byte("first_key"), 0); err != nil {
 		t.Error(err)
 	}
-	if err := rs.Build(); err == nil {
+	if err := rs.Build(context.Background()); err == nil {
 		t.Errorf("test is expected to fail, duplicate key")
 	}
 }
@@ -119,7 +120,7 @@ func TestIndexLookup(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
-	if err := rs.Build(); err != nil {
+	if err := rs.Build(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 	idx := MustOpen(indexFile)
@@ -154,7 +155,7 @@ func TestTwoLayerIndex(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
-	if err := rs.Build(); err != nil {
+	if err := rs.Build(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 

state/domain.go

@@ -1144,7 +1144,7 @@ func buildIndex(ctx context.Context, d *compress.Decompressor, idxPath, tmpdir s
 
 			p.Processed.Add(1)
 		}
-		if err = rs.Build(); err != nil {
+		if err = rs.Build(ctx); err != nil {
 			if rs.Collision() {
 				logger.Info("Building recsplit. Collision happened. It's ok. Restarting...")
 				rs.ResetNextSalt()

state/history.go

@@ -464,7 +464,7 @@ func buildVi(ctx context.Context, historyItem, iiItem *filesItem, historyIdxPath
 
 			p.Processed.Add(1)
 		}
-		if err = rs.Build(); err != nil {
+		if err = rs.Build(ctx); err != nil {
 			if rs.Collision() {
 				logger.Info("Building recsplit. Collision happened. It's ok. Restarting...")
 				rs.ResetNextSalt()
@@ -940,7 +940,7 @@ func (h *History) buildFiles(ctx context.Context, step uint64, collation History
 				valOffset, _ = g.Skip()
 			}
 		}
-		if err = rs.Build(); err != nil {
+		if err = rs.Build(ctx); err != nil {
 			if rs.Collision() {
 				log.Info("Building recsplit. Collision happened. It's ok. Restarting...")
 				rs.ResetNextSalt()

state/locality_index.go

@@ -353,7 +353,7 @@ func (li *LocalityIndex) buildFiles(ctx context.Context, ic *InvertedIndexContex
 			return nil, err
 		}
 
-		if err = rs.Build(); err != nil {
+		if err = rs.Build(ctx); err != nil {
 			if rs.Collision() {
 				li.logger.Debug("Building recsplit. Collision happened. It's ok. Restarting...")
 				rs.ResetNextSalt()

state/merge.go

@@ -986,7 +986,7 @@ func (h *History) mergeFiles(ctx context.Context, indexFiles, historyFiles []*fi
 				}
 				p.Processed.Add(1)
 			}
-			if err = rs.Build(); err != nil {
+			if err = rs.Build(ctx); err != nil {
 				if rs.Collision() {
 					log.Info("Building recsplit. Collision happened. It's ok. Restarting...")
 					rs.ResetNextSalt()