flatfs.go 25.7 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 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

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

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

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

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

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

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

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

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

144 145
type ShardFunc func(string) string

146
type opT int
147

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
// 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()
}
196

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

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
	shardId, err := ReadShardFunc(path)
241
	if err != nil {
242
		return nil, err
243 244
	}

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

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

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

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

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

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

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

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

	// 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
308 309 310 311
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
312 313 314 315
	}
	return nil
}

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

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

331 332 333 334 335 336 337 338 339 340 341 342 343 344 345
// 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.
346
	err = os.Rename(tmpPath, path)
347 348 349 350
	fs.updateDiskUsage(path, true)
	return err
}

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

353 354 355 356 357 358 359 360 361 362
// 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.
363
func (fs *Datastore) Put(key datastore.Key, value []byte) error {
364 365 366 367
	if !keyIsValid(key) {
		return fmt.Errorf("key not supported by flatfs: '%q'", key)
	}

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

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

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

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

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

	return nil
}

405 406 407 408 409 410 411 412 413 414 415 416 417
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")
	}
}

418 419
// doWrite optimizes out write operations (put/delete) to the same
// key by queueing them and succeeding all queued
420
// operations if one of them does. In such case,
421
// we assume that the first succeeding operation
422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441
// 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
}

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

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

	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
469
	if fs.sync {
470
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
471 472
			return err
		}
473
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
474 475 476 477 478
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

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

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

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

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

502 503 504 505 506
	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
507
			val := ops[fi]
508 509 510 511 512 513 514
			switch val {
			case 0:
				_ = fi.Close()
				fallthrough
			case 1:
				_ = os.Remove(fi.Name())
			}
Jeromy's avatar
Jeromy committed
515
		}
516 517 518
	}()

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

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

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

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

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

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

		// signify closed
		ops[fi] = 1
	}

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

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

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

	return nil
}

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

Tommi Virtanen's avatar
Tommi Virtanen committed
592 593 594 595 596 597 598 599 600 601 602 603 604
	_, 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) {
605 606 607 608 609
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return false, nil
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
610 611 612 613 614 615 616 617 618
	_, 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
619 620
}

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

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

638 639 640
// 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
641
func (fs *Datastore) Delete(key datastore.Key) error {
642 643 644 645 646
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil
	}

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

653 654 655 656 657 658 659 660 661 662
	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
663
	_, path := fs.encode(key)
664 665 666

	fSize := fileSize(path)

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

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
Steven Allen's avatar
Steven Allen committed
680
	prefix := datastore.NewKey(q.Prefix).String()
681 682 683 684 685 686
	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.
		return query.ResultsWithEntries(q, nil), nil
	}
687

Steven Allen's avatar
Steven Allen committed
688 689 690 691 692 693 694
	// 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
695
		}
Steven Allen's avatar
Steven Allen committed
696 697 698 699 700
		select {
		case b.Output <- query.Result{Error: errors.New("walk failed: " + err.Error())}:
		case <-p.Closing():
		}
	})
Steven Allen's avatar
Steven Allen committed
701
	go b.Process.CloseAfterChildren() //nolint
Steven Allen's avatar
Steven Allen committed
702

703 704 705 706 707 708 709 710 711 712 713 714
	results := b.Results()

	if len(q.Filters) > 0 ||
		len(q.Orders) > 0 ||
		q.Limit > 0 ||
		q.Offset > 0 {
		nq := q
		nq.Prefix = "" // already handled
		results = query.NaiveQueryApply(nq, results)
	}

	return results, nil
Tommi Virtanen's avatar
Tommi Virtanen committed
715 716
}

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

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

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

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

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

	stat, err := folder.Stat()
	if err != nil {
761
		return 0, "", err
762 763 764 765
	}

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

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

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

792 793 794 795
		if i >= totalFiles || filesProcessed >= maxFiles {
			if filesProcessed >= maxFiles {
				accuracy = approxA
			}
796 797 798 799 800 801 802 803
			break
		}

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

		// Find folder size recursively
		if st.IsDir() {
809
			du2, acc, err := folderSize(filepath.Join(subpath), deadline)
810
			if err != nil {
811
				return 0, "", err
812
			}
813
			accuracy = combineAccuracy(acc, accuracy)
814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835
			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)
836
	return du, accuracy, nil
837 838 839 840
}

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

849 850 851 852 853 854 855
	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()
856
	deadline := time.Now().Add(DiskUsageCalcTimeout)
857
	du, accuracy, err := folderSize(fs.path, deadline)
858 859 860
	if err != nil {
		return err
	}
861 862 863
	if !msgTimer.Stop() {
		<-msgDone
	}
864
	if accuracy == timedoutA {
865 866 867 868 869 870 871
		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")
	}

872
	fs.storedValue.Accuracy = accuracy
873
	fs.diskUsage = du
874
	fs.writeDiskUsageFile(du, true)
875

876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896
	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 {
897 898
		atomic.AddInt64(&fs.diskUsage, fsize)
		fs.checkpointDiskUsage()
899 900 901
	}
}

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

911
func (fs *Datastore) checkpointLoop() {
Steven Allen's avatar
Steven Allen committed
912 913
	defer close(fs.done)

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

954
func (fs *Datastore) writeDiskUsageFile(du int64, doSync bool) {
955
	tmp, err := ioutil.TempFile(fs.path, "du-")
956 957 958 959 960
	if err != nil {
		log.Warningf("cound not write disk usage: %v", err)
		return
	}

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

969 970
	toWrite := fs.storedValue
	toWrite.DiskUsage = du
971
	encoder := json.NewEncoder(tmp)
972
	if err := encoder.Encode(&toWrite); err != nil {
973
		log.Warningf("cound not write disk usage: %v", err)
974 975
		return
	}
976

977 978 979 980 981 982 983
	if doSync {
		if err := tmp.Sync(); err != nil {
			log.Warningf("cound not sync %s: %v", DiskUsageFile, err)
			return
		}
	}

984
	if err := tmp.Close(); err != nil {
985
		log.Warningf("cound not write disk usage: %v", err)
986 987 988
		return
	}

989
	if err := os.Rename(tmp.Name(), filepath.Join(fs.path, DiskUsageFile)); err != nil {
990
		log.Warningf("cound not write disk usage: %v", err)
991
		return
992
	}
993
	removed = true
994

995
	fs.storedValue = toWrite
996
	fs.dirty = false
997 998
}

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

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

1031 1032 1033 1034
// 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 {
1035
	return string(fs.storedValue.Accuracy)
1036 1037
}

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

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

1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069
	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 {
1070
			log.Warningf("failed to decode flatfs entry: %s", fn)
1071 1072 1073
			continue
		}

1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095
		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
1096
		select {
1097 1098
		case qrb.Output <- result:
		case <-qrb.Process.Closing():
Steven Allen's avatar
Steven Allen committed
1099
			return nil
1100 1101 1102 1103 1104
		}
	}
	return nil
}

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

1119
func (fs *Datastore) Close() error {
Steven Allen's avatar
Steven Allen committed
1120 1121
	fs.deactivate()
	return nil
1122 1123
}

Jeromy's avatar
Jeromy committed
1124
type flatfsBatch struct {
1125
	puts    map[datastore.Key][]byte
Jeromy's avatar
Jeromy committed
1126 1127 1128 1129 1130
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

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

1139
func (bt *flatfsBatch) Put(key datastore.Key, val []byte) error {
1140 1141 1142
	if !keyIsValid(key) {
		return fmt.Errorf("key not supported by flatfs: '%q'", key)
	}
Jeromy's avatar
Jeromy committed
1143 1144 1145 1146
	bt.puts[key] = val
	return nil
}

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

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

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

	return nil
}