From 4420d721968f78ad02e52bf9bd7a3f0dc202221e Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Mon, 25 Mar 2019 14:03:10 -0500
Subject: [PATCH 1/2] test concurrent value imports, fix race

---
 fragment.go               |  2 ++
 fragment_internal_test.go | 21 +++++++++++++++++++++
 2 files changed, 23 insertions(+)

diff --git a/fragment.go b/fragment.go
index 2b2925e16..f864fdc1c 100644
--- a/fragment.go
+++ b/fragment.go
@@ -1785,7 +1785,9 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
 		for i := uint(0); i < bitDepth+1; i++ {
 			rowSet[uint64(i)] = struct{}{}
 		}
+		f.mu.Lock()
 		err := f.importPositions(toSet, toClear, rowSet)
+		f.mu.Unlock()
 		return errors.Wrap(err, "importing positions")
 	}
 	err := f.snapshot()
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index b726d4fe5..71d3b5ebe 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -3024,3 +3024,24 @@ func TestSmallImportRestart(t *testing.T) {
 		t.Errorf("row 1 should be [1], but got %v", r1)
 	}
 }
+
+func TestImportValueConcurrent(t *testing.T) {
+	f := mustOpenFragment("i", "f", viewBSIGroupPrefix+"foo", 0, "none")
+	eg := &errgroup.Group{}
+	for i := 0; i < 4; i++ {
+		i := i
+		eg.Go(func() error {
+			for j := uint64(0); j < 10; j++ {
+				err := f.importValue([]uint64{j}, []uint64{uint64(rand.Int63n(1000))}, 10, i%2 == 0)
+				if err != nil {
+					return err
+				}
+			}
+			return nil
+		})
+	}
+	err := eg.Wait()
+	if err != nil {
+		t.Fatalf("concurrently importing values: %v", err)
+	}
+}

From 714f89c65c5b3677d85226868ded2c8349bba80d Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Mon, 25 Mar 2019 14:27:05 -0500
Subject: [PATCH 2/2] simplify locking in importValue

may be a slight perf cost, but the simplicity is well worth it
---
 fragment.go | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/fragment.go b/fragment.go
index f864fdc1c..1718d4acb 100644
--- a/fragment.go
+++ b/fragment.go
@@ -1733,12 +1733,15 @@ func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error {
 
 // importValue bulk imports a set of range-encoded values.
 func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear bool) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
 	// Verify that there are an equal number of column ids and values.
 	if len(columnIDs) != len(values) {
 		return fmt.Errorf("mismatch of column/value len: %d != %d", len(columnIDs), len(values))
 	}
 	var toSet, toClear []uint64
-	f.mu.RLock()
+
 	smallWrite := false
 	if len(columnIDs)*int(bitDepth+1)+f.opN < f.MaxOpN {
 		smallWrite = true
@@ -1751,14 +1754,11 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
 		toSet = make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8))
 		toClear = make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8))
 	}
-	f.mu.RUnlock()
 
 	if !smallWrite {
-		f.mu.Lock()
-		defer f.mu.Unlock()
-
 		f.storage.OpWriter = nil
 	}
+
 	// Process every value.
 	// If an error occurs then reopen the storage.
 	if err := func() (err error) {
@@ -1785,9 +1785,7 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
 		for i := uint(0); i < bitDepth+1; i++ {
 			rowSet[uint64(i)] = struct{}{}
 		}
-		f.mu.Lock()
 		err := f.importPositions(toSet, toClear, rowSet)
-		f.mu.Unlock()
 		return errors.Wrap(err, "importing positions")
 	}
 	err := f.snapshot()