storage/sqlite/sqlite-storage.go

   1 //go:build cgo
   2 // +build cgo
   3
   4 package sqliteStorage
   5
   6 import (
   7         "bytes"
   8         "context"
   9         "errors"
  10         "expvar"
  11         "fmt"
  12         "io"
  13         "log"
  14         "net/url"
  15         "os"
  16         "runtime"
  17         "runtime/pprof"
  18         "strings"
  19         "sync"
  20         "time"
  21
  22         "crawshaw.io/sqlite"
  23         "crawshaw.io/sqlite/sqlitex"
  24         "github.com/anacrolix/missinggo/iter"
  25         "github.com/anacrolix/missinggo/v2/resource"
  26
  27         "github.com/anacrolix/torrent/storage"
  28 )
  29
  30 type conn = *sqlite.Conn
  31
  32 type InitConnOpts struct {
  33         SetSynchronous int
  34         SetJournalMode string
  35         MmapSizeOk     bool  // If false, a package-specific default will be used.
  36         MmapSize       int64 // If MmapSizeOk is set, use sqlite default if < 0, otherwise this value.
  37 }
  38
  39 type UnexpectedJournalMode struct {
  40         JournalMode string
  41 }
  42
  43 func (me UnexpectedJournalMode) Error() string {
  44         return fmt.Sprintf("unexpected journal mode: %q", me.JournalMode)
  45 }
  46
  47 func setSynchronous(conn conn, syncInt int) (err error) {
  48         err = sqlitex.ExecTransient(conn, fmt.Sprintf(`pragma synchronous=%v`, syncInt), nil)
  49         if err != nil {
  50                 return err
  51         }
  52         var (
  53                 actual   int
  54                 actualOk bool
  55         )
  56         err = sqlitex.ExecTransient(conn, `pragma synchronous`, func(stmt *sqlite.Stmt) error {
  57                 actual = stmt.ColumnInt(0)
  58                 actualOk = true
  59                 return nil
  60         })
  61         if err != nil {
  62                 return
  63         }
  64         if !actualOk {
  65                 return errors.New("synchronous setting query didn't return anything")
  66         }
  67         if actual != syncInt {
  68                 return fmt.Errorf("set synchronous %q, got %q", syncInt, actual)
  69         }
  70         return nil
  71 }
  72
  73 func initConn(conn conn, opts InitConnOpts) (err error) {
  74         err = setSynchronous(conn, opts.SetSynchronous)
  75         if err != nil {
  76                 return
  77         }
  78         // Recursive triggers are required because we need to trim the blob_meta size after trimming to
  79         // capacity. Hopefully we don't hit the recursion limit, and if we do, there's an error thrown.
  80         err = sqlitex.ExecTransient(conn, "pragma recursive_triggers=on", nil)
  81         if err != nil {
  82                 return err
  83         }
  84         if opts.SetJournalMode != "" {
  85                 err = sqlitex.ExecTransient(conn, fmt.Sprintf(`pragma journal_mode=%s`, opts.SetJournalMode), func(stmt *sqlite.Stmt) error {
  86                         ret := stmt.ColumnText(0)
  87                         if ret != opts.SetJournalMode {
  88                                 return UnexpectedJournalMode{ret}
  89                         }
  90                         return nil
  91                 })
  92                 if err != nil {
  93                         return err
  94                 }
  95         }
  96         if !opts.MmapSizeOk {
  97                 // Set the default. Currently it seems the library picks reasonable defaults, especially for
  98                 // wal.
  99                 opts.MmapSize = -1
 100                 //opts.MmapSize = 1 << 24 // 8 MiB
 101         }
 102         if opts.MmapSize >= 0 {
 103                 err = sqlitex.ExecTransient(conn, fmt.Sprintf(`pragma mmap_size=%d`, opts.MmapSize), nil)
 104                 if err != nil {
 105                         return err
 106                 }
 107         }
 108         return nil
 109 }
 110
 111 func setPageSize(conn conn, pageSize int) error {
 112         if pageSize == 0 {
 113                 return nil
 114         }
 115         var retSize int64
 116         err := sqlitex.ExecTransient(conn, fmt.Sprintf(`pragma page_size=%d`, pageSize), nil)
 117         if err != nil {
 118                 return err
 119         }
 120         err = sqlitex.ExecTransient(conn, "pragma page_size", func(stmt *sqlite.Stmt) error {
 121                 retSize = stmt.ColumnInt64(0)
 122                 return nil
 123         })
 124         if err != nil {
 125                 return err
 126         }
 127         if retSize != int64(pageSize) {
 128                 return fmt.Errorf("requested page size %v but got %v", pageSize, retSize)
 129         }
 130         return nil
 131 }
 132
 133 func InitSchema(conn conn, pageSize int, triggers bool) error {
 134         err := setPageSize(conn, pageSize)
 135         if err != nil {
 136                 return fmt.Errorf("setting page size: %w", err)
 137         }
 138         err = sqlitex.ExecScript(conn, `
 139                 -- We have to opt into this before creating any tables, or before a vacuum to enable it. It means we
 140                 -- can trim the database file size with partial vacuums without having to do a full vacuum, which
 141                 -- locks everything.
 142                 pragma auto_vacuum=incremental;
 143
 144                 create table if not exists blob (
 145                         name text,
 146                         last_used timestamp default (datetime('now')),
 147                         data blob,
 148                         verified bool,
 149                         primary key (name)
 150                 );
 151
 152                 create table if not exists blob_meta (
 153                         key text primary key,
 154                         value
 155                 );
 156
 157                 create index if not exists blob_last_used on blob(last_used);
 158
 159                 -- While sqlite *seems* to be faster to get sum(length(data)) instead of
 160                 -- sum(length(data)), it may still require a large table scan at start-up or with a
 161                 -- cold-cache. With this we can be assured that it doesn't.
 162                 insert or ignore into blob_meta values ('size', 0);
 163
 164                 create table if not exists setting (
 165                         name primary key on conflict replace,
 166                         value
 167                 );
 168
 169                 create view if not exists deletable_blob as
 170                 with recursive excess (
 171                         usage_with,
 172                         last_used,
 173                         blob_rowid,
 174                         data_length
 175                 ) as (
 176                         select *
 177                         from (
 178                                 select
 179                                         (select value from blob_meta where key='size') as usage_with,
 180                                         last_used,
 181                                         rowid,
 182                                         length(data)
 183                                 from blob order by last_used, rowid limit 1
 184                         )
 185                         where usage_with > (select value from setting where name='capacity')
 186                         union all
 187                         select
 188                                 usage_with-data_length as new_usage_with,
 189                                 blob.last_used,
 190                                 blob.rowid,
 191                                 length(data)
 192                         from excess join blob
 193                         on blob.rowid=(select rowid from blob where (last_used, rowid) > (excess.last_used, blob_rowid))
 194                         where new_usage_with > (select value from setting where name='capacity')
 195                 )
 196                 select * from excess;
 197         `)
 198         if err != nil {
 199                 return err
 200         }
 201         if triggers {
 202                 err := sqlitex.ExecScript(conn, `
 203                         create trigger if not exists after_insert_blob
 204                         after insert on blob
 205                         begin
 206                                 update blob_meta set value=value+length(cast(new.data as blob)) where key='size';
 207                                 delete from blob where rowid in (select blob_rowid from deletable_blob);
 208                         end;
 209
 210                         create trigger if not exists after_update_blob
 211                         after update of data on blob
 212                         begin
 213                                 update blob_meta set value=value+length(cast(new.data as blob))-length(cast(old.data as blob)) where key='size';
 214                                 delete from blob where rowid in (select blob_rowid from deletable_blob);
 215                         end;
 216
 217                         create trigger if not exists after_delete_blob
 218                         after delete on blob
 219                         begin
 220                                 update blob_meta set value=value-length(cast(old.data as blob)) where key='size';
 221                         end;
 222                 `)
 223                 if err != nil {
 224                         return err
 225                 }
 226         }
 227         return nil
 228 }
 229
 230 type NewPiecesStorageOpts struct {
 231         NewPoolOpts
 232         InitDbOpts
 233         ProvOpts    func(*ProviderOpts)
 234         StorageOpts func(*storage.ResourcePiecesOpts)
 235 }
 236
 237 type NewPoolOpts struct {
 238         NewConnOpts
 239         InitConnOpts
 240         NumConns int
 241 }
 242
 243 type InitDbOpts struct {
 244         DontInitSchema bool
 245         PageSize       int
 246         // If non-zero, overrides the existing setting.
 247         Capacity   int64
 248         NoTriggers bool
 249 }
 250
 251 // There's some overlap here with NewPoolOpts, and I haven't decided what needs to be done. For now,
 252 // the fact that the pool opts are a superset, means our helper NewPiecesStorage can just take the
 253 // top-level option type.
 254 type PoolConf struct {
 255         NumConns    int
 256         JournalMode string
 257 }
 258
 259 // Remove any capacity limits.
 260 func UnlimitCapacity(conn conn) error {
 261         return sqlitex.Exec(conn, "delete from setting where key='capacity'", nil)
 262 }
 263
 264 // Set the capacity limit to exactly this value.
 265 func SetCapacity(conn conn, cap int64) error {
 266         return sqlitex.Exec(conn, "insert into setting values ('capacity', ?)", nil, cap)
 267 }
 268
 269 type NewConnOpts struct {
 270         // See https://www.sqlite.org/c3ref/open.html. NB: "If the filename is an empty string, then a
 271         // private, temporary on-disk database will be created. This private database will be
 272         // automatically deleted as soon as the database connection is closed."
 273         Path   string
 274         Memory bool
 275         // Whether multiple blobs will not be read simultaneously. Enables journal mode other than WAL,
 276         // and NumConns < 2.
 277         NoConcurrentBlobReads bool
 278 }
 279
 280 func newOpenUri(opts NewConnOpts) string {
 281         path := url.PathEscape(opts.Path)
 282         if opts.Memory {
 283                 path = ":memory:"
 284         }
 285         values := make(url.Values)
 286         if opts.NoConcurrentBlobReads || opts.Memory {
 287                 values.Add("cache", "shared")
 288         }
 289         return fmt.Sprintf("file:%s?%s", path, values.Encode())
 290 }
 291
 292 func initDatabase(conn conn, opts InitDbOpts) (err error) {
 293         if !opts.DontInitSchema {
 294                 err = InitSchema(conn, opts.PageSize, !opts.NoTriggers)
 295                 if err != nil {
 296                         return
 297                 }
 298         }
 299         if opts.Capacity != 0 {
 300                 err = SetCapacity(conn, opts.Capacity)
 301                 if err != nil {
 302                         return
 303                 }
 304         }
 305         return
 306 }
 307
 308 func initPoolDatabase(pool ConnPool, opts InitDbOpts) (err error) {
 309         withPoolConn(pool, func(c conn) {
 310                 err = initDatabase(c, opts)
 311         })
 312         return
 313 }
 314
 315 // Go fmt, why you so shit?
 316 const openConnFlags = 0 |
 317         sqlite.SQLITE_OPEN_READWRITE |
 318         sqlite.SQLITE_OPEN_CREATE |
 319         sqlite.SQLITE_OPEN_URI |
 320         sqlite.SQLITE_OPEN_NOMUTEX
 321
 322 func newConn(opts NewConnOpts) (conn, error) {
 323         return sqlite.OpenConn(newOpenUri(opts), openConnFlags)
 324 }
 325
 326 type poolWithNumConns struct {
 327         *sqlitex.Pool
 328         numConns int
 329 }
 330
 331 func (me poolWithNumConns) NumConns() int {
 332         return me.numConns
 333 }
 334
 335 func NewPool(opts NewPoolOpts) (_ ConnPool, err error) {
 336         if opts.NumConns == 0 {
 337                 opts.NumConns = runtime.NumCPU()
 338         }
 339         switch opts.NumConns {
 340         case 1:
 341                 conn, err := newConn(opts.NewConnOpts)
 342                 return &poolFromConn{conn: conn}, err
 343         default:
 344                 _pool, err := sqlitex.Open(newOpenUri(opts.NewConnOpts), openConnFlags, opts.NumConns)
 345                 return poolWithNumConns{_pool, opts.NumConns}, err
 346         }
 347 }
 348
 349 // Emulates a ConnPool from a single Conn. Might be faster than using a sqlitex.Pool.
 350 type poolFromConn struct {
 351         mu   sync.Mutex
 352         conn conn
 353 }
 354
 355 func (me *poolFromConn) Get(ctx context.Context) conn {
 356         me.mu.Lock()
 357         return me.conn
 358 }
 359
 360 func (me *poolFromConn) Put(conn conn) {
 361         if conn != me.conn {
 362                 panic("expected to same conn")
 363         }
 364         me.mu.Unlock()
 365 }
 366
 367 func (me *poolFromConn) Close() error {
 368         return me.conn.Close()
 369 }
 370
 371 func (poolFromConn) NumConns() int { return 1 }
 372
 373 type ProviderOpts struct {
 374         BatchWrites bool
 375 }
 376
 377 // Needs the ConnPool size so it can initialize all the connections with pragmas. Takes ownership of
 378 // the ConnPool (since it has to initialize all the connections anyway).
 379 func NewProvider(pool ConnPool, opts ProviderOpts) (_ *provider, err error) {
 380         prov := &provider{pool: pool, opts: opts}
 381         if opts.BatchWrites {
 382                 writes := make(chan writeRequest)
 383                 prov.writes = writes
 384                 // This is retained for backwards compatibility. It may not be necessary.
 385                 runtime.SetFinalizer(prov, func(p *provider) {
 386                         p.Close()
 387                 })
 388                 go providerWriter(writes, prov.pool)
 389         }
 390         return prov, nil
 391 }
 392
 393 type InitPoolOpts struct {
 394         NumConns int
 395         InitConnOpts
 396 }
 397
 398 func initPoolConns(ctx context.Context, pool ConnPool, opts InitConnOpts) (err error) {
 399         var conns []conn
 400         defer func() {
 401                 for _, c := range conns {
 402                         pool.Put(c)
 403                 }
 404         }()
 405         for range iter.N(pool.NumConns()) {
 406                 conn := pool.Get(ctx)
 407                 if conn == nil {
 408                         break
 409                 }
 410                 conns = append(conns, conn)
 411                 err = initConn(conn, opts)
 412                 if err != nil {
 413                         err = fmt.Errorf("initing conn %v: %w", len(conns), err)
 414                         return
 415                 }
 416         }
 417         return
 418 }
 419
 420 type ConnPool interface {
 421         Get(context.Context) conn
 422         Put(conn)
 423         Close() error
 424         NumConns() int
 425 }
 426
 427 func withPoolConn(pool ConnPool, with func(conn)) {
 428         c := pool.Get(context.TODO())
 429         defer pool.Put(c)
 430         with(c)
 431 }
 432
 433 type provider struct {
 434         pool     ConnPool
 435         writes   chan<- writeRequest
 436         opts     ProviderOpts
 437         closeMu  sync.RWMutex
 438         closed   bool
 439         closeErr error
 440 }
 441
 442 var _ storage.ConsecutiveChunkReader = (*provider)(nil)
 443
 444 func (p *provider) ReadConsecutiveChunks(prefix string) (io.ReadCloser, error) {
 445         p.closeMu.RLock()
 446         runner, err := p.getReadWithConnRunner()
 447         if err != nil {
 448                 p.closeMu.RUnlock()
 449                 return nil, err
 450         }
 451         r, w := io.Pipe()
 452         go func() {
 453                 defer p.closeMu.RUnlock()
 454                 err = runner(func(_ context.Context, conn conn) error {
 455                         var written int64
 456                         err = sqlitex.Exec(conn, `
 457                                 select
 458                                         data,
 459                                         cast(substr(name, ?+1) as integer) as offset
 460                                 from blob
 461                                 where name like ?||'%'
 462                                 order by offset`,
 463                                 func(stmt *sqlite.Stmt) error {
 464                                         offset := stmt.ColumnInt64(1)
 465                                         if offset != written {
 466                                                 return fmt.Errorf("got chunk at offset %v, expected offset %v", offset, written)
 467                                         }
 468                                         // TODO: Avoid intermediate buffers here
 469                                         r := stmt.ColumnReader(0)
 470                                         w1, err := io.Copy(w, r)
 471                                         written += w1
 472                                         return err
 473                                 },
 474                                 len(prefix),
 475                                 prefix,
 476                         )
 477                         return err
 478                 })
 479                 w.CloseWithError(err)
 480         }()
 481         return r, nil
 482 }
 483
 484 func (me *provider) Close() error {
 485         me.closeMu.Lock()
 486         defer me.closeMu.Unlock()
 487         if me.closed {
 488                 return me.closeErr
 489         }
 490         if me.writes != nil {
 491                 close(me.writes)
 492         }
 493         me.closeErr = me.pool.Close()
 494         me.closed = true
 495         return me.closeErr
 496 }
 497
 498 type writeRequest struct {
 499         query  withConn
 500         done   chan<- error
 501         labels pprof.LabelSet
 502 }
 503
 504 var expvars = expvar.NewMap("sqliteStorage")
 505
 506 func runQueryWithLabels(query withConn, labels pprof.LabelSet, conn conn) (err error) {
 507         pprof.Do(context.Background(), labels, func(ctx context.Context) {
 508                 // We pass in the context in the hope that the CPU profiler might incorporate sqlite
 509                 // activity the action that triggered it. It doesn't seem that way, since those calls don't
 510                 // take a context.Context themselves. It may come in useful in the goroutine profiles
 511                 // though, and doesn't hurt to expose it here for other purposes should things change.
 512                 err = query(ctx, conn)
 513         })
 514         return
 515 }
 516
 517 // Intentionally avoids holding a reference to *provider to allow it to use a finalizer, and to have
 518 // stronger typing on the writes channel.
 519 func providerWriter(writes <-chan writeRequest, pool ConnPool) {
 520         conn := pool.Get(context.TODO())
 521         if conn == nil {
 522                 return
 523         }
 524         defer pool.Put(conn)
 525         for {
 526                 first, ok := <-writes
 527                 if !ok {
 528                         return
 529                 }
 530                 var buf []func()
 531                 var cantFail error
 532                 func() {
 533                         defer sqlitex.Save(conn)(&cantFail)
 534                         firstErr := runQueryWithLabels(first.query, first.labels, conn)
 535                         buf = append(buf, func() { first.done <- firstErr })
 536                         for {
 537                                 select {
 538                                 case wr, ok := <-writes:
 539                                         if ok {
 540                                                 err := runQueryWithLabels(wr.query, wr.labels, conn)
 541                                                 buf = append(buf, func() { wr.done <- err })
 542                                                 continue
 543                                         }
 544                                 default:
 545                                 }
 546                                 break
 547                         }
 548                 }()
 549                 // Not sure what to do if this failed.
 550                 if cantFail != nil {
 551                         expvars.Add("batchTransactionErrors", 1)
 552                 }
 553                 // Signal done after we know the transaction succeeded.
 554                 for _, done := range buf {
 555                         done()
 556                 }
 557                 expvars.Add("batchTransactions", 1)
 558                 expvars.Add("batchedQueries", int64(len(buf)))
 559                 //log.Printf("batched %v write queries", len(buf))
 560         }
 561 }
 562
 563 func (p *provider) NewInstance(s string) (resource.Instance, error) {
 564         return instance{s, p}, nil
 565 }
 566
 567 type instance struct {
 568         location string
 569         p        *provider
 570 }
 571
 572 func getLabels(skip int) pprof.LabelSet {
 573         return pprof.Labels("sqlite-storage-action", func() string {
 574                 var pcs [8]uintptr
 575                 runtime.Callers(skip+3, pcs[:])
 576                 fs := runtime.CallersFrames(pcs[:])
 577                 f, _ := fs.Next()
 578                 funcName := f.Func.Name()
 579                 funcName = funcName[strings.LastIndexByte(funcName, '.')+1:]
 580                 //log.Printf("func name: %q", funcName)
 581                 return funcName
 582         }())
 583 }
 584
 585 func (p *provider) withConn(with withConn, write bool, skip int) error {
 586         p.closeMu.RLock()
 587         // I think we need to check this here because it may not be valid to send to the writes channel
 588         // if we're already closed. So don't try to move this check into getReadWithConnRunner.
 589         if p.closed {
 590                 p.closeMu.RUnlock()
 591                 return errors.New("closed")
 592         }
 593         if write && p.opts.BatchWrites {
 594                 done := make(chan error)
 595                 p.writes <- writeRequest{
 596                         query:  with,
 597                         done:   done,
 598                         labels: getLabels(skip + 1),
 599                 }
 600                 p.closeMu.RUnlock()
 601                 return <-done
 602         } else {
 603                 defer p.closeMu.RUnlock()
 604                 runner, err := p.getReadWithConnRunner()
 605                 if err != nil {
 606                         return err
 607                 }
 608                 return runner(with)
 609         }
 610 }
 611
 612 // Obtains a DB conn and returns a withConn for executing with it. If no error is returned from this
 613 // function, the runner *must* be used or the conn is leaked. You should check the provider isn't
 614 // closed before using this.
 615 func (p *provider) getReadWithConnRunner() (with func(withConn) error, err error) {
 616         conn := p.pool.Get(context.TODO())
 617         if conn == nil {
 618                 err = errors.New("couldn't get pool conn")
 619                 return
 620         }
 621         with = func(with withConn) error {
 622                 defer p.pool.Put(conn)
 623                 return runQueryWithLabels(with, getLabels(1), conn)
 624         }
 625         return
 626 }
 627
 628 type withConn func(context.Context, conn) error
 629
 630 func (i instance) withConn(with withConn, write bool) error {
 631         return i.p.withConn(with, write, 1)
 632 }
 633
 634 func (i instance) getConn() *sqlite.Conn {
 635         return i.p.pool.Get(context.TODO())
 636 }
 637
 638 func (i instance) putConn(conn *sqlite.Conn) {
 639         i.p.pool.Put(conn)
 640 }
 641
 642 func (i instance) Readdirnames() (names []string, err error) {
 643         prefix := i.location + "/"
 644         err = i.withConn(func(_ context.Context, conn conn) error {
 645                 return sqlitex.Exec(conn, "select name from blob where name like ?", func(stmt *sqlite.Stmt) error {
 646                         names = append(names, stmt.ColumnText(0)[len(prefix):])
 647                         return nil
 648                 }, prefix+"%")
 649         }, false)
 650         //log.Printf("readdir %q gave %q", i.location, names)
 651         return
 652 }
 653
 654 func (i instance) getBlobRowid(conn conn) (rowid int64, err error) {
 655         rows := 0
 656         err = sqlitex.Exec(conn, "select rowid from blob where name=?", func(stmt *sqlite.Stmt) error {
 657                 rowid = stmt.ColumnInt64(0)
 658                 rows++
 659                 return nil
 660         }, i.location)
 661         if err != nil {
 662                 return
 663         }
 664         if rows == 1 {
 665                 return
 666         }
 667         if rows == 0 {
 668                 err = errors.New("blob not found")
 669                 return
 670         }
 671         panic(rows)
 672 }
 673
 674 type connBlob struct {
 675         *sqlite.Blob
 676         onClose func()
 677 }
 678
 679 func (me connBlob) Close() error {
 680         err := me.Blob.Close()
 681         me.onClose()
 682         return err
 683 }
 684
 685 func (i instance) Get() (ret io.ReadCloser, err error) {
 686         conn := i.getConn()
 687         if conn == nil {
 688                 panic("nil sqlite conn")
 689         }
 690         blob, err := i.openBlob(conn, false, true)
 691         if err != nil {
 692                 i.putConn(conn)
 693                 return
 694         }
 695         var once sync.Once
 696         return connBlob{blob, func() {
 697                 once.Do(func() { i.putConn(conn) })
 698         }}, nil
 699 }
 700
 701 func (i instance) openBlob(conn conn, write, updateAccess bool) (*sqlite.Blob, error) {
 702         rowid, err := i.getBlobRowid(conn)
 703         if err != nil {
 704                 return nil, err
 705         }
 706         // This seems to cause locking issues with in-memory databases. Is it something to do with not
 707         // having WAL?
 708         if updateAccess {
 709                 err = sqlitex.Exec(conn, "update blob set last_used=datetime('now') where rowid=?", nil, rowid)
 710                 if err != nil {
 711                         err = fmt.Errorf("updating last_used: %w", err)
 712                         return nil, err
 713                 }
 714                 if conn.Changes() != 1 {
 715                         panic(conn.Changes())
 716                 }
 717         }
 718         return conn.OpenBlob("main", "blob", "data", rowid, write)
 719 }
 720
 721 func (i instance) PutSized(reader io.Reader, size int64) (err error) {
 722         err = i.withConn(func(_ context.Context, conn conn) error {
 723                 err := sqlitex.Exec(conn, "insert or replace into blob(name, data) values(?, zeroblob(?))",
 724                         nil,
 725                         i.location, size)
 726                 if err != nil {
 727                         return err
 728                 }
 729                 blob, err := i.openBlob(conn, true, false)
 730                 if err != nil {
 731                         return err
 732                 }
 733                 defer blob.Close()
 734                 _, err = io.Copy(blob, reader)
 735                 return err
 736         }, true)
 737         return
 738 }
 739
 740 func (i instance) Put(reader io.Reader) (err error) {
 741         var buf bytes.Buffer
 742         _, err = io.Copy(&buf, reader)
 743         if err != nil {
 744                 return err
 745         }
 746         if false {
 747                 return i.PutSized(&buf, int64(buf.Len()))
 748         } else {
 749                 return i.withConn(func(_ context.Context, conn conn) error {
 750                         for range iter.N(10) {
 751                                 err = sqlitex.Exec(conn,
 752                                         "insert or replace into blob(name, data) values(?, cast(? as blob))",
 753                                         nil,
 754                                         i.location, buf.Bytes())
 755                                 if err, ok := err.(sqlite.Error); ok && err.Code == sqlite.SQLITE_BUSY {
 756                                         log.Print("sqlite busy")
 757                                         time.Sleep(time.Second)
 758                                         continue
 759                                 }
 760                                 break
 761                         }
 762                         return err
 763                 }, true)
 764         }
 765 }
 766
 767 type fileInfo struct {
 768         size int64
 769 }
 770
 771 func (f fileInfo) Name() string {
 772         panic("implement me")
 773 }
 774
 775 func (f fileInfo) Size() int64 {
 776         return f.size
 777 }
 778
 779 func (f fileInfo) Mode() os.FileMode {
 780         panic("implement me")
 781 }
 782
 783 func (f fileInfo) ModTime() time.Time {
 784         panic("implement me")
 785 }
 786
 787 func (f fileInfo) IsDir() bool {
 788         panic("implement me")
 789 }
 790
 791 func (f fileInfo) Sys() interface{} {
 792         panic("implement me")
 793 }
 794
 795 func (i instance) Stat() (ret os.FileInfo, err error) {
 796         err = i.withConn(func(_ context.Context, conn conn) error {
 797                 var blob *sqlite.Blob
 798                 blob, err = i.openBlob(conn, false, false)
 799                 if err != nil {
 800                         return err
 801                 }
 802                 defer blob.Close()
 803                 ret = fileInfo{blob.Size()}
 804                 return nil
 805         }, false)
 806         return
 807 }
 808
 809 func (i instance) ReadAt(p []byte, off int64) (n int, err error) {
 810         err = i.withConn(func(_ context.Context, conn conn) error {
 811                 if false {
 812                         var blob *sqlite.Blob
 813                         blob, err = i.openBlob(conn, false, true)
 814                         if err != nil {
 815                                 return err
 816                         }
 817                         defer blob.Close()
 818                         if off >= blob.Size() {
 819                                 err = io.EOF
 820                                 return err
 821                         }
 822                         if off+int64(len(p)) > blob.Size() {
 823                                 p = p[:blob.Size()-off]
 824                         }
 825                         n, err = blob.ReadAt(p, off)
 826                 } else {
 827                         gotRow := false
 828                         err = sqlitex.Exec(
 829                                 conn,
 830                                 "select substr(data, ?, ?) from blob where name=?",
 831                                 func(stmt *sqlite.Stmt) error {
 832                                         if gotRow {
 833                                                 panic("found multiple matching blobs")
 834                                         } else {
 835                                                 gotRow = true
 836                                         }
 837                                         n = stmt.ColumnBytes(0, p)
 838                                         return nil
 839                                 },
 840                                 off+1, len(p), i.location,
 841                         )
 842                         if err != nil {
 843                                 return err
 844                         }
 845                         if !gotRow {
 846                                 err = errors.New("blob not found")
 847                                 return err
 848                         }
 849                         if n < len(p) {
 850                                 err = io.EOF
 851                         }
 852                 }
 853                 return nil
 854         }, false)
 855         return
 856 }
 857
 858 func (i instance) WriteAt(bytes []byte, i2 int64) (int, error) {
 859         panic("implement me")
 860 }
 861
 862 func (i instance) Delete() error {
 863         return i.withConn(func(_ context.Context, conn conn) error {
 864                 return sqlitex.Exec(conn, "delete from blob where name=?", nil, i.location)
 865         }, true)
 866 }