Repository: bmatsuo/lmdb-go (c)

Issue:
lmdbsync: Remove exp/cmd/lmdb_example_resize
The command is basically a test meant to be run in parallel in background processes of a shell or in a multiplexer. The example should be pared down as much as possible, rerooted under exp/lmdbsync, and run by `go test` for that package (it must run as two forked processes because only one open map of an environment can exist in a process).

Pull request:
lmdbsync: real test for map resizing
Fixes #7

Patch:
diff --git a/exp/cmd/lmdb_example_resize/main.go b/exp/lmdbsync/testresize/main.go
similarity index 67%
rename from exp/cmd/lmdb_example_resize/main.go
rename to exp/lmdbsync/testresize/main.go
index fbc8f32..42d8173 100644
--- a/exp/cmd/lmdb_example_resize/main.go
+++ b/exp/lmdbsync/testresize/main.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bufio"
 	"crypto/rand"
 	"flag"
 	"fmt"
@@ -14,6 +15,8 @@ import (
 )
 
 func main() {
+	numitems := flag.Int64("n", 5<<10, "the number of items to write")
+	chunksize := flag.Int64("c", 100, "the number of items to write per txn")
 	flag.Parse()
 
 	failed := false
@@ -27,7 +30,7 @@ func main() {
 		log.Print(err)
 	}
 
-	err := WriteRandomItems("db", 1<<20, 10<<10)
+	err := WriteRandomItems("db", *numitems, *chunksize)
 	if err != nil {
 		fail(err)
 	} else {
@@ -37,7 +40,7 @@ func main() {
 
 // WriteRandomItems writes numitem items with chunksize sized values full of
 // random data.
-func WriteRandomItems(path string, numitem, chunksize int64) error {
+func WriteRandomItems(path string, numitem, chunksize int64) (err error) {
 	env, err := OpenEnv(path)
 	if err != nil {
 		return err
@@ -45,21 +48,38 @@ func WriteRandomItems(path string, numitem, chunksize int64) error {
 	defer env.Close()
 
 	numResize := 0
+	numResized := 0
 	defer func() {
 		log.Printf("%d resizes", numResize)
+		log.Printf("%d size adoptions", numResized)
+		if err == nil {
+			if numResize == 0 {
+				err = fmt.Errorf("process did not resize the memory map")
+			} else if numResized == 0 {
+				err = fmt.Errorf("process did not adopt a new map size")
+			}
+		}
 	}()
+	mapResizedLogger := func(b lmdbsync.Bag, err error) (lmdbsync.Bag, error) {
+		if lmdb.IsMapResized(err) {
+			log.Printf("map resized")
+			numResized++
+		}
+		return b, err
+	}
 	mapFullLogger := func(b lmdbsync.Bag, err error) (lmdbsync.Bag, error) {
 		if lmdb.IsMapFull(err) {
-			log.Printf("resize required: %v", err)
+			log.Printf("resize required")
 			numResize++
 		}
 		return b, err
 	}
 	env.Handlers = env.Handlers.Append(
+		handlerFunc(mapResizedLogger),
 		lmdbsync.MapResizedHandler(2, func(int) time.Duration { return 100 * time.Microsecond }),
 		handlerFunc(mapFullLogger),
 		lmdbsync.MapFullHandler(func(size int64) (int64, bool) {
-			newsize := size * 2
+			newsize := size + 128<<10 // linear scale is bad -- but useful to test
 			log.Printf("oldsize=%d newsize=%d", size, newsize)
 			return newsize, true
 		}),
@@ -67,7 +87,12 @@ func WriteRandomItems(path string, numitem, chunksize int64) error {
 
 	pid := os.Getpid()
 
+	scanner := bufio.NewScanner(os.Stdin)
 	for i := int64(0); i < numitem; {
+		if !scanner.Scan() {
+			return scanner.Err()
+		}
+
 		start := i
 		chunkmax := i + chunksize
 		if chunkmax > numitem {
@@ -78,7 +103,6 @@ func WriteRandomItems(path string, numitem, chunksize int64) error {
 		if err != nil {
 			return err
 		}
-		log.Printf("i=%d", i)
 		err = env.Update(func(txn *lmdb.Txn) (err error) {
 			dbi, err := txn.OpenRoot(0)
 			if err != nil {
@@ -86,7 +110,7 @@ func WriteRandomItems(path string, numitem, chunksize int64) error {
 			}
 
 			for i = start; i < chunkmax; i++ {
-				k := fmt.Sprintf("%d-%016x", pid, i)
+				k := fmt.Sprintf("%06d-%016x", pid, i)
 				err = txn.Put(dbi, []byte(k), v, 0)
 				if err != nil {
 					return err
@@ -98,6 +122,7 @@ func WriteRandomItems(path string, numitem, chunksize int64) error {
 		if err != nil {
 			return err
 		}
+		fmt.Println("ok")
 	}
 
 	return nil


Test patch:
diff --git a/exp/lmdbsync/testresize_test.go b/exp/lmdbsync/testresize_test.go
new file mode 100644
index 0000000..ba4b004
--- /dev/null
+++ b/exp/lmdbsync/testresize_test.go
@@ -0,0 +1,176 @@
+package lmdbsync
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/bmatsuo/lmdb-go/lmdb"
+)
+
+func TestResize(t *testing.T) {
+	tempdir, err := ioutil.TempDir("", "lmdbsync_testresize")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tempdir)
+
+	dbpath := filepath.Join(tempdir, "db")
+	err = os.Mkdir(dbpath, 0755)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	env, err := NewEnv(nil)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer env.Close()
+	err = env.Open(dbpath, 0, 0644)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	var root lmdb.DBI
+	env.Update(func(txn *lmdb.Txn) (err error) {
+		root, err = txn.OpenRoot(0)
+		if err != nil {
+			return err
+		}
+		return txn.Put(root, []byte("_start"), []byte(time.Now().String()), 0)
+	})
+
+	before, err := env.Info()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	bin := filepath.Join(tempdir, "testresize")
+	build := exec.Command("go", "build", "-o", bin, "./testresize")
+	build.Stderr = os.Stderr
+	err = build.Run()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	r1, w1, err := os.Pipe()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	r2, w2, err := os.Pipe()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	closePipes := func() {
+		w1.Close()
+		w2.Close()
+		r1.Close()
+		r2.Close()
+	}
+
+	cmd1 := exec.Command(bin)
+	cmd1.Dir = tempdir
+	cmd2 := exec.Command(bin)
+	cmd2.Dir = tempdir
+
+	cmd1.Stdin = r1
+	cmd2.Stdout = w1
+
+	cmd2.Stdin = r2
+	cmd1.Stdout = w2
+
+	stderr1, err := cmd1.StderrPipe()
+	if err != nil {
+		t.Error(err)
+		closePipes()
+		return
+	}
+	stderr2, err := cmd2.StderrPipe()
+	if err != nil {
+		t.Error(err)
+		closePipes()
+		return
+	}
+	stderr := io.MultiReader(stderr1, stderr2)
+
+	err = cmd1.Start()
+	if err != nil {
+		t.Errorf("start 1: %v", err)
+		closePipes()
+		return
+	}
+	err = cmd2.Start()
+	if err != nil {
+		t.Errorf("start 2: %v", err)
+		closePipes()
+		return
+	}
+	go fmt.Fprintln(w1, "start")
+
+	scanner := bufio.NewScanner(stderr)
+	for scanner.Scan() {
+		t.Log(scanner.Text())
+	}
+	closePipes()
+
+	err1 := cmd1.Wait()
+	if err1 != nil {
+		t.Errorf("err 1: %v", err1)
+	}
+	err2 := cmd2.Wait()
+	if err2 != nil {
+		t.Errorf("err 2: %v", err2)
+	}
+
+	trace := &resizeTracer{}
+	runner := env.WithHandler(HandlerChain{
+		trace,
+		MapResizedHandler(2, func(retry int) time.Duration {
+			t.Error("unable to reopen")
+			return time.Millisecond
+		}),
+	})
+	err = runner.Update(func(txn *lmdb.Txn) (err error) {
+		return txn.Put(root, []byte("_finish"), []byte(time.Now().String()), 0)
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
+	if trace.resized == 0 {
+		t.Errorf("no resize detected")
+	}
+
+	after, err := env.Info()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if after.MapSize <= before.MapSize {
+		t.Errorf("mapsize: %d (<= %d)", after.MapSize, before.MapSize)
+	}
+}
+
+type resizeTracer struct {
+	resized int
+}
+
+func (t *resizeTracer) HandleTxnErr(b Bag, err error) (Bag, error) {
+	if lmdb.IsMapResized(err) {
+		t.resized++
+	}
+	return b, err
+}