From d8acb127a3f3b06f78f875da86239a87f9793476 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Thu, 29 Nov 2018 15:37:12 +0100 Subject: [PATCH] swarm/storage/localstore: most basic database --- swarm/storage/localstore/accessor.go | 74 ++++++ swarm/storage/localstore/accessor_test.go | 98 ++++++++ swarm/storage/localstore/localstore.go | 244 ++++++++++++++++++++ swarm/storage/localstore/localstore_test.go | 61 +++++ swarm/storage/localstore/mode.go | 182 +++++++++++++++ 5 files changed, 659 insertions(+) create mode 100644 swarm/storage/localstore/accessor.go create mode 100644 swarm/storage/localstore/accessor_test.go create mode 100644 swarm/storage/localstore/localstore.go create mode 100644 swarm/storage/localstore/localstore_test.go create mode 100644 swarm/storage/localstore/mode.go diff --git a/swarm/storage/localstore/accessor.go b/swarm/storage/localstore/accessor.go new file mode 100644 index 0000000000..9cba713d08 --- /dev/null +++ b/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 . + +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, + } +} diff --git a/swarm/storage/localstore/accessor_test.go b/swarm/storage/localstore/accessor_test.go new file mode 100644 index 0000000000..20a38c0445 --- /dev/null +++ b/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 . + +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) + } + }) +} diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go new file mode 100644 index 0000000000..9c455507bf --- /dev/null +++ b/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 . + +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() +} diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go new file mode 100644 index 0000000000..1a12c27703 --- /dev/null +++ b/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 . + +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) +} diff --git a/swarm/storage/localstore/mode.go b/swarm/storage/localstore/mode.go new file mode 100644 index 0000000000..63dac3626a --- /dev/null +++ b/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 . + +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 +}