diff options
author | Balint Gabor <balint.g@gmail.com> | 2018-09-13 17:42:19 +0800 |
---|---|---|
committer | GitHub <noreply@github.com> | 2018-09-13 17:42:19 +0800 |
commit | 3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e (patch) | |
tree | 62a2896b3b824449595272f0b92dda877ba1c58d /swarm/storage/hasherstore.go | |
parent | ff3a5d24d2e40fd66f7813173e9cfc31144f3c53 (diff) | |
download | go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.tar go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.tar.gz go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.tar.bz2 go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.tar.lz go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.tar.xz go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.tar.zst go-tangerine-3ff2f756368f2d8ec0d1d9d25f6ba9cdabd7383e.zip |
swarm: Chunk refactor (#17659)
Co-authored-by: Janos Guljas <janos@resenje.org>
Co-authored-by: Balint Gabor <balint.g@gmail.com>
Co-authored-by: Anton Evangelatov <anton.evangelatov@gmail.com>
Co-authored-by: Viktor TrĂ³n <viktor.tron@gmail.com>
Diffstat (limited to 'swarm/storage/hasherstore.go')
-rw-r--r-- | swarm/storage/hasherstore.go | 112 |
1 files changed, 68 insertions, 44 deletions
diff --git a/swarm/storage/hasherstore.go b/swarm/storage/hasherstore.go index 766207eae..879622b9a 100644 --- a/swarm/storage/hasherstore.go +++ b/swarm/storage/hasherstore.go @@ -19,10 +19,10 @@ package storage import ( "context" "fmt" - "sync" + "sync/atomic" "github.com/ethereum/go-ethereum/crypto/sha3" - "github.com/ethereum/go-ethereum/swarm/chunk" + ch "github.com/ethereum/go-ethereum/swarm/chunk" "github.com/ethereum/go-ethereum/swarm/storage/encryption" ) @@ -30,31 +30,36 @@ type hasherStore struct { store ChunkStore toEncrypt bool hashFunc SwarmHasher - hashSize int // content hash size - refSize int64 // reference size (content hash + possibly encryption key) - wg *sync.WaitGroup - closed chan struct{} + hashSize int // content hash size + refSize int64 // reference size (content hash + possibly encryption key) + nrChunks uint64 // number of chunks to store + errC chan error // global error channel + doneC chan struct{} // closed by Close() call to indicate that count is the final number of chunks + quitC chan struct{} // closed to quit unterminated routines } // NewHasherStore creates a hasherStore object, which implements Putter and Getter interfaces. // With the HasherStore you can put and get chunk data (which is just []byte) into a ChunkStore // and the hasherStore will take core of encryption/decryption of data if necessary -func NewHasherStore(chunkStore ChunkStore, hashFunc SwarmHasher, toEncrypt bool) *hasherStore { +func NewHasherStore(store ChunkStore, hashFunc SwarmHasher, toEncrypt bool) *hasherStore { hashSize := hashFunc().Size() refSize := int64(hashSize) if toEncrypt { refSize += encryption.KeyLength } - return &hasherStore{ - store: chunkStore, + h := &hasherStore{ + store: store, toEncrypt: toEncrypt, hashFunc: hashFunc, hashSize: hashSize, refSize: refSize, - wg: &sync.WaitGroup{}, - closed: make(chan struct{}), + errC: make(chan error), + doneC: make(chan struct{}), + quitC: make(chan struct{}), } + + return h } // Put stores the chunkData into the ChunkStore of the hasherStore and returns the reference. @@ -62,7 +67,6 @@ func NewHasherStore(chunkStore ChunkStore, hashFunc SwarmHasher, toEncrypt bool) // Asynchronous function, the data will not necessarily be stored when it returns. func (h *hasherStore) Put(ctx context.Context, chunkData ChunkData) (Reference, error) { c := chunkData - size := chunkData.Size() var encryptionKey encryption.Key if h.toEncrypt { var err error @@ -71,29 +75,28 @@ func (h *hasherStore) Put(ctx context.Context, chunkData ChunkData) (Reference, return nil, err } } - chunk := h.createChunk(c, size) - + chunk := h.createChunk(c) h.storeChunk(ctx, chunk) - return Reference(append(chunk.Addr, encryptionKey...)), nil + return Reference(append(chunk.Address(), encryptionKey...)), nil } // Get returns data of the chunk with the given reference (retrieved from the ChunkStore of hasherStore). // If the data is encrypted and the reference contains an encryption key, it will be decrypted before // return. func (h *hasherStore) Get(ctx context.Context, ref Reference) (ChunkData, error) { - key, encryptionKey, err := parseReference(ref, h.hashSize) + addr, encryptionKey, err := parseReference(ref, h.hashSize) if err != nil { return nil, err } - toDecrypt := (encryptionKey != nil) - chunk, err := h.store.Get(ctx, key) + chunk, err := h.store.Get(ctx, addr) if err != nil { return nil, err } - chunkData := chunk.SData + chunkData := ChunkData(chunk.Data()) + toDecrypt := (encryptionKey != nil) if toDecrypt { var err error chunkData, err = h.decryptChunkData(chunkData, encryptionKey) @@ -107,16 +110,40 @@ func (h *hasherStore) Get(ctx context.Context, ref Reference) (ChunkData, error) // Close indicates that no more chunks will be put with the hasherStore, so the Wait // function can return when all the previously put chunks has been stored. func (h *hasherStore) Close() { - close(h.closed) + close(h.doneC) } // Wait returns when // 1) the Close() function has been called and // 2) all the chunks which has been Put has been stored func (h *hasherStore) Wait(ctx context.Context) error { - <-h.closed - h.wg.Wait() - return nil + defer close(h.quitC) + var nrStoredChunks uint64 // number of stored chunks + var done bool + doneC := h.doneC + for { + select { + // if context is done earlier, just return with the error + case <-ctx.Done(): + return ctx.Err() + // doneC is closed if all chunks have been submitted, from then we just wait until all of them are also stored + case <-doneC: + done = true + doneC = nil + // a chunk has been stored, if err is nil, then successfully, so increase the stored chunk counter + case err := <-h.errC: + if err != nil { + return err + } + nrStoredChunks++ + } + // if all the chunks have been submitted and all of them are stored, then we can return + if done { + if nrStoredChunks >= atomic.LoadUint64(&h.nrChunks) { + return nil + } + } + } } func (h *hasherStore) createHash(chunkData ChunkData) Address { @@ -126,12 +153,9 @@ func (h *hasherStore) createHash(chunkData ChunkData) Address { return hasher.Sum(nil) } -func (h *hasherStore) createChunk(chunkData ChunkData, chunkSize int64) *Chunk { +func (h *hasherStore) createChunk(chunkData ChunkData) *chunk { hash := h.createHash(chunkData) - chunk := NewChunk(hash, nil) - chunk.SData = chunkData - chunk.Size = chunkSize - + chunk := NewChunk(hash, chunkData) return chunk } @@ -162,10 +186,10 @@ func (h *hasherStore) decryptChunkData(chunkData ChunkData, encryptionKey encryp // removing extra bytes which were just added for padding length := ChunkData(decryptedSpan).Size() - for length > chunk.DefaultSize { - length = length + (chunk.DefaultSize - 1) - length = length / chunk.DefaultSize - length *= h.refSize + for length > ch.DefaultSize { + length = length + (ch.DefaultSize - 1) + length = length / ch.DefaultSize + length *= uint64(h.refSize) } c := make(ChunkData, length+8) @@ -205,32 +229,32 @@ func (h *hasherStore) decrypt(chunkData ChunkData, key encryption.Key) ([]byte, } func (h *hasherStore) newSpanEncryption(key encryption.Key) encryption.Encryption { - return encryption.New(key, 0, uint32(chunk.DefaultSize/h.refSize), sha3.NewKeccak256) + return encryption.New(key, 0, uint32(ch.DefaultSize/h.refSize), sha3.NewKeccak256) } func (h *hasherStore) newDataEncryption(key encryption.Key) encryption.Encryption { - return encryption.New(key, int(chunk.DefaultSize), 0, sha3.NewKeccak256) + return encryption.New(key, int(ch.DefaultSize), 0, sha3.NewKeccak256) } -func (h *hasherStore) storeChunk(ctx context.Context, chunk *Chunk) { - h.wg.Add(1) +func (h *hasherStore) storeChunk(ctx context.Context, chunk *chunk) { + atomic.AddUint64(&h.nrChunks, 1) go func() { - <-chunk.dbStoredC - h.wg.Done() + select { + case h.errC <- h.store.Put(ctx, chunk): + case <-h.quitC: + } }() - h.store.Put(ctx, chunk) } func parseReference(ref Reference, hashSize int) (Address, encryption.Key, error) { - encryptedKeyLength := hashSize + encryption.KeyLength + encryptedRefLength := hashSize + encryption.KeyLength switch len(ref) { - case KeyLength: + case AddressLength: return Address(ref), nil, nil - case encryptedKeyLength: + case encryptedRefLength: encKeyIdx := len(ref) - encryption.KeyLength return Address(ref[:encKeyIdx]), encryption.Key(ref[encKeyIdx:]), nil default: - return nil, nil, fmt.Errorf("Invalid reference length, expected %v or %v got %v", hashSize, encryptedKeyLength, len(ref)) + return nil, nil, fmt.Errorf("Invalid reference length, expected %v or %v got %v", hashSize, encryptedRefLength, len(ref)) } - } |