flatfs.go 27.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"
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
55 56 57 58 59 60 61 62
)

const (
	opPut = iota
	opDelete
	opRename
)

63 64 65
type initAccuracy string

const (
66
	unknownA  initAccuracy = "unknown"
67 68 69 70 71 72
	exactA    initAccuracy = "initial-exact"
	approxA   initAccuracy = "initial-approximate"
	timedoutA initAccuracy = "initial-timed-out"
)

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

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

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
112
type Datastore struct {
113 114 115 116 117
	// 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

118 119
	path     string
	tempPath string
120

121 122
	shardStr string
	getDir   ShardFunc
Jeromy's avatar
Jeromy committed
123 124 125

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

127 128 129
	// these values should only be used during internalization or
	// inside the checkpoint loop
	dirty       bool
130
	storedValue diskUsageValue
131

Steven Allen's avatar
Steven Allen committed
132
	// Used to trigger a checkpoint.
Kevin Atkinson's avatar
Kevin Atkinson committed
133 134
	checkpointCh chan struct{}
	done         chan struct{}
135

Steven Allen's avatar
Steven Allen committed
136 137 138
	shutdownLock sync.RWMutex
	shutdown     bool

139 140 141
	// opMap handles concurrent write operations (put/delete)
	// to the same key
	opMap *opMap
Tommi Virtanen's avatar
Tommi Virtanen committed
142 143
}

144
type diskUsageValue struct {
145 146
	DiskUsage int64        `json:"diskUsage"`
	Accuracy  initAccuracy `json:"accuracy"`
147 148
}

149 150
type ShardFunc func(string) string

151
type opT int
152

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

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

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

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

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

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

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

255
	shardId, err := ReadShardFunc(path)
256
	if err != nil {
257
		return nil, err
258 259
	}

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

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

Kevin Atkinson's avatar
Kevin Atkinson committed
283
	go fs.checkpointLoop()
284 285 286
	return fs, nil
}

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

296 297
func (fs *Datastore) ShardStr() string {
	return fs.shardStr
298 299
}

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

307
func (fs *Datastore) decode(file string) (key datastore.Key, ok bool) {
308 309 310 311 312 313
	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)
		}
314 315 316
		return datastore.Key{}, false
	}
	name := file[:len(file)-len(extension)]
Jeromy's avatar
Jeromy committed
317
	return datastore.NewKey(name), true
318 319
}

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

	// 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
329 330 331 332
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
333 334 335 336
	}
	return nil
}

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

	// Track DiskUsage of this NEW folder
	fs.updateDiskUsage(dir, true)
Jeromy's avatar
Jeromy committed
349 350 351
	return nil
}

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

Jeromy's avatar
Jeromy committed
372
var putMaxRetries = 6
373

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

Steven Allen's avatar
Steven Allen committed
389 390 391 392 393 394
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

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

		if !strings.Contains(err.Error(), "too many open files") {
Jeromy's avatar
Jeromy committed
407
			break
408 409
		}

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

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

	return nil
}

426 427 428 429 430 431 432 433 434 435 436 437 438
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
439 440 441 442 443 444 445 446 447
func isTooManyFDError(err error) bool {
	perr, ok := err.(*os.PathError)
	if ok && perr.Err == syscall.EMFILE {
		return true
	}

	return false
}

448 449
// doWrite optimizes out write operations (put/delete) to the same
// key by queueing them and succeeding all queued
450
// operations if one of them does. In such case,
451
// we assume that the first succeeding operation
452 453
// on that key was the last one to happen after
// all successful others.
454 455 456 457
//
// 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) {
458 459 460 461
	keyStr := oper.key.String()

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

	// Do the operation
Will's avatar
Will committed
466 467 468 469 470 471 472 473
	for i := 0; i < 6; i++ {
		err = fs.doOp(oper)

		if err == nil || !isTooManyFDError(err) {
			break
		}
		time.Sleep(time.Duration(i+1) * RetryDelay)
	}
474 475 476 477 478

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

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

Tommi Virtanen's avatar
Tommi Virtanen committed
484
	dir, path := fs.encode(key)
485
	if err := fs.makeDir(dir); err != nil {
486
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
487 488
	}

489
	tmp, err := fs.tempFile()
Tommi Virtanen's avatar
Tommi Virtanen committed
490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508
	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
509
	if fs.sync {
510
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
511 512
			return err
		}
513
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
514 515 516 517 518
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

519
	err = fs.renameAndUpdateDiskUsage(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
520 521 522 523 524
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
525 526 527 528
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
529
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
530 531 532
	return nil
}

533
func (fs *Datastore) putMany(data map[datastore.Key][]byte) error {
Steven Allen's avatar
Steven Allen committed
534 535 536 537 538 539
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

Jeromy's avatar
Jeromy committed
540 541
	var dirsToSync []string

542 543 544 545 546
	files := make(map[*os.File]*op, len(data))
	ops := make(map[*os.File]int, len(data))

	defer func() {
		for fi := range files {
Steven Allen's avatar
Steven Allen committed
547
			val := ops[fi]
548 549 550 551 552 553 554
			switch val {
			case 0:
				_ = fi.Close()
				fallthrough
			case 1:
				_ = os.Remove(fi.Name())
			}
Jeromy's avatar
Jeromy committed
555
		}
556 557
	}()

Will's avatar
Will committed
558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579
	closer := func() error {
		for fi := range files {
			if ops[fi] != 0 {
				continue
			}

			if fs.sync {
				if err := syncFile(fi); err != nil {
					return err
				}
			}

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

			// signify closed
			ops[fi] = 1
		}
		return nil
	}

580
	for key, value := range data {
Jeromy's avatar
Jeromy committed
581
		dir, path := fs.encode(key)
582
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
583 584 585 586
			return err
		}
		dirsToSync = append(dirsToSync, dir)

587
		tmp, err := fs.tempFile()
Will's avatar
Will committed
588 589 590 591 592 593 594 595 596 597 598 599 600 601 602

		// Fallback retry for temporary error.
		if err != nil && isTooManyFDError(err) {
			if err = closer(); err != nil {
				return err
			}
			for i := 0; i < 6; i++ {
				time.Sleep(time.Duration(i+1) * RetryDelay)

				tmp, err = fs.tempFile()
				if err == nil || !isTooManyFDError(err) {
					break
				}
			}
		}
Jeromy's avatar
Jeromy committed
603 604 605 606
		if err != nil {
			return err
		}

607
		if _, err := tmp.Write(value); err != nil {
Jeromy's avatar
Jeromy committed
608 609 610
			return err
		}

611 612 613 614 615 616
		files[tmp] = &op{
			typ:  opRename,
			path: path,
			tmp:  tmp.Name(),
			key:  key,
		}
Jeromy's avatar
Jeromy committed
617 618 619 620
	}

	// Now we sync everything
	// sync and close files
Will's avatar
Will committed
621 622 623
	err := closer()
	if err != nil {
		return err
Jeromy's avatar
Jeromy committed
624 625 626
	}

	// move files to their proper places
627
	for fi, op := range files {
628
		done, err := fs.doWriteOp(op)
629 630
		if err != nil {
			return err
631 632 633
		} else if done {
			// signify removed
			ops[fi] = 2
634
		}
Jeromy's avatar
Jeromy committed
635 636 637
	}

	// now sync the dirs for those files
Jeromy's avatar
Jeromy committed
638 639 640 641 642
	if fs.sync {
		for _, dir := range dirsToSync {
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
643 644
		}

Jeromy's avatar
Jeromy committed
645 646 647 648
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
649 650 651 652 653
	}

	return nil
}

654
func (fs *Datastore) Get(key datastore.Key) (value []byte, err error) {
655 656 657 658 659
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil, datastore.ErrNotFound
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
660
	_, path := fs.encode(key)
661
	data, err := readFile(path)
Tommi Virtanen's avatar
Tommi Virtanen committed
662 663 664 665 666 667 668 669 670 671 672
	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) {
673 674 675 676 677
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return false, nil
	}

Tommi Virtanen's avatar
Tommi Virtanen committed
678 679 680 681 682 683 684 685 686
	_, 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
687 688
}

Steven Allen's avatar
Steven Allen committed
689
func (fs *Datastore) GetSize(key datastore.Key) (size int, err error) {
690 691 692 693 694
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return -1, datastore.ErrNotFound
	}

Steven Allen's avatar
Steven Allen committed
695 696 697 698 699 700 701 702 703 704 705
	_, 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
	}
}

706 707 708
// 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
709
func (fs *Datastore) Delete(key datastore.Key) error {
710 711 712 713 714
	// Can't exist in datastore.
	if !keyIsValid(key) {
		return nil
	}

Steven Allen's avatar
Steven Allen committed
715 716 717 718 719 720
	fs.shutdownLock.RLock()
	defer fs.shutdownLock.RUnlock()
	if fs.shutdown {
		return ErrClosed
	}

721
	_, err := fs.doWriteOp(&op{
722 723 724 725
		typ: opDelete,
		key: key,
		v:   nil,
	})
726
	return err
727 728 729 730 731
}

// 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
732
	_, path := fs.encode(key)
733 734 735

	fSize := fileSize(path)

Tommi Virtanen's avatar
Tommi Virtanen committed
736 737
	switch err := os.Remove(path); {
	case err == nil:
738 739
		atomic.AddInt64(&fs.diskUsage, -fSize)
		fs.checkpointDiskUsage()
Tommi Virtanen's avatar
Tommi Virtanen committed
740 741
		return nil
	case os.IsNotExist(err):
Steven Allen's avatar
Steven Allen committed
742
		return nil
Tommi Virtanen's avatar
Tommi Virtanen committed
743 744 745
	default:
		return err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
746 747 748
}

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

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

777 778 779
	// 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
780 781
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1041
	}()
1042

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

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

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

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

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

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

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

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

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

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

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

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

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

1191
func (fs *Datastore) Close() error {
Steven Allen's avatar
Steven Allen committed
1192 1193
	fs.deactivate()
	return nil
1194 1195
}

Jeromy's avatar
Jeromy committed
1196
type flatfsBatch struct {
1197
	puts    map[datastore.Key][]byte
Jeromy's avatar
Jeromy committed
1198 1199 1200 1201 1202
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

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

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

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

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

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

	return nil
}