flatfs.go 24 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 97 98 99 100 101 102 103 104 105 106
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)
)

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

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

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

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

Kevin Atkinson's avatar
Kevin Atkinson committed
126 127
	checkpointCh chan struct{}
	done         chan struct{}
128 129 130 131

	// opMap handles concurrent write operations (put/delete)
	// to the same key
	opMap *opMap
Tommi Virtanen's avatar
Tommi Virtanen committed
132 133
}

134
type diskUsageValue struct {
135 136
	DiskUsage int64        `json:"diskUsage"`
	Accuracy  initAccuracy `json:"accuracy"`
137 138
}

139 140
type ShardFunc func(string) string

141
type opT int
142

143 144 145 146 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
// 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()
}
191

192
func Create(path string, fun *ShardIdV1) error {
193

kpcyrd's avatar
kpcyrd committed
194
	err := os.Mkdir(path, 0755)
195 196
	if err != nil && !os.IsExist(err) {
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
197 198
	}

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

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

227
func Open(path string, syncFiles bool) (*Datastore, error) {
228 229
	_, err := os.Stat(path)
	if os.IsNotExist(err) {
Kevin Atkinson's avatar
Kevin Atkinson committed
230
		return nil, ErrDatastoreDoesNotExist
231 232 233 234
	} else if err != nil {
		return nil, err
	}

235
	shardId, err := ReadShardFunc(path)
236
	if err != nil {
237
		return nil, err
238 239
	}

240
	fs := &Datastore{
241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257
		path:      path,
		shardStr:  shardId.String(),
		getDir:    shardId.Func(),
		sync:      syncFiles,
		diskUsage: 0,
		opMap:     new(opMap),
	}

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

Kevin Atkinson's avatar
Kevin Atkinson committed
260 261 262
	fs.checkpointCh = make(chan struct{}, 1)
	fs.done = make(chan struct{})
	go fs.checkpointLoop()
263 264 265
	return fs, nil
}

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

275 276
func (fs *Datastore) ShardStr() string {
	return fs.shardStr
277 278
}

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

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

294 295
func (fs *Datastore) makeDir(dir string) error {
	if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
296
		return err
297 298 299 300 301 302
	}

	// 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
303 304 305 306
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
307 308 309 310
	}
	return nil
}

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

	// Track DiskUsage of this NEW folder
	fs.updateDiskUsage(dir, true)
Jeromy's avatar
Jeromy committed
323 324 325
	return nil
}

326 327 328 329 330 331 332 333 334 335 336 337 338 339 340
// 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.
341
	err = os.Rename(tmpPath, path)
342 343 344 345
	fs.updateDiskUsage(path, true)
	return err
}

Jeromy's avatar
Jeromy committed
346
var putMaxRetries = 6
347

348 349 350 351 352 353 354 355 356 357
// 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.
358
func (fs *Datastore) Put(key datastore.Key, value []byte) error {
359
	var err error
Jeromy's avatar
Jeromy committed
360
	for i := 1; i <= putMaxRetries; i++ {
361 362 363
		err = fs.doWriteOp(&op{
			typ: opPut,
			key: key,
364
			v:   value,
365
		})
366
		if err == nil {
Jeromy's avatar
Jeromy committed
367
			break
368 369 370
		}

		if !strings.Contains(err.Error(), "too many open files") {
Jeromy's avatar
Jeromy committed
371
			break
372 373
		}

Or Rikon's avatar
Or Rikon committed
374
		log.Errorf("too many open files, retrying in %dms", 100*i)
Jeromy's avatar
Jeromy committed
375
		time.Sleep(time.Millisecond * 100 * time.Duration(i))
376 377 378 379
	}
	return err
}

380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416
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")
	}
}

// doWrite optmizes out write operations (put/delete) to the same
// key by queueing them and suceeding all queued
// operations if one of them does. In such case,
// we assume that the first suceeding operation
// 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
}

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

Tommi Virtanen's avatar
Tommi Virtanen committed
419
	dir, path := fs.encode(key)
420
	if err := fs.makeDir(dir); err != nil {
421
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443
	}

	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
444
	if fs.sync {
445
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
446 447
			return err
		}
448
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
449 450 451 452 453
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

454
	err = fs.renameAndUpdateDiskUsage(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
455 456 457 458 459
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
460 461 462 463
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
464
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
465 466 467
	return nil
}

Jeromy's avatar
Jeromy committed
468 469
func (fs *Datastore) putMany(data map[datastore.Key]interface{}) error {
	var dirsToSync []string
470
	files := make(map[*os.File]*op)
Jeromy's avatar
Jeromy committed
471 472 473 474 475 476 477

	for key, value := range data {
		val, ok := value.([]byte)
		if !ok {
			return datastore.ErrInvalidType
		}
		dir, path := fs.encode(key)
478
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
479 480 481 482 483 484 485 486 487 488 489 490 491
			return err
		}
		dirsToSync = append(dirsToSync, dir)

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

		if _, err := tmp.Write(val); err != nil {
			return err
		}

492 493 494 495 496 497
		files[tmp] = &op{
			typ:  opRename,
			path: path,
			tmp:  tmp.Name(),
			key:  key,
		}
Jeromy's avatar
Jeromy committed
498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517
	}

	ops := make(map[*os.File]int)

	defer func() {
		for fi, _ := range files {
			val, _ := ops[fi]
			switch val {
			case 0:
				_ = fi.Close()
				fallthrough
			case 1:
				_ = os.Remove(fi.Name())
			}
		}
	}()

	// Now we sync everything
	// sync and close files
	for fi, _ := range files {
Jeromy's avatar
Jeromy committed
518
		if fs.sync {
519
			if err := syncFile(fi); err != nil {
Jeromy's avatar
Jeromy committed
520 521
				return err
			}
Jeromy's avatar
Jeromy committed
522 523 524 525 526 527 528 529 530 531 532
		}

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

		// signify closed
		ops[fi] = 1
	}

	// move files to their proper places
533 534
	for fi, op := range files {
		fs.doWriteOp(op)
Jeromy's avatar
Jeromy committed
535 536 537 538 539
		// signify removed
		ops[fi] = 2
	}

	// now sync the dirs for those files
Jeromy's avatar
Jeromy committed
540 541 542 543 544
	if fs.sync {
		for _, dir := range dirsToSync {
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
545 546
		}

Jeromy's avatar
Jeromy committed
547 548 549 550
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
551 552 553 554 555
	}

	return nil
}

556
func (fs *Datastore) Get(key datastore.Key) (value []byte, err error) {
Tommi Virtanen's avatar
Tommi Virtanen committed
557 558 559 560 561 562 563 564 565 566 567 568 569
	_, 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) {
Tommi Virtanen's avatar
Tommi Virtanen committed
570 571 572 573 574 575 576 577 578
	_, 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
579 580
}

Steven Allen's avatar
Steven Allen committed
581 582 583 584 585 586 587 588 589 590 591 592
func (fs *Datastore) GetSize(key datastore.Key) (size int, err error) {
	_, 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
	}
}

593 594 595
// 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
596
func (fs *Datastore) Delete(key datastore.Key) error {
597 598 599 600 601 602 603 604 605 606
	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
607
	_, path := fs.encode(key)
608 609 610

	fSize := fileSize(path)

Tommi Virtanen's avatar
Tommi Virtanen committed
611 612
	switch err := os.Remove(path); {
	case err == nil:
613 614
		atomic.AddInt64(&fs.diskUsage, -fSize)
		fs.checkpointDiskUsage()
Tommi Virtanen's avatar
Tommi Virtanen committed
615 616 617 618 619 620
		return nil
	case os.IsNotExist(err):
		return datastore.ErrNotFound
	default:
		return err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
621 622 623
}

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
624 625 626 627 628 629 630 631 632 633 634
	if (q.Prefix != "" && q.Prefix != "/") ||
		len(q.Filters) > 0 ||
		len(q.Orders) > 0 ||
		q.Limit > 0 ||
		q.Offset > 0 ||
		!q.KeysOnly {
		// TODO this is overly simplistic, but the only caller is
		// `ipfs refs local` for now, and this gets us moving.
		return nil, errors.New("flatfs only supports listing all keys in random order")
	}

Steven Allen's avatar
Steven Allen committed
635 636 637 638 639 640 641
	// 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
642
		}
Steven Allen's avatar
Steven Allen committed
643 644 645 646 647 648 649 650
		select {
		case b.Output <- query.Result{Error: errors.New("walk failed: " + err.Error())}:
		case <-p.Closing():
		}
	})
	go b.Process.CloseAfterChildren()

	return b.Results(), nil
Tommi Virtanen's avatar
Tommi Virtanen committed
651 652
}

Steven Allen's avatar
Steven Allen committed
653
func (fs *Datastore) walkTopLevel(path string, result *query.ResultBuilder) error {
654 655 656 657
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
658
	defer dir.Close()
659 660 661 662 663
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return err
	}
	for _, dir := range names {
Kevin Atkinson's avatar
Kevin Atkinson committed
664 665 666 667
		if len(dir) == 0 || dir[0] == '.' {
			continue
		}

Steven Allen's avatar
Steven Allen committed
668
		err = fs.walk(filepath.Join(path, dir), result)
669 670 671
		if err != nil {
			return err
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
672

Steven Allen's avatar
Steven Allen committed
673 674 675 676 677 678
		// Are we closing?
		select {
		case <-result.Process.Closing():
			return nil
		default:
		}
679 680 681 682
	}
	return nil
}

683 684 685
// folderSize estimates the diskUsage of a folder by reading
// up to DiskUsageFilesAverage entries in it and assumming any
// other files will have an avereage size.
686
func folderSize(path string, deadline time.Time) (int64, initAccuracy, error) {
687 688 689 690
	var du int64

	folder, err := os.Open(path)
	if err != nil {
691
		return 0, "", err
692 693 694 695 696
	}
	defer folder.Close()

	stat, err := folder.Stat()
	if err != nil {
697
		return 0, "", err
698 699 700 701
	}

	files, err := folder.Readdirnames(-1)
	if err != nil {
702
		return 0, "", err
703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719
	}

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

720
	accuracy := exactA
721
	for {
722 723 724
		// Do not process any files after deadline is over
		if time.Now().After(deadline) {
			accuracy = timedoutA
725 726 727
			break
		}

728 729 730 731
		if i >= totalFiles || filesProcessed >= maxFiles {
			if filesProcessed >= maxFiles {
				accuracy = approxA
			}
732 733 734 735 736 737 738 739
			break
		}

		// Stat the file
		fname := files[i]
		subpath := filepath.Join(path, fname)
		st, err := os.Stat(subpath)
		if err != nil {
740
			return 0, "", err
741 742 743 744
		}

		// Find folder size recursively
		if st.IsDir() {
745
			du2, acc, err := folderSize(filepath.Join(subpath), deadline)
746
			if err != nil {
747
				return 0, "", err
748
			}
749
			accuracy = combineAccuracy(acc, accuracy)
750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771
			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)
772
	return du, accuracy, nil
773 774 775 776
}

// calculateDiskUsage tries to read the DiskUsageFile for a cached
// diskUsage value, otherwise walks the datastore files.
777
// it is only safe to call in Open()
778 779 780
func (fs *Datastore) calculateDiskUsage() error {
	// Try to obtain a previously stored value from disk
	if persDu := fs.readDiskUsageFile(); persDu > 0 {
781
		fs.diskUsage = persDu
782 783 784
		return nil
	}

785 786 787 788 789 790 791
	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()
792
	deadline := time.Now().Add(DiskUsageCalcTimeout)
793
	du, accuracy, err := folderSize(fs.path, deadline)
794 795 796
	if err != nil {
		return err
	}
797 798 799
	if !msgTimer.Stop() {
		<-msgDone
	}
800
	if accuracy == timedoutA {
801 802 803 804 805 806 807
		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")
	}

808
	fs.storedValue.Accuracy = accuracy
809
	fs.diskUsage = du
810
	fs.writeDiskUsageFile(du, true)
811

812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832
	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 {
833 834
		atomic.AddInt64(&fs.diskUsage, fsize)
		fs.checkpointDiskUsage()
835 836 837
	}
}

838
func (fs *Datastore) checkpointDiskUsage() {
839
	select {
Kevin Atkinson's avatar
Kevin Atkinson committed
840
	case fs.checkpointCh <- struct{}{}:
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
841
		// msg sent
842 843
	default:
		// checkpoint request already pending
844 845 846
	}
}

847
func (fs *Datastore) checkpointLoop() {
848 849
	timerActive := true
	timer := time.NewTimer(0)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
850
	defer timer.Stop()
851
	for {
852 853 854
		select {
		case _, more := <-fs.checkpointCh:
			du := atomic.LoadInt64(&fs.diskUsage)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
855
			fs.dirty = true
856
			if !more { // shutting down
857
				fs.writeDiskUsageFile(du, true)
858
				if fs.dirty {
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
859
					log.Errorf("could not store final value of disk usage to file, future estimates may be inaccurate")
860
				}
Kevin Atkinson's avatar
Kevin Atkinson committed
861
				fs.done <- struct{}{}
862 863 864 865 866
				return
			}
			// If the difference between the checkpointed disk usage and
			// current one is larger than than `diskUsageCheckpointPercent`
			// of the checkpointed: store it.
867
			newDu := float64(du)
868
			lastCheckpointDu := float64(fs.storedValue.DiskUsage)
869
			diff := math.Abs(newDu - lastCheckpointDu)
870 871
			if lastCheckpointDu*diskUsageCheckpointPercent < diff*100.0 {
				fs.writeDiskUsageFile(du, false)
872
			}
873 874 875 876
			// 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
877
				timerActive = true
878 879 880 881 882
			}
		case <-timer.C:
			timerActive = false
			if fs.dirty {
				du := atomic.LoadInt64(&fs.diskUsage)
883
				fs.writeDiskUsageFile(du, false)
884
			}
885
		}
886
	}
887 888
}

889
func (fs *Datastore) writeDiskUsageFile(du int64, doSync bool) {
890
	tmp, err := ioutil.TempFile(fs.path, "du-")
891 892 893 894 895
	if err != nil {
		log.Warningf("cound not write disk usage: %v", err)
		return
	}

896 897 898 899 900 901 902
	removed := false
	defer func() {
		if !removed {
			// silence errcheck
			_ = os.Remove(tmp.Name())
		}
	}()
903

904 905
	toWrite := fs.storedValue
	toWrite.DiskUsage = du
906
	encoder := json.NewEncoder(tmp)
907
	if err := encoder.Encode(&toWrite); err != nil {
908
		log.Warningf("cound not write disk usage: %v", err)
909 910
		return
	}
911

912 913 914 915 916 917 918
	if doSync {
		if err := tmp.Sync(); err != nil {
			log.Warningf("cound not sync %s: %v", DiskUsageFile, err)
			return
		}
	}

919
	if err := tmp.Close(); err != nil {
920
		log.Warningf("cound not write disk usage: %v", err)
921 922 923
		return
	}

924
	if err := os.Rename(tmp.Name(), filepath.Join(fs.path, DiskUsageFile)); err != nil {
925
		log.Warningf("cound not write disk usage: %v", err)
926
		return
927
	}
928
	removed = true
929

930
	fs.storedValue = toWrite
931
	fs.dirty = false
932 933
}

934
// readDiskUsageFile is only safe to call in Open()
935 936 937 938 939 940
func (fs *Datastore) readDiskUsageFile() int64 {
	fpath := filepath.Join(fs.path, DiskUsageFile)
	duB, err := ioutil.ReadFile(fpath)
	if err != nil {
		return 0
	}
941
	err = json.Unmarshal(duB, &fs.storedValue)
942 943 944
	if err != nil {
		return 0
	}
945
	return fs.storedValue.DiskUsage
946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965
}

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

966 967 968 969
// 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 {
970
	return string(fs.storedValue.Accuracy)
971 972
}

Steven Allen's avatar
Steven Allen committed
973
func (fs *Datastore) walk(path string, result *query.ResultBuilder) error {
974 975
	dir, err := os.Open(path)
	if err != nil {
976 977 978 979
		if os.IsNotExist(err) {
			// not an error if the file disappeared
			return nil
		}
980 981
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
982
	defer dir.Close()
983 984 985 986 987 988 989 990 991 992

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

993 994 995 996 997 998 999 1000 1001 1002 1003 1004
	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 {
1005
			log.Warningf("failed to decode flatfs entry: %s", fn)
1006 1007 1008
			continue
		}

Steven Allen's avatar
Steven Allen committed
1009 1010
		select {
		case result.Output <- query.Result{
1011 1012 1013
			Entry: query.Entry{
				Key: key.String(),
			},
Steven Allen's avatar
Steven Allen committed
1014 1015 1016
		}:
		case <-result.Process.Closing():
			return nil
1017 1018 1019 1020 1021
		}
	}
	return nil
}

1022 1023 1024 1025 1026 1027 1028 1029 1030
// Deactivate closes background maintenance threads, most write
// operations will fail but readonly operations will continue to
// function
func (fs *Datastore) deactivate() error {
	if fs.checkpointCh != nil {
		close(fs.checkpointCh)
		<-fs.done
		fs.checkpointCh = nil
	}
Jeromy's avatar
Jeromy committed
1031 1032 1033
	return nil
}

1034 1035 1036 1037
func (fs *Datastore) Close() error {
	return fs.deactivate()
}

Jeromy's avatar
Jeromy committed
1038
type flatfsBatch struct {
Jeromy's avatar
Jeromy committed
1039 1040 1041 1042 1043 1044
	puts    map[datastore.Key]interface{}
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

Jeromy's avatar
Jeromy committed
1045
func (fs *Datastore) Batch() (datastore.Batch, error) {
Jeromy's avatar
Jeromy committed
1046
	return &flatfsBatch{
Jeromy's avatar
Jeromy committed
1047 1048 1049
		puts:    make(map[datastore.Key]interface{}),
		deletes: make(map[datastore.Key]struct{}),
		ds:      fs,
Jeromy's avatar
Jeromy committed
1050
	}, nil
Jeromy's avatar
Jeromy committed
1051 1052
}

1053
func (bt *flatfsBatch) Put(key datastore.Key, val []byte) error {
Jeromy's avatar
Jeromy committed
1054 1055 1056 1057
	bt.puts[key] = val
	return nil
}

Jeromy's avatar
Jeromy committed
1058
func (bt *flatfsBatch) Delete(key datastore.Key) error {
Jeromy's avatar
Jeromy committed
1059 1060 1061 1062
	bt.deletes[key] = struct{}{}
	return nil
}

Jeromy's avatar
Jeromy committed
1063
func (bt *flatfsBatch) Commit() error {
Jeromy's avatar
Jeromy committed
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076
	if err := bt.ds.putMany(bt.puts); err != nil {
		return err
	}

	for k, _ := range bt.deletes {
		if err := bt.ds.Delete(k); err != nil {
			return err
		}
	}

	return nil
}

Tommi Virtanen's avatar
Tommi Virtanen committed
1077 1078 1079
var _ datastore.ThreadSafeDatastore = (*Datastore)(nil)

func (*Datastore) IsThreadSafe() {}