syncthing/internal/files/concurrency_test.go
2014-11-17 12:54:42 +01:00

243 lines
4.5 KiB
Go

// Copyright (C) 2014 The Syncthing Authors.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along
// with this program. If not, see <http://www.gnu.org/licenses/>.
package files_test
import (
"crypto/rand"
"log"
"os"
"sync"
"testing"
"time"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/opt"
"github.com/syndtr/goleveldb/leveldb/util"
)
var keys [][]byte
func init() {
for i := 0; i < nItems; i++ {
keys = append(keys, randomData(1))
}
}
const nItems = 10000
func randomData(prefix byte) []byte {
data := make([]byte, 1+32+64+32)
_, err := rand.Reader.Read(data)
if err != nil {
panic(err)
}
return append([]byte{prefix}, data...)
}
func setItems(db *leveldb.DB) error {
batch := new(leveldb.Batch)
for _, k1 := range keys {
k2 := randomData(2)
// k2 -> data
batch.Put(k2, randomData(42))
// k1 -> k2
batch.Put(k1, k2)
}
if testing.Verbose() {
log.Printf("batch write (set) %p", batch)
}
return db.Write(batch, nil)
}
func clearItems(db *leveldb.DB) error {
snap, err := db.GetSnapshot()
if err != nil {
return err
}
defer snap.Release()
// Iterate over k2
it := snap.NewIterator(util.BytesPrefix([]byte{1}), nil)
defer it.Release()
batch := new(leveldb.Batch)
for it.Next() {
k1 := it.Key()
k2 := it.Value()
// k2 should exist
_, err := snap.Get(k2, nil)
if err != nil {
return err
}
// Delete the k1 => k2 mapping first
batch.Delete(k1)
// Then the k2 => data mapping
batch.Delete(k2)
}
if testing.Verbose() {
log.Printf("batch write (clear) %p", batch)
}
return db.Write(batch, nil)
}
func scanItems(db *leveldb.DB) error {
snap, err := db.GetSnapshot()
if testing.Verbose() {
log.Printf("snap create %p", snap)
}
if err != nil {
return err
}
defer func() {
if testing.Verbose() {
log.Printf("snap release %p", snap)
}
snap.Release()
}()
// Iterate from the start of k2 space to the end
it := snap.NewIterator(util.BytesPrefix([]byte{1}), nil)
defer it.Release()
i := 0
for it.Next() {
// k2 => k1 => data
k1 := it.Key()
k2 := it.Value()
_, err := snap.Get(k2, nil)
if err != nil {
log.Printf("k1: %x", k1)
log.Printf("k2: %x (missing)", k2)
return err
}
i++
}
if testing.Verbose() {
log.Println("scanned", i)
}
return nil
}
func TestConcurrentSetClear(t *testing.T) {
if testing.Short() {
return
}
dur := 30 * time.Second
t0 := time.Now()
var wg sync.WaitGroup
os.RemoveAll("testdata/concurrent-set-clear.db")
db, err := leveldb.OpenFile("testdata/concurrent-set-clear.db", &opt.Options{CachedOpenFiles: 10})
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll("testdata/concurrent-set-clear.db")
errChan := make(chan error, 3)
wg.Add(1)
go func() {
defer wg.Done()
for time.Since(t0) < dur {
if err := setItems(db); err != nil {
errChan <- err
return
}
if err := clearItems(db); err != nil {
errChan <- err
return
}
}
}()
wg.Add(1)
go func() {
defer wg.Done()
for time.Since(t0) < dur {
if err := scanItems(db); err != nil {
errChan <- err
return
}
}
}()
go func() {
wg.Wait()
errChan <- nil
}()
err = <-errChan
if err != nil {
t.Error(err)
}
db.Close()
}
func TestConcurrentSetOnly(t *testing.T) {
if testing.Short() {
return
}
dur := 30 * time.Second
t0 := time.Now()
var wg sync.WaitGroup
os.RemoveAll("testdata/concurrent-set-only.db")
db, err := leveldb.OpenFile("testdata/concurrent-set-only.db", &opt.Options{CachedOpenFiles: 10})
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll("testdata/concurrent-set-only.db")
errChan := make(chan error, 3)
wg.Add(1)
go func() {
defer wg.Done()
for time.Since(t0) < dur {
if err := setItems(db); err != nil {
errChan <- err
return
}
}
}()
wg.Add(1)
go func() {
defer wg.Done()
for time.Since(t0) < dur {
if err := scanItems(db); err != nil {
errChan <- err
return
}
}
}()
go func() {
wg.Wait()
errChan <- nil
}()
err = <-errChan
if err != nil {
t.Error(err)
}
}