storage/sqlite/direct.go

   1 //go:build cgo
   2 // +build cgo
   3
   4 package sqliteStorage
   5
   6 import (
   7         "errors"
   8         "fmt"
   9         "runtime"
  10         "sync"
  11         "time"
  12
  13         "crawshaw.io/sqlite"
  14         "crawshaw.io/sqlite/sqlitex"
  15
  16         "github.com/anacrolix/torrent/metainfo"
  17         "github.com/anacrolix/torrent/storage"
  18 )
  19
  20 type NewDirectStorageOpts struct {
  21         NewConnOpts
  22         InitDbOpts
  23         InitConnOpts
  24         GcBlobs           bool
  25         NoCacheBlobs      bool
  26         BlobFlushInterval time.Duration
  27 }
  28
  29 // A convenience function that creates a connection pool, resource provider, and a pieces storage
  30 // ClientImpl and returns them all with a Close attached.
  31 func NewDirectStorage(opts NewDirectStorageOpts) (_ storage.ClientImplCloser, err error) {
  32         conn, err := newConn(opts.NewConnOpts)
  33         if err != nil {
  34                 return
  35         }
  36         if opts.PageSize == 0 {
  37                 // The largest size sqlite supports. I think we want this to be the smallest piece size we
  38                 // can expect, which is probably 1<<17.
  39                 opts.PageSize = 1 << 16
  40         }
  41         err = initDatabase(conn, opts.InitDbOpts)
  42         if err != nil {
  43                 conn.Close()
  44                 return
  45         }
  46         err = initConn(conn, opts.InitConnOpts)
  47         if err != nil {
  48                 conn.Close()
  49                 return
  50         }
  51         if opts.BlobFlushInterval == 0 && !opts.GcBlobs {
  52                 // This is influenced by typical busy timeouts, of 5-10s. We want to give other connections
  53                 // a few chances at getting a transaction through.
  54                 opts.BlobFlushInterval = time.Second
  55         }
  56         cl := &client{
  57                 conn:  conn,
  58                 blobs: make(map[string]*sqlite.Blob),
  59                 opts:  opts,
  60         }
  61         // Avoid race with cl.blobFlusherFunc
  62         cl.l.Lock()
  63         defer cl.l.Unlock()
  64         if opts.BlobFlushInterval != 0 {
  65                 cl.blobFlusher = time.AfterFunc(opts.BlobFlushInterval, cl.blobFlusherFunc)
  66         }
  67         cl.capacity = cl.getCapacity
  68         return cl, nil
  69 }
  70
  71 func (cl *client) getCapacity() (ret *int64) {
  72         cl.l.Lock()
  73         defer cl.l.Unlock()
  74         err := sqlitex.Exec(cl.conn, "select value from setting where name='capacity'", func(stmt *sqlite.Stmt) error {
  75                 ret = new(int64)
  76                 *ret = stmt.ColumnInt64(0)
  77                 return nil
  78         })
  79         if err != nil {
  80                 panic(err)
  81         }
  82         return
  83 }
  84
  85 type client struct {
  86         l           sync.Mutex
  87         conn        conn
  88         blobs       map[string]*sqlite.Blob
  89         blobFlusher *time.Timer
  90         opts        NewDirectStorageOpts
  91         closed      bool
  92         capacity    func() *int64
  93 }
  94
  95 func (c *client) blobFlusherFunc() {
  96         c.l.Lock()
  97         defer c.l.Unlock()
  98         c.flushBlobs()
  99         if !c.closed {
 100                 c.blobFlusher.Reset(c.opts.BlobFlushInterval)
 101         }
 102 }
 103
 104 func (c *client) flushBlobs() {
 105         for key, b := range c.blobs {
 106                 // Need the lock to prevent racing with the GC finalizers.
 107                 b.Close()
 108                 delete(c.blobs, key)
 109         }
 110 }
 111
 112 func (c *client) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage.TorrentImpl, error) {
 113         t := torrent{c}
 114         return storage.TorrentImpl{Piece: t.Piece, Close: t.Close, Capacity: &c.capacity}, nil
 115 }
 116
 117 func (c *client) Close() error {
 118         c.l.Lock()
 119         defer c.l.Unlock()
 120         c.flushBlobs()
 121         c.closed = true
 122         if c.opts.BlobFlushInterval != 0 {
 123                 c.blobFlusher.Stop()
 124         }
 125         return c.conn.Close()
 126 }
 127
 128 type torrent struct {
 129         c *client
 130 }
 131
 132 func rowidForBlob(c conn, name string, length int64, create bool) (rowid int64, err error) {
 133         rowidOk := false
 134         err = sqlitex.Exec(c, "select rowid from blob where name=?", func(stmt *sqlite.Stmt) error {
 135                 if rowidOk {
 136                         panic("expected at most one row")
 137                 }
 138                 // TODO: How do we know if we got this wrong?
 139                 rowid = stmt.ColumnInt64(0)
 140                 rowidOk = true
 141                 return nil
 142         }, name)
 143         if err != nil {
 144                 return
 145         }
 146         if rowidOk {
 147                 return
 148         }
 149         if !create {
 150                 err = errors.New("no existing row")
 151                 return
 152         }
 153         err = sqlitex.Exec(c, "insert into blob(name, data) values(?, zeroblob(?))", nil, name, length)
 154         if err != nil {
 155                 return
 156         }
 157         rowid = c.LastInsertRowID()
 158         return
 159 }
 160
 161 func (t torrent) Piece(p metainfo.Piece) storage.PieceImpl {
 162         t.c.l.Lock()
 163         defer t.c.l.Unlock()
 164         name := p.Hash().HexString()
 165         return piece{
 166                 name,
 167                 p.Length(),
 168                 t.c,
 169         }
 170 }
 171
 172 func (t torrent) Close() error {
 173         return nil
 174 }
 175
 176 type piece struct {
 177         name   string
 178         length int64
 179         *client
 180 }
 181
 182 func (p piece) doAtIoWithBlob(
 183         atIo func(*sqlite.Blob) func([]byte, int64) (int, error),
 184         b []byte,
 185         off int64,
 186         create bool,
 187 ) (n int, err error) {
 188         p.l.Lock()
 189         defer p.l.Unlock()
 190         if p.opts.NoCacheBlobs {
 191                 defer p.forgetBlob()
 192         }
 193         blob, err := p.getBlob(create)
 194         if err != nil {
 195                 err = fmt.Errorf("getting blob: %w", err)
 196                 return
 197         }
 198         n, err = atIo(blob)(b, off)
 199         if err == nil {
 200                 return
 201         }
 202         var se sqlite.Error
 203         if !errors.As(err, &se) {
 204                 return
 205         }
 206         // "ABORT" occurs if the row the blob is on is modified elsewhere. "ERROR: invalid blob" occurs
 207         // if the blob has been closed. We don't forget blobs that are closed by our GC finalizers,
 208         // because they may be attached to names that have since moved on to another blob.
 209         if se.Code != sqlite.SQLITE_ABORT && !(p.opts.GcBlobs && se.Code == sqlite.SQLITE_ERROR && se.Msg == "invalid blob") {
 210                 return
 211         }
 212         p.forgetBlob()
 213         // Try again, this time we're guaranteed to get a fresh blob, and so errors are no excuse. It
 214         // might be possible to skip to this version if we don't cache blobs.
 215         blob, err = p.getBlob(create)
 216         if err != nil {
 217                 err = fmt.Errorf("getting blob: %w", err)
 218                 return
 219         }
 220         return atIo(blob)(b, off)
 221 }
 222
 223 func (p piece) ReadAt(b []byte, off int64) (n int, err error) {
 224         return p.doAtIoWithBlob(func(blob *sqlite.Blob) func([]byte, int64) (int, error) {
 225                 return blob.ReadAt
 226         }, b, off, false)
 227 }
 228
 229 func (p piece) WriteAt(b []byte, off int64) (n int, err error) {
 230         return p.doAtIoWithBlob(func(blob *sqlite.Blob) func([]byte, int64) (int, error) {
 231                 return blob.WriteAt
 232         }, b, off, true)
 233 }
 234
 235 func (p piece) MarkComplete() error {
 236         p.l.Lock()
 237         defer p.l.Unlock()
 238         err := sqlitex.Exec(p.conn, "update blob set verified=true where name=?", nil, p.name)
 239         if err != nil {
 240                 return err
 241         }
 242         changes := p.conn.Changes()
 243         if changes != 1 {
 244                 panic(changes)
 245         }
 246         return nil
 247 }
 248
 249 func (p piece) forgetBlob() {
 250         blob, ok := p.blobs[p.name]
 251         if !ok {
 252                 return
 253         }
 254         blob.Close()
 255         delete(p.blobs, p.name)
 256 }
 257
 258 func (p piece) MarkNotComplete() error {
 259         p.l.Lock()
 260         defer p.l.Unlock()
 261         return sqlitex.Exec(p.conn, "update blob set verified=false where name=?", nil, p.name)
 262 }
 263
 264 func (p piece) Completion() (ret storage.Completion) {
 265         p.l.Lock()
 266         defer p.l.Unlock()
 267         err := sqlitex.Exec(p.conn, "select verified from blob where name=?", func(stmt *sqlite.Stmt) error {
 268                 ret.Complete = stmt.ColumnInt(0) != 0
 269                 return nil
 270         }, p.name)
 271         ret.Ok = err == nil
 272         if err != nil {
 273                 panic(err)
 274         }
 275         return
 276 }
 277
 278 func (p piece) getBlob(create bool) (*sqlite.Blob, error) {
 279         blob, ok := p.blobs[p.name]
 280         if !ok {
 281                 rowid, err := rowidForBlob(p.conn, p.name, p.length, create)
 282                 if err != nil {
 283                         return nil, fmt.Errorf("getting rowid for blob: %w", err)
 284                 }
 285                 blob, err = p.conn.OpenBlob("main", "blob", "data", rowid, true)
 286                 if err != nil {
 287                         panic(err)
 288                 }
 289                 if p.opts.GcBlobs {
 290                         herp := new(byte)
 291                         runtime.SetFinalizer(herp, func(*byte) {
 292                                 p.l.Lock()
 293                                 defer p.l.Unlock()
 294                                 // Note there's no guarantee that the finalizer fired while this blob is the same
 295                                 // one in the blob cache. It might be possible to rework this so that we check, or
 296                                 // strip finalizers as appropriate.
 297                                 blob.Close()
 298                         })
 299                 }
 300                 p.blobs[p.name] = blob
 301         }
 302         return blob, nil
 303 }