flatfs.go 23.5 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"
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 96 97 98 99 100 101 102 103 104 105
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
106
type Datastore struct {
107 108 109 110 111
	// 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
112
	path string
113

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

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

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

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

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

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

138 139
type ShardFunc func(string) string

140
type opT int
141

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

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

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

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

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

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

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

239
	fs := &Datastore{
240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256
		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
257
	}
258

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

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

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

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

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

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

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

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

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

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

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

347 348 349 350 351 352 353 354 355 356
// 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.
Tommi Virtanen's avatar
Tommi Virtanen committed
357 358 359 360 361 362
func (fs *Datastore) Put(key datastore.Key, value interface{}) error {
	val, ok := value.([]byte)
	if !ok {
		return datastore.ErrInvalidType
	}

363
	var err error
Jeromy's avatar
Jeromy committed
364
	for i := 1; i <= putMaxRetries; i++ {
365 366 367 368 369
		err = fs.doWriteOp(&op{
			typ: opPut,
			key: key,
			v:   val,
		})
370
		if err == nil {
Jeromy's avatar
Jeromy committed
371
			break
372 373 374
		}

		if !strings.Contains(err.Error(), "too many open files") {
Jeromy's avatar
Jeromy committed
375
			break
376 377
		}

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

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 417 418 419 420
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
}

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

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

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

458
	err = fs.renameAndUpdateDiskUsage(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
459 460 461 462 463
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
464 465 466 467
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
468
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
469 470 471
	return nil
}

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

	for key, value := range data {
		val, ok := value.([]byte)
		if !ok {
			return datastore.ErrInvalidType
		}
		dir, path := fs.encode(key)
482
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
483 484 485 486 487 488 489 490 491 492 493 494 495
			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
		}

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

	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
522
		if fs.sync {
523
			if err := syncFile(fi); err != nil {
Jeromy's avatar
Jeromy committed
524 525
				return err
			}
Jeromy's avatar
Jeromy committed
526 527 528 529 530 531 532 533 534 535 536
		}

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

		// signify closed
		ops[fi] = 1
	}

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

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

Jeromy's avatar
Jeromy committed
551 552 553 554
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
555 556 557 558 559
	}

	return nil
}

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

585 586 587
// 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
588
func (fs *Datastore) Delete(key datastore.Key) error {
589 590 591 592 593 594 595 596 597 598
	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
599
	_, path := fs.encode(key)
600 601 602

	fSize := fileSize(path)

Tommi Virtanen's avatar
Tommi Virtanen committed
603 604
	switch err := os.Remove(path); {
	case err == nil:
605 606
		atomic.AddInt64(&fs.diskUsage, -fSize)
		fs.checkpointDiskUsage()
Tommi Virtanen's avatar
Tommi Virtanen committed
607 608 609 610 611 612
		return nil
	case os.IsNotExist(err):
		return datastore.ErrNotFound
	default:
		return err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
613 614 615
}

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
616 617 618 619 620 621 622 623 624 625 626
	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")
	}

627
	reschan := make(chan query.Result, query.KeysOnlyBufSize)
Jeromy's avatar
Jeromy committed
628 629
	go func() {
		defer close(reschan)
630
		err := fs.walkTopLevel(fs.path, reschan)
Jeromy's avatar
Jeromy committed
631
		if err != nil {
632
			reschan <- query.Result{Error: errors.New("walk failed: " + err.Error())}
633
		}
Jeromy's avatar
Jeromy committed
634 635
	}()
	return query.ResultsWithChan(q, reschan), nil
Tommi Virtanen's avatar
Tommi Virtanen committed
636 637
}

638 639 640 641 642
func (fs *Datastore) walkTopLevel(path string, reschan chan query.Result) error {
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
643
	defer dir.Close()
644 645 646 647 648
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return err
	}
	for _, dir := range names {
Kevin Atkinson's avatar
Kevin Atkinson committed
649 650 651 652 653

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

654 655 656 657
		err = fs.walk(filepath.Join(path, dir), reschan)
		if err != nil {
			return err
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
658

659 660 661 662
	}
	return nil
}

663 664 665
// 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.
666
func folderSize(path string, deadline time.Time) (int64, initAccuracy, error) {
667 668 669 670
	var du int64

	folder, err := os.Open(path)
	if err != nil {
671
		return 0, "", err
672 673 674 675 676
	}
	defer folder.Close()

	stat, err := folder.Stat()
	if err != nil {
677
		return 0, "", err
678 679 680 681
	}

	files, err := folder.Readdirnames(-1)
	if err != nil {
682
		return 0, "", err
683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699
	}

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

700
	accuracy := exactA
701
	for {
702 703 704
		// Do not process any files after deadline is over
		if time.Now().After(deadline) {
			accuracy = timedoutA
705 706 707
			break
		}

708 709 710 711
		if i >= totalFiles || filesProcessed >= maxFiles {
			if filesProcessed >= maxFiles {
				accuracy = approxA
			}
712 713 714 715 716 717 718 719
			break
		}

		// Stat the file
		fname := files[i]
		subpath := filepath.Join(path, fname)
		st, err := os.Stat(subpath)
		if err != nil {
720
			return 0, "", err
721 722 723 724
		}

		// Find folder size recursively
		if st.IsDir() {
725
			du2, acc, err := folderSize(filepath.Join(subpath), deadline)
726
			if err != nil {
727
				return 0, "", err
728
			}
729
			accuracy = combineAccuracy(acc, accuracy)
730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751
			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)
752
	return du, accuracy, nil
753 754 755 756
}

// calculateDiskUsage tries to read the DiskUsageFile for a cached
// diskUsage value, otherwise walks the datastore files.
757
// it is only safe to call in Open()
758 759 760
func (fs *Datastore) calculateDiskUsage() error {
	// Try to obtain a previously stored value from disk
	if persDu := fs.readDiskUsageFile(); persDu > 0 {
761
		fs.diskUsage = persDu
762 763 764
		return nil
	}

765 766 767 768 769 770 771
	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()
772
	deadline := time.Now().Add(DiskUsageCalcTimeout)
773
	du, accuracy, err := folderSize(fs.path, deadline)
774 775 776
	if err != nil {
		return err
	}
777 778 779
	if !msgTimer.Stop() {
		<-msgDone
	}
780
	if accuracy == timedoutA {
781 782 783 784 785 786 787
		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")
	}

788
	fs.storedValue.Accuracy = accuracy
789
	fs.diskUsage = du
790
	fs.writeDiskUsageFile(du, true)
791

792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812
	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 {
813 814
		atomic.AddInt64(&fs.diskUsage, fsize)
		fs.checkpointDiskUsage()
815 816 817
	}
}

818
func (fs *Datastore) checkpointDiskUsage() {
819
	select {
Kevin Atkinson's avatar
Kevin Atkinson committed
820
	case fs.checkpointCh <- struct{}{}:
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
821
		// msg sent
822 823
	default:
		// checkpoint request already pending
824 825 826
	}
}

827
func (fs *Datastore) checkpointLoop() {
828 829
	timerActive := true
	timer := time.NewTimer(0)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
830
	defer timer.Stop()
831
	for {
832 833 834
		select {
		case _, more := <-fs.checkpointCh:
			du := atomic.LoadInt64(&fs.diskUsage)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
835
			fs.dirty = true
836
			if !more { // shutting down
837
				fs.writeDiskUsageFile(du, true)
838
				if fs.dirty {
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
839
					log.Errorf("could not store final value of disk usage to file, future estimates may be inaccurate")
840
				}
Kevin Atkinson's avatar
Kevin Atkinson committed
841
				fs.done <- struct{}{}
842 843 844 845 846
				return
			}
			// If the difference between the checkpointed disk usage and
			// current one is larger than than `diskUsageCheckpointPercent`
			// of the checkpointed: store it.
847
			newDu := float64(du)
848
			lastCheckpointDu := float64(fs.storedValue.DiskUsage)
849
			diff := math.Abs(newDu - lastCheckpointDu)
850 851
			if lastCheckpointDu*diskUsageCheckpointPercent < diff*100.0 {
				fs.writeDiskUsageFile(du, false)
852
			}
853 854 855 856
			// 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
857
				timerActive = true
858 859 860 861 862
			}
		case <-timer.C:
			timerActive = false
			if fs.dirty {
				du := atomic.LoadInt64(&fs.diskUsage)
863
				fs.writeDiskUsageFile(du, false)
864
			}
865
		}
866
	}
867 868
}

869
func (fs *Datastore) writeDiskUsageFile(du int64, doSync bool) {
870
	tmp, err := ioutil.TempFile(fs.path, "du-")
871 872 873 874 875 876 877
	removed := false
	defer func() {
		if !removed {
			// silence errcheck
			_ = os.Remove(tmp.Name())
		}
	}()
878
	if err != nil {
879
		log.Warningf("cound not write disk usage: %v", err)
880 881
		return
	}
882

883 884
	toWrite := fs.storedValue
	toWrite.DiskUsage = du
885
	encoder := json.NewEncoder(tmp)
886
	if err := encoder.Encode(&toWrite); err != nil {
887
		log.Warningf("cound not write disk usage: %v", err)
888 889
		return
	}
890

891 892 893 894 895 896 897
	if doSync {
		if err := tmp.Sync(); err != nil {
			log.Warningf("cound not sync %s: %v", DiskUsageFile, err)
			return
		}
	}

898
	if err := tmp.Close(); err != nil {
899
		log.Warningf("cound not write disk usage: %v", err)
900 901 902
		return
	}

903
	if err := os.Rename(tmp.Name(), filepath.Join(fs.path, DiskUsageFile)); err != nil {
904
		log.Warningf("cound not write disk usage: %v", err)
905
		return
906
	}
907
	removed = true
908

909
	fs.storedValue = toWrite
910
	fs.dirty = false
911 912
}

913
// readDiskUsageFile is only safe to call in Open()
914 915 916 917 918 919
func (fs *Datastore) readDiskUsageFile() int64 {
	fpath := filepath.Join(fs.path, DiskUsageFile)
	duB, err := ioutil.ReadFile(fpath)
	if err != nil {
		return 0
	}
920
	err = json.Unmarshal(duB, &fs.storedValue)
921 922 923
	if err != nil {
		return 0
	}
924
	return fs.storedValue.DiskUsage
925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944
}

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

945 946 947 948
// 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 {
949
	return string(fs.storedValue.Accuracy)
950 951
}

952 953 954
func (fs *Datastore) walk(path string, reschan chan query.Result) error {
	dir, err := os.Open(path)
	if err != nil {
955 956 957 958
		if os.IsNotExist(err) {
			// not an error if the file disappeared
			return nil
		}
959 960
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
961
	defer dir.Close()
962 963 964 965 966 967 968 969 970 971

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

972 973 974 975 976 977 978 979 980 981 982 983
	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 {
984
			log.Warningf("failed to decode flatfs entry: %s", fn)
985 986 987 988 989 990 991 992 993 994 995 996
			continue
		}

		reschan <- query.Result{
			Entry: query.Entry{
				Key: key.String(),
			},
		}
	}
	return nil
}

997 998 999 1000 1001 1002 1003 1004 1005
// 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
1006 1007 1008
	return nil
}

1009 1010 1011 1012
func (fs *Datastore) Close() error {
	return fs.deactivate()
}

Jeromy's avatar
Jeromy committed
1013
type flatfsBatch struct {
Jeromy's avatar
Jeromy committed
1014 1015 1016 1017 1018 1019
	puts    map[datastore.Key]interface{}
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

Jeromy's avatar
Jeromy committed
1020
func (fs *Datastore) Batch() (datastore.Batch, error) {
Jeromy's avatar
Jeromy committed
1021
	return &flatfsBatch{
Jeromy's avatar
Jeromy committed
1022 1023 1024
		puts:    make(map[datastore.Key]interface{}),
		deletes: make(map[datastore.Key]struct{}),
		ds:      fs,
Jeromy's avatar
Jeromy committed
1025
	}, nil
Jeromy's avatar
Jeromy committed
1026 1027
}

Jeromy's avatar
Jeromy committed
1028
func (bt *flatfsBatch) Put(key datastore.Key, val interface{}) error {
Jeromy's avatar
Jeromy committed
1029 1030 1031 1032
	bt.puts[key] = val
	return nil
}

Jeromy's avatar
Jeromy committed
1033
func (bt *flatfsBatch) Delete(key datastore.Key) error {
Jeromy's avatar
Jeromy committed
1034 1035 1036 1037
	bt.deletes[key] = struct{}{}
	return nil
}

Jeromy's avatar
Jeromy committed
1038
func (bt *flatfsBatch) Commit() error {
Jeromy's avatar
Jeromy committed
1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051
	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
1052 1053 1054
var _ datastore.ThreadSafeDatastore = (*Datastore)(nil)

func (*Datastore) IsThreadSafe() {}