swarm/storage/localstore: most basic database

swarm/storage/localstore/accessor.go

@@ -0,0 +1,74 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"context"
+
+	"github.com/syndtr/goleveldb/leveldb"
+
+	"github.com/ethereum/go-ethereum/swarm/shed"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// Accessor implements ChunkStore to manage data
+// in DB with different modes of access and update.
+type Accessor struct {
+	db   *DB
+	mode Mode
+}
+
+// Accessor returns a new Accessor with a specified Mode.
+func (db *DB) Accessor(mode Mode) *Accessor {
+	return &Accessor{
+		mode: mode,
+		db:   db,
+	}
+}
+
+// Put overwrites the underlying DB Put method for the specific mode of update.
+func (u *Accessor) Put(ctx context.Context, ch storage.Chunk) error {
+	return u.db.update(ctx, u.mode, chunkToItem(ch))
+}
+
+// Get overwrites the underlying DB Get method for the specific mode of access.
+func (u *Accessor) Get(_ context.Context, addr storage.Address) (chunk storage.Chunk, err error) {
+	item := addressToItem(addr)
+	out, err := u.db.access(u.mode, item)
+	if err != nil {
+		if err == leveldb.ErrNotFound {
+			return nil, storage.ErrChunkNotFound
+		}
+		return nil, err
+	}
+	return storage.NewChunk(out.Address, out.Data), nil
+}
+
+// chunkToItem creates new IndexItem with data provided by the Chunk.
+func chunkToItem(ch storage.Chunk) shed.IndexItem {
+	return shed.IndexItem{
+		Address: ch.Address(),
+		Data:    ch.Data(),
+	}
+}
+
+// addressToItem creates new IndexItem with a provided address.
+func addressToItem(addr storage.Address) shed.IndexItem {
+	return shed.IndexItem{
+		Address: addr,
+	}
+}

swarm/storage/localstore/accessor_test.go

@@ -0,0 +1,98 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// TestAccessors tests most basic Put and Get functionalities
+// for different accessors. This test validates that the chunk
+// is retrievable from the database, not if all indexes are set
+// correctly.
+func TestAccessors(t *testing.T) {
+	db, cleanupFunc := newTestDB(t)
+	defer cleanupFunc()
+
+	for _, m := range []Mode{
+		ModeSyncing,
+		ModeUpload,
+		ModeRequest,
+		ModeSynced,
+		ModeAccess,
+	} {
+		t.Run(ModeName(m), func(t *testing.T) {
+			a := db.Accessor(m)
+
+			want := generateRandomChunk()
+
+			err := a.Put(context.Background(), want)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			got, err := a.Get(context.Background(), want.Address())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !bytes.Equal(got.Data(), want.Data()) {
+				t.Errorf("got chunk data %x, want %x", got.Data(), want.Data())
+			}
+		})
+	}
+
+	// Removal mode is a special case as it removes the chunk
+	// from the database.
+	t.Run(ModeName(ModeRemoval), func(t *testing.T) {
+		a := db.Accessor(ModeUpload)
+
+		want := generateRandomChunk()
+
+		// first put a random chunk to the database
+		err := a.Put(context.Background(), want)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		got, err := a.Get(context.Background(), want.Address())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !bytes.Equal(got.Data(), want.Data()) {
+			t.Errorf("got chunk data %x, want %x", got.Data(), want.Data())
+		}
+
+		a = db.Accessor(ModeRemoval)
+
+		// removal accessor actually removes the chunk on Put
+		err = a.Put(context.Background(), want)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// chunk should not be found
+		wantErr := storage.ErrChunkNotFound
+		_, err = a.Get(context.Background(), want.Address())
+		if err != wantErr {
+			t.Errorf("got error %v, expected %v", err, wantErr)
+		}
+	})
+}

swarm/storage/localstore/localstore.go

@@ -0,0 +1,244 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"encoding/binary"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/swarm/shed"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+const (
+	// maximal time for DB.Close must return
+	closeTimeout = 10 * time.Second
+)
+
+var (
+	// ErrInvalidMode is retuned when an unkonw Mode
+	// is provided to the function.
+	ErrInvalidMode = errors.New("invalid mode")
+	// ErrDBClosed is returned when database is closed.
+	ErrDBClosed = errors.New("db closed")
+)
+
+// DB is the local store implementation and holds
+// database related objects.
+type DB struct {
+	shed *shed.DB
+
+	// fields and indexes
+	schemaName     shed.StringField
+	sizeCounter    shed.Uint64Field
+	retrievalIndex shed.Index
+	pushIndex      shed.Index
+	pullIndex      shed.Index
+	gcIndex        shed.Index
+
+	baseKey []byte
+
+	batch        *batch        // current batch
+	mu           sync.RWMutex  // mutex for accessing current batch
+	writeTrigger chan struct{} // channel to signal current write batch
+	writeDone    chan struct{} // closed when writeBatches function returns
+	close        chan struct{} // closed on Close, signals other goroutines to terminate
+}
+
+// New returns a new DB.  All fields and indexes are initialized
+// and possible conflicts with schema from existing database is checked.
+// One goroutine for writing batches is created.
+func New(path string, baseKey []byte) (db *DB, err error) {
+	db = &DB{
+		baseKey:      baseKey,
+		batch:        newBatch(),
+		writeTrigger: make(chan struct{}, 1),
+		close:        make(chan struct{}),
+		writeDone:    make(chan struct{}),
+	}
+	db.shed, err = shed.NewDB(path)
+	if err != nil {
+		return nil, err
+	}
+	// Identify current storage schema by arbitrary name.
+	db.schemaName, err = db.shed.NewStringField("schema-name")
+	if err != nil {
+		return nil, err
+	}
+	db.sizeCounter, err = db.shed.NewUint64Field("size")
+	if err != nil {
+		return nil, err
+	}
+	db.retrievalIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{
+		EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+			return fields.Address, nil
+		},
+		DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+			e.Address = key
+			return e, nil
+		},
+		EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+			b := make([]byte, 16)
+			binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
+			binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
+			value = append(b, fields.Data...)
+			return value, nil
+		},
+		DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+			e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
+			e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
+			e.Data = value[16:]
+			return e, nil
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	// pull index allows history and live syncing per po bin
+	db.pullIndex, err = db.shed.NewIndex("PO|StoredTimestamp|Hash->nil", shed.IndexFuncs{
+		EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+			key = make([]byte, 41)
+			key[0] = db.po(fields.Address)
+			binary.BigEndian.PutUint64(key[1:9], uint64(fields.StoreTimestamp))
+			copy(key[9:], fields.Address[:])
+			return key, nil
+		},
+		DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+			e.Address = key[9:]
+			e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[1:9]))
+			return e, nil
+		},
+		EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+			return nil, nil
+		},
+		DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+			return e, nil
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	// push index contains as yet unsynced chunks
+	db.pushIndex, err = db.shed.NewIndex("StoredTimestamp|Hash->nil", shed.IndexFuncs{
+		EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+			key = make([]byte, 40)
+			binary.BigEndian.PutUint64(key[:8], uint64(fields.StoreTimestamp))
+			copy(key[8:], fields.Address[:])
+			return key, nil
+		},
+		DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+			e.Address = key[8:]
+			e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
+			return e, nil
+		},
+		EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+			return nil, nil
+		},
+		DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+			e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
+			return e, nil
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	// gc index for removable chunk ordered by ascending last access time
+	db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|StoredTimestamp|Hash->nil", shed.IndexFuncs{
+		EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+			b := make([]byte, 16, 16+len(fields.Address))
+			binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp))
+			binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp))
+			key = append(b, fields.Address...)
+			return key, nil
+		},
+		DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+			e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
+			e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16]))
+			e.Address = key[16:]
+			return e, nil
+		},
+		EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+			return nil, nil
+		},
+		DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+			return e, nil
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	// start goroutine what writes batches
+	go db.writeBatches()
+	return db, nil
+}
+
+// Close closes the underlying database.
+func (db *DB) Close() (err error) {
+	// signal other goroutines that
+	// the database is closing
+	close(db.close)
+	select {
+	// wait for writeBatches to write
+	// the last batch
+	case <-db.writeDone:
+	// closing timeout
+	case <-time.After(closeTimeout):
+	}
+	return db.shed.Close()
+}
+
+// writeBatches is a forever loop handing out the current batch apply
+// the batch when the db is free.
+func (db *DB) writeBatches() {
+	// close the writeDone channel
+	// so the DB.Close can return
+	defer close(db.writeDone)
+
+	write := func() {
+		db.mu.Lock()
+		b := db.batch
+		db.batch = newBatch()
+		db.mu.Unlock()
+		b.Err = db.shed.WriteBatch(b.Batch)
+		close(b.Done)
+	}
+	for {
+		select {
+		case <-db.writeTrigger:
+			write()
+		case <-db.close:
+			// check it there is a batch
+			// left to be written
+			write()
+			return
+		}
+	}
+}
+
+// po computes the proximity order between the address
+// and database base key.
+func (db *DB) po(addr storage.Address) (bin uint8) {
+	return uint8(storage.Proximity(db.baseKey, addr))
+}
+
+// now is a helper function that returns a current unix timestamp
+// in UTC timezone.
+func now() (t int64) {
+	return time.Now().UTC().UnixNano()
+}

swarm/storage/localstore/localstore_test.go

@@ -0,0 +1,61 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"crypto/rand"
+	"io/ioutil"
+	"os"
+	"testing"
+
+	ch "github.com/ethereum/go-ethereum/swarm/chunk"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// newTestDB is a helper function that constructs a
+// temporary database and returns a cleanup function that must
+// be called to remove the data.
+func newTestDB(t *testing.T) (db *DB, cleanupFunc func()) {
+	t.Helper()
+
+	dir, err := ioutil.TempDir("", "shed-test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupFunc = func() { os.RemoveAll(dir) }
+	baseKey := make([]byte, 32)
+	if _, err := rand.Read(baseKey); err != nil {
+		t.Fatal(err)
+	}
+	db, err = New(dir, baseKey)
+	if err != nil {
+		cleanupFunc()
+		t.Fatal(err)
+	}
+	cleanupFunc = func() {
+		err := db.Close()
+		if err != nil {
+			t.Error(err)
+		}
+		os.RemoveAll(dir)
+	}
+	return db, cleanupFunc
+}
+
+func generateRandomChunk() storage.Chunk {
+	return storage.GenerateRandomChunk(ch.DefaultSize)
+}

swarm/storage/localstore/mode.go

@@ -0,0 +1,182 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/swarm/shed"
+	"github.com/syndtr/goleveldb/leveldb"
+)
+
+// Mode enumerates different modes of access and update
+// operations on a database.
+type Mode int
+
+// Modes of access and update.
+const (
+	ModeSyncing Mode = iota
+	ModeUpload
+	ModeRequest
+	ModeSynced
+	ModeAccess
+	// Q: this mode is not needed,
+	// as it will be used only internally for GC.
+	ModeRemoval
+)
+
+// ModeName returns a descriptive name of a Mode.
+// If the Mode is not know, a blank string is returned.
+func ModeName(m Mode) (name string) {
+	switch m {
+	case ModeSyncing:
+		return "syncing"
+	case ModeUpload:
+		return "upload"
+	case ModeRequest:
+		return "request"
+	case ModeSynced:
+		return "synced"
+	case ModeAccess:
+		return "access"
+	case ModeRemoval:
+		return "removal"
+	}
+	return ""
+}
+
+// access is called by an Accessor with a specific Mode.
+// This function utilizes differnet indexes depending on
+// the Mode.
+func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err error) {
+	out, err = db.retrievalIndex.Get(item)
+	if err != nil {
+		return out, err
+	}
+	switch mode {
+	case ModeRequest:
+		// update the access counter
+		// Q: can we do this asynchronously
+		return out, db.update(context.TODO(), mode, item)
+	default:
+		// all other modes are not updating the index
+	}
+	return out, nil
+}
+
+// update is called by an Accessor with a specific Mode.
+// This function calles updateBatch to perform operations
+// on indexes and fields within a single batch.
+func (db *DB) update(ctx context.Context, mode Mode, item shed.IndexItem) error {
+	db.mu.RLock()
+	b := db.batch
+	db.mu.RUnlock()
+
+	// check if the database is not closed
+	select {
+	case <-db.close:
+		return ErrDBClosed
+	default:
+	}
+
+	// call the update with the provided mode
+	err := db.updateBatch(b, mode, item)
+	if err != nil {
+		return err
+	}
+	// trigger the writeBatches loop
+	select {
+	case db.writeTrigger <- struct{}{}:
+	default:
+	}
+	// wait for batch to be written and return batch error
+	// this is in order for Put calls to be synchronous
+	select {
+	case <-b.Done:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	return b.Err
+}
+
+// batch wraps leveldb.Batch extending it with a done channel.
+type batch struct {
+	*leveldb.Batch
+	Done chan struct{} // to signal when batch is written
+	Err  error         // error resulting from write
+}
+
+// newBatch constructs a new batch.
+func newBatch() *batch {
+	return &batch{
+		Batch: new(leveldb.Batch),
+		Done:  make(chan struct{}),
+	}
+}
+
+// updateBatch performs different operations on fields and indexes
+// depending on the provided Mode.
+func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) {
+	switch mode {
+	case ModeSyncing:
+		// put to indexes: retrieve, pull
+		item.StoreTimestamp = now()
+		item.AccessTimestamp = now()
+		db.retrievalIndex.PutInBatch(b.Batch, item)
+		db.pullIndex.PutInBatch(b.Batch, item)
+		db.sizeCounter.IncInBatch(b.Batch)
+
+	case ModeUpload:
+		// put to indexes: retrieve, push, pull
+		item.StoreTimestamp = now()
+		item.AccessTimestamp = now()
+		db.retrievalIndex.PutInBatch(b.Batch, item)
+		db.pullIndex.PutInBatch(b.Batch, item)
+		db.pushIndex.PutInBatch(b.Batch, item)
+
+	case ModeRequest:
+		// put to indexes: retrieve, gc
+		item.StoreTimestamp = now()
+		item.AccessTimestamp = now()
+		db.retrievalIndex.PutInBatch(b.Batch, item)
+		db.gcIndex.PutInBatch(b.Batch, item)
+
+	case ModeSynced:
+		// delete from push, insert to gc
+		item.StoreTimestamp = now()
+		db.retrievalIndex.PutInBatch(b.Batch, item)
+		db.pushIndex.DeleteInBatch(b.Batch, item)
+		db.gcIndex.PutInBatch(b.Batch, item)
+
+	case ModeAccess:
+		// update accessTimeStamp in retrieve, gc
+		db.gcIndex.DeleteInBatch(b.Batch, item)
+		item.AccessTimestamp = now()
+		db.retrievalIndex.PutInBatch(b.Batch, item)
+		db.gcIndex.PutInBatch(b.Batch, item)
+
+	case ModeRemoval:
+		// delete from retrieve, pull, gc
+		db.retrievalIndex.DeleteInBatch(b.Batch, item)
+		db.pullIndex.DeleteInBatch(b.Batch, item)
+		db.gcIndex.DeleteInBatch(b.Batch, item)
+
+	default:
+		return ErrInvalidMode
+	}
+	return nil
+}