Add multi-threaded database benchmarks

db/db_test.go

@@ -6,6 +6,7 @@ import (
 	"math/rand"
 	"os"
 	"os/exec"
+	"sync"
 	"testing"
 	"time"
 
@@ -163,7 +164,7 @@ func BenchmarkWrite_Sqlite(b *testing.B) {
 	runWriteBench(b, "sqlite")
 }
 
-func runReadBenchmark(b *testing.B, driver string, eventsPerIP int) {
+func runReadBenchmark(b *testing.B, driver string, eventsPerIP int, threadCounts []int) {
 	dir, err := ioutil.TempDir("", "")
 	if err != nil {
 		b.Fatal(err)
@@ -200,9 +201,9 @@ func runReadBenchmark(b *testing.B, driver string, eventsPerIP int) {
 	}
 
 	deadline := time.Now().Add(-1 * time.Hour)
-	b.ResetTimer()
 
-	ipCount := 0
+	fn := func(b *testing.B, off int) {
+		ipCount := (off * 13) % len(refIPs)
 		for i := 0; i < b.N; i++ {
 			refIP := refIPs[ipCount]
 			ipCount++
@@ -222,18 +223,36 @@ func runReadBenchmark(b *testing.B, driver string, eventsPerIP int) {
 		}
 	}
 
+	// Now run separate benchmarks for each of the threadCounts values.
+	for _, numThreads := range threadCounts {
+		b.Run(fmt.Sprintf("threads:%d", numThreads), func(b *testing.B) {
+			var wg sync.WaitGroup
+			for i := 0; i < numThreads; i++ {
+				wg.Add(1)
+				go func(i int) {
+					fn(b, i)
+					wg.Done()
+				}(i)
+			}
+			wg.Wait()
+		})
+	}
+}
+
+var threadTests = []int{1, 5, 50}
+
 func BenchmarkRead_LevelDB_SmallAggregate(b *testing.B) {
-	runReadBenchmark(b, "leveldb", 5)
+	runReadBenchmark(b, "leveldb", 5, threadTests)
 }
 
 func BenchmarkRead_LevelDB_LargeAggregate(b *testing.B) {
-	runReadBenchmark(b, "leveldb", 1000)
+	runReadBenchmark(b, "leveldb", 1000, threadTests)
 }
 
 func BenchmarkRead_Sqlite_SmallAggregate(b *testing.B) {
-	runReadBenchmark(b, "sqlite", 5)
+	runReadBenchmark(b, "sqlite", 5, threadTests)
 }
 
 func BenchmarkRead_Sqlite_LargeAggregate(b *testing.B) {
-	runReadBenchmark(b, "sqlite", 1000)
+	runReadBenchmark(b, "sqlite", 1000, threadTests)
 }