Add concurrency tests and fix race conditions

syncstore/store.go

@@ -4,6 +4,7 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"fmt"
+	"sync"
 	"time"
 )
 
@@ -14,11 +15,12 @@ type StoreDriver interface {
 }
 
 type Store struct {
-	drv StoreDriver
+	drv  StoreDriver
+	lock sync.Mutex
 }
 
 func NewStore(drv StoreDriver) Store {
-	return Store{drv}
+	return Store{drv: drv}
 }
 
 type CreateResp struct {
@@ -35,6 +37,8 @@ type GetResp struct {
 
 func (store *Store) Create(version string) (*CreateResp, error) {
 	var sid string
+	store.lock.Lock()
+	defer store.lock.Unlock()
 	failed := true
 	for i := 0; i < 5; i++ {
 		var idb = make([]byte, 16)
@@ -66,6 +70,8 @@ func (store *Store) Create(version string) (*CreateResp, error) {
 }
 
 func (store *Store) Update(id string, bookmarks string, lastUpdated time.Time) (time.Time, error) {
+	store.lock.Lock()
+	defer store.lock.Unlock()
 	t, err := store.drv.RawLoad(id)
 	if err != nil {
 		return time.Time{}, err
@@ -83,6 +89,8 @@ func (store *Store) Update(id string, bookmarks string, lastUpdated time.Time) (
 }
 
 func (store *Store) Get(id string) (*GetResp, error) {
+	store.lock.Lock()
+	defer store.lock.Unlock()
 	s, err := store.drv.RawLoad(id)
 	if err != nil {
 		return nil, err

syncstore/store_test.go

@@ -1,12 +1,16 @@
 package syncstore
 
 import (
+	"fmt"
+	"math/rand"
 	"regexp"
+	"strconv"
+	"sync"
 	"testing"
 	"time"
 )
 
-func testStore_helper(t *testing.T, drv StoreDriver) {
+func testStoreHelper(t *testing.T, drv StoreDriver) {
 	store := NewStore(drv)
 
 	v1 := "0.1.2"
@@ -78,29 +82,94 @@ func testStore_helper(t *testing.T, drv StoreDriver) {
 
 func TestStore_Mem(t *testing.T) {
 	drv, _ := NewMemStore()
-	testStore_helper(t, drv)
+	testStoreHelper(t, drv)
 }
 
 func TestStore_FS(t *testing.T) {
 	drv, _ := NewFSStore(t.TempDir())
-	testStore_helper(t, drv)
+	testStoreHelper(t, drv)
 }
 
-func benchCreate_helper(b *testing.B, drv StoreDriver) {
+const concurrencyWorkers = 5
+
+func testStoreConcurrencyWorker(wg *sync.WaitGroup, t *testing.T, store *Store, sid string, rounds int) {
+	defer wg.Done()
+	for i := 0; i < rounds; i++ {
+		for {
+			s, err := store.Get(sid)
+			if err != nil {
+				t.Error("Get() failed", err)
+			}
+			n := 0
+			if s.Bookmarks != "" {
+				n, err = strconv.Atoi(s.Bookmarks)
+				if err != nil {
+					t.Error("string to int conversion failed", err)
+				}
+			}
+			_, err = store.Update(sid, fmt.Sprintf("%d", n+1), s.LastUpdated)
+			if err == nil {
+				break
+			} else {
+				time.Sleep(time.Duration(10000 * (rand.Int31()%300 + 10)))
+			}
+		}
+	}
+}
+
+func testStoreConcurrencyHelper(t *testing.T, drv StoreDriver, workers, rounds int) {
+	var wg sync.WaitGroup
+	store := NewStore(drv)
+
+	s, err := store.Create("0.1.2")
+	if err != nil {
+		t.Error("Create() failed")
+		return
+	}
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go testStoreConcurrencyWorker(&wg, t, &store, s.ID, rounds)
+	}
+	wg.Wait()
+	data, err := store.Get(s.ID)
+	if err != nil {
+		t.Error("Get() failed: ", err)
+	}
+	n, err := strconv.Atoi(data.Bookmarks)
+	if err != nil {
+		t.Error("failed to parse an integer: ", err)
+	}
+	expected := workers * rounds
+	if n != expected {
+		t.Errorf("inconsistent result: expected %d, got %d", expected, n)
+	}
+}
+
+func TestStore_Concurrency_Mem(t *testing.T) {
+	drv, _ := NewMemStore()
+	testStoreConcurrencyHelper(t, drv, 5, 50000)
+}
+
+func TestStore_Concurrency_FS(t *testing.T) {
+	drv, _ := NewFSStore(t.TempDir())
+	testStoreConcurrencyHelper(t, drv, 5, 1000)
+}
+
+func benchCreateHelper(b *testing.B, drv StoreDriver) {
 	store := NewStore(drv)
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		store.Create("0.1.2")
+		_, _ = store.Create("0.1.2")
 	}
 }
 
 func BenchmarkStore_Create_Mem(b *testing.B) {
 	drv, _ := NewMemStore()
-	benchCreate_helper(b, drv)
+	benchCreateHelper(b, drv)
 }
 
 func BenchmarkStore_Create_FS(b *testing.B) {
 	drv, _ := NewFSStore(b.TempDir())
-	benchCreate_helper(b, drv)
+	benchCreateHelper(b, drv)
 }