flatfs.go 25.9 KB
Newer Older
Tommi Virtanen's avatar
Tommi Virtanen committed
1 2 3 4 5 6
// Package flatfs is a Datastore implementation that stores all
// objects in a two-level directory structure in the local file
// system, regardless of the hierarchy of the keys.
package flatfs

import (
7
	"encoding/json"
Tommi Virtanen's avatar
Tommi Virtanen committed
8
	"errors"
9
	"fmt"
Tommi Virtanen's avatar
Tommi Virtanen committed
10
	"io/ioutil"
11 12
	"math"
	"math/rand"
Tommi Virtanen's avatar
Tommi Virtanen committed
13
	"os"
Jeromy's avatar
Jeromy committed
14
	"path/filepath"
Tommi Virtanen's avatar
Tommi Virtanen committed
15
	"strings"
16 17
	"sync"
	"sync/atomic"
18
	"time"
Tommi Virtanen's avatar
Tommi Virtanen committed
19

Jeromy's avatar
Jeromy committed
20 21
	"github.com/ipfs/go-datastore"
	"github.com/ipfs/go-datastore/query"
Steven Allen's avatar
Steven Allen committed
22
	"github.com/jbenet/goprocess"
23

Jakub Sztandera's avatar
Jakub Sztandera committed
24
	logging "github.com/ipfs/go-log"
Tommi Virtanen's avatar
Tommi Virtanen committed
25 26
)

27 28
var log = logging.Logger("flatfs")

Tommi Virtanen's avatar
Tommi Virtanen committed
29
const (
30
	extension                  = ".data"
31
	diskUsageMessageTimeout    = 5 * time.Second
32
	diskUsageCheckpointPercent = 1.0
33
	diskUsageCheckpointTimeout = 2 * time.Second
Tommi Virtanen's avatar
Tommi Virtanen committed
34 35
)

36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59
var (
	// DiskUsageFile is the name of the file to cache the size of the
	// datastore in disk
	DiskUsageFile = "diskUsage.cache"
	// DiskUsageFilesAverage is the maximum number of files per folder
	// to stat in order to calculate the size of the datastore.
	// The size of the rest of the files in a folder will be assumed
	// to be the average of the values obtained. This includes
	// regular files and directories.
	DiskUsageFilesAverage = 2000
	// DiskUsageCalcTimeout is the maximum time to spend
	// calculating the DiskUsage upon a start when no
	// DiskUsageFile is present.
	// If this period did not suffice to read the size of the datastore,
	// the remaining sizes will be stimated.
	DiskUsageCalcTimeout = 5 * time.Minute
)

const (
	opPut = iota
	opDelete
	opRename
)

60 61 62
type initAccuracy string

const (
63
	unknownA  initAccuracy = "unknown"
64 65 66 67 68 69
	exactA    initAccuracy = "initial-exact"
	approxA   initAccuracy = "initial-approximate"
	timedoutA initAccuracy = "initial-timed-out"
)

func combineAccuracy(a, b initAccuracy) initAccuracy {
70 71 72
	if a == unknownA || b == unknownA {
		return unknownA
	}
73 74 75 76 77 78 79 80 81
	if a == timedoutA || b == timedoutA {
		return timedoutA
	}
	if a == approxA || b == approxA {
		return approxA
	}
	if a == exactA && b == exactA {
		return exactA
	}
82 83 84 85 86 87 88
	if a == "" {
		return b
	}
	if b == "" {
		return a
	}
	return unknownA
89 90
}

91 92 93 94 95 96
var _ datastore.Datastore = (*Datastore)(nil)

var (
	ErrDatastoreExists       = errors.New("datastore already exists")
	ErrDatastoreDoesNotExist = errors.New("datastore directory does not exist")
	ErrShardingFileMissing   = fmt.Errorf("%s file not found in datastore", SHARDING_FN)
Steven Allen's avatar
Steven Allen committed
97
	ErrClosed                = errors.New("datastore closed")
98
	ErrInvalidKey            = errors.New("key not supported by flatfs")
99 100 101 102 103 104 105 106 107 108
)

func init() {
	rand.Seed(time.Now().UTC().UnixNano())
}

// Datastore implements the go-datastore Interface.
// Note this datastore cannot guarantee order of concurrent
// write operations to the same key. See the explanation in
// Put().
Tommi Virtanen's avatar
Tommi Virtanen committed
109
type Datastore struct {
110 111 112 113 114
	// atmoic operations should always be used with diskUsage.
	// Must be first in struct to ensure correct alignment
	// (see https://golang.org/pkg/sync/atomic/#pkg-note-BUG)
	diskUsage int64

Tommi Virtanen's avatar
Tommi Virtanen committed
115
	path string
116

117 118
	shardStr string
	getDir   ShardFunc
Jeromy's avatar
Jeromy committed
119 120 121

	// sychronize all writes and directory changes for added safety
	sync bool
122

123 124 125
	// these values should only be used during internalization or
	// inside the checkpoint loop
	dirty       bool
126
	storedValue diskUsageValue
127

Steven Allen's avatar
Steven Allen committed
128
	// Used to trigger a checkpoint.
Kevin Atkinson's avatar
Kevin Atkinson committed
129 130
	checkpointCh chan struct{}
	done         chan struct{}
131

Steven Allen's avatar
Steven Allen committed
132 133 134
	shutdownLock sync.RWMutex
	shutdown     bool

135 136 137
	// opMap handles concurrent write operations (put/delete)
	// to the same key
	opMap *opMap
Tommi Virtanen's avatar
Tommi Virtanen committed
138 139
}

140
type diskUsageValue struct {
141 142
	DiskUsage int64        `json:"diskUsage"`
	Accuracy  initAccuracy `json:"accuracy"`
143 144
}

145 146
type ShardFunc func(string) string

147
type opT int
148

149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196
// op wraps useful arguments of write operations
type op struct {
	typ  opT           // operation type
	key  datastore.Key // datastore key. Mandatory.
	tmp  string        // temp file path
	path string        // file path
	v    []byte        // value
}

type opMap struct {
	ops sync.Map
}

type opResult struct {
	mu      sync.RWMutex
	success bool

	opMap *opMap
	name  string
}

// Returns nil if there's nothing to do.
func (m *opMap) Begin(name string) *opResult {
	for {
		myOp := &opResult{opMap: m, name: name}
		myOp.mu.Lock()
		opIface, loaded := m.ops.LoadOrStore(name, myOp)
		if !loaded { // no one else doing ops with this key
			return myOp
		}

		op := opIface.(*opResult)
		// someone else doing ops with this key, wait for
		// the result
		op.mu.RLock()
		if op.success {
			return nil
		}

		// if we are here, we will retry the operation
	}
}

func (o *opResult) Finish(ok bool) {
	o.success = ok
	o.opMap.ops.Delete(o.name)
	o.mu.Unlock()
}
197

198
func Create(path string, fun *ShardIdV1) error {
199

kpcyrd's avatar
kpcyrd committed
200
	err := os.Mkdir(path, 0755)
201 202
	if err != nil && !os.IsExist(err) {
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
203 204
	}

205 206
	dsFun, err := ReadShardFunc(path)
	switch err {
Kevin Atkinson's avatar
Kevin Atkinson committed
207
	case ErrShardingFileMissing:
208 209 210 211 212
		isEmpty, err := DirIsEmpty(path)
		if err != nil {
			return err
		}
		if !isEmpty {
213
			return fmt.Errorf("directory missing %s file: %s", SHARDING_FN, path)
214 215 216
		}

		err = WriteShardFunc(path, fun)
217 218 219
		if err != nil {
			return err
		}
220 221
		err = WriteReadme(path, fun)
		return err
222
	case nil:
223
		if fun.String() != dsFun.String() {
224
			return fmt.Errorf("specified shard func '%s' does not match repo shard func '%s'",
225
				fun.String(), dsFun.String())
226
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
227
		return ErrDatastoreExists
228
	default:
229
		return err
230
	}
231 232
}

233
func Open(path string, syncFiles bool) (*Datastore, error) {
234 235
	_, err := os.Stat(path)
	if os.IsNotExist(err) {
Kevin Atkinson's avatar
Kevin Atkinson committed
236
		return nil, ErrDatastoreDoesNotExist
237 238 239 240
	} else if err != nil {
		return nil, err
	}

241
	shardId, err := ReadShardFunc(path)
242
	if err != nil {
243
		return nil, err
244 245
	}

246
	fs := &Datastore{
Steven Allen's avatar
Steven Allen committed
247 248 249 250 251 252 253 254
		path:         path,
		shardStr:     shardId.String(),
		getDir:       shardId.Func(),
		sync:         syncFiles,
		checkpointCh: make(chan struct{}, 1),
		done:         make(chan struct{}),
		diskUsage:    0,
		opMap:        new(opMap),
255 256 257 258 259 260 261 262 263 264 265
	}

	// This sets diskUsage to the correct value
	// It might be slow, but allowing it to happen
	// while the datastore is usable might
	// cause diskUsage to not be accurate.
	err = fs.calculateDiskUsage()
	if err != nil {
		// Cannot stat() all
		// elements in the datastore.
		return nil, err
266
	}
267

Kevin Atkinson's avatar
Kevin Atkinson committed
268
	go fs.checkpointLoop()
269 270 271
	return fs, nil
}

272
// convenience method
273
func CreateOrOpen(path string, fun *ShardIdV1, sync bool) (*Datastore, error) {
274
	err := Create(path, fun)
Kevin Atkinson's avatar
Kevin Atkinson committed
275
	if err != nil && err != ErrDatastoreExists {
276 277 278 279 280
		return nil, err
	}
	return Open(path, sync)
}

281 282
func (fs *Datastore) ShardStr() string {
	return fs.shardStr
283 284
}

Tommi Virtanen's avatar
Tommi Virtanen committed
285
func (fs *Datastore) encode(key datastore.Key) (dir, file string) {
286
	noslash := key.String()[1:]
Jeromy's avatar
Jeromy committed
287 288
	dir = filepath.Join(fs.path, fs.getDir(noslash))
	file = filepath.Join(dir, noslash+extension)
Tommi Virtanen's avatar
Tommi Virtanen committed
289 290 291
	return dir, file
}

292
func (fs *Datastore) decode(file string) (key datastore.Key, ok bool) {
Jeromy's avatar
Jeromy committed
293
	if filepath.Ext(file) != extension {
294 295 296
		return datastore.Key{}, false
	}
	name := file[:len(file)-len(extension)]
Jeromy's avatar
Jeromy committed
297
	return datastore.NewKey(name), true
298 299
}

300 301
func (fs *Datastore) makeDir(dir string) error {
	if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
302
		return err
303 304 305 306 307 308
	}

	// In theory, if we create a new prefix dir and add a file to
	// it, the creation of the prefix dir itself might not be
	// durable yet. Sync the root dir after a successful mkdir of
	// a prefix dir, just to be paranoid.
Jeromy's avatar
Jeromy committed
309 310 311 312
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
313 314 315 316
	}
	return nil
}

317
func (fs *Datastore) makeDirNoSync(dir string) error {
kpcyrd's avatar
kpcyrd committed
318
	if err := os.Mkdir(dir, 0755); err != nil {
Jeromy's avatar
Jeromy committed
319 320 321 322 323
		// EEXIST is safe to ignore here, that just means the prefix
		// directory already existed.
		if !os.IsExist(err) {
			return err
		}
324
		return nil
Jeromy's avatar
Jeromy committed
325
	}
326 327 328

	// Track DiskUsage of this NEW folder
	fs.updateDiskUsage(dir, true)
Jeromy's avatar
Jeromy committed
329 330 331
	return nil
}

332 333 334 335 336 337 338 339 340 341 342 343 344 345 346
// This function always runs under an opLock. Therefore, only one thread is
// touching the affected files.
func (fs *Datastore) renameAndUpdateDiskUsage(tmpPath, path string) error {
	fi, err := os.Stat(path)

	// Destination exists, we need to discount it from diskUsage
	if fs != nil && err == nil {
		atomic.AddInt64(&fs.diskUsage, -fi.Size())
	} else if !os.IsNotExist(err) {
		return err
	}

	// Rename and add new file's diskUsage. If the rename fails,
	// it will either a) Re-add the size of an existing file, which
	// was sustracted before b) Add 0 if there is no existing file.
347
	err = os.Rename(tmpPath, path)
348 349 350 351
	fs.updateDiskUsage(path, true)
	return err
}

Jeromy's avatar
Jeromy committed
352
var putMaxRetries = 6
353

354 355 356 357 358 359 360 361 362 363
// Put stores a key/value in the datastore.
//
// Note, that we do not guarantee order of write operations (Put or Delete)
// to the same key in this datastore.
//
// For example. i.e. in the case of two concurrent Put, we only guarantee
// that one of them will come through, but cannot assure which one even if
// one arrived slightly later than the other. In the case of a
// concurrent Put and a Delete operation, we cannot guarantee which one
// will win.
364
func (fs *Datastore) Put(key datastore.Key, value []byte) error {
365
	if !keyIsValid(key) {
366
		return fmt.Errorf("when putting '%q': %w", key, ErrInvalidKey)
367 368
	}

Steven Allen's avatar
Steven Allen committed
369 370 371 372 373 374
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

375
	var err error
Jeromy's avatar
Jeromy committed
376
	for i := 1; i <= putMaxRetries; i++ {
377 378 379
		err = fs.doWriteOp(&op{
			typ: opPut,
			key: key,
380
			v:   value,
381
		})
382
		if err == nil {
Jeromy's avatar
Jeromy committed
383
			break
384 385 386
		}

		if !strings.Contains(err.Error(), "too many open files") {
Jeromy's avatar
Jeromy committed
387
			break
388 389
		}

Or Rikon's avatar
Or Rikon committed
390
		log.Errorf("too many open files, retrying in %dms", 100*i)
Jeromy's avatar
Jeromy committed
391
		time.Sleep(time.Millisecond * 100 * time.Duration(i))
392 393 394 395
	}
	return err
}

Adin Schmahmann's avatar
Adin Schmahmann committed
396 397 398 399 400 401 402 403 404 405
func (fs *Datastore) Sync(prefix datastore.Key) error {
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

	return nil
}

406 407 408 409 410 411 412 413 414 415 416 417 418
func (fs *Datastore) doOp(oper *op) error {
	switch oper.typ {
	case opPut:
		return fs.doPut(oper.key, oper.v)
	case opDelete:
		return fs.doDelete(oper.key)
	case opRename:
		return fs.renameAndUpdateDiskUsage(oper.tmp, oper.path)
	default:
		panic("bad operation, this is a bug")
	}
}

419 420
// doWrite optimizes out write operations (put/delete) to the same
// key by queueing them and succeeding all queued
421
// operations if one of them does. In such case,
422
// we assume that the first succeeding operation
423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442
// on that key was the last one to happen after
// all successful others.
func (fs *Datastore) doWriteOp(oper *op) error {
	keyStr := oper.key.String()

	opRes := fs.opMap.Begin(keyStr)
	if opRes == nil { // nothing to do, a concurrent op succeeded
		return nil
	}

	// Do the operation
	err := fs.doOp(oper)

	// Finish it. If no error, it will signal other operations
	// waiting on this result to succeed. Otherwise, they will
	// retry.
	opRes.Finish(err == nil)
	return err
}

443
func (fs *Datastore) doPut(key datastore.Key, val []byte) error {
444

Tommi Virtanen's avatar
Tommi Virtanen committed
445
	dir, path := fs.encode(key)
446
	if err := fs.makeDir(dir); err != nil {
447
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469
	}

	tmp, err := ioutil.TempFile(dir, "put-")
	if err != nil {
		return err
	}
	closed := false
	removed := false
	defer func() {
		if !closed {
			// silence errcheck
			_ = tmp.Close()
		}
		if !removed {
			// silence errcheck
			_ = os.Remove(tmp.Name())
		}
	}()

	if _, err := tmp.Write(val); err != nil {
		return err
	}
Jeromy's avatar
Jeromy committed
470
	if fs.sync {
471
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
472 473
			return err
		}
474
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
475 476 477 478 479
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

480
	err = fs.renameAndUpdateDiskUsage(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
481 482 483 484 485
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
486 487 488 489
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
490
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
491 492 493
	return nil
}

494
func (fs *Datastore) putMany(data map[datastore.Key][]byte) error {
Steven Allen's avatar
Steven Allen committed
495 496 497 498 499 500
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

Jeromy's avatar
Jeromy committed
501 502
	var dirsToSync []string

503 504 505 506 507
	files := make(map[*os.File]*op, len(data))
	ops := make(map[*os.File]int, len(data))

	defer func() {
		for fi := range files {
Steven Allen's avatar
Steven Allen committed
508
			val := ops[fi]
509 510 511 512 513 514 515
			switch val {
			case 0:
				_ = fi.Close()
				fallthrough
			case 1:
				_ = os.Remove(fi.Name())
			}
Jeromy's avatar
Jeromy committed
516
		}
517 518 519
	}()

	for key, value := range data {
Jeromy's avatar
Jeromy committed
520
		dir, path := fs.encode(key)
521
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
522 523 524 525 526 527 528 529 530
			return err
		}
		dirsToSync = append(dirsToSync, dir)

		tmp, err := ioutil.TempFile(dir, "put-")
		if err != nil {
			return err
		}

531
		if _, err := tmp.Write(value); err != nil {
Jeromy's avatar
Jeromy committed
532 533 534
			return err
		}

535 536 537 538 539 540
		files[tmp] = &op{
			typ:  opRename,
			path: path,
			tmp:  tmp.Name(),
			key:  key,
		}
Jeromy's avatar
Jeromy committed
541 542 543 544
	}

	// Now we sync everything
	// sync and close files
545
	for fi := range files {
Jeromy's avatar
Jeromy committed
546
		if fs.sync {
547
			if err := syncFile(fi); err != nil {
Jeromy's avatar
Jeromy committed
548 549
				return err
			}
Jeromy's avatar
Jeromy committed
550 551 552 553 554 555 556 557 558 559 560
		}

		if err := fi.Close(); err != nil {
			return err
		}

		// signify closed
		ops[fi] = 1
	}

	// move files to their proper places
561
	for fi, op := range files {
562 563 564 565
		err := fs.doWriteOp(op)
		if err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
566 567 568 569 570
		// signify removed
		ops[fi] = 2
	}

	// now sync the dirs for those files
Jeromy's avatar
Jeromy committed
571 572 573 574 575
	if fs.sync {
		for _, dir := range dirsToSync {
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
576 577
		}

Jeromy's avatar
Jeromy committed
578 579 580 581
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
582 583 584 585 586
	}

	return nil
}

587
func (fs *Datastore) Get(key datastore.Key) (value []byte, err error) {
588 589 590 591 592
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil, datastore.ErrNotFound
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
593 594 595 596 597 598 599 600 601 602 603 604 605
	_, path := fs.encode(key)
	data, err := ioutil.ReadFile(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, datastore.ErrNotFound
		}
		// no specific error to return, so just pass it through
		return nil, err
	}
	return data, nil
}

func (fs *Datastore) Has(key datastore.Key) (exists bool, err error) {
606 607 608 609 610
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return false, nil
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
611 612 613 614 615 616 617 618 619
	_, path := fs.encode(key)
	switch _, err := os.Stat(path); {
	case err == nil:
		return true, nil
	case os.IsNotExist(err):
		return false, nil
	default:
		return false, err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
620 621
}

Steven Allen's avatar
Steven Allen committed
622
func (fs *Datastore) GetSize(key datastore.Key) (size int, err error) {
623 624 625 626 627
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return -1, datastore.ErrNotFound
	}

Steven Allen's avatar
Steven Allen committed
628 629 630 631 632 633 634 635 636 637 638
	_, path := fs.encode(key)
	switch s, err := os.Stat(path); {
	case err == nil:
		return int(s.Size()), nil
	case os.IsNotExist(err):
		return -1, datastore.ErrNotFound
	default:
		return -1, err
	}
}

639 640 641
// Delete removes a key/value from the Datastore. Please read
// the Put() explanation about the handling of concurrent write
// operations to the same key.
Tommi Virtanen's avatar
Tommi Virtanen committed
642
func (fs *Datastore) Delete(key datastore.Key) error {
643 644 645 646 647
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil
	}

Steven Allen's avatar
Steven Allen committed
648 649 650 651 652 653
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

654 655 656 657 658 659 660 661 662 663
	return fs.doWriteOp(&op{
		typ: opDelete,
		key: key,
		v:   nil,
	})
}

// This function always runs within an opLock for the given
// key, and not concurrently.
func (fs *Datastore) doDelete(key datastore.Key) error {
Tommi Virtanen's avatar
Tommi Virtanen committed
664
	_, path := fs.encode(key)
665 666 667

	fSize := fileSize(path)

Tommi Virtanen's avatar
Tommi Virtanen committed
668 669
	switch err := os.Remove(path); {
	case err == nil:
670 671
		atomic.AddInt64(&fs.diskUsage, -fSize)
		fs.checkpointDiskUsage()
Tommi Virtanen's avatar
Tommi Virtanen committed
672 673
		return nil
	case os.IsNotExist(err):
Steven Allen's avatar
Steven Allen committed
674
		return nil
Tommi Virtanen's avatar
Tommi Virtanen committed
675 676 677
	default:
		return err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
678 679 680
}

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
Steven Allen's avatar
Steven Allen committed
681
	prefix := datastore.NewKey(q.Prefix).String()
682 683 684 685
	if prefix != "/" {
		// This datastore can't include keys with multiple components.
		// Therefore, it's always correct to return an empty result when
		// the user requests a filter by prefix.
686 687 688 689 690
		log.Warnw(
			"flatfs was queried with a key prefix but flatfs only supports keys at the root",
			"prefix", q.Prefix,
			"query", q,
		)
691 692
		return query.ResultsWithEntries(q, nil), nil
	}
693

Steven Allen's avatar
Steven Allen committed
694 695 696 697 698 699 700
	// Replicates the logic in ResultsWithChan but actually respects calls
	// to `Close`.
	b := query.NewResultBuilder(q)
	b.Process.Go(func(p goprocess.Process) {
		err := fs.walkTopLevel(fs.path, b)
		if err == nil {
			return
701
		}
Steven Allen's avatar
Steven Allen committed
702 703 704 705 706
		select {
		case b.Output <- query.Result{Error: errors.New("walk failed: " + err.Error())}:
		case <-p.Closing():
		}
	})
Steven Allen's avatar
Steven Allen committed
707
	go b.Process.CloseAfterChildren() //nolint
Steven Allen's avatar
Steven Allen committed
708

709 710 711
	// We don't apply _any_ of the query logic ourselves so we'll leave it
	// all up to the naive query engine.
	return query.NaiveQueryApply(q, b.Results()), nil
Tommi Virtanen's avatar
Tommi Virtanen committed
712 713
}

Steven Allen's avatar
Steven Allen committed
714
func (fs *Datastore) walkTopLevel(path string, result *query.ResultBuilder) error {
715 716 717 718
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
719
	defer dir.Close()
720 721 722 723 724
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return err
	}
	for _, dir := range names {
Kevin Atkinson's avatar
Kevin Atkinson committed
725 726 727 728
		if len(dir) == 0 || dir[0] == '.' {
			continue
		}

Steven Allen's avatar
Steven Allen committed
729
		err = fs.walk(filepath.Join(path, dir), result)
730 731 732
		if err != nil {
			return err
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
733

Steven Allen's avatar
Steven Allen committed
734 735 736 737 738 739
		// Are we closing?
		select {
		case <-result.Process.Closing():
			return nil
		default:
		}
740 741 742 743
	}
	return nil
}

744
// folderSize estimates the diskUsage of a folder by reading
745 746
// up to DiskUsageFilesAverage entries in it and assuming any
// other files will have an average size.
747
func folderSize(path string, deadline time.Time) (int64, initAccuracy, error) {
748 749 750 751
	var du int64

	folder, err := os.Open(path)
	if err != nil {
752
		return 0, "", err
753 754 755 756 757
	}
	defer folder.Close()

	stat, err := folder.Stat()
	if err != nil {
758
		return 0, "", err
759 760 761 762
	}

	files, err := folder.Readdirnames(-1)
	if err != nil {
763
		return 0, "", err
764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780
	}

	totalFiles := len(files)
	i := 0
	filesProcessed := 0
	maxFiles := DiskUsageFilesAverage
	if maxFiles <= 0 {
		maxFiles = totalFiles
	}

	// randomize file order
	// https://stackoverflow.com/a/42776696
	for i := len(files) - 1; i > 0; i-- {
		j := rand.Intn(i + 1)
		files[i], files[j] = files[j], files[i]
	}

781
	accuracy := exactA
782
	for {
783 784 785
		// Do not process any files after deadline is over
		if time.Now().After(deadline) {
			accuracy = timedoutA
786 787 788
			break
		}

789 790 791 792
		if i >= totalFiles || filesProcessed >= maxFiles {
			if filesProcessed >= maxFiles {
				accuracy = approxA
			}
793 794 795 796 797 798 799 800
			break
		}

		// Stat the file
		fname := files[i]
		subpath := filepath.Join(path, fname)
		st, err := os.Stat(subpath)
		if err != nil {
801
			return 0, "", err
802 803 804 805
		}

		// Find folder size recursively
		if st.IsDir() {
806
			du2, acc, err := folderSize(filepath.Join(subpath), deadline)
807
			if err != nil {
808
				return 0, "", err
809
			}
810
			accuracy = combineAccuracy(acc, accuracy)
811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832
			du += du2
			filesProcessed++
		} else { // in any other case, add the file size
			du += st.Size()
			filesProcessed++
		}

		i++
	}

	nonProcessed := totalFiles - filesProcessed

	// Avg is total size in this folder up to now / total files processed
	// it includes folders ant not folders
	avg := 0.0
	if filesProcessed > 0 {
		avg = float64(du) / float64(filesProcessed)
	}
	duEstimation := int64(avg * float64(nonProcessed))
	du += duEstimation
	du += stat.Size()
	//fmt.Println(path, "total:", totalFiles, "totalStat:", i, "totalFile:", filesProcessed, "left:", nonProcessed, "avg:", int(avg), "est:", int(duEstimation), "du:", du)
833
	return du, accuracy, nil
834 835 836 837
}

// calculateDiskUsage tries to read the DiskUsageFile for a cached
// diskUsage value, otherwise walks the datastore files.
838
// it is only safe to call in Open()
839 840 841
func (fs *Datastore) calculateDiskUsage() error {
	// Try to obtain a previously stored value from disk
	if persDu := fs.readDiskUsageFile(); persDu > 0 {
842
		fs.diskUsage = persDu
843 844 845
		return nil
	}

846 847 848 849 850 851 852
	msgDone := make(chan struct{}, 1) // prevent race condition
	msgTimer := time.AfterFunc(diskUsageMessageTimeout, func() {
		fmt.Printf("Calculating datastore size. This might take %s at most and will happen only once\n",
			DiskUsageCalcTimeout.String())
		msgDone <- struct{}{}
	})
	defer msgTimer.Stop()
853
	deadline := time.Now().Add(DiskUsageCalcTimeout)
854
	du, accuracy, err := folderSize(fs.path, deadline)
855 856 857
	if err != nil {
		return err
	}
858 859 860
	if !msgTimer.Stop() {
		<-msgDone
	}
861
	if accuracy == timedoutA {
862 863 864 865 866 867 868
		fmt.Println("WARN: It took to long to calculate the datastore size")
		fmt.Printf("WARN: The total size (%d) is an estimation. You can fix errors by\n", du)
		fmt.Printf("WARN: replacing the %s file with the right disk usage in bytes and\n",
			filepath.Join(fs.path, DiskUsageFile))
		fmt.Println("WARN: re-opening the datastore")
	}

869
	fs.storedValue.Accuracy = accuracy
870
	fs.diskUsage = du
871
	fs.writeDiskUsageFile(du, true)
872

873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893
	return nil
}

func fileSize(path string) int64 {
	fi, err := os.Stat(path)
	if err != nil {
		return 0
	}
	return fi.Size()
}

// updateDiskUsage reads the size of path and atomically
// increases or decreases the diskUsage variable.
// setting add to false will subtract from disk usage.
func (fs *Datastore) updateDiskUsage(path string, add bool) {
	fsize := fileSize(path)
	if !add {
		fsize = -fsize
	}

	if fsize != 0 {
894 895
		atomic.AddInt64(&fs.diskUsage, fsize)
		fs.checkpointDiskUsage()
896 897 898
	}
}

899
func (fs *Datastore) checkpointDiskUsage() {
900
	select {
Kevin Atkinson's avatar
Kevin Atkinson committed
901
	case fs.checkpointCh <- struct{}{}:
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
902
		// msg sent
903 904
	default:
		// checkpoint request already pending
905 906 907
	}
}

908
func (fs *Datastore) checkpointLoop() {
Steven Allen's avatar
Steven Allen committed
909 910
	defer close(fs.done)

911 912
	timerActive := true
	timer := time.NewTimer(0)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
913
	defer timer.Stop()
914
	for {
915 916 917
		select {
		case _, more := <-fs.checkpointCh:
			du := atomic.LoadInt64(&fs.diskUsage)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
918
			fs.dirty = true
919
			if !more { // shutting down
920
				fs.writeDiskUsageFile(du, true)
921
				if fs.dirty {
Steven Allen's avatar
Steven Allen committed
922
					log.Error("could not store final value of disk usage to file, future estimates may be inaccurate")
923
				}
924 925 926 927 928
				return
			}
			// If the difference between the checkpointed disk usage and
			// current one is larger than than `diskUsageCheckpointPercent`
			// of the checkpointed: store it.
929
			newDu := float64(du)
930
			lastCheckpointDu := float64(fs.storedValue.DiskUsage)
931
			diff := math.Abs(newDu - lastCheckpointDu)
932 933
			if lastCheckpointDu*diskUsageCheckpointPercent < diff*100.0 {
				fs.writeDiskUsageFile(du, false)
934
			}
935 936 937 938
			// Otherwise insure the value will be written to disk after
			// `diskUsageCheckpointTimeout`
			if fs.dirty && !timerActive {
				timer.Reset(diskUsageCheckpointTimeout)
Kevin Atkinson's avatar
Kevin Atkinson committed
939
				timerActive = true
940 941 942 943 944
			}
		case <-timer.C:
			timerActive = false
			if fs.dirty {
				du := atomic.LoadInt64(&fs.diskUsage)
945
				fs.writeDiskUsageFile(du, false)
946
			}
947
		}
948
	}
949 950
}

951
func (fs *Datastore) writeDiskUsageFile(du int64, doSync bool) {
952
	tmp, err := ioutil.TempFile(fs.path, "du-")
953
	if err != nil {
Steven Allen's avatar
Steven Allen committed
954
		log.Warnw("could not write disk usage", "error", err)
955 956 957
		return
	}

958 959 960 961 962 963 964
	removed := false
	defer func() {
		if !removed {
			// silence errcheck
			_ = os.Remove(tmp.Name())
		}
	}()
965

966 967
	toWrite := fs.storedValue
	toWrite.DiskUsage = du
968
	encoder := json.NewEncoder(tmp)
969
	if err := encoder.Encode(&toWrite); err != nil {
Steven Allen's avatar
Steven Allen committed
970
		log.Warnw("cound not write disk usage", "error", err)
971 972
		return
	}
973

974 975
	if doSync {
		if err := tmp.Sync(); err != nil {
Steven Allen's avatar
Steven Allen committed
976
			log.Warnw("cound not sync", "error", err, "file", DiskUsageFile)
977 978 979 980
			return
		}
	}

981
	if err := tmp.Close(); err != nil {
Steven Allen's avatar
Steven Allen committed
982
		log.Warnw("cound not write disk usage", "error", err)
983 984 985
		return
	}

986
	if err := os.Rename(tmp.Name(), filepath.Join(fs.path, DiskUsageFile)); err != nil {
Steven Allen's avatar
Steven Allen committed
987
		log.Warnw("cound not write disk usage", "error", err)
988
		return
989
	}
990
	removed = true
991

992
	fs.storedValue = toWrite
993
	fs.dirty = false
994 995
}

996
// readDiskUsageFile is only safe to call in Open()
997 998 999 1000 1001 1002
func (fs *Datastore) readDiskUsageFile() int64 {
	fpath := filepath.Join(fs.path, DiskUsageFile)
	duB, err := ioutil.ReadFile(fpath)
	if err != nil {
		return 0
	}
1003
	err = json.Unmarshal(duB, &fs.storedValue)
1004 1005 1006
	if err != nil {
		return 0
	}
1007
	return fs.storedValue.DiskUsage
1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027
}

// DiskUsage implements the PersistentDatastore interface
// and returns the current disk usage in bytes used by
// this datastore.
//
// The size is approximative and may slightly differ from
// the real disk values.
func (fs *Datastore) DiskUsage() (uint64, error) {
	// it may differ from real disk values if
	// the filesystem has allocated for blocks
	// for a directory because it has many files in it
	// we don't account for "resized" directories.
	// In a large datastore, the differences should be
	// are negligible though.

	du := atomic.LoadInt64(&fs.diskUsage)
	return uint64(du), nil
}

1028 1029 1030 1031
// Accuracy returns a string representing the accuracy of the
// DiskUsage() result, the value returned is implementation defined
// and for informational purposes only
func (fs *Datastore) Accuracy() string {
1032
	return string(fs.storedValue.Accuracy)
1033 1034
}

1035
func (fs *Datastore) walk(path string, qrb *query.ResultBuilder) error {
1036 1037
	dir, err := os.Open(path)
	if err != nil {
1038 1039 1040 1041
		if os.IsNotExist(err) {
			// not an error if the file disappeared
			return nil
		}
1042 1043
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
1044
	defer dir.Close()
1045 1046 1047 1048 1049 1050 1051 1052 1053 1054

	// ignore non-directories
	fileInfo, err := dir.Stat()
	if err != nil {
		return err
	}
	if !fileInfo.IsDir() {
		return nil
	}

1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return err
	}
	for _, fn := range names {

		if len(fn) == 0 || fn[0] == '.' {
			continue
		}

		key, ok := fs.decode(fn)
		if !ok {
Steven Allen's avatar
Steven Allen committed
1067
			log.Warnw("failed to decode flatfs entry", "file", fn)
1068 1069 1070
			continue
		}

1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092
		var result query.Result
		result.Key = key.String()
		if !qrb.Query.KeysOnly {
			value, err := ioutil.ReadFile(filepath.Join(path, fn))
			if err != nil {
				result.Error = err
			} else {
				// NOTE: Don't set the value/size on error. We
				// don't want to return partial values.
				result.Value = value
				result.Size = len(value)
			}
		} else if qrb.Query.ReturnsSizes {
			var stat os.FileInfo
			stat, err := os.Stat(filepath.Join(path, fn))
			if err != nil {
				result.Error = err
			} else {
				result.Size = int(stat.Size())
			}
		}

Steven Allen's avatar
Steven Allen committed
1093
		select {
1094 1095
		case qrb.Output <- result:
		case <-qrb.Process.Closing():
Steven Allen's avatar
Steven Allen committed
1096
			return nil
1097 1098 1099 1100 1101
		}
	}
	return nil
}

1102 1103 1104
// Deactivate closes background maintenance threads, most write
// operations will fail but readonly operations will continue to
// function
Steven Allen's avatar
Steven Allen committed
1105
func (fs *Datastore) deactivate() {
Steven Allen's avatar
Steven Allen committed
1106 1107 1108
	fs.shutdownLock.Lock()
	defer fs.shutdownLock.Unlock()
	if fs.shutdown {
Steven Allen's avatar
Steven Allen committed
1109
		return
1110
	}
Steven Allen's avatar
Steven Allen committed
1111 1112 1113
	fs.shutdown = true
	close(fs.checkpointCh)
	<-fs.done
Jeromy's avatar
Jeromy committed
1114 1115
}

1116
func (fs *Datastore) Close() error {
Steven Allen's avatar
Steven Allen committed
1117 1118
	fs.deactivate()
	return nil
1119 1120
}

Jeromy's avatar
Jeromy committed
1121
type flatfsBatch struct {
1122
	puts    map[datastore.Key][]byte
Jeromy's avatar
Jeromy committed
1123 1124 1125 1126 1127
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

Jeromy's avatar
Jeromy committed
1128
func (fs *Datastore) Batch() (datastore.Batch, error) {
Jeromy's avatar
Jeromy committed
1129
	return &flatfsBatch{
1130
		puts:    make(map[datastore.Key][]byte),
Jeromy's avatar
Jeromy committed
1131 1132
		deletes: make(map[datastore.Key]struct{}),
		ds:      fs,
Jeromy's avatar
Jeromy committed
1133
	}, nil
Jeromy's avatar
Jeromy committed
1134 1135
}

1136
func (bt *flatfsBatch) Put(key datastore.Key, val []byte) error {
1137
	if !keyIsValid(key) {
1138
		return fmt.Errorf("when putting '%q': %w", key, ErrInvalidKey)
1139
	}
Jeromy's avatar
Jeromy committed
1140 1141 1142 1143
	bt.puts[key] = val
	return nil
}

Jeromy's avatar
Jeromy committed
1144
func (bt *flatfsBatch) Delete(key datastore.Key) error {
1145 1146 1147
	if keyIsValid(key) {
		bt.deletes[key] = struct{}{}
	} // otherwise, delete is a no-op anyways.
Jeromy's avatar
Jeromy committed
1148 1149 1150
	return nil
}

Jeromy's avatar
Jeromy committed
1151
func (bt *flatfsBatch) Commit() error {
Jeromy's avatar
Jeromy committed
1152 1153 1154 1155
	if err := bt.ds.putMany(bt.puts); err != nil {
		return err
	}

Steven Allen's avatar
Steven Allen committed
1156
	for k := range bt.deletes {
Jeromy's avatar
Jeromy committed
1157 1158 1159 1160 1161 1162 1163
		if err := bt.ds.Delete(k); err != nil {
			return err
		}
	}

	return nil
}