func benchmarkTransactionSpeed(b *testing.B, numThreads int, appendUpdate bool) {
	b.Logf("Running benchmark with %v threads", numThreads)
	wt, err := newWALTester(b.Name(), &dependencyProduction{})
	if err != nil {
		b.Error(err)
	}
	defer wt.Close()
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(4000),  
	})
	f := func() (latency time.Duration, err error) {
		startTime := time.Now()
		txn, err := wt.wal.NewTransaction(updates)
		if err != nil {
			return
		}
		if appendUpdate {
			if err = <-txn.Append(updates); err != nil {
				return
			}
		}
		if err = <-txn.SignalSetupComplete(); err != nil {
			return
		}
		latency = time.Since(startTime)
		if err = txn.SignalUpdatesApplied(); err != nil {
			return
		}
		return latency, nil
	}
	stop := make(chan struct{})
	var atomicNumTxns uint64
	var atomicNumErr uint64
	var wg sync.WaitGroup
	latencies := make([]time.Duration, numThreads, numThreads)
	for i := 0; i < numThreads; i++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			for {
				select {
				case <-stop:
					return
				default:
				}
				latency, err := f()
				if err != nil {
					atomic.AddUint64(&atomicNumErr, 1)
					return
				}
				atomic.AddUint64(&atomicNumTxns, 1)
				if latency > latencies[j] {
					latencies[j] = latency
				}
			}
		}(i)
	}
	select {
	case <-time.After(time.Minute):
		close(stop)
	}
	wg.Wait()
	if atomicNumErr > 0 {
		b.Fatalf("%v errors happened during execution", atomicNumErr)
	}
	fi, err := os.Stat(wt.path)
	if os.IsNotExist(err) {
		b.Errorf("wal was deleted but shouldn't have")
	}
	var maxLatency time.Duration
	for i := 0; i < numThreads; i++ {
		if latencies[i] > maxLatency {
			maxLatency = latencies[i]
		}
	}
	b.Logf("filesize: %v mb", float64(fi.Size())/float64(1e+6))
	b.Logf("used pages: %v", wt.wal.filePageCount)
	b.Logf("total transactions: %v", atomicNumTxns)
	b.Logf("txn/s: %v", float64(atomicNumTxns)/60.0)
	b.Logf("maxLatency: %v", maxLatency)
}
benchmarkTransactionSpeed is a helper function to create benchmarks that run
for 1 min to find out how many transactions can be applied to the wal and
how large the wal grows during that time using a certain number of threads.
When appendUpdate is set to 'true', a second update will be appended to the
transaction before it is committed.
package writeaheadlog
import (
	"os"
	"path/filepath"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
	"time"
	"github.com/NebulousLabs/fastrand"
)
func retry(tries int, durationBetweenAttempts time.Duration, fn func() error) (err error) {
	for i := 1; i < tries; i++ {
		err = fn()
		if err == nil {
			return nil
		}
		time.Sleep(durationBetweenAttempts)
	}
	return fn()
}
func tempDir(dirs ...string) string {
	path := filepath.Join(os.TempDir(), "wal", filepath.Join(dirs...))
	os.RemoveAll(path)  
	return path
}
type walTester struct {
	wal *WAL
	recoveredTxns []*Transaction
	path          string
}
func (wt *walTester) Close() error {
	return wt.wal.Close()
}
func newWALTester(name string, deps dependencies) (*walTester, error) {
	testdir := tempDir(name)
	err := os.MkdirAll(testdir, 0700)
	if err != nil {
		return nil, err
	}
	path := filepath.Join(testdir, "test.wal")
	recoveredTxns, wal, err := newWal(path, deps)
	if err != nil {
		return nil, err
	}
	for _, txn := range recoveredTxns {
		if err := txn.SignalUpdatesApplied(); err != nil {
			wal.Close()
			return nil, err
		}
	}
	cmt := &walTester{
		wal:           wal,
		recoveredTxns: recoveredTxns,
		path:          path,
	}
	return cmt, nil
}
func transactionPages(txn *Transaction) (pages []page) {
	page := txn.firstPage
	for page != nil {
		pages = append(pages, *page)
		page = page.nextPage
	}
	return
}
func TestCommitFailed(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyCommitFail{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(1234),
	})
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	wait := txn.SignalSetupComplete()
	if err := <-wait; err == nil {
		t.Error("SignalSetupComplete should have failed but didn't")
	}
	wt.Close()
	if _, err := os.Stat(wt.path); os.IsNotExist(err) {
		t.Errorf("wal was deleted at %v", wt.path)
	}
	updates2, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Close()
	if len(updates2) != 0 {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			0, len(updates2))
	}
}
func TestMisleadingWrite(t *testing.T) {
	t.Skip("not implemented yet")
}
func BenchmarkMarshalUpdates(b *testing.B) {
	updates := make([]Update, 100)
	for i := range updates {
		updates[i] = Update{
			Name:         "test",
			Instructions: fastrand.Bytes(1234),
		}
	}
	b.ResetTimer()
	b.ReportAllocs()
	for i := 0; i < b.N; i++ {
		marshalUpdates(updates)
	}
}
func BenchmarkUnmarshalUpdates(b *testing.B) {
	updates := make([]Update, 100)
	for i := range updates {
		updates[i] = Update{
			Name:         "test",
			Instructions: fastrand.Bytes(1234),
		}
	}
	data := marshalUpdates(updates)
	b.ResetTimer()
	b.ReportAllocs()
	for i := 0; i < b.N; i++ {
		_, err := unmarshalUpdates(data)
		if err != nil {
			b.Fatal(err)
		}
	}
}
func TestReleaseFailed(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyReleaseFail{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(1234),
	})
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	wait := txn.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete failed %v", err)
	}
	if err := txn.SignalUpdatesApplied(); err == nil {
		t.Error("SignalUpdatesApplies should have failed but didn't")
	}
	wt.Close()
	if _, err := os.Stat(wt.path); os.IsNotExist(err) {
		t.Errorf("wal was deleted at %v", wt.path)
	}
	updates2, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Close()
	if len(updates2) != 1 {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			1, len(updates2))
	}
}
func TestReleaseNotCalled(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyUncleanShutdown{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(1234),
	})
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	txn2, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	wait := txn.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete for the first transaction failed %v", err)
	}
	wait2 := txn2.SignalSetupComplete()
	if err := <-wait2; err != nil {
		t.Errorf("SignalSetupComplete for the second transaction failed")
	}
	if err := txn2.SignalUpdatesApplied(); err != nil {
		t.Errorf("SignalApplyComplete for the second transaction failed")
	}
	wt.Close()
	if _, err := os.Stat(wt.path); os.IsNotExist(err) {
		t.Errorf("wal was deleted at %v", wt.path)
	}
	updates2, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Close()
	if len(updates2) != len(updates) {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			len(updates), len(updates2))
	}
}
func TestPayloadCorrupted(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyUncleanShutdown{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(1234),
	})
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	txn2, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	wait := txn.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete failed %v", err)
	}
	wait = txn2.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete failed %v", err)
	}
	txn.firstPage.payload = fastrand.Bytes(2000)
	_, err = txn.wal.logFile.WriteAt(txn.firstPage.appendTo(nil), int64(txn.firstPage.offset))
	if err != nil {
		t.Errorf("Corrupting the page failed %v", err)
	}
	wt.Close()
	if _, err := os.Stat(wt.path); os.IsNotExist(err) {
		t.Errorf("wal was deleted at %v", wt.path)
	}
	updates2, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Close()
	if len(updates2) != 1 {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			1, len(updates2))
	}
}
func TestPayloadCorrupted2(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyUncleanShutdown{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(1234),
	})
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	txn2, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	wait := txn.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete failed %v", err)
	}
	wait = txn2.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete failed %v", err)
	}
	txn2.firstPage.payload = fastrand.Bytes(2000)
	_, err = txn2.wal.logFile.WriteAt(txn2.firstPage.appendTo(nil), int64(txn2.firstPage.offset))
	if err != nil {
		t.Errorf("Corrupting the page failed %v", err)
	}
	wt.Close()
	if _, err := os.Stat(wt.path); os.IsNotExist(err) {
		t.Errorf("wal was deleted at %v", wt.path)
	}
	updates2, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Close()
	if len(updates2) != 1 {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			0, len(updates2))
	}
}
func TestWalParallel(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyProduction{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(1234),
	})
	done := make(chan error)
	f := func() {
		txn, err := wt.wal.NewTransaction(updates)
		if err != nil {
			done <- err
			return
		}
		if err := <-txn.SignalSetupComplete(); err != nil {
			done <- err
			return
		}
		if err := txn.SignalUpdatesApplied(); err != nil {
			done <- err
			return
		}
		done <- nil
	}
	numThreads := 1000
	for i := 0; i < numThreads; i++ {
		go f()
	}
	for i := 0; i < numThreads; i++ {
		err := <-done
		if err != nil {
			t.Errorf("Thread %v failed: %v", i, err)
		}
	}
	if wt.wal.filePageCount != uint64(len(wt.wal.availablePages)) {
		t.Errorf("number of available pages doesn't match the number of created ones. Expected %v, but was %v",
			wt.wal.availablePages, wt.wal.filePageCount)
	}
	wt.Close()
	fi, err := os.Stat(wt.path)
	if os.IsNotExist(err) {
		t.Fatalf("wal was deleted but shouldn't have")
	}
	t.Logf("filesize: %v mb", float64(fi.Size())/float64(1e+6))
	t.Logf("used pages: %v", wt.wal.filePageCount)
	updates2, w, err := New(wt.path)
	if err != nil {
		t.Error(err)
	}
	defer w.Close()
	if len(updates2) != 0 {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			0, len(updates2))
	}
}
func TestPageRecycling(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyProduction{})
	if err != nil {
		t.Error(err)
	}
	defer wt.Close()
	updates := []Update{}
	updates = append(updates, Update{
		Name:         "test",
		Instructions: fastrand.Bytes(5000),
	})
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	if err := <-txn.SignalSetupComplete(); err != nil {
		t.Errorf("SignalSetupComplete failed: %v", err)
	}
	if len(wt.wal.availablePages) != 0 {
		t.Errorf("Number of available pages should be 0 but was %v", len(wt.wal.availablePages))
	}
	if err := txn.SignalUpdatesApplied(); err != nil {
		t.Errorf("SignalApplyComplete failed: %v", err)
	}
	usedPages := wt.wal.filePageCount
	availablePages := len(wt.wal.availablePages)
	if usedPages == 0 {
		t.Errorf("The number of used pages should be greater than 0")
	}
	if usedPages != uint64(availablePages) {
		t.Errorf("number of used pages doesn't match number of available pages")
	}
	txn2, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	if err := <-txn2.SignalSetupComplete(); err != nil {
		t.Errorf("SignalSetupComplete failed: %v", err)
	}
	if len(wt.wal.availablePages) != 0 {
		t.Errorf("Number of available pages should be 0 but was %v", len(wt.wal.availablePages))
	}
	if err := txn2.SignalUpdatesApplied(); err != nil {
		t.Errorf("SignalApplyComplete failed: %v", err)
	}
	if wt.wal.filePageCount != usedPages || len(wt.wal.availablePages) != availablePages {
		t.Errorf("expected used pages %v, was %v", usedPages, wt.wal.filePageCount)
		t.Errorf("expected available pages %v, was %v", availablePages, len(wt.wal.availablePages))
	}
}
func TestRestoreTransactions(t *testing.T) {
	t.Skip("broken")
}
func TestRecoveryFailed(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyUncleanShutdown{})
	if err != nil {
		t.Error(err)
	}
	numUpdates := 10
	updates := []Update{}
	for i := 0; i < numUpdates; i++ {
		updates = append(updates, Update{
			Name:         "test",
			Instructions: fastrand.Bytes(10000),
		})
	}
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		return
	}
	if err = <-txn.SignalSetupComplete(); err != nil {
		return
	}
	if err := wt.Close(); err == nil {
		t.Error("There should have been an error but there wasn't")
	}
	recoveredTxns2, w, err := newWal(wt.path, &dependencyRecoveryFail{})
	if err != nil {
		t.Fatal(err)
	}
	numRecoveredUpdates := 0
	for _, txn := range recoveredTxns2 {
		numRecoveredUpdates += len(txn.Updates)
	}
	if numRecoveredUpdates != numUpdates {
		t.Errorf("There should be %v updates but there were %v", numUpdates, numRecoveredUpdates)
	}
	for _, txn := range recoveredTxns2 {
		if err := txn.SignalUpdatesApplied(); err != nil {
			t.Errorf("Failed to signal applied updates: %v", err)
		}
	}
	if err := w.Close(); err != nil {
		t.Errorf("Failed to close wal: %v", err)
	}
	recoveredTxns3, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	if len(recoveredTxns3) != 0 {
		t.Errorf("There should be %v updates but there were %v", 0, len(recoveredTxns3))
	}
	mdData := make([]byte, pageSize)
	if _, err := w.logFile.ReadAt(mdData, 0); err != nil {
		t.Fatal(err)
	}
	recoveryState, err := readWALMetadata(mdData)
	if err != nil {
		t.Fatal(err)
	}
	if recoveryState != recoveryStateUnclean {
		t.Errorf("recoveryState should be %v but was %v",
			recoveryStateUnclean, recoveryState)
	}
	if err := w.Close(); err != nil {
		t.Errorf("Failed to close wal: %v", err)
	}
	_, err = os.Stat(wt.path)
	if os.IsNotExist(err) {
		t.Errorf("wal was deleted but shouldn't have")
	}
}
func TestTransactionAppend(t *testing.T) {
	wt, err := newWALTester(t.Name(), &dependencyProduction{})
	if err != nil {
		t.Fatal(err)
	}
	updates := []Update{{
		Name:         "test",
		Instructions: fastrand.Bytes(3000),
	}}
	txn, err := wt.wal.NewTransaction(updates)
	if err != nil {
		t.Fatal(err)
	}
	if err := <-txn.Append(updates); err != nil {
		t.Errorf("Append failed: %v", err)
	}
	wait := txn.SignalSetupComplete()
	if err := <-wait; err != nil {
		t.Errorf("SignalSetupComplete for the first transaction failed %v", err)
	}
	wt.Close()
	recoveredTxns2, w, err := New(wt.path)
	if err != nil {
		t.Fatal(err)
	}
	defer w.Close()
	if len(recoveredTxns2[0].Updates) != len(updates)*2 {
		t.Errorf("Number of updates after restart didn't match. Expected %v, but was %v",
			len(updates)*2, len(recoveredTxns2[0].Updates))
	}
}