flatfs.go 9.61 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

43 44 45 46
var DatastoreExists = errors.New("Datastore already exist")
var DatastoreDoesNotExist = errors.New("Datastore directory does not exist")
var ShardingFileMissing = errors.New("SHARDING file not found in datastore")

47
const IPFS_DEF_SHARD = "/repo/flatfs/shard/v1/next-to-last/2"
48

49
func Create(path string, funStr string) error {
50

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

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

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

func Open(path string, sync bool) (*Datastore, error) {
	_, err := os.Stat(path)
	if os.IsNotExist(err) {
		return nil, DatastoreDoesNotExist
	} else if err != nil {
		return nil, err
	}

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

96 97
	if err != nil {
		return nil, fmt.Errorf("unable to parse shard func: %v", err)
98
	}
Tommi Virtanen's avatar
Tommi Virtanen committed
99

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

109 110 111 112 113 114 115 116 117
// convince method, fixme: maybe just call it "New"?
func CreateOrOpen(path string, fun string, sync bool) (*Datastore, error) {
	err := Create(path, fun)
	if err != nil && err != DatastoreExists {
		return nil, err
	}
	return Open(path, sync)
}

118 119
func (fs *Datastore) ShardFunc() string {
	return fs.shardFunc
120 121
}

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

129 130 131 132 133
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
134
	return datastore.NewKey(name), true
135 136
}

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

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

154
func (fs *Datastore) makeDirNoSync(dir string) error {
Jeromy's avatar
Jeromy committed
155 156 157 158 159 160 161 162 163 164
	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
165
var putMaxRetries = 6
166

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

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

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

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

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

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

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

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

Jeromy's avatar
Jeromy committed
240 241 242 243 244 245 246 247 248 249
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)
250
		if err := fs.makeDirNoSync(dir); err != nil {
Jeromy's avatar
Jeromy committed
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 283 284
			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
285
		if fs.sync {
286
			if err := syncFile(fi); err != nil {
Jeromy's avatar
Jeromy committed
287 288
				return err
			}
Jeromy's avatar
Jeromy committed
289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
		}

		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
310 311 312 313 314
	if fs.sync {
		for _, dir := range dirsToSync {
			if err := syncDir(dir); err != nil {
				return err
			}
Jeromy's avatar
Jeromy committed
315 316
		}

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

	return nil
}

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

func (fs *Datastore) Delete(key datastore.Key) error {
Tommi Virtanen's avatar
Tommi Virtanen committed
352 353 354 355 356 357 358 359 360
	_, 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
361 362 363
}

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

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

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

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

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

407 408 409 410 411 412 413 414 415
	}
	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
416
	defer dir.Close()
417 418 419 420 421 422 423 424 425 426

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

427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451
	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
452 453 454 455
func (fs *Datastore) Close() error {
	return nil
}

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

	ds *Datastore
}

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

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

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

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

func (*Datastore) IsThreadSafe() {}