forked from ethereum/go-ethereum
-
Notifications
You must be signed in to change notification settings - Fork 0
/
types.go
232 lines (197 loc) · 7.56 KB
/
types.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
// Copyright 2016 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 storage
import (
"bytes"
"crypto"
"fmt"
"hash"
"io"
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto/sha3"
)
type Hasher func() hash.Hash
// Peer is the recorded as Source on the chunk
// should probably not be here? but network should wrap chunk object
type Peer interface{}
type Key []byte
func (x Key) Size() uint {
return uint(len(x))
}
func (x Key) isEqual(y Key) bool {
return bytes.Equal(x, y)
}
func (h Key) bits(i, j uint) uint {
ii := i >> 3
jj := i & 7
if ii >= h.Size() {
return 0
}
if jj+j <= 8 {
return uint((h[ii] >> jj) & ((1 << j) - 1))
}
res := uint(h[ii] >> jj)
jj = 8 - jj
j -= jj
for j != 0 {
ii++
if j < 8 {
res += uint(h[ii]&((1<<j)-1)) << jj
return res
}
res += uint(h[ii]) << jj
jj += 8
j -= 8
}
return res
}
func IsZeroKey(key Key) bool {
return len(key) == 0 || bytes.Equal(key, ZeroKey)
}
var ZeroKey = Key(common.Hash{}.Bytes())
func MakeHashFunc(hash string) Hasher {
switch hash {
case "SHA256":
return crypto.SHA256.New
case "SHA3":
return sha3.NewKeccak256
}
return nil
}
func (key Key) Hex() string {
return fmt.Sprintf("%064x", []byte(key[:]))
}
func (key Key) Log() string {
if len(key[:]) < 4 {
return fmt.Sprintf("%x", []byte(key[:]))
}
return fmt.Sprintf("%08x", []byte(key[:4]))
}
func (key Key) String() string {
return fmt.Sprintf("%064x", []byte(key)[:])
}
func (key Key) MarshalJSON() (out []byte, err error) {
return []byte(`"` + key.String() + `"`), nil
}
func (key *Key) UnmarshalJSON(value []byte) error {
s := string(value)
*key = make([]byte, 32)
h := common.Hex2Bytes(s[1 : len(s)-1])
copy(*key, h)
return nil
}
// each chunk when first requested opens a record associated with the request
// next time a request for the same chunk arrives, this record is updated
// this request status keeps track of the request ID-s as well as the requesting
// peers and has a channel that is closed when the chunk is retrieved. Multiple
// local callers can wait on this channel (or combined with a timeout, block with a
// select).
type RequestStatus struct {
Key Key
Source Peer
C chan bool
Requesters map[uint64][]interface{}
}
func newRequestStatus(key Key) *RequestStatus {
return &RequestStatus{
Key: key,
Requesters: make(map[uint64][]interface{}),
C: make(chan bool),
}
}
// Chunk also serves as a request object passed to ChunkStores
// in case it is a retrieval request, Data is nil and Size is 0
// Note that Size is not the size of the data chunk, which is Data.Size()
// but the size of the subtree encoded in the chunk
// 0 if request, to be supplied by the dpa
type Chunk struct {
Key Key // always
SData []byte // nil if request, to be supplied by dpa
Size int64 // size of the data covered by the subtree encoded in this chunk
Source Peer // peer
C chan bool // to signal data delivery by the dpa
Req *RequestStatus // request Status needed by netStore
wg *sync.WaitGroup // wg to synchronize
dbStored chan bool // never remove a chunk from memStore before it is written to dbStore
}
func NewChunk(key Key, rs *RequestStatus) *Chunk {
return &Chunk{Key: key, Req: rs}
}
/*
The ChunkStore interface is implemented by :
- MemStore: a memory cache
- DbStore: local disk/db store
- LocalStore: a combination (sequence of) memStore and dbStore
- NetStore: cloud storage abstraction layer
- DPA: local requests for swarm storage and retrieval
*/
type ChunkStore interface {
Put(*Chunk) // effectively there is no error even if there is an error
Get(Key) (*Chunk, error)
}
/*
Chunker is the interface to a component that is responsible for disassembling and assembling larger data and indended to be the dependency of a DPA storage system with fixed maximum chunksize.
It relies on the underlying chunking model.
When calling Split, the caller provides a channel (chan *Chunk) on which it receives chunks to store. The DPA delegates to storage layers (implementing ChunkStore interface).
Split returns an error channel, which the caller can monitor.
After getting notified that all the data has been split (the error channel is closed), the caller can safely read or save the root key. Optionally it times out if not all chunks get stored or not the entire stream of data has been processed. By inspecting the errc channel the caller can check if any explicit errors (typically IO read/write failures) occurred during splitting.
When calling Join with a root key, the caller gets returned a seekable lazy reader. The caller again provides a channel on which the caller receives placeholder chunks with missing data. The DPA is supposed to forward this to the chunk stores and notify the chunker if the data has been delivered (i.e. retrieved from memory cache, disk-persisted db or cloud based swarm delivery). As the seekable reader is used, the chunker then puts these together the relevant parts on demand.
*/
type Splitter interface {
/*
When splitting, data is given as a SectionReader, and the key is a hashSize long byte slice (Key), the root hash of the entire content will fill this once processing finishes.
New chunks to store are coming to caller via the chunk storage channel, which the caller provides.
wg is a Waitgroup (can be nil) that can be used to block until the local storage finishes
The caller gets returned an error channel, if an error is encountered during splitting, it is fed to errC error channel.
A closed error signals process completion at which point the key can be considered final if there were no errors.
*/
Split(io.Reader, int64, chan *Chunk, *sync.WaitGroup, *sync.WaitGroup) (Key, error)
}
type Joiner interface {
/*
Join reconstructs original content based on a root key.
When joining, the caller gets returned a Lazy SectionReader, which is
seekable and implements on-demand fetching of chunks as and where it is read.
New chunks to retrieve are coming to caller via the Chunk channel, which the caller provides.
If an error is encountered during joining, it appears as a reader error.
The SectionReader.
As a result, partial reads from a document are possible even if other parts
are corrupt or lost.
The chunks are not meant to be validated by the chunker when joining. This
is because it is left to the DPA to decide which sources are trusted.
*/
Join(key Key, chunkC chan *Chunk) LazySectionReader
}
type Chunker interface {
Joiner
Splitter
// returns the key length
// KeySize() int64
}
// Size, Seek, Read, ReadAt
type LazySectionReader interface {
Size(chan bool) (int64, error)
io.Seeker
io.Reader
io.ReaderAt
}
type LazyTestSectionReader struct {
*io.SectionReader
}
func (self *LazyTestSectionReader) Size(chan bool) (int64, error) {
return self.SectionReader.Size(), nil
}