storage/sqlite/sqlite-storage_test.go

   1 //go:build cgo
   2 // +build cgo
   3
   4 package sqliteStorage
   5
   6 import (
   7         "bytes"
   8         "context"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "io/ioutil"
  13         "path/filepath"
  14         "sync"
  15         "testing"
  16         "time"
  17
  18         _ "github.com/anacrolix/envpprof"
  19         "github.com/dustin/go-humanize"
  20         qt "github.com/frankban/quicktest"
  21         "github.com/stretchr/testify/assert"
  22         "github.com/stretchr/testify/require"
  23
  24         "github.com/anacrolix/torrent/storage"
  25         test_storage "github.com/anacrolix/torrent/storage/test"
  26 )
  27
  28 func newConnsAndProv(t *testing.T, opts NewPoolOpts) (ConnPool, *provider) {
  29         opts.Path = filepath.Join(t.TempDir(), "sqlite3.db")
  30         pool, err := NewPool(opts)
  31         qt.Assert(t, err, qt.IsNil)
  32         // sqlitex.Pool.Close doesn't like being called more than once. Let it slide for now.
  33         //t.Cleanup(func() { pool.Close() })
  34         qt.Assert(t, initPoolDatabase(pool, InitDbOpts{}), qt.IsNil)
  35         if !opts.Memory && opts.SetJournalMode == "" {
  36                 opts.SetJournalMode = "wal"
  37         }
  38         qt.Assert(t, initPoolConns(context.TODO(), pool, opts.InitConnOpts), qt.IsNil)
  39         prov, err := NewProvider(pool, ProviderOpts{BatchWrites: pool.NumConns() > 1})
  40         require.NoError(t, err)
  41         t.Cleanup(func() { prov.Close() })
  42         return pool, prov
  43 }
  44
  45 func TestTextBlobSize(t *testing.T) {
  46         _, prov := newConnsAndProv(t, NewPoolOpts{})
  47         a, _ := prov.NewInstance("a")
  48         err := a.Put(bytes.NewBufferString("\x00hello"))
  49         qt.Assert(t, err, qt.IsNil)
  50         fi, err := a.Stat()
  51         qt.Assert(t, err, qt.IsNil)
  52         assert.EqualValues(t, 6, fi.Size())
  53 }
  54
  55 func TestSimultaneousIncrementalBlob(t *testing.T) {
  56         _, p := newConnsAndProv(t, NewPoolOpts{
  57                 NumConns: 3,
  58         })
  59         a, err := p.NewInstance("a")
  60         require.NoError(t, err)
  61         const contents = "hello, world"
  62         require.NoError(t, a.Put(bytes.NewReader([]byte("hello, world"))))
  63         rc1, err := a.Get()
  64         require.NoError(t, err)
  65         rc2, err := a.Get()
  66         require.NoError(t, err)
  67         var b1, b2 []byte
  68         var e1, e2 error
  69         var wg sync.WaitGroup
  70         doRead := func(b *[]byte, e *error, rc io.ReadCloser, n int) {
  71                 defer wg.Done()
  72                 defer rc.Close()
  73                 *b, *e = ioutil.ReadAll(rc)
  74                 require.NoError(t, *e, n)
  75                 assert.EqualValues(t, contents, *b)
  76         }
  77         wg.Add(2)
  78         go doRead(&b2, &e2, rc2, 2)
  79         go doRead(&b1, &e1, rc1, 1)
  80         wg.Wait()
  81 }
  82
  83 func BenchmarkMarkComplete(b *testing.B) {
  84         const pieceSize = test_storage.DefaultPieceSize
  85         const noTriggers = false
  86         var capacity int64 = test_storage.DefaultNumPieces * pieceSize / 2
  87         if noTriggers {
  88                 // Since we won't push out old pieces, we have to mark them incomplete manually.
  89                 capacity = 0
  90         }
  91         runBench := func(b *testing.B, ci storage.ClientImpl) {
  92                 test_storage.BenchmarkPieceMarkComplete(b, ci, pieceSize, test_storage.DefaultNumPieces, capacity)
  93         }
  94         c := qt.New(b)
  95         b.Run("CustomDirect", func(b *testing.B) {
  96                 var opts NewDirectStorageOpts
  97                 opts.Capacity = capacity
  98                 opts.NoTriggers = noTriggers
  99                 benchOpts := func(b *testing.B) {
 100                         opts.Path = filepath.Join(b.TempDir(), "storage.db")
 101                         ci, err := NewDirectStorage(opts)
 102                         c.Assert(err, qt.IsNil)
 103                         defer ci.Close()
 104                         runBench(b, ci)
 105                 }
 106                 b.Run("Default", benchOpts)
 107         })
 108         for _, memory := range []bool{false, true} {
 109                 b.Run(fmt.Sprintf("Memory=%v", memory), func(b *testing.B) {
 110                         b.Run("Direct", func(b *testing.B) {
 111                                 var opts NewDirectStorageOpts
 112                                 opts.Memory = memory
 113                                 opts.Capacity = capacity
 114                                 //opts.GcBlobs = true
 115                                 opts.BlobFlushInterval = time.Second
 116                                 opts.NoTriggers = noTriggers
 117                                 directBench := func(b *testing.B) {
 118                                         opts.Path = filepath.Join(b.TempDir(), "storage.db")
 119                                         ci, err := NewDirectStorage(opts)
 120                                         var ujm UnexpectedJournalMode
 121                                         if errors.As(err, &ujm) {
 122                                                 b.Skipf("setting journal mode %q: %v", opts.SetJournalMode, err)
 123                                         }
 124                                         c.Assert(err, qt.IsNil)
 125                                         defer ci.Close()
 126                                         runBench(b, ci)
 127                                 }
 128                                 for _, journalMode := range []string{"", "wal", "off", "truncate", "delete", "persist", "memory"} {
 129                                         opts.SetJournalMode = journalMode
 130                                         b.Run("JournalMode="+journalMode, func(b *testing.B) {
 131                                                 for _, mmapSize := range []int64{-1} {
 132                                                         if memory && mmapSize >= 0 {
 133                                                                 continue
 134                                                         }
 135                                                         b.Run(fmt.Sprintf("MmapSize=%s", func() string {
 136                                                                 if mmapSize < 0 {
 137                                                                         return "default"
 138                                                                 } else {
 139                                                                         return humanize.IBytes(uint64(mmapSize))
 140                                                                 }
 141                                                         }()), func(b *testing.B) {
 142                                                                 opts.MmapSize = mmapSize
 143                                                                 opts.MmapSizeOk = true
 144                                                                 directBench(b)
 145                                                         })
 146                                                 }
 147                                         })
 148                                 }
 149                         })
 150                 })
 151         }
 152 }