storage/sqlite/sqlite-storage_test.go

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