flatfs.go 9.55 KB
Newer Older
Tommi Virtanen's avatar
Tommi Virtanen committed
1 2 3 4 5 6 7
// 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 (
	"errors"
8
	"fmt"
Tommi Virtanen's avatar
Tommi Virtanen committed
9 10 11
	"io/ioutil"
	"os"
	"path"
Jeromy's avatar
Jeromy committed
12
	"path/filepath"
Tommi Virtanen's avatar
Tommi Virtanen committed
13
	"strings"
14
	"time"
Tommi Virtanen's avatar
Tommi Virtanen committed
15

Jeromy's avatar
Jeromy committed
16 17
	"github.com/ipfs/go-datastore"
	"github.com/ipfs/go-datastore/query"
Jakub Sztandera's avatar
Jakub Sztandera committed
18
	"github.com/jbenet/go-os-rename"
19

Jakub Sztandera's avatar
Jakub Sztandera committed
20
	logging "github.com/ipfs/go-log"
Tommi Virtanen's avatar
Tommi Virtanen committed
21 22
)

23 24
var log = logging.Logger("flatfs")

Tommi Virtanen's avatar
Tommi Virtanen committed
25
const (
26
	extension = ".data"
Tommi Virtanen's avatar
Tommi Virtanen committed
27 28 29 30
)

type Datastore struct {
	path string
31

32 33
	shardFunc string
	getDir    ShardFunc
Jeromy's avatar
Jeromy committed
34 35 36

	// sychronize all writes and directory changes for added safety
	sync bool
Tommi Virtanen's avatar
Tommi Virtanen committed
37 38 39 40
}

var _ datastore.Datastore = (*Datastore)(nil)

41 42
type ShardFunc func(string) string

Kevin Atkinson's avatar
Kevin Atkinson committed
43
var (
44
	ErrDatastoreExists       = errors.New("Datastore already exist")
Kevin Atkinson's avatar
Kevin Atkinson committed
45
	ErrDatastoreDoesNotExist = errors.New("Datastore directory does not exist")
46
	ErrShardingFileMissing   = errors.New("SHARDING file not found in datastore")
Kevin Atkinson's avatar
Kevin Atkinson committed
47
)
48

49
const IPFS_DEF_SHARD = "/repo/flatfs/shard/v1/next-to-last/2"
50

51
func Create(path string, funStr string) error {
52

53 54 55
	err := os.Mkdir(path, 0777)
	if err != nil && !os.IsExist(err) {
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
56 57
	}

58 59 60 61 62
	fun, err := ParseShardFunc(funStr)
	if err != nil {
		return err
	}

63 64
	dsFun, err := ReadShardFunc(path)
	switch err {
Kevin Atkinson's avatar
Kevin Atkinson committed
65
	case ErrShardingFileMissing:
66 67
		// fixme: make sure directory is empty and return an error if
		// it is not
68
		err := WriteShardFunc(path, fun)
69 70 71
		if err != nil {
			return err
		}
72 73
		err = WriteReadme(path, fun)
		return err
74
	case nil:
75
		if fun.String() != dsFun.String() {
76
			return fmt.Errorf("specified shard func '%s' does not match repo shard func '%s'",
77
				fun.String(), dsFun.String())
78
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
79
		return ErrDatastoreExists
80
	default:
81
		return err
82
	}
83 84 85 86 87
}

func Open(path string, sync bool) (*Datastore, error) {
	_, err := os.Stat(path)
	if os.IsNotExist(err) {
Kevin Atkinson's avatar
Kevin Atkinson committed
88
		return nil, ErrDatastoreDoesNotExist
89 90 91 92
	} else if err != nil {
		return nil, err
	}

93
	shardId, err := ReadShardFunc(path)
94
	if err != nil {
95
		return nil, err
96 97
	}

98 99
	fs := &Datastore{
		path:      path,
100 101
		shardFunc: shardId.String(),
		getDir:    shardId.Func(),
102
		sync:      sync,
103
	}
104 105 106
	return fs, nil
}

107 108 109
// convince method, fixme: maybe just call it "New"?
func CreateOrOpen(path string, fun string, sync bool) (*Datastore, error) {
	err := Create(path, fun)
Kevin Atkinson's avatar
Kevin Atkinson committed
110
	if err != nil && err != ErrDatastoreExists {
111 112 113 114 115
		return nil, err
	}
	return Open(path, sync)
}

116 117
func (fs *Datastore) ShardFunc() string {
	return fs.shardFunc
118 119
}

Tommi Virtanen's avatar
Tommi Virtanen committed
120
func (fs *Datastore) encode(key datastore.Key) (dir, file string) {
121
	noslash := key.String()[1:]
122
	dir = path.Join(fs.path, fs.getDir(noslash))
123
	file = path.Join(dir, noslash+extension)
Tommi Virtanen's avatar
Tommi Virtanen committed
124 125 126
	return dir, file
}

127 128 129 130 131
func (fs *Datastore) decode(file string) (key datastore.Key, ok bool) {
	if path.Ext(file) != extension {
		return datastore.Key{}, false
	}
	name := file[:len(file)-len(extension)]
Jeromy's avatar
Jeromy committed
132
	return datastore.NewKey(name), true
133 134
}

135 136
func (fs *Datastore) makeDir(dir string) error {
	if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
137
		return err
138 139 140 141 142 143
	}

	// 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
144 145 146 147
	if fs.sync {
		if err := syncDir(fs.path); err != nil {
			return err
		}
148 149 150 151
	}
	return nil
}

152
func (fs *Datastore) makeDirNoSync(dir string) error {
Jeromy's avatar
Jeromy committed
153 154 155 156 157 158 159 160 161 162
	if err := os.Mkdir(dir, 0777); err != nil {
		// EEXIST is safe to ignore here, that just means the prefix
		// directory already existed.
		if !os.IsExist(err) {
			return err
		}
	}
	return nil
}

Jeromy's avatar
Jeromy committed
163
var putMaxRetries = 6
164

Tommi Virtanen's avatar
Tommi Virtanen committed
165 166 167 168 169 170
func (fs *Datastore) Put(key datastore.Key, value interface{}) error {
	val, ok := value.([]byte)
	if !ok {
		return datastore.ErrInvalidType
	}

171
	var err error
Jeromy's avatar
Jeromy committed
172
	for i := 1; i <= putMaxRetries; i++ {
173 174
		err = fs.doPut(key, val)
		if err == nil {
Jeromy's avatar
Jeromy committed
175
			break
176 177 178
		}

		if !strings.Contains(err.Error(), "too many open files") {
Jeromy's avatar
Jeromy committed
179
			break
180 181
		}

Or Rikon's avatar
Or Rikon committed
182
		log.Errorf("too many open files, retrying in %dms", 100*i)
Jeromy's avatar
Jeromy committed
183
		time.Sleep(time.Millisecond * 100 * time.Duration(i))
184 185 186 187 188
	}
	return err
}

func (fs *Datastore) doPut(key datastore.Key, val []byte) error {
Tommi Virtanen's avatar
Tommi Virtanen committed
189
	dir, path := fs.encode(key)
190
	if err := fs.makeDir(dir); err != nil {
191
		return err
Tommi Virtanen's avatar
Tommi Virtanen committed
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213
	}

	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
214
	if fs.sync {
215
		if err := syncFile(tmp); err != nil {
Jeromy's avatar
Jeromy committed
216 217
			return err
		}
218
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
219 220 221 222 223
	if err := tmp.Close(); err != nil {
		return err
	}
	closed = true

224
	err = osrename.Rename(tmp.Name(), path)
Tommi Virtanen's avatar
Tommi Virtanen committed
225 226 227 228 229
	if err != nil {
		return err
	}
	removed = true

Jeromy's avatar
Jeromy committed
230 231 232 233
	if fs.sync {
		if err := syncDir(dir); err != nil {
			return err
		}
234
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
235 236 237
	return nil
}

Jeromy's avatar
Jeromy committed
238 239 240 241 242 243 244 245 246 247
func (fs *Datastore) putMany(data map[datastore.Key]interface{}) error {
	var dirsToSync []string
	files := make(map[*os.File]string)

	for key, value := range data {
		val, ok := value.([]byte)
		if !ok {
			return datastore.ErrInvalidType
		}
		dir, path := fs.encode(key)
248
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
			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
		}

		files[tmp] = path
	}

	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
283
		if fs.sync {
284
			if err := syncFile(fi); err != nil {
Jeromy's avatar
Jeromy committed
285 286
				return err
			}
Jeromy's avatar
Jeromy committed
287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307
		}

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

		// signify closed
		ops[fi] = 1
	}

	// move files to their proper places
	for fi, path := range files {
		if err := osrename.Rename(fi.Name(), path); err != nil {
			return err
		}

		// signify removed
		ops[fi] = 2
	}

	// now sync the dirs for those files
Jeromy's avatar
Jeromy committed
308 309 310 311 312
	if fs.sync {
		for _, dir := range dirsToSync {
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
313 314
		}

Jeromy's avatar
Jeromy committed
315 316 317 318
		// sync top flatfs dir
		if err := syncDir(fs.path); err != nil {
			return err
		}
Jeromy's avatar
Jeromy committed
319 320 321 322 323
	}

	return nil
}

Tommi Virtanen's avatar
Tommi Virtanen committed
324 325 326 327 328 329 330 331 332 333 334 335 336 337
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
338 339 340 341 342 343 344 345 346
	_, 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
347 348 349
}

func (fs *Datastore) Delete(key datastore.Key) error {
Tommi Virtanen's avatar
Tommi Virtanen committed
350 351 352 353 354 355 356 357 358
	_, path := fs.encode(key)
	switch err := os.Remove(path); {
	case err == nil:
		return nil
	case os.IsNotExist(err):
		return datastore.ErrNotFound
	default:
		return err
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
359 360 361
}

func (fs *Datastore) Query(q query.Query) (query.Results, error) {
362 363 364 365 366 367 368 369 370 371 372
	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")
	}

373
	reschan := make(chan query.Result, query.KeysOnlyBufSize)
Jeromy's avatar
Jeromy committed
374 375
	go func() {
		defer close(reschan)
376
		err := fs.walkTopLevel(fs.path, reschan)
Jeromy's avatar
Jeromy committed
377
		if err != nil {
378
			reschan <- query.Result{Error: errors.New("walk failed: " + err.Error())}
379
		}
Jeromy's avatar
Jeromy committed
380 381
	}()
	return query.ResultsWithChan(q, reschan), nil
Tommi Virtanen's avatar
Tommi Virtanen committed
382 383
}

384 385 386 387 388
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
389
	defer dir.Close()
390 391 392 393 394
	names, err := dir.Readdirnames(-1)
	if err != nil {
		return err
	}
	for _, dir := range names {
Kevin Atkinson's avatar
Kevin Atkinson committed
395 396 397 398 399

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

400 401 402 403
		err = fs.walk(filepath.Join(path, dir), reschan)
		if err != nil {
			return err
		}
Kevin Atkinson's avatar
Kevin Atkinson committed
404

405 406 407 408 409 410 411 412 413
	}
	return nil
}

func (fs *Datastore) walk(path string, reschan chan query.Result) error {
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
Kevin Atkinson's avatar
Kevin Atkinson committed
414
	defer dir.Close()
415 416 417 418 419 420 421 422 423 424

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

425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449
	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 {
			log.Warning("failed to decode entry in flatfs")
			continue
		}

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

Jeromy's avatar
Jeromy committed
450 451 452 453
func (fs *Datastore) Close() error {
	return nil
}

Jeromy's avatar
Jeromy committed
454
type flatfsBatch struct {
Jeromy's avatar
Jeromy committed
455 456 457 458 459 460
	puts    map[datastore.Key]interface{}
	deletes map[datastore.Key]struct{}

	ds *Datastore
}

Jeromy's avatar
Jeromy committed
461
func (fs *Datastore) Batch() (datastore.Batch, error) {
Jeromy's avatar
Jeromy committed
462
	return &flatfsBatch{
Jeromy's avatar
Jeromy committed
463 464 465
		puts:    make(map[datastore.Key]interface{}),
		deletes: make(map[datastore.Key]struct{}),
		ds:      fs,
Jeromy's avatar
Jeromy committed
466
	}, nil
Jeromy's avatar
Jeromy committed
467 468
}

Jeromy's avatar
Jeromy committed
469
func (bt *flatfsBatch) Put(key datastore.Key, val interface{}) error {
Jeromy's avatar
Jeromy committed
470 471 472 473
	bt.puts[key] = val
	return nil
}

Jeromy's avatar
Jeromy committed
474
func (bt *flatfsBatch) Delete(key datastore.Key) error {
Jeromy's avatar
Jeromy committed
475 476 477 478
	bt.deletes[key] = struct{}{}
	return nil
}

Jeromy's avatar
Jeromy committed
479
func (bt *flatfsBatch) Commit() error {
Jeromy's avatar
Jeromy committed
480 481 482 483 484 485 486 487 488 489 490 491 492
	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
493 494 495
var _ datastore.ThreadSafeDatastore = (*Datastore)(nil)

func (*Datastore) IsThreadSafe() {}