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

import (
7
	"encoding/json"
Tommi Virtanen's avatar
Tommi Virtanen committed
8
	"errors"
9
	"fmt"
10 11
	"math"
	"math/rand"
Tommi Virtanen's avatar
Tommi Virtanen committed
12
	"os"
Jeromy's avatar
Jeromy committed
13
	"path/filepath"
Tommi Virtanen's avatar
Tommi Virtanen committed
14
	"strings"
15 16
	"sync"
	"sync/atomic"
Will's avatar
Will committed
17
	"syscall"
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
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
Will's avatar
Will committed
52 53 54
	// RetryDelay is a timeout for a backoff on retrying operations
	// that fail due to transient errors like too many file descriptors open.
	RetryDelay = time.Millisecond * 200
Steven Allen's avatar
Steven Allen committed
55 56 57 58

	// RetryAttempts is the maximum number of retries that will be attempted
	// before giving up.
	RetryAttempts = 6
59 60 61 62 63 64 65 66
)

const (
	opPut = iota
	opDelete
	opRename
)

67 68 69
type initAccuracy string

const (
70
	unknownA  initAccuracy = "unknown"
71 72 73 74 75 76
	exactA    initAccuracy = "initial-exact"
	approxA   initAccuracy = "initial-approximate"
	timedoutA initAccuracy = "initial-timed-out"
)

func combineAccuracy(a, b initAccuracy) initAccuracy {
77 78 79
	if a == unknownA || b == unknownA {
		return unknownA
	}
80 81 82 83 84 85 86 87 88
	if a == timedoutA || b == timedoutA {
		return timedoutA
	}
	if a == approxA || b == approxA {
		return approxA
	}
	if a == exactA && b == exactA {
		return exactA
	}
89 90 91 92 93 94 95
	if a == "" {
		return b
	}
	if b == "" {
		return a
	}
	return unknownA
96 97
}

98 99 100 101 102 103
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
104
	ErrClosed                = errors.New("datastore closed")
105
	ErrInvalidKey            = errors.New("key not supported by flatfs")
106 107 108 109 110 111 112 113 114 115
)

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
116
type Datastore struct {
117 118 119 120 121
	// 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

122 123
	path     string
	tempPath string
124

125 126
	shardStr string
	getDir   ShardFunc
Jeromy's avatar
Jeromy committed
127 128 129

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

131 132 133
	// these values should only be used during internalization or
	// inside the checkpoint loop
	dirty       bool
134
	storedValue diskUsageValue
135

Steven Allen's avatar
Steven Allen committed
136
	// Used to trigger a checkpoint.
Kevin Atkinson's avatar
Kevin Atkinson committed
137 138
	checkpointCh chan struct{}
	done         chan struct{}
139

Steven Allen's avatar
Steven Allen committed
140 141 142
	shutdownLock sync.RWMutex
	shutdown     bool

143 144 145
	// opMap handles concurrent write operations (put/delete)
	// to the same key
	opMap *opMap
Tommi Virtanen's avatar
Tommi Virtanen committed
146 147
}

148
type diskUsageValue struct {
149 150
	DiskUsage int64        `json:"diskUsage"`
	Accuracy  initAccuracy `json:"accuracy"`
151 152
}

153 154
type ShardFunc func(string) string

155
type opT int
156

157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204
// 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()
}
205

206
func Create(path string, fun *ShardIdV1) error {
kpcyrd's avatar
kpcyrd committed
207
	err := os.Mkdir(path, 0755)
208 209
	if err != nil && !os.IsExist(err) {
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
210 211
	}

212 213
	dsFun, err := ReadShardFunc(path)
	switch err {
Kevin Atkinson's avatar
Kevin Atkinson committed
214
	case ErrShardingFileMissing:
215 216 217 218 219
		isEmpty, err := DirIsEmpty(path)
		if err != nil {
			return err
		}
		if !isEmpty {
220
			return fmt.Errorf("directory missing %s file: %s", SHARDING_FN, path)
221 222 223
		}

		err = WriteShardFunc(path, fun)
224 225 226
		if err != nil {
			return err
		}
227 228
		err = WriteReadme(path, fun)
		return err
229
	case nil:
230
		if fun.String() != dsFun.String() {
231
			return fmt.Errorf("specified shard func '%s' does not match repo shard func '%s'",
232
				fun.String(), dsFun.String())
233
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
234
		return ErrDatastoreExists
235
	default:
236
		return err
237
	}
238 239
}

240
func Open(path string, syncFiles bool) (*Datastore, error) {
241 242
	_, err := os.Stat(path)
	if os.IsNotExist(err) {
Kevin Atkinson's avatar
Kevin Atkinson committed
243
		return nil, ErrDatastoreDoesNotExist
244 245 246 247
	} else if err != nil {
		return nil, err
	}

248 249 250
	tempPath := filepath.Join(path, ".temp")
	err = os.RemoveAll(tempPath)
	if err != nil && !os.IsNotExist(err) {
251
		return nil, fmt.Errorf("failed to remove temporary directory: %v", err)
252 253 254 255
	}

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

259
	shardId, err := ReadShardFunc(path)
260
	if err != nil {
261
		return nil, err
262 263
	}

264
	fs := &Datastore{
Steven Allen's avatar
Steven Allen committed
265
		path:         path,
266
		tempPath:     tempPath,
Steven Allen's avatar
Steven Allen committed
267 268 269 270 271 272 273
		shardStr:     shardId.String(),
		getDir:       shardId.Func(),
		sync:         syncFiles,
		checkpointCh: make(chan struct{}, 1),
		done:         make(chan struct{}),
		diskUsage:    0,
		opMap:        new(opMap),
274 275 276 277 278 279 280 281 282 283 284
	}

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

Kevin Atkinson's avatar
Kevin Atkinson committed
287
	go fs.checkpointLoop()
288 289 290
	return fs, nil
}

291
// convenience method
292
func CreateOrOpen(path string, fun *ShardIdV1, sync bool) (*Datastore, error) {
293
	err := Create(path, fun)
Kevin Atkinson's avatar
Kevin Atkinson committed
294
	if err != nil && err != ErrDatastoreExists {
295 296 297 298 299
		return nil, err
	}
	return Open(path, sync)
}

300 301
func (fs *Datastore) ShardStr() string {
	return fs.shardStr
302 303
}

Tommi Virtanen's avatar
Tommi Virtanen committed
304
func (fs *Datastore) encode(key datastore.Key) (dir, file string) {
305
	noslash := key.String()[1:]
Jeromy's avatar
Jeromy committed
306 307
	dir = filepath.Join(fs.path, fs.getDir(noslash))
	file = filepath.Join(dir, noslash+extension)
Tommi Virtanen's avatar
Tommi Virtanen committed
308 309 310
	return dir, file
}

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

324 325
func (fs *Datastore) makeDir(dir string) error {
	if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
326
		return err
327 328 329 330 331 332
	}

	// 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
333 334 335 336
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
337 338 339 340
	}
	return nil
}

341
func (fs *Datastore) makeDirNoSync(dir string) error {
kpcyrd's avatar
kpcyrd committed
342
	if err := os.Mkdir(dir, 0755); err != nil {
Jeromy's avatar
Jeromy committed
343 344 345 346 347
		// EEXIST is safe to ignore here, that just means the prefix
		// directory already existed.
		if !os.IsExist(err) {
			return err
		}
348
		return nil
Jeromy's avatar
Jeromy committed
349
	}
350 351 352

	// Track DiskUsage of this NEW folder
	fs.updateDiskUsage(dir, true)
Jeromy's avatar
Jeromy committed
353 354 355
	return nil
}

356 357 358 359 360 361 362 363 364 365 366 367 368 369 370
// 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.
Steven Allen's avatar
Steven Allen committed
371
	for i := 0; i < RetryAttempts; i++ {
372
		err = rename(tmpPath, path)
Steven Allen's avatar
Steven Allen committed
373 374 375 376 377 378 379 380 381
		// if there's no error, or the source file doesn't exist, abort.
		if err == nil || os.IsNotExist(err) {
			break
		}
		// Otherwise, this could be a transient error due to some other
		// process holding open one of the files. Wait a bit and then
		// retry.
		time.Sleep(time.Duration(i+1) * RetryDelay)
	}
382 383 384 385 386 387 388 389 390 391 392 393 394 395
	fs.updateDiskUsage(path, true)
	return err
}

// 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.
396
func (fs *Datastore) Put(key datastore.Key, value []byte) error {
397
	if !keyIsValid(key) {
398
		return fmt.Errorf("when putting '%q': %v", key, ErrInvalidKey)
399 400
	}

Steven Allen's avatar
Steven Allen committed
401 402 403 404 405 406
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

Steven Allen's avatar
Steven Allen committed
407 408 409 410 411
	_, err := fs.doWriteOp(&op{
		typ: opPut,
		key: key,
		v:   value,
	})
412 413 414
	return err
}

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

	return nil
}

425 426 427 428 429 430 431 432 433 434 435 436 437
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")
	}
}

Will's avatar
Will committed
438 439 440 441 442 443 444 445 446
func isTooManyFDError(err error) bool {
	perr, ok := err.(*os.PathError)
	if ok && perr.Err == syscall.EMFILE {
		return true
	}

	return false
}

447 448
// doWrite optimizes out write operations (put/delete) to the same
// key by queueing them and succeeding all queued
449
// operations if one of them does. In such case,
450
// we assume that the first succeeding operation
451 452
// on that key was the last one to happen after
// all successful others.
453 454 455 456
//
// done is true if we actually performed the operation, false if we skipped or
// failed.
func (fs *Datastore) doWriteOp(oper *op) (done bool, err error) {
457 458 459 460
	keyStr := oper.key.String()

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

Steven Allen's avatar
Steven Allen committed
464
	err = fs.doOp(oper)
465 466 467 468 469

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

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

Tommi Virtanen's avatar
Tommi Virtanen committed
475
	dir, path := fs.encode(key)
476
	if err := fs.makeDir(dir); err != nil {
477
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
478 479
	}

480
	tmp, err := fs.tempFile()
Tommi Virtanen's avatar
Tommi Virtanen committed
481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499
	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
500
	if fs.sync {
501
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
502 503
			return err
		}
504
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
505 506 507 508 509
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

510
	err = fs.renameAndUpdateDiskUsage(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
511 512 513 514 515
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
516 517 518 519
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
520
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
521 522 523
	return nil
}

524
func (fs *Datastore) putMany(data map[datastore.Key][]byte) error {
Steven Allen's avatar
Steven Allen committed
525 526 527 528 529 530
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

Steven Allen's avatar
Steven Allen committed
531 532 533 534 535 536
	type putManyOp struct {
		key     datastore.Key
		file    *os.File
		dstPath string
		srcPath string
	}
Jeromy's avatar
Jeromy committed
537

Steven Allen's avatar
Steven Allen committed
538 539 540 541 542 543
	var (
		dirsToSync = make(map[string]struct{}, len(data))
		files      = make([]putManyOp, 0, len(data))
		closed     int
		removed    int
	)
544 545

	defer func() {
Steven Allen's avatar
Steven Allen committed
546 547 548 549 550 551 552
		for closed < len(files) {
			files[closed].file.Close()
			closed++
		}
		for removed < len(files) {
			_ = os.Remove(files[removed].srcPath)
			removed++
Jeromy's avatar
Jeromy committed
553
		}
554 555
	}()

Will's avatar
Will committed
556
	closer := func() error {
Steven Allen's avatar
Steven Allen committed
557 558
		for closed < len(files) {
			fi := files[closed].file
Will's avatar
Will committed
559 560 561 562 563 564 565 566
			if fs.sync {
				if err := syncFile(fi); err != nil {
					return err
				}
			}
			if err := fi.Close(); err != nil {
				return err
			}
Steven Allen's avatar
Steven Allen committed
567
			closed++
Will's avatar
Will committed
568 569 570 571
		}
		return nil
	}

572
	for key, value := range data {
Jeromy's avatar
Jeromy committed
573
		dir, path := fs.encode(key)
574
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
575 576
			return err
		}
Steven Allen's avatar
Steven Allen committed
577
		dirsToSync[dir] = struct{}{}
Jeromy's avatar
Jeromy committed
578

Steven Allen's avatar
Steven Allen committed
579
		tmp, err := fs.tempFileOnce()
Will's avatar
Will committed
580

Steven Allen's avatar
Steven Allen committed
581 582 583
		// If we have too many files open, try closing some, then try
		// again repeatedly.
		if isTooManyFDError(err) {
Will's avatar
Will committed
584 585 586
			if err = closer(); err != nil {
				return err
			}
Steven Allen's avatar
Steven Allen committed
587
			tmp, err = fs.tempFile()
Will's avatar
Will committed
588
		}
Steven Allen's avatar
Steven Allen committed
589

Jeromy's avatar
Jeromy committed
590 591 592 593
		if err != nil {
			return err
		}

Steven Allen's avatar
Steven Allen committed
594 595 596 597 598 599 600 601
		// Do this _first_ so we close it if writing fails.
		files = append(files, putManyOp{
			key:     key,
			file:    tmp,
			dstPath: path,
			srcPath: tmp.Name(),
		})

602
		if _, err := tmp.Write(value); err != nil {
Jeromy's avatar
Jeromy committed
603 604 605 606 607 608
			return err
		}
	}

	// Now we sync everything
	// sync and close files
Will's avatar
Will committed
609 610 611
	err := closer()
	if err != nil {
		return err
Jeromy's avatar
Jeromy committed
612 613 614
	}

	// move files to their proper places
Steven Allen's avatar
Steven Allen committed
615 616 617 618 619 620 621
	for _, pop := range files {
		done, err := fs.doWriteOp(&op{
			typ:  opRename,
			key:  pop.key,
			tmp:  pop.srcPath,
			path: pop.dstPath,
		})
622 623
		if err != nil {
			return err
Steven Allen's avatar
Steven Allen committed
624 625
		} else if !done {
			_ = os.Remove(pop.file.Name())
626
		}
Steven Allen's avatar
Steven Allen committed
627
		removed++
Jeromy's avatar
Jeromy committed
628 629 630
	}

	// now sync the dirs for those files
Jeromy's avatar
Jeromy committed
631
	if fs.sync {
Steven Allen's avatar
Steven Allen committed
632
		for dir := range dirsToSync {
Jeromy's avatar
Jeromy committed
633 634 635
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
636 637
		}

Jeromy's avatar
Jeromy committed
638 639 640 641
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
642 643 644 645 646
	}

	return nil
}

647
func (fs *Datastore) Get(key datastore.Key) (value []byte, err error) {
648 649 650 651 652
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil, datastore.ErrNotFound
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
653
	_, path := fs.encode(key)
654
	data, err := readFile(path)
Tommi Virtanen's avatar
Tommi Virtanen committed
655 656 657 658 659 660 661 662 663 664 665
	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) {
666 667 668 669 670
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return false, nil
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
671 672 673 674 675 676 677 678 679
	_, 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
680 681
}

Steven Allen's avatar
Steven Allen committed
682
func (fs *Datastore) GetSize(key datastore.Key) (size int, err error) {
683 684 685 686 687
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return -1, datastore.ErrNotFound
	}

Steven Allen's avatar
Steven Allen committed
688 689 690 691 692 693 694 695 696 697 698
	_, 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
	}
}

699 700 701
// 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
702
func (fs *Datastore) Delete(key datastore.Key) error {
703 704 705 706 707
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil
	}

Steven Allen's avatar
Steven Allen committed
708 709 710 711 712 713
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

714
	_, err := fs.doWriteOp(&op{
715 716 717 718
		typ: opDelete,
		key: key,
		v:   nil,
	})
719
	return err
720 721 722 723 724
}

// 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
725
	_, path := fs.encode(key)
726 727 728

	fSize := fileSize(path)

Steven Allen's avatar
Steven Allen committed
729 730 731 732 733 734 735 736 737 738 739
	var err error
	for i := 0; i < RetryAttempts; i++ {
		err = os.Remove(path)
		if err == nil {
			break
		} else if os.IsNotExist(err) {
			return nil
		}
	}

	if err == nil {
740 741
		atomic.AddInt64(&fs.diskUsage, -fSize)
		fs.checkpointDiskUsage()
Tommi Virtanen's avatar
Tommi Virtanen committed
742
	}
Steven Allen's avatar
Steven Allen committed
743 744

	return err
Tommi Virtanen's avatar
Tommi Virtanen committed
745 746 747
}

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
Steven Allen's avatar
Steven Allen committed
748
	prefix := datastore.NewKey(q.Prefix).String()
749 750 751 752
	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.
753 754 755 756 757
		log.Warnw(
			"flatfs was queried with a key prefix but flatfs only supports keys at the root",
			"prefix", q.Prefix,
			"query", q,
		)
758 759
		return query.ResultsWithEntries(q, nil), nil
	}
760

Steven Allen's avatar
Steven Allen committed
761 762 763 764 765 766 767
	// 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
768
		}
Steven Allen's avatar
Steven Allen committed
769 770 771 772 773
		select {
		case b.Output <- query.Result{Error: errors.New("walk failed: " + err.Error())}:
		case <-p.Closing():
		}
	})
Steven Allen's avatar
Steven Allen committed
774
	go b.Process.CloseAfterChildren() //nolint
Steven Allen's avatar
Steven Allen committed
775

776 777 778
	// We don't apply _any_ of the query logic ourselves so we'll leave it
	// all up to the naive query engine.
	return query.NaiveQueryApply(q, b.Results()), nil
Tommi Virtanen's avatar
Tommi Virtanen committed
779 780
}

Steven Allen's avatar
Steven Allen committed
781
func (fs *Datastore) walkTopLevel(path string, result *query.ResultBuilder) error {
782 783 784 785
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
786
	defer dir.Close()
Steven Allen's avatar
Steven Allen committed
787
	entries, err := dir.Readdir(-1)
788 789 790
	if err != nil {
		return err
	}
Steven Allen's avatar
Steven Allen committed
791 792 793 794 795
	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		dir := entry.Name()
Kevin Atkinson's avatar
Kevin Atkinson committed
796 797 798 799
		if len(dir) == 0 || dir[0] == '.' {
			continue
		}

Steven Allen's avatar
Steven Allen committed
800
		err = fs.walk(filepath.Join(path, dir), result)
801 802 803
		if err != nil {
			return err
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
804

Steven Allen's avatar
Steven Allen committed
805 806 807 808 809 810
		// Are we closing?
		select {
		case <-result.Process.Closing():
			return nil
		default:
		}
811 812 813 814
	}
	return nil
}

815
// folderSize estimates the diskUsage of a folder by reading
816 817
// up to DiskUsageFilesAverage entries in it and assuming any
// other files will have an average size.
818
func folderSize(path string, deadline time.Time) (int64, initAccuracy, error) {
819 820 821 822
	var du int64

	folder, err := os.Open(path)
	if err != nil {
823
		return 0, "", err
824 825 826 827 828
	}
	defer folder.Close()

	stat, err := folder.Stat()
	if err != nil {
829
		return 0, "", err
830 831 832 833
	}

	files, err := folder.Readdirnames(-1)
	if err != nil {
834
		return 0, "", err
835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851
	}

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

852
	accuracy := exactA
853
	for {
854 855 856
		// Do not process any files after deadline is over
		if time.Now().After(deadline) {
			accuracy = timedoutA
857 858 859
			break
		}

860 861 862 863
		if i >= totalFiles || filesProcessed >= maxFiles {
			if filesProcessed >= maxFiles {
				accuracy = approxA
			}
864 865 866 867 868 869 870 871
			break
		}

		// Stat the file
		fname := files[i]
		subpath := filepath.Join(path, fname)
		st, err := os.Stat(subpath)
		if err != nil {
872
			return 0, "", err
873 874 875 876
		}

		// Find folder size recursively
		if st.IsDir() {
877
			du2, acc, err := folderSize(filepath.Join(subpath), deadline)
878
			if err != nil {
879
				return 0, "", err
880
			}
881
			accuracy = combineAccuracy(acc, accuracy)
882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903
			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)
904
	return du, accuracy, nil
905 906 907 908
}

// calculateDiskUsage tries to read the DiskUsageFile for a cached
// diskUsage value, otherwise walks the datastore files.
909
// it is only safe to call in Open()
910 911 912
func (fs *Datastore) calculateDiskUsage() error {
	// Try to obtain a previously stored value from disk
	if persDu := fs.readDiskUsageFile(); persDu > 0 {
913
		fs.diskUsage = persDu
914 915 916
		return nil
	}

917 918 919 920 921 922 923
	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()
924
	deadline := time.Now().Add(DiskUsageCalcTimeout)
925
	du, accuracy, err := folderSize(fs.path, deadline)
926 927 928
	if err != nil {
		return err
	}
929 930 931
	if !msgTimer.Stop() {
		<-msgDone
	}
932
	if accuracy == timedoutA {
933 934 935 936 937 938 939
		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")
	}

940
	fs.storedValue.Accuracy = accuracy
941
	fs.diskUsage = du
942
	fs.writeDiskUsageFile(du, true)
943

944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964
	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 {
965 966
		atomic.AddInt64(&fs.diskUsage, fsize)
		fs.checkpointDiskUsage()
967 968 969
	}
}

970
func (fs *Datastore) checkpointDiskUsage() {
971
	select {
Kevin Atkinson's avatar
Kevin Atkinson committed
972
	case fs.checkpointCh <- struct{}{}:
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
973
		// msg sent
974 975
	default:
		// checkpoint request already pending
976 977 978
	}
}

979
func (fs *Datastore) checkpointLoop() {
Steven Allen's avatar
Steven Allen committed
980 981
	defer close(fs.done)

982 983
	timerActive := true
	timer := time.NewTimer(0)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
984
	defer timer.Stop()
985
	for {
986 987 988
		select {
		case _, more := <-fs.checkpointCh:
			du := atomic.LoadInt64(&fs.diskUsage)
Kevin Atkinson's avatar
Tweaks.  
Kevin Atkinson committed
989
			fs.dirty = true
990
			if !more { // shutting down
991
				fs.writeDiskUsageFile(du, true)
992
				if fs.dirty {
Steven Allen's avatar
Steven Allen committed
993
					log.Error("could not store final value of disk usage to file, future estimates may be inaccurate")
994
				}
995 996 997 998 999
				return
			}
			// If the difference between the checkpointed disk usage and
			// current one is larger than than `diskUsageCheckpointPercent`
			// of the checkpointed: store it.
1000
			newDu := float64(du)
1001
			lastCheckpointDu := float64(fs.storedValue.DiskUsage)
1002
			diff := math.Abs(newDu - lastCheckpointDu)
1003 1004
			if lastCheckpointDu*diskUsageCheckpointPercent < diff*100.0 {
				fs.writeDiskUsageFile(du, false)
1005
			}
1006 1007 1008 1009
			// 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
1010
				timerActive = true
1011 1012 1013 1014 1015
			}
		case <-timer.C:
			timerActive = false
			if fs.dirty {
				du := atomic.LoadInt64(&fs.diskUsage)
1016
				fs.writeDiskUsageFile(du, false)
1017
			}
1018
		}
1019
	}
1020 1021
}

1022
func (fs *Datastore) writeDiskUsageFile(du int64, doSync bool) {
1023
	tmp, err := fs.tempFile()
1024
	if err != nil {
Steven Allen's avatar
Steven Allen committed
1025
		log.Warnw("could not write disk usage", "error", err)
1026 1027 1028
		return
	}

1029
	removed := false
1030
	closed := false
1031
	defer func() {
1032 1033 1034
		if !closed {
			_ = tmp.Close()
		}
1035 1036 1037 1038
		if !removed {
			// silence errcheck
			_ = os.Remove(tmp.Name())
		}
1039

1040
	}()
1041

1042 1043
	toWrite := fs.storedValue
	toWrite.DiskUsage = du
1044
	encoder := json.NewEncoder(tmp)
1045
	if err := encoder.Encode(&toWrite); err != nil {
Steven Allen's avatar
Steven Allen committed
1046
		log.Warnw("cound not write disk usage", "error", err)
1047 1048
		return
	}
1049

1050 1051
	if doSync {
		if err := tmp.Sync(); err != nil {
Steven Allen's avatar
Steven Allen committed
1052
			log.Warnw("cound not sync", "error", err, "file", DiskUsageFile)
1053 1054 1055 1056
			return
		}
	}

1057
	if err := tmp.Close(); err != nil {
Steven Allen's avatar
Steven Allen committed
1058
		log.Warnw("cound not write disk usage", "error", err)
1059 1060
		return
	}
1061
	closed = true
1062

1063
	if err := rename(tmp.Name(), filepath.Join(fs.path, DiskUsageFile)); err != nil {
Steven Allen's avatar
Steven Allen committed
1064
		log.Warnw("cound not write disk usage", "error", err)
1065
		return
1066
	}
1067
	removed = true
1068

1069
	fs.storedValue = toWrite
1070
	fs.dirty = false
1071 1072
}

1073
// readDiskUsageFile is only safe to call in Open()
1074 1075
func (fs *Datastore) readDiskUsageFile() int64 {
	fpath := filepath.Join(fs.path, DiskUsageFile)
1076
	duB, err := readFile(fpath)
1077 1078 1079
	if err != nil {
		return 0
	}
1080
	err = json.Unmarshal(duB, &fs.storedValue)
1081 1082 1083
	if err != nil {
		return 0
	}
1084
	return fs.storedValue.DiskUsage
1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104
}

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

1105 1106 1107 1108
// 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 {
1109
	return string(fs.storedValue.Accuracy)
1110 1111
}

1112
func (fs *Datastore) tempFile() (*os.File, error) {
1113
	file, err := tempFile(fs.tempPath, "temp-")
1114 1115 1116
	return file, err
}

Steven Allen's avatar
Steven Allen committed
1117
func (fs *Datastore) tempFileOnce() (*os.File, error) {
Steven Allen's avatar
Steven Allen committed
1118
	return tempFileOnce(fs.tempPath, "temp-")
Steven Allen's avatar
Steven Allen committed
1119 1120
}

Steven Allen's avatar
Steven Allen committed
1121
// only call this on directories.
1122
func (fs *Datastore) walk(path string, qrb *query.ResultBuilder) error {
1123 1124
	dir, err := os.Open(path)
	if err != nil {
1125 1126 1127 1128
		if os.IsNotExist(err) {
			// not an error if the file disappeared
			return nil
		}
1129 1130
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
1131
	defer dir.Close()
1132

1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144
	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 {
1145
			// not a block.
1146 1147 1148
			continue
		}

1149 1150 1151
		var result query.Result
		result.Key = key.String()
		if !qrb.Query.KeysOnly {
1152
			value, err := readFile(filepath.Join(path, fn))
1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170
			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
1171
		select {
1172 1173
		case qrb.Output <- result:
		case <-qrb.Process.Closing():
Steven Allen's avatar
Steven Allen committed
1174
			return nil
1175 1176 1177 1178 1179
		}
	}
	return nil
}

1180 1181 1182
// Deactivate closes background maintenance threads, most write
// operations will fail but readonly operations will continue to
// function
Steven Allen's avatar
Steven Allen committed
1183
func (fs *Datastore) deactivate() {
Steven Allen's avatar
Steven Allen committed
1184 1185 1186
	fs.shutdownLock.Lock()
	defer fs.shutdownLock.Unlock()
	if fs.shutdown {
Steven Allen's avatar
Steven Allen committed
1187
		return
1188
	}
Steven Allen's avatar
Steven Allen committed
1189 1190 1191
	fs.shutdown = true
	close(fs.checkpointCh)
	<-fs.done
Jeromy's avatar
Jeromy committed
1192 1193
}

1194
func (fs *Datastore) Close() error {
Steven Allen's avatar
Steven Allen committed
1195 1196
	fs.deactivate()
	return nil
1197 1198
}

Jeromy's avatar
Jeromy committed
1199
type flatfsBatch struct {
1200
	puts    map[datastore.Key][]byte
Jeromy's avatar
Jeromy committed
1201 1202 1203 1204 1205
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

Jeromy's avatar
Jeromy committed
1206
func (fs *Datastore) Batch() (datastore.Batch, error) {
Jeromy's avatar
Jeromy committed
1207
	return &flatfsBatch{
1208
		puts:    make(map[datastore.Key][]byte),
Jeromy's avatar
Jeromy committed
1209 1210
		deletes: make(map[datastore.Key]struct{}),
		ds:      fs,
Jeromy's avatar
Jeromy committed
1211
	}, nil
Jeromy's avatar
Jeromy committed
1212 1213
}

1214
func (bt *flatfsBatch) Put(key datastore.Key, val []byte) error {
1215
	if !keyIsValid(key) {
1216
		return fmt.Errorf("when putting '%q': %v", key, ErrInvalidKey)
1217
	}
Jeromy's avatar
Jeromy committed
1218 1219 1220 1221
	bt.puts[key] = val
	return nil
}

Jeromy's avatar
Jeromy committed
1222
func (bt *flatfsBatch) Delete(key datastore.Key) error {
1223 1224 1225
	if keyIsValid(key) {
		bt.deletes[key] = struct{}{}
	} // otherwise, delete is a no-op anyways.
Jeromy's avatar
Jeromy committed
1226 1227 1228
	return nil
}

Jeromy's avatar
Jeromy committed
1229
func (bt *flatfsBatch) Commit() error {
Jeromy's avatar
Jeromy committed
1230 1231 1232 1233
	if err := bt.ds.putMany(bt.puts); err != nil {
		return err
	}

Steven Allen's avatar
Steven Allen committed
1234
	for k := range bt.deletes {
Jeromy's avatar
Jeromy committed
1235 1236 1237 1238 1239 1240 1241
		if err := bt.ds.Delete(k); err != nil {
			return err
		}
	}

	return nil
}