flatfs.go 26.4 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"
10 11
	"math"
	"math/rand"
Tommi Virtanen's avatar
Tommi Virtanen committed
12
	"os"
Jeromy's avatar
Jeromy committed
13
	"path/filepath"
Tommi Virtanen's avatar
Tommi Virtanen committed
14
	"strings"
15 16
	"sync"
	"sync/atomic"
17
	"time"
Tommi Virtanen's avatar
Tommi Virtanen committed
18

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

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

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

Tommi Virtanen's avatar
Tommi Virtanen committed
28
const (
29
	extension                  = ".data"
30
	diskUsageMessageTimeout    = 5 * time.Second
31
	diskUsageCheckpointPercent = 1.0
32
	diskUsageCheckpointTimeout = 2 * time.Second
Tommi Virtanen's avatar
Tommi Virtanen committed
33 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
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
)

59 60 61
type initAccuracy string

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

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

90 91 92 93 94 95
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
96
	ErrClosed                = errors.New("datastore closed")
97
	ErrInvalidKey            = errors.New("key not supported by flatfs")
98 99 100 101 102 103 104 105 106 107
)

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
108
type Datastore struct {
109 110 111 112 113
	// 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

114 115
	path     string
	tempPath 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 {
kpcyrd's avatar
kpcyrd committed
199
	err := os.Mkdir(path, 0755)
200 201
	if err != nil && !os.IsExist(err) {
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
202 203
	}

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

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

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

240 241 242 243 244 245 246 247 248 249 250
	tempPath := filepath.Join(path, ".temp")
	err = os.RemoveAll(tempPath)
	if err != nil && !os.IsNotExist(err) {
		return nil, fmt.Errorf("failed to remove temporary directory: %w", err)
	}

	err = os.Mkdir(tempPath, 0755)
	if err != nil {
		return nil, fmt.Errorf("failed to create temporary directory: %w", err)
	}

251
	shardId, err := ReadShardFunc(path)
252
	if err != nil {
253
		return nil, err
254 255
	}

256
	fs := &Datastore{
Steven Allen's avatar
Steven Allen committed
257
		path:         path,
258
		tempPath:     tempPath,
Steven Allen's avatar
Steven Allen committed
259 260 261 262 263 264 265
		shardStr:     shardId.String(),
		getDir:       shardId.Func(),
		sync:         syncFiles,
		checkpointCh: make(chan struct{}, 1),
		done:         make(chan struct{}),
		diskUsage:    0,
		opMap:        new(opMap),
266 267 268 269 270 271 272 273 274 275 276
	}

	// 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
277
	}
278

Kevin Atkinson's avatar
Kevin Atkinson committed
279
	go fs.checkpointLoop()
280 281 282
	return fs, nil
}

283
// convenience method
284
func CreateOrOpen(path string, fun *ShardIdV1, sync bool) (*Datastore, error) {
285
	err := Create(path, fun)
Kevin Atkinson's avatar
Kevin Atkinson committed
286
	if err != nil && err != ErrDatastoreExists {
287 288 289 290 291
		return nil, err
	}
	return Open(path, sync)
}

292 293
func (fs *Datastore) ShardStr() string {
	return fs.shardStr
294 295
}

Tommi Virtanen's avatar
Tommi Virtanen committed
296
func (fs *Datastore) encode(key datastore.Key) (dir, file string) {
297
	noslash := key.String()[1:]
Jeromy's avatar
Jeromy committed
298 299
	dir = filepath.Join(fs.path, fs.getDir(noslash))
	file = filepath.Join(dir, noslash+extension)
Tommi Virtanen's avatar
Tommi Virtanen committed
300 301 302
	return dir, file
}

303
func (fs *Datastore) decode(file string) (key datastore.Key, ok bool) {
304 305 306 307 308 309
	if !strings.HasSuffix(file, extension) {
		// We expect random files like "put-". Log when we encounter
		// others.
		if !strings.HasPrefix(file, "put-") {
			log.Warnw("failed to decode flatfs filename", "file", file)
		}
310 311 312
		return datastore.Key{}, false
	}
	name := file[:len(file)-len(extension)]
Jeromy's avatar
Jeromy committed
313
	return datastore.NewKey(name), true
314 315
}

316 317
func (fs *Datastore) makeDir(dir string) error {
	if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
318
		return err
319 320 321 322 323 324
	}

	// 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
325 326 327 328
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
329 330 331 332
	}
	return nil
}

333
func (fs *Datastore) makeDirNoSync(dir string) error {
kpcyrd's avatar
kpcyrd committed
334
	if err := os.Mkdir(dir, 0755); err != nil {
Jeromy's avatar
Jeromy committed
335 336 337 338 339
		// EEXIST is safe to ignore here, that just means the prefix
		// directory already existed.
		if !os.IsExist(err) {
			return err
		}
340
		return nil
Jeromy's avatar
Jeromy committed
341
	}
342 343 344

	// Track DiskUsage of this NEW folder
	fs.updateDiskUsage(dir, true)
Jeromy's avatar
Jeromy committed
345 346 347
	return nil
}

348 349 350 351 352 353 354 355 356 357 358 359 360 361 362
// 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.
363
	err = os.Rename(tmpPath, path)
364 365 366 367
	fs.updateDiskUsage(path, true)
	return err
}

Jeromy's avatar
Jeromy committed
368
var putMaxRetries = 6
369

370 371 372 373 374 375 376 377 378 379
// 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.
380
func (fs *Datastore) Put(key datastore.Key, value []byte) error {
381
	if !keyIsValid(key) {
382
		return fmt.Errorf("when putting '%q': %w", key, ErrInvalidKey)
383 384
	}

Steven Allen's avatar
Steven Allen committed
385 386 387 388 389 390
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

391
	var err error
Jeromy's avatar
Jeromy committed
392
	for i := 1; i <= putMaxRetries; i++ {
393 394 395
		err = fs.doWriteOp(&op{
			typ: opPut,
			key: key,
396
			v:   value,
397
		})
398
		if err == nil {
Jeromy's avatar
Jeromy committed
399
			break
400 401 402
		}

		if !strings.Contains(err.Error(), "too many open files") {
Jeromy's avatar
Jeromy committed
403
			break
404 405
		}

Or Rikon's avatar
Or Rikon committed
406
		log.Errorf("too many open files, retrying in %dms", 100*i)
Jeromy's avatar
Jeromy committed
407
		time.Sleep(time.Millisecond * 100 * time.Duration(i))
408 409 410 411
	}
	return err
}

Adin Schmahmann's avatar
Adin Schmahmann committed
412 413 414 415 416 417 418 419 420 421
func (fs *Datastore) Sync(prefix datastore.Key) error {
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

	return nil
}

422 423 424 425 426 427 428 429 430 431 432 433 434
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")
	}
}

435 436
// doWrite optimizes out write operations (put/delete) to the same
// key by queueing them and succeeding all queued
437
// operations if one of them does. In such case,
438
// we assume that the first succeeding operation
439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458
// 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
}

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

Tommi Virtanen's avatar
Tommi Virtanen committed
461
	dir, path := fs.encode(key)
462
	if err := fs.makeDir(dir); err != nil {
463
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
464 465
	}

466
	tmp, err := fs.tempFile()
Tommi Virtanen's avatar
Tommi Virtanen committed
467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485
	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
486
	if fs.sync {
487
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
488 489
			return err
		}
490
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
491 492 493 494 495
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

496
	err = fs.renameAndUpdateDiskUsage(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
497 498 499 500 501
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
502 503 504 505
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
506
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
507 508 509
	return nil
}

510
func (fs *Datastore) putMany(data map[datastore.Key][]byte) error {
Steven Allen's avatar
Steven Allen committed
511 512 513 514 515 516
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

Jeromy's avatar
Jeromy committed
517 518
	var dirsToSync []string

519 520 521 522 523
	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
524
			val := ops[fi]
525 526 527 528 529 530 531
			switch val {
			case 0:
				_ = fi.Close()
				fallthrough
			case 1:
				_ = os.Remove(fi.Name())
			}
Jeromy's avatar
Jeromy committed
532
		}
533 534 535
	}()

	for key, value := range data {
Jeromy's avatar
Jeromy committed
536
		dir, path := fs.encode(key)
537
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
538 539 540 541
			return err
		}
		dirsToSync = append(dirsToSync, dir)

542
		tmp, err := fs.tempFile()
Jeromy's avatar
Jeromy committed
543 544 545 546
		if err != nil {
			return err
		}

547
		if _, err := tmp.Write(value); err != nil {
Jeromy's avatar
Jeromy committed
548 549 550
			return err
		}

551 552 553 554 555 556
		files[tmp] = &op{
			typ:  opRename,
			path: path,
			tmp:  tmp.Name(),
			key:  key,
		}
Jeromy's avatar
Jeromy committed
557 558 559 560
	}

	// Now we sync everything
	// sync and close files
561
	for fi := range files {
Jeromy's avatar
Jeromy committed
562
		if fs.sync {
563
			if err := syncFile(fi); err != nil {
Jeromy's avatar
Jeromy committed
564 565
				return err
			}
Jeromy's avatar
Jeromy committed
566 567 568 569 570 571 572 573 574 575 576
		}

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

		// signify closed
		ops[fi] = 1
	}

	// move files to their proper places
577
	for fi, op := range files {
578 579 580 581
		err := fs.doWriteOp(op)
		if err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
582 583 584 585 586
		// signify removed
		ops[fi] = 2
	}

	// now sync the dirs for those files
Jeromy's avatar
Jeromy committed
587 588 589 590 591
	if fs.sync {
		for _, dir := range dirsToSync {
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
592 593
		}

Jeromy's avatar
Jeromy committed
594 595 596 597
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
598 599 600 601 602
	}

	return nil
}

603
func (fs *Datastore) Get(key datastore.Key) (value []byte, err error) {
604 605 606 607 608
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil, datastore.ErrNotFound
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
609
	_, path := fs.encode(key)
610
	data, err := readFile(path)
Tommi Virtanen's avatar
Tommi Virtanen committed
611 612 613 614 615 616 617 618 619 620 621
	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) {
622 623 624 625 626
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return false, nil
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
627 628 629 630 631 632 633 634 635
	_, 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
636 637
}

Steven Allen's avatar
Steven Allen committed
638
func (fs *Datastore) GetSize(key datastore.Key) (size int, err error) {
639 640 641 642 643
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return -1, datastore.ErrNotFound
	}

Steven Allen's avatar
Steven Allen committed
644 645 646 647 648 649 650 651 652 653 654
	_, 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
	}
}

655 656 657
// 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
658
func (fs *Datastore) Delete(key datastore.Key) error {
659 660 661 662 663
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil
	}

Steven Allen's avatar
Steven Allen committed
664 665 666 667 668 669
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

670 671 672 673 674 675 676 677 678 679
	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
680
	_, path := fs.encode(key)
681 682 683

	fSize := fileSize(path)

Tommi Virtanen's avatar
Tommi Virtanen committed
684 685
	switch err := os.Remove(path); {
	case err == nil:
686 687
		atomic.AddInt64(&fs.diskUsage, -fSize)
		fs.checkpointDiskUsage()
Tommi Virtanen's avatar
Tommi Virtanen committed
688 689
		return nil
	case os.IsNotExist(err):
Steven Allen's avatar
Steven Allen committed
690
		return nil
Tommi Virtanen's avatar
Tommi Virtanen committed
691 692 693
	default:
		return err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
694 695 696
}

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
Steven Allen's avatar
Steven Allen committed
697
	prefix := datastore.NewKey(q.Prefix).String()
698 699 700 701
	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.
702 703 704 705 706
		log.Warnw(
			"flatfs was queried with a key prefix but flatfs only supports keys at the root",
			"prefix", q.Prefix,
			"query", q,
		)
707 708
		return query.ResultsWithEntries(q, nil), nil
	}
709

Steven Allen's avatar
Steven Allen committed
710 711 712 713 714 715 716
	// 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
717
		}
Steven Allen's avatar
Steven Allen committed
718 719 720 721 722
		select {
		case b.Output <- query.Result{Error: errors.New("walk failed: " + err.Error())}:
		case <-p.Closing():
		}
	})
Steven Allen's avatar
Steven Allen committed
723
	go b.Process.CloseAfterChildren() //nolint
Steven Allen's avatar
Steven Allen committed
724

725 726 727
	// 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
728 729
}

Steven Allen's avatar
Steven Allen committed
730
func (fs *Datastore) walkTopLevel(path string, result *query.ResultBuilder) error {
731 732 733 734
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
735
	defer dir.Close()
736 737 738 739 740
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return err
	}
	for _, dir := range names {
Kevin Atkinson's avatar
Kevin Atkinson committed
741 742 743 744
		if len(dir) == 0 || dir[0] == '.' {
			continue
		}

Steven Allen's avatar
Steven Allen committed
745
		err = fs.walk(filepath.Join(path, dir), result)
746 747 748
		if err != nil {
			return err
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
749

Steven Allen's avatar
Steven Allen committed
750 751 752 753 754 755
		// Are we closing?
		select {
		case <-result.Process.Closing():
			return nil
		default:
		}
756 757 758 759
	}
	return nil
}

760
// folderSize estimates the diskUsage of a folder by reading
761 762
// up to DiskUsageFilesAverage entries in it and assuming any
// other files will have an average size.
763
func folderSize(path string, deadline time.Time) (int64, initAccuracy, error) {
764 765 766 767
	var du int64

	folder, err := os.Open(path)
	if err != nil {
768
		return 0, "", err
769 770 771 772 773
	}
	defer folder.Close()

	stat, err := folder.Stat()
	if err != nil {
774
		return 0, "", err
775 776 777 778
	}

	files, err := folder.Readdirnames(-1)
	if err != nil {
779
		return 0, "", err
780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796
	}

	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]
	}

797
	accuracy := exactA
798
	for {
799 800 801
		// Do not process any files after deadline is over
		if time.Now().After(deadline) {
			accuracy = timedoutA
802 803 804
			break
		}

805 806 807 808
		if i >= totalFiles || filesProcessed >= maxFiles {
			if filesProcessed >= maxFiles {
				accuracy = approxA
			}
809 810 811 812 813 814 815 816
			break
		}

		// Stat the file
		fname := files[i]
		subpath := filepath.Join(path, fname)
		st, err := os.Stat(subpath)
		if err != nil {
817
			return 0, "", err
818 819 820 821
		}

		// Find folder size recursively
		if st.IsDir() {
822
			du2, acc, err := folderSize(filepath.Join(subpath), deadline)
823
			if err != nil {
824
				return 0, "", err
825
			}
826
			accuracy = combineAccuracy(acc, accuracy)
827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848
			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)
849
	return du, accuracy, nil
850 851 852 853
}

// calculateDiskUsage tries to read the DiskUsageFile for a cached
// diskUsage value, otherwise walks the datastore files.
854
// it is only safe to call in Open()
855 856 857
func (fs *Datastore) calculateDiskUsage() error {
	// Try to obtain a previously stored value from disk
	if persDu := fs.readDiskUsageFile(); persDu > 0 {
858
		fs.diskUsage = persDu
859 860 861
		return nil
	}

862 863 864 865 866 867 868
	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()
869
	deadline := time.Now().Add(DiskUsageCalcTimeout)
870
	du, accuracy, err := folderSize(fs.path, deadline)
871 872 873
	if err != nil {
		return err
	}
874 875 876
	if !msgTimer.Stop() {
		<-msgDone
	}
877
	if accuracy == timedoutA {
878 879 880 881 882 883 884
		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")
	}

885
	fs.storedValue.Accuracy = accuracy
886
	fs.diskUsage = du
887
	fs.writeDiskUsageFile(du, true)
888

889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909
	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 {
910 911
		atomic.AddInt64(&fs.diskUsage, fsize)
		fs.checkpointDiskUsage()
912 913 914
	}
}

915
func (fs *Datastore) checkpointDiskUsage() {
916
	select {
Kevin Atkinson's avatar
Kevin Atkinson committed
917
	case fs.checkpointCh <- struct{}{}:
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
918
		// msg sent
919 920
	default:
		// checkpoint request already pending
921 922 923
	}
}

924
func (fs *Datastore) checkpointLoop() {
Steven Allen's avatar
Steven Allen committed
925 926
	defer close(fs.done)

927 928
	timerActive := true
	timer := time.NewTimer(0)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
929
	defer timer.Stop()
930
	for {
931 932 933
		select {
		case _, more := <-fs.checkpointCh:
			du := atomic.LoadInt64(&fs.diskUsage)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
934
			fs.dirty = true
935
			if !more { // shutting down
936
				fs.writeDiskUsageFile(du, true)
937
				if fs.dirty {
Steven Allen's avatar
Steven Allen committed
938
					log.Error("could not store final value of disk usage to file, future estimates may be inaccurate")
939
				}
940 941 942 943 944
				return
			}
			// If the difference between the checkpointed disk usage and
			// current one is larger than than `diskUsageCheckpointPercent`
			// of the checkpointed: store it.
945
			newDu := float64(du)
946
			lastCheckpointDu := float64(fs.storedValue.DiskUsage)
947
			diff := math.Abs(newDu - lastCheckpointDu)
948 949
			if lastCheckpointDu*diskUsageCheckpointPercent < diff*100.0 {
				fs.writeDiskUsageFile(du, false)
950
			}
951 952 953 954
			// 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
955
				timerActive = true
956 957 958 959 960
			}
		case <-timer.C:
			timerActive = false
			if fs.dirty {
				du := atomic.LoadInt64(&fs.diskUsage)
961
				fs.writeDiskUsageFile(du, false)
962
			}
963
		}
964
	}
965 966
}

967
func (fs *Datastore) writeDiskUsageFile(du int64, doSync bool) {
968
	tmp, err := fs.tempFile()
969
	if err != nil {
Steven Allen's avatar
Steven Allen committed
970
		log.Warnw("could not write disk usage", "error", err)
971 972 973
		return
	}

974 975 976 977 978 979 980
	removed := false
	defer func() {
		if !removed {
			// silence errcheck
			_ = os.Remove(tmp.Name())
		}
	}()
981

982 983
	toWrite := fs.storedValue
	toWrite.DiskUsage = du
984
	encoder := json.NewEncoder(tmp)
985
	if err := encoder.Encode(&toWrite); err != nil {
Steven Allen's avatar
Steven Allen committed
986
		log.Warnw("cound not write disk usage", "error", err)
987 988
		return
	}
989

990 991
	if doSync {
		if err := tmp.Sync(); err != nil {
Steven Allen's avatar
Steven Allen committed
992
			log.Warnw("cound not sync", "error", err, "file", DiskUsageFile)
993 994 995 996
			return
		}
	}

997
	if err := tmp.Close(); err != nil {
Steven Allen's avatar
Steven Allen committed
998
		log.Warnw("cound not write disk usage", "error", err)
999 1000 1001
		return
	}

1002
	if err := os.Rename(tmp.Name(), filepath.Join(fs.path, DiskUsageFile)); err != nil {
Steven Allen's avatar
Steven Allen committed
1003
		log.Warnw("cound not write disk usage", "error", err)
1004
		return
1005
	}
1006
	removed = true
1007

1008
	fs.storedValue = toWrite
1009
	fs.dirty = false
1010 1011
}

1012
// readDiskUsageFile is only safe to call in Open()
1013 1014
func (fs *Datastore) readDiskUsageFile() int64 {
	fpath := filepath.Join(fs.path, DiskUsageFile)
1015
	duB, err := readFile(fpath)
1016 1017 1018
	if err != nil {
		return 0
	}
1019
	err = json.Unmarshal(duB, &fs.storedValue)
1020 1021 1022
	if err != nil {
		return 0
	}
1023
	return fs.storedValue.DiskUsage
1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043
}

// 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
}

1044 1045 1046 1047
// 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 {
1048
	return string(fs.storedValue.Accuracy)
1049 1050
}

1051
func (fs *Datastore) tempFile() (*os.File, error) {
1052
	file, err := tempFile(fs.tempPath, "temp-")
1053 1054 1055
	return file, err
}

1056
func (fs *Datastore) walk(path string, qrb *query.ResultBuilder) error {
1057 1058
	dir, err := os.Open(path)
	if err != nil {
1059 1060 1061 1062
		if os.IsNotExist(err) {
			// not an error if the file disappeared
			return nil
		}
1063 1064
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
1065
	defer dir.Close()
1066 1067 1068 1069 1070 1071 1072 1073 1074 1075

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

1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087
	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 {
1088
			// not a block.
1089 1090 1091
			continue
		}

1092 1093 1094
		var result query.Result
		result.Key = key.String()
		if !qrb.Query.KeysOnly {
1095
			value, err := readFile(filepath.Join(path, fn))
1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113
			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
1114
		select {
1115 1116
		case qrb.Output <- result:
		case <-qrb.Process.Closing():
Steven Allen's avatar
Steven Allen committed
1117
			return nil
1118 1119 1120 1121 1122
		}
	}
	return nil
}

1123 1124 1125
// Deactivate closes background maintenance threads, most write
// operations will fail but readonly operations will continue to
// function
Steven Allen's avatar
Steven Allen committed
1126
func (fs *Datastore) deactivate() {
Steven Allen's avatar
Steven Allen committed
1127 1128 1129
	fs.shutdownLock.Lock()
	defer fs.shutdownLock.Unlock()
	if fs.shutdown {
Steven Allen's avatar
Steven Allen committed
1130
		return
1131
	}
Steven Allen's avatar
Steven Allen committed
1132 1133 1134
	fs.shutdown = true
	close(fs.checkpointCh)
	<-fs.done
Jeromy's avatar
Jeromy committed
1135 1136
}

1137
func (fs *Datastore) Close() error {
Steven Allen's avatar
Steven Allen committed
1138 1139
	fs.deactivate()
	return nil
1140 1141
}

Jeromy's avatar
Jeromy committed
1142
type flatfsBatch struct {
1143
	puts    map[datastore.Key][]byte
Jeromy's avatar
Jeromy committed
1144 1145 1146 1147 1148
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

Jeromy's avatar
Jeromy committed
1149
func (fs *Datastore) Batch() (datastore.Batch, error) {
Jeromy's avatar
Jeromy committed
1150
	return &flatfsBatch{
1151
		puts:    make(map[datastore.Key][]byte),
Jeromy's avatar
Jeromy committed
1152 1153
		deletes: make(map[datastore.Key]struct{}),
		ds:      fs,
Jeromy's avatar
Jeromy committed
1154
	}, nil
Jeromy's avatar
Jeromy committed
1155 1156
}

1157
func (bt *flatfsBatch) Put(key datastore.Key, val []byte) error {
1158
	if !keyIsValid(key) {
1159
		return fmt.Errorf("when putting '%q': %w", key, ErrInvalidKey)
1160
	}
Jeromy's avatar
Jeromy committed
1161 1162 1163 1164
	bt.puts[key] = val
	return nil
}

Jeromy's avatar
Jeromy committed
1165
func (bt *flatfsBatch) Delete(key datastore.Key) error {
1166 1167 1168
	if keyIsValid(key) {
		bt.deletes[key] = struct{}{}
	} // otherwise, delete is a no-op anyways.
Jeromy's avatar
Jeromy committed
1169 1170 1171
	return nil
}

Jeromy's avatar
Jeromy committed
1172
func (bt *flatfsBatch) Commit() error {
Jeromy's avatar
Jeromy committed
1173 1174 1175 1176
	if err := bt.ds.putMany(bt.puts); err != nil {
		return err
	}

Steven Allen's avatar
Steven Allen committed
1177
	for k := range bt.deletes {
Jeromy's avatar
Jeromy committed
1178 1179 1180 1181 1182 1183 1184
		if err := bt.ds.Delete(k); err != nil {
			return err
		}
	}

	return nil
}