diff --git a/cmd/swarm/cleandb.go b/cmd/swarm/cleandb.go
index 2680760624..9f5edd7231 100644
--- a/cmd/swarm/cleandb.go
+++ b/cmd/swarm/cleandb.go
@@ -17,22 +17,63 @@
package main
import (
+ "encoding/json"
+ "fmt"
+ "io/ioutil"
+ "path"
+ "path/filepath"
+
"github.com/ethereum/go-ethereum/cmd/utils"
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/log"
+ "github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/storage"
"gopkg.in/urfave/cli.v1"
)
func cleandb(ctx *cli.Context) {
- args := ctx.Args()
- if len(args) != 1 {
- utils.Fatalf("Need path to chunks database as the first and only argument")
- }
-
- chunkDbPath := args[0]
- hash := storage.MakeHashFunc("SHA3")
- dbStore, err := storage.NewDbStore(chunkDbPath, hash, 10000000, 0)
+ dbStore, err := setupDb(ctx)
if err != nil {
utils.Fatalf("Cannot initialise dbstore: %v", err)
}
dbStore.Cleanup()
}
+
+func dumpdb(ctx *cli.Context) {
+ dbStore, err := setupDb(ctx)
+ if err != nil {
+ utils.Fatalf("Cannot initialise dbstore: %v", err)
+ }
+ dbStore.Dump()
+}
+
+func setupDb(ctx *cli.Context) (*storage.DbStore, error) {
+ args := ctx.Args()
+ if len(args) != 0 {
+ utils.Fatalf("Takes no argument")
+ }
+
+ hash := storage.MakeHashFunc("SHA3")
+
+ var (
+ bzzaccount = ctx.GlobalString(SwarmAccountFlag.Name)
+ datadir = ctx.GlobalString(utils.DataDirFlag.Name)
+ )
+
+ bzzdir := fmt.Sprintf("%s/swarm/bzz-%s", path.Clean(datadir), bzzaccount)
+
+ configdata, err := ioutil.ReadFile(bzzdir + "/config.json")
+ if err != nil {
+ utils.Fatalf("Could not open source config file '%s'", filepath.Join(bzzdir, "/config.json"))
+ }
+
+ var sourceconfig api.Config
+ err = json.Unmarshal(configdata, &sourceconfig)
+ if err != nil {
+ utils.Fatalf("Corrupt or invalid source config file '%s'", filepath.Join(bzzdir, "/config.json"))
+ }
+ log.Trace(fmt.Sprintf("bzzkey %v", sourceconfig.BzzKey))
+
+ basekey := common.HexToHash(sourceconfig.BzzKey[2:])
+ return storage.NewDbStore(filepath.Join(bzzdir, "chunks"), hash, 10000000, func(k storage.Key) (ret uint8) { return uint8(storage.Proximity(basekey[:], k[:])) })
+}
diff --git a/cmd/swarm/import.go b/cmd/swarm/import.go
new file mode 100644
index 0000000000..6fd438e279
--- /dev/null
+++ b/cmd/swarm/import.go
@@ -0,0 +1,97 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of go-ethereum.
+//
+// go-ethereum is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// go-ethereum 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with go-ethereum. If not, see .
+
+package main
+
+import (
+ "encoding/json"
+ "fmt"
+ "io/ioutil"
+ "path"
+
+ "github.com/ethereum/go-ethereum/cmd/utils"
+
+ "github.com/ethereum/go-ethereum/swarm/api"
+ "github.com/ethereum/go-ethereum/swarm/storage"
+
+ "github.com/ethereum/go-ethereum/log"
+ "gopkg.in/urfave/cli.v1"
+)
+
+func import_ldb(ctx *cli.Context) {
+ var importdatadir string
+ var importbzzaccount string
+ var configdata []byte
+ var sourceconfig api.Config
+ var targetconfig api.Config
+
+ var sourcedatadirfull string
+ var targetdatadirfull string
+
+ var err error
+
+ args := ctx.Args()
+ if len(args) == 0 {
+ utils.Fatalf("need at least one argument []")
+ }
+
+ importdatadir = args[0]
+ if len(args) > 0 {
+ importbzzaccount = args[1]
+ }
+
+ var (
+ bzzaccount = ctx.GlobalString(SwarmAccountFlag.Name)
+ datadir = ctx.GlobalString(utils.DataDirFlag.Name)
+ )
+
+ if importdatadir == "" {
+ utils.Fatalf("--importdir must be specificed")
+ }
+ if datadir == "" {
+ utils.Fatalf("--datadir must be specificed")
+ }
+
+ sourcedatadirfull = fmt.Sprintf("%s/swarm/bzz-%s", path.Clean(importdatadir), importbzzaccount)
+ targetdatadirfull = fmt.Sprintf("%s/swarm/bzz-%s", path.Clean(datadir), bzzaccount)
+
+ configdata, err = ioutil.ReadFile(sourcedatadirfull + "/config.json")
+ if err != nil {
+ utils.Fatalf("Could not open source config file '%s'", sourcedatadirfull+"/config.json")
+ }
+ err = json.Unmarshal(configdata, &sourceconfig)
+ if err != nil {
+ utils.Fatalf("Corrupt or invalid source config file '%s'", sourcedatadirfull+"/config.json")
+ }
+ log.Trace(fmt.Sprintf("Sourceconfig has bzzkey %v", sourceconfig.BzzKey))
+
+ configdata, err = ioutil.ReadFile(targetdatadirfull + "/config.json")
+ if err != nil {
+ utils.Fatalf("Could not open target config file '%s'", targetdatadirfull+"/config.json")
+ }
+ err = json.Unmarshal(configdata, &targetconfig)
+ if err != nil {
+ utils.Fatalf("Corrupt or invalid source config file '%s'", targetdatadirfull+"/config.json")
+ }
+
+ chunkcount, err := storage.Import(sourceconfig.ChunkDbPath, targetconfig.ChunkDbPath, sourceconfig.BzzKey, targetconfig.BzzKey)
+ if err != nil {
+ utils.Fatalf("import failed: %s", err)
+ }
+
+ log.Trace(fmt.Sprintf("Chunks imported: %d", chunkcount))
+
+}
diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go
index 6b4feb0bcc..4c9358aee1 100644
--- a/cmd/swarm/main.go
+++ b/cmd/swarm/main.go
@@ -245,7 +245,25 @@ Removes a path from the manifest
Usage: "Cleans database of corrupted entries",
ArgsUsage: " ",
Description: `
-Cleans database of corrupted entries.
+Cleans database of corrupted entries. needs account flag
+`,
+ },
+ {
+ Action: dumpdb,
+ Name: "dumpdb",
+ Usage: "Dump hashes of all chunks",
+ ArgsUsage: " ",
+ Description: `
+Dump hashes of all chunks.
+`,
+ },
+ {
+ Action: import_ldb,
+ Name: "import",
+ Usage: "Imports chunks from one datastore to another",
+ ArgsUsage: " [source-bzzaccount]",
+ Description: `
+Imports chunks from one datastore to another
`,
},
}
diff --git a/swarm/api/api_test.go b/swarm/api/api_test.go
index f9caed27f5..e29ec57b8a 100644
--- a/swarm/api/api_test.go
+++ b/swarm/api/api_test.go
@@ -36,7 +36,7 @@ func testApi(t *testing.T, f func(*Api)) {
}
os.RemoveAll(datadir)
defer os.RemoveAll(datadir)
- dpa, err := storage.NewLocalDPA(datadir)
+ dpa, err := storage.NewLocalDPA(datadir, nil)
if err != nil {
return
}
diff --git a/swarm/network/hive.go b/swarm/network/hive.go
index 70652c4509..a3b270a6a8 100644
--- a/swarm/network/hive.go
+++ b/swarm/network/hive.go
@@ -326,7 +326,7 @@ func loadSync(record *kademlia.NodeRecord, node kademlia.Node) error {
}
if record.Meta == nil {
log.Debug(fmt.Sprintf("no sync state for node record %v setting default", record))
- p.syncState = &syncState{DbSyncState: &storage.DbSyncState{}}
+ p.syncState = &syncState{}
return nil
}
state, err := decodeSync(record.Meta)
diff --git a/swarm/network/kademlia/address.go b/swarm/network/kademlia/address.go
index 16c5ce5322..6d7107921d 100644
--- a/swarm/network/kademlia/address.go
+++ b/swarm/network/kademlia/address.go
@@ -48,6 +48,9 @@ func (a Address) Bin() string {
return strings.Join(bs, "")
}
+func Proximity(one, other Address) (ret uint8) {
+ return uint8(proximity(one, other))
+}
/*
Proximity(x, y) returns the proximity order of the MSB distance between x and y
diff --git a/swarm/network/kademlia/kademlia.go b/swarm/network/kademlia/kademlia.go
index bf976a3e1d..036282f653 100644
--- a/swarm/network/kademlia/kademlia.go
+++ b/swarm/network/kademlia/kademlia.go
@@ -240,6 +240,13 @@ func (self *Kademlia) setProxLimit(r int, on bool) {
}
}
+/*
+returns the current proxLimit
+*/
+func (self *Kademlia) GetProxLimit() int {
+ return self.proxLimit
+}
+
/*
returns the list of nodes belonging to the same proximity bin
as the target. The most proximate bin will be the union of the bins between
diff --git a/swarm/network/messages.go b/swarm/network/messages.go
index d3858c4240..d25c384a68 100644
--- a/swarm/network/messages.go
+++ b/swarm/network/messages.go
@@ -34,14 +34,14 @@ BZZ protocol Message Types and Message Data Types
// bzz protocol message codes
const (
- statusMsg = iota // 0x01
- storeRequestMsg // 0x02
- retrieveRequestMsg // 0x03
- peersMsg // 0x04
- syncRequestMsg // 0x05
- deliveryRequestMsg // 0x06
- unsyncedKeysMsg // 0x07
- paymentMsg // 0x08
+ statusMsg = iota // 0x00
+ storeRequestMsg // 0x01
+ retrieveRequestMsg // 0x02
+ peersMsg // 0x03
+ syncRequestMsg // 0x04
+ deliveryRequestMsg // 0x05
+ unsyncedKeysMsg // 0x06
+ paymentMsg // 0x07
)
/*
diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go
index d013b3109b..0409944fdb 100644
--- a/swarm/network/protocol.go
+++ b/swarm/network/protocol.go
@@ -41,6 +41,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/swarm/network/kademlia"
bzzswap "github.com/ethereum/go-ethereum/swarm/services/swap"
"github.com/ethereum/go-ethereum/swarm/services/swap/swap"
"github.com/ethereum/go-ethereum/swarm/storage"
@@ -364,14 +365,21 @@ func (self *bzz) handleStatus() (err error) {
}
log.Info(fmt.Sprintf("Peer %08x is capable (%d/%d)", self.remoteAddr.Addr[:4], status.Version, status.NetworkId))
+
err = self.hive.addPeer(&peer{bzz: self})
if err != nil {
return err
}
// hive sets syncstate so sync should start after node added
- log.Info(fmt.Sprintf("syncronisation request sent with %v", self.syncState))
+ self.syncState.PO = kademlia.Proximity(self.hive.addr, self.remoteAddr.Addr)
+
+ if self.syncState.PO >= uint8(self.hive.kad.GetProxLimit()) {
+ self.syncState.IncludeCloser = true
+ }
+
self.syncRequest()
+ log.Debug(fmt.Sprintf("syncronisation request sent with %v", self.syncState))
return nil
}
@@ -382,28 +390,23 @@ func (self *bzz) sync(state *syncState) error {
return errors.New("sync request can only be sent once")
}
- cnt := self.dbAccess.counter()
+ cnt := self.dbAccess.currentStorageIndex()
remoteaddr := self.remoteAddr.Addr
- start, stop := self.hive.kad.KeyRange(remoteaddr)
// an explicitly received nil syncstate disables syncronisation
if state == nil {
self.syncEnabled = false
log.Warn(fmt.Sprintf("syncronisation disabled for peer %v", self))
- state = &syncState{DbSyncState: &storage.DbSyncState{}, Synced: true}
+ state = &syncState{Synced: true}
} else {
- state.synced = make(chan bool)
state.SessionAt = cnt
- if storage.IsZeroKey(state.Stop) && state.Synced {
- state.Start = storage.Key(start[:])
- state.Stop = storage.Key(stop[:])
- }
log.Debug(fmt.Sprintf("syncronisation requested by peer %v at state %v", self, state))
}
var err error
self.syncer, err = newSyncer(
self.requestDb,
storage.Key(remoteaddr[:]),
+ self.hive.kad.GetProxLimit,
self.dbAccess,
self.unsyncedKeys, self.store,
self.syncParams, state, func() bool { return self.syncEnabled },
@@ -502,7 +505,7 @@ func (self *bzz) peers(req *peersMsgData) error {
func (self *bzz) send(msg uint64, data interface{}) error {
if self.hive.blockWrite {
- return fmt.Errorf("network write blocked")
+ log.Warn("network write blocked")
}
log.Trace(fmt.Sprintf("-> %v: %v (%T) to %v", msg, data, data, self))
err := p2p.Send(self.rw, msg, data)
diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go
index 20129c2a81..0d0ec7306f 100644
--- a/swarm/network/syncer.go
+++ b/swarm/network/syncer.go
@@ -1,4 +1,4 @@
-// Copyright 2016 The go-ethereum Authors
+/// 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
@@ -21,6 +21,7 @@ import (
"encoding/json"
"fmt"
"path/filepath"
+ "time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/storage"
@@ -31,6 +32,7 @@ const (
requestDbBatchSize = 512 // size of batch before written to request db
keyBufferSize = 1024 // size of buffer for unsynced keys
syncBatchSize = 128 // maximum batchsize for outgoing requests
+ historyBufferSize = 128 // maximum size for history iteration buffer
syncBufferSize = 128 // size of buffer for delivery requests
syncCacheSize = 1024 // cache capacity to store request queue in memory
)
@@ -54,11 +56,16 @@ const (
// json serialisable struct to record the syncronisation state between 2 peers
type syncState struct {
- *storage.DbSyncState // embeds the following 4 fields:
- // Start Key // lower limit of address space
- // Stop Key // upper limit of address space
- // First uint64 // counter taken from last sync state
- // Last uint64 // counter of remote peer dbStore at the time of last connection
+ SessionAt uint64 // set at the time of connection
+ Since uint64 // requested start index
+ PO uint8 // the requested proximity order (wrt to requester's address)
+ Last uint64 // index of last synced chunk
+ Synced bool // true iff Sync is done up to session at
+ IncludeCloser bool // whether to include all keys that are closer to the peer than the requested PO
+}
+
+// json serialisable struct to record the syncronisation state between 2 peers
+type legacySyncState struct {
SessionAt uint64 // set at the time of connection
LastSeenAt uint64 // set at the time of connection
Latest storage.Key // cursor of dbstore when last (continuously set by syncer)
@@ -82,40 +89,18 @@ func (self *DbAccess) get(key storage.Key) (*storage.Chunk, error) {
}
// current storage counter of chunk db
-func (self *DbAccess) counter() uint64 {
- return self.db.Counter()
+func (self *DbAccess) currentStorageIndex() uint64 {
+ return self.db.CurrentStorageIndex()
}
-// implemented by dbStoreSyncIterator
-type keyIterator interface {
- Next() storage.Key
-}
-
-// generator function for iteration by address range and storage counter
-func (self *DbAccess) iterator(s *syncState) keyIterator {
- it, err := self.db.NewSyncIterator(*(s.DbSyncState))
- if err != nil {
- return nil
- }
- return keyIterator(it)
+// iteration storage counter and proximity order
+func (self *DbAccess) iterator(since uint64, until uint64, po uint8, f func(storage.Key, uint64) bool) error {
+ return self.db.SyncIterator(since, until, po, f)
}
func (self syncState) String() string {
- if self.Synced {
- return fmt.Sprintf(
- "session started at: %v, last seen at: %v, latest key: %v",
- self.SessionAt, self.LastSeenAt,
- self.Latest.Log(),
- )
- } else {
- return fmt.Sprintf(
- "address: %v-%v, index: %v-%v, session started at: %v, last seen at: %v, latest key: %v",
- self.Start.Log(), self.Stop.Log(),
- self.First, self.Last,
- self.SessionAt, self.LastSeenAt,
- self.Latest.Log(),
- )
- }
+ return fmt.Sprintf("synced: session started at: %v, requested sync since: %v, latest key: %v, includecloser: %v",
+ self.SessionAt, self.Since, self.Last, self.IncludeCloser)
}
// syncer parameters (global, not peer specific)
@@ -146,14 +131,15 @@ func NewSyncParams(bzzdir string) *SyncParams {
// syncer is the agent that manages content distribution/storage replication/chunk storeRequest forwarding
type syncer struct {
- *SyncParams // sync parameters
- syncF func() bool // if syncing is needed
- key storage.Key // remote peers address key
- state *syncState // sync state for our dbStore
- syncStates chan *syncState // different stages of sync
- deliveryRequest chan bool // one of two triggers needed to send unsyncedKeys
- newUnsyncedKeys chan bool // one of two triggers needed to send unsynced keys
- quit chan bool // signal to quit loops
+ *SyncParams // sync parameters
+ syncF func() bool // if syncing is needed
+ key storage.Key // remote peers address key
+ proxLimit func() int // kademlia proxlimit retrieval function
+ state *syncState // sync state for our dbStore
+ //syncStates chan *syncState // different stages of sync
+ deliveryRequest chan bool // one of two triggers needed to send unsyncedKeys
+ newUnsyncedKeys chan bool // one of two triggers needed to send unsynced keys
+ quit chan bool // signal to quit loops
// DB related fields
dbAccess *DbAccess // access to dbStore
@@ -175,6 +161,7 @@ type syncer struct {
// by the forwarder
func newSyncer(
db *storage.LDBDatabase, remotekey storage.Key,
+ proxLimit func() int,
dbAccess *DbAccess,
unsyncedKeys func([]*syncRequest, *syncState) error,
store func(*storeRequestMsgData) error,
@@ -190,8 +177,8 @@ func newSyncer(
self := &syncer{
syncF: syncF,
key: remotekey,
+ proxLimit: proxLimit,
dbAccess: dbAccess,
- syncStates: make(chan *syncState, 20),
deliveryRequest: make(chan bool, 1),
newUnsyncedKeys: make(chan bool, 1),
SyncParams: params,
@@ -213,8 +200,29 @@ func newSyncer(
go self.syncDeliveries()
// launch sync task manager
if self.syncF() {
- go self.sync()
+ /*
+ * first all items left in the request Db are replayed
+ * type = StaleSync
+ * Mode: by default once again via confirmation roundtrip
+ * Priority: the items are replayed as the proirity specified for StaleSync
+ * but within the order respects earlier priority level of request
+ * after all items are consumed for a priority level, then the respective
+ queue for delivery requests is open (this way new reqs not written to db)
+ * the sync state provided by the remote peer is used to sync history
+ sync is called from the syncer constructor and is not supposed to be used externally
+ */
+ if state.SessionAt == 0 {
+ log.Trace(fmt.Sprintf("syncer[%v]: nothing to sync", self.key.Log()))
+ return self, nil
+ }
+ log.Trace(fmt.Sprintf("syncer[%v]: start replaying stale requests from request db", self.key.Log()))
+ for p := priorities - 1; p >= 0; p-- {
+ self.queues[p].dbRead(false, 0, self.replay())
+ }
+ log.Trace(fmt.Sprintf("syncer[%v]: done replaying stale requests from request db", self.key.Log()))
+
}
+
// process unsynced keys to broadcast
go self.syncUnsyncedKeys()
@@ -239,96 +247,11 @@ func decodeSync(meta *json.RawMessage) (*syncState, error) {
if len(data) == 0 {
return nil, fmt.Errorf("unable to deserialise sync state from ")
}
- state := &syncState{DbSyncState: &storage.DbSyncState{}}
+ state := &syncState{}
err := json.Unmarshal(data, state)
return state, err
}
-/*
- sync implements the syncing script
- * first all items left in the request Db are replayed
- * type = StaleSync
- * Mode: by default once again via confirmation roundtrip
- * Priority: the items are replayed as the proirity specified for StaleSync
- * but within the order respects earlier priority level of request
- * after all items are consumed for a priority level, the the respective
- queue for delivery requests is open (this way new reqs not written to db)
- (TODO: this should be checked)
- * the sync state provided by the remote peer is used to sync history
- * all the backlog from earlier (aborted) syncing is completed starting from latest
- * if Last < LastSeenAt then all items in between then process all
- backlog from upto last disconnect
- * if Last > 0 &&
-
- sync is called from the syncer constructor and is not supposed to be used externally
-*/
-func (self *syncer) sync() {
- state := self.state
- // sync finished
- defer close(self.syncStates)
-
- // 0. first replay stale requests from request db
- if state.SessionAt == 0 {
- log.Debug(fmt.Sprintf("syncer[%v]: nothing to sync", self.key.Log()))
- return
- }
- log.Debug(fmt.Sprintf("syncer[%v]: start replaying stale requests from request db", self.key.Log()))
- for p := priorities - 1; p >= 0; p-- {
- self.queues[p].dbRead(false, 0, self.replay())
- }
- log.Debug(fmt.Sprintf("syncer[%v]: done replaying stale requests from request db", self.key.Log()))
-
- // unless peer is synced sync unfinished history beginning on
- if !state.Synced {
- start := state.Start
-
- if !storage.IsZeroKey(state.Latest) {
- // 1. there is unfinished earlier sync
- state.Start = state.Latest
- log.Debug(fmt.Sprintf("syncer[%v]: start syncronising backlog (unfinished sync: %v)", self.key.Log(), state))
- // blocks while the entire history upto state is synced
- self.syncState(state)
- if state.Last < state.SessionAt {
- state.First = state.Last + 1
- }
- }
- state.Latest = storage.ZeroKey
- state.Start = start
- // 2. sync up to last disconnect1
- if state.First < state.LastSeenAt {
- state.Last = state.LastSeenAt
- log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history upto last disconnect at %v: %v", self.key.Log(), state.LastSeenAt, state))
- self.syncState(state)
- state.First = state.LastSeenAt
- }
- state.Latest = storage.ZeroKey
-
- } else {
- // synchronisation starts at end of last session
- state.First = state.LastSeenAt
- }
-
- // 3. sync up to current session start
- // if there have been new chunks since last session
- if state.LastSeenAt < state.SessionAt {
- state.Last = state.SessionAt
- log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history since last disconnect at %v up until session start at %v: %v", self.key.Log(), state.LastSeenAt, state.SessionAt, state))
- // blocks until state syncing is finished
- self.syncState(state)
- }
- log.Info(fmt.Sprintf("syncer[%v]: syncing all history complete", self.key.Log()))
-
-}
-
-// wait till syncronised block uptil state is synced
-func (self *syncer) syncState(state *syncState) {
- self.syncStates <- state
- select {
- case <-state.synced:
- case <-self.quit:
- }
-}
-
// stop quits both request processor and saves the request cache to disk
func (self *syncer) stop() {
close(self.quit)
@@ -360,40 +283,69 @@ func (self *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error)
// serves historical items from the DB
// * read is on demand, blocking unless history channel is read
-// * accepts sync requests (syncStates) to create new db iterator
-// * closes the channel one iteration finishes
+// * closes the channel once iteration finishes
func (self *syncer) syncHistory(state *syncState) chan interface{} {
- var n uint
- history := make(chan interface{})
- log.Debug(fmt.Sprintf("syncer[%v]: syncing history between %v - %v for chunk addresses %v - %v", self.key.Log(), state.First, state.Last, state.Start, state.Stop))
- it := self.dbAccess.iterator(state)
- if it != nil {
- go func() {
- // signal end of the iteration ended
- defer close(history)
- IT:
- for {
- key := it.Next()
- if key == nil {
- break IT
- }
+ var roundCnt, stateCnt, totalCnt uint
+ var quit, wait bool
+ history := make(chan interface{}, historyBufferSize)
+
+ go func() {
+ // signal end of the iteration ended
+ defer close(history)
+ last := state.Last
+ since := state.Since
+ for {
+ log.Debug(fmt.Sprintf("syncer[%v]: syncing history since %v for chunks of proximity order %v", self.key.Log(), since, state.PO))
+ err := self.dbAccess.iterator(since, state.SessionAt, state.PO, func(key storage.Key, idx uint64) bool {
select {
- // blocking until history channel is read from
- case history <- storage.Key(key):
- n++
- log.Trace(fmt.Sprintf("syncer[%v]: history: %v (%v keys)", self.key.Log(), key.Log(), n))
- state.Latest = key
+ // if history channel cannot be written to, we fall through to default
+ // and release the iterator
+ // last is not set to idx so the lost key will be retrieved in the next
+ // batch given Since is set to last
+ case history <- key:
+ roundCnt++
+ stateCnt++
+ totalCnt++
+ last = idx
+ return true
+ case <-self.quit:
+ quit = true
+ return false
+ default:
+ wait = true
+ return false
+ }
+ // return true //dummy return.
+ })
+ if err != nil {
+ log.Debug(fmt.Sprintf("syncer[%v]: sync for %v failed: %v: ..abort syncing", self.key.Log(), state, err))
+ return
+ }
+ if quit {
+ return
+ }
+ // advancing syncstate
+ since = last
+ if !wait {
+ log.Debug(fmt.Sprintf("syncer[%v]: sync for %v failed: %v: ..abort syncing", self.key.Log(), state, err))
+ break
+ }
+ // if history channel is no longer contented, continue outer loop
+ // and read yet another batch
+ for len(history) > historyBufferSize/5 {
+ t := time.NewTimer(100 * time.Millisecond)
+ select {
+ case <-t.C:
case <-self.quit:
- return
}
}
- log.Debug(fmt.Sprintf("syncer[%v]: finished syncing history between %v - %v for chunk addresses %v - %v (at %v) (chunks = %v)", self.key.Log(), state.First, state.Last, state.Start, state.Stop, state.Latest, n))
- }()
- }
+ }
+ roundCnt = 0
+ }()
return history
}
-// triggers key syncronisation
+// triggers key synchronisation
func (self *syncer) sendUnsyncedKeys() {
select {
case self.deliveryRequest <- true:
@@ -412,16 +364,19 @@ func (self *syncer) syncUnsyncedKeys() {
var unsynced []*syncRequest
var more, justSynced bool
var keyCount, historyCnt int
- var history chan interface{}
priority := High
keys := self.keys[priority]
var newUnsyncedKeys, deliveryRequest chan bool
keyCounts := make([]int, priorities)
histPrior := self.SyncPriorities[HistoryReq]
- syncStates := self.syncStates
state := self.state
+ if state.IncludeCloser {
+ state.PO = 255
+ }
+ history := self.syncHistory(self.state)
+
LOOP:
for {
@@ -454,7 +409,7 @@ LOOP:
// if peer ready to receive but nothing to send
if keys == nil && deliveryRequest == nil {
// if no items left and switch to waiting mode
- log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", self.key.Log()))
+ log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting (keys %v (len %v) deliveryrequest %v)", self.key.Log(), keys, len(keys), deliveryRequest))
newUnsyncedKeys = self.newUnsyncedKeys
}
@@ -464,8 +419,7 @@ LOOP:
// * batch full OR
// * all history have been consumed, synced)
if deliveryRequest == nil &&
- (justSynced ||
- len(unsynced) > 0 && keys == nil ||
+ (justSynced && len(unsynced) > 0 ||
len(unsynced) == int(self.SyncBatchSize)) {
justSynced = false
// listen to requests
@@ -473,11 +427,9 @@ LOOP:
newUnsyncedKeys = nil // not care about data until next req comes in
// set sync to current counter
// (all nonhistorical outgoing traffic sheduled and persisted
- state.LastSeenAt = self.dbAccess.counter()
- state.Latest = storage.ZeroKey
- log.Trace(fmt.Sprintf("syncer[%v]: sending %v", self.key.Log(), unsynced))
- // send the unsynced keys
stateCopy := *state
+ // actually sending unsynced keys with a state
+ log.Trace(fmt.Sprintf("syncer[%v]: sending %v", self.key.Log(), unsynced))
err := self.unsyncedKeys(unsynced, &stateCopy)
if err != nil {
log.Warn(fmt.Sprintf("syncer[%v]: unable to send unsynced keys: %v", self.key.Log(), err))
@@ -495,16 +447,24 @@ LOOP:
case req, more = <-keys:
if keys == history && !more {
log.Trace(fmt.Sprintf("syncer[%v]: syncing history segment complete", self.key.Log()))
- // history channel is closed, waiting for new state (called from sync())
- syncStates = self.syncStates
- state.Synced = true // this signals that the current segment is complete
- select {
- case state.synced <- false:
- case <-self.quit:
- break LOOP
- }
+ // moved to closing of channel in syncHistory
justSynced = true
- history = nil
+
+ log.Trace(fmt.Sprintf("syncer[%v]: start synchronising history since last disconnect at %v up until session start at %v: %v", self.key.Log(), state.Last, state.SessionAt, state))
+
+ if state.IncludeCloser && state.PO > uint8(self.proxLimit()) {
+ log.Trace(fmt.Sprintf("syncer[%v]: PO is now %d", self.key.Log(), state.PO))
+ state.PO--
+ history = self.syncHistory(state)
+
+ } else {
+
+ history = nil
+ state.Synced = true
+ state.Last = self.dbAccess.currentStorageIndex()
+ log.Trace(fmt.Sprintf("syncer[%v]: syncing all history complete", self.key.Log()))
+ }
+
}
case <-deliveryRequest:
log.Trace(fmt.Sprintf("syncer[%v]: peer ready to receive", self.key.Log()))
@@ -520,40 +480,27 @@ LOOP:
// signals that data is available to send if peer is ready to receive
newUnsyncedKeys = nil
keys = self.keys[High]
-
- case state, more = <-syncStates:
- // this resets the state
- if !more {
- state = self.state
- log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", self.key.Log(), priority, state))
- state.Synced = true
- syncStates = nil
- } else {
- log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing history upto %v priority %v)", self.key.Log(), priority, state, histPrior))
- state.Synced = false
- history = self.syncHistory(state)
- // only one history at a time, only allow another one once the
- // history channel is closed
- syncStates = nil
- }
}
if req == nil {
continue LOOP
}
- log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) added to unsynced keys: %v", self.key.Log(), priority, req))
+ sreq, err := self.newSyncRequest(req, priority)
+ if err != nil {
+ log.Warn(fmt.Sprintf("syncer[%v]: (priority %v): error creating request for %v: %v)", self.key.Log(), priority, req, state.Synced, err))
+ continue
+ }
+
+ unsynced = append(unsynced, sreq)
+
keyCounts[priority]++
keyCount++
+
if keys == history {
- log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) history item %v (synced = %v)", self.key.Log(), priority, req, state.Synced))
historyCnt++
- }
- if sreq, err := self.newSyncRequest(req, priority); err == nil {
- // extract key from req
- log.Trace(fmt.Sprintf("syncer[%v]: (priority %v): request %v (synced = %v)", self.key.Log(), priority, req, state.Synced))
- unsynced = append(unsynced, sreq)
+ log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) history item %v (synced = %v)", self.key.Log(), priority, req, state.Synced))
} else {
- log.Warn(fmt.Sprintf("syncer[%v]: (priority %v): error creating request for %v: %v)", self.key.Log(), priority, req, err))
+ log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) added to unsynced keys: %v", self.key.Log(), priority, req))
}
}
diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go
index 44d1dd1f72..a2910d7d6e 100644
--- a/swarm/storage/common_test.go
+++ b/swarm/storage/common_test.go
@@ -63,7 +63,7 @@ func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) {
return
}
-func testStore(m ChunkStore, l int64, branches int64, t *testing.T) {
+func testStore(m ChunkStore, indata io.Reader, l int64, branches int64, t *testing.T) {
chunkC := make(chan *Chunk)
go func() {
@@ -79,7 +79,7 @@ func testStore(m ChunkStore, l int64, branches int64, t *testing.T) {
Hash: defaultHash,
})
swg := &sync.WaitGroup{}
- key, _ := chunker.Split(rand.Reader, l, chunkC, swg, nil)
+ key, _ := chunker.Split(indata, l, chunkC, swg, nil)
swg.Wait()
close(chunkC)
chunkC = make(chan *Chunk)
@@ -114,3 +114,28 @@ func testStore(m ChunkStore, l int64, branches int64, t *testing.T) {
close(chunkC)
<-quit
}
+
+// only put, but fills an array supplied by the caller with the keys
+func testSplit(m ChunkStore, l int64, branches int64, chunkkeys []Key, t *testing.T) Key {
+ var i int
+ chunkC := make(chan *Chunk)
+ go func() {
+ for chunk := range chunkC {
+ chunkkeys[i] = chunk.Key
+ i++
+ m.Put(chunk)
+ if chunk.wg != nil {
+ chunk.wg.Done()
+ }
+ }
+ }()
+ chunker := NewTreeChunker(&ChunkerParams{
+ Branches: branches,
+ Hash: defaultHash,
+ })
+ swg := &sync.WaitGroup{}
+ key, _ := chunker.Split(rand.Reader, l, chunkC, swg, nil)
+ swg.Wait()
+ close(chunkC)
+ return key
+}
diff --git a/swarm/storage/database.go b/swarm/storage/database.go
index 2532490cc9..7d3b57265e 100644
--- a/swarm/storage/database.go
+++ b/swarm/storage/database.go
@@ -29,6 +29,7 @@ import (
)
const openFileLimit = 128
+//const openFileLimit = -1
type LDBDatabase struct {
db *leveldb.DB
diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go
index 31ff5b64e6..eaef7aea16 100644
--- a/swarm/storage/dbstore.go
+++ b/swarm/storage/dbstore.go
@@ -25,13 +25,15 @@ package storage
import (
"bytes"
"encoding/binary"
+ "encoding/hex"
"fmt"
"sync"
+ "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
"github.com/syndtr/goleveldb/leveldb"
- "github.com/syndtr/goleveldb/leveldb/iterator"
+ "github.com/syndtr/goleveldb/leveldb/opt"
)
const (
@@ -47,10 +49,13 @@ const (
)
var (
- keyAccessCnt = []byte{2}
- keyEntryCnt = []byte{3}
- keyDataIdx = []byte{4}
- keyGCPos = []byte{5}
+ keyOldData = byte(1)
+ keyAccessCnt = []byte{2}
+ keyEntryCnt = []byte{3}
+ keyDataIdx = []byte{4}
+ keyGCPos = []byte{5}
+ keyData = byte(6)
+ keyDistanceCnt = byte(7)
)
type gcItem struct {
@@ -64,25 +69,32 @@ type DbStore struct {
// this should be stored in db, accessed transactionally
entryCnt, accessCnt, dataIdx, capacity uint64
+ bucketCnt []uint64
gcPos, gcStartPos []byte
gcArray []*gcItem
hashfunc Hasher
-
- lock sync.Mutex
+ po func(Key) uint8
+ lock sync.Mutex
}
-func NewDbStore(path string, hash Hasher, capacity uint64, radius int) (s *DbStore, err error) {
- s = new(DbStore)
+// TODO: Instead of passing the distance function, just pass the address from which distances are calculated
+// to avoid the appearance of a pluggable distance metric and opportunities of bugs associated with providing
+// a function diferent from the one that is actually used.
+func NewDbStore(path string, hash Hasher, capacity uint64, po func(Key) uint8) (*DbStore, error) {
- s.hashfunc = hash
-
- s.db, err = NewLDBDatabase(path)
+ db, err := NewLDBDatabase(path)
if err != nil {
- return
+ return nil, err
}
+ s := &DbStore{
+ hashfunc: hash,
+ db: db,
+ }
+
+ s.po = po
s.setCapacity(capacity)
s.gcStartPos = make([]byte, 1)
@@ -91,15 +103,26 @@ func NewDbStore(path string, hash Hasher, capacity uint64, radius int) (s *DbSto
data, _ := s.db.Get(keyEntryCnt)
s.entryCnt = BytesToU64(data)
+ s.bucketCnt = make([]uint64, 0x100)
+ for i := 0; i < 0x100; i++ {
+ k := make([]byte, 2)
+ k[0] = keyDistanceCnt
+ k[1] = byte(uint8(i))
+ cnt, _ := s.db.Get(k)
+ s.bucketCnt[i] = BytesToU64(cnt)
+ }
data, _ = s.db.Get(keyAccessCnt)
- s.accessCnt = BytesToU64(data)
+ //s.accessCnt = BytesToU64(data)
+ if len(data) == 8 {
+ s.accessCnt = binary.LittleEndian.Uint64(data)
+ }
data, _ = s.db.Get(keyDataIdx)
s.dataIdx = BytesToU64(data)
s.gcPos, _ = s.db.Get(keyGCPos)
if s.gcPos == nil {
s.gcPos = s.gcStartPos
}
- return
+ return s, nil
}
type dpaDBIndex struct {
@@ -111,12 +134,14 @@ func BytesToU64(data []byte) uint64 {
if len(data) < 8 {
return 0
}
- return binary.LittleEndian.Uint64(data)
+ //return binary.LittleEndian.Uint64(data)
+ return binary.BigEndian.Uint64(data)
}
func U64ToBytes(val uint64) []byte {
data := make([]byte, 8)
- binary.LittleEndian.PutUint64(data, val)
+ //binary.LittleEndian.PutUint64(data, val)
+ binary.BigEndian.PutUint64(data, val)
return data
}
@@ -129,38 +154,52 @@ func (s *DbStore) updateIndexAccess(index *dpaDBIndex) {
}
func getIndexKey(hash Key) []byte {
- HashSize := len(hash)
- key := make([]byte, HashSize+1)
+ hashSize := len(hash)
+ key := make([]byte, hashSize+1)
key[0] = 0
copy(key[1:], hash[:])
return key
}
-func getDataKey(idx uint64) []byte {
+func getOldDataKey(idx uint64) []byte {
key := make([]byte, 9)
- key[0] = 1
+ key[0] = keyOldData
binary.BigEndian.PutUint64(key[1:9], idx)
return key
}
+func getDataKey(idx uint64, po uint8) []byte {
+ key := make([]byte, 10)
+ key[0] = keyData
+ key[1] = byte(po)
+ binary.BigEndian.PutUint64(key[2:], idx)
+
+ return key
+}
+
func encodeIndex(index *dpaDBIndex) []byte {
data, _ := rlp.EncodeToBytes(index)
return data
}
func encodeData(chunk *Chunk) []byte {
- return chunk.SData
+ return append(chunk.Key[:], chunk.SData...)
}
-func decodeIndex(data []byte, index *dpaDBIndex) {
+func decodeIndex(data []byte, index *dpaDBIndex) error {
dec := rlp.NewStream(bytes.NewReader(data), 0)
- dec.Decode(index)
+ return dec.Decode(index)
}
func decodeData(data []byte, chunk *Chunk) {
+ chunk.SData = data[32:]
+ chunk.Size = int64(binary.BigEndian.Uint64(data[32:40]))
+}
+
+func decodeOldData(data []byte, chunk *Chunk) {
chunk.SData = data
- chunk.Size = int64(binary.LittleEndian.Uint64(data[0:8]))
+ chunk.Size = int64(binary.BigEndian.Uint64(data[0:8]))
}
func gcListPartition(list []*gcItem, left int, right int, pivotIndex int) int {
@@ -246,17 +285,13 @@ func (s *DbStore) collectGarbage(ratio float32) {
cutidx := gcListSelect(s.gcArray, 0, gcnt-1, int(float32(gcnt)*ratio))
cutval := s.gcArray[cutidx].value
- // fmt.Print(gcnt, " ", s.entryCnt, " ")
-
// actual gc
for i := 0; i < gcnt; i++ {
if s.gcArray[i].value <= cutval {
- s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey)
+ s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey, s.po(Key(s.gcPos[1:])))
}
}
- // fmt.Println(s.entryCnt)
-
s.db.Put(keyGCPos, s.gcPos)
}
@@ -274,21 +309,23 @@ func (s *DbStore) Cleanup() {
}
total++
var index dpaDBIndex
- decodeIndex(it.Value(), &index)
-
- data, err := s.db.Get(getDataKey(index.Idx))
+ err := decodeIndex(it.Value(), &index)
+ if err != nil {
+ it.Next()
+ continue
+ }
+ data, err := s.db.Get(getDataKey(index.Idx, s.po(Key(key[1:]))))
if err != nil {
log.Warn(fmt.Sprintf("Chunk %x found but could not be accessed: %v", key[:], err))
- s.delete(index.Idx, getIndexKey(key[1:]))
+ s.delete(index.Idx, getIndexKey(key[1:]), s.po(Key(key[1:])))
errorsFound++
} else {
hasher := s.hashfunc()
- hasher.Write(data)
+ hasher.Write(data[32:])
hash := hasher.Sum(nil)
if !bytes.Equal(hash, key[1:]) {
log.Warn(fmt.Sprintf("Found invalid chunk. Hash mismatch. hash=%x, key=%x", hash, key[:]))
- s.delete(index.Idx, getIndexKey(key[1:]))
- errorsFound++
+ s.delete(index.Idx, getIndexKey(key[1:]), s.po(Key(key[1:])))
}
}
it.Next()
@@ -297,16 +334,90 @@ func (s *DbStore) Cleanup() {
log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total))
}
-func (s *DbStore) delete(idx uint64, idxKey []byte) {
+func (s *DbStore) Dump() {
+ //Iterates over the database and checks that there are no faulty chunks
+ it := s.db.NewIterator()
+ startPosition := []byte{kpIndex}
+ it.Seek(startPosition)
+ var key []byte
+ var total int
+ for it.Valid() {
+ key = it.Key()
+ if (key == nil) || (key[0] != kpIndex) {
+ break
+ }
+ total++
+ fmt.Printf("%x\n", key[1:])
+ it.Next()
+ }
+ it.Release()
+ log.Warn(fmt.Sprintf("logged %v chunks", total))
+}
+
+func (s *DbStore) ReIndex() {
+ //Iterates over the database and checks that there are no faulty chunks
+ it := s.db.NewIterator()
+ startPosition := []byte{keyOldData}
+ it.Seek(startPosition)
+ var key []byte
+ var errorsFound, total int
+ for it.Valid() {
+ key = it.Key()
+ if (key == nil) || (key[0] != keyOldData) {
+ break
+ }
+ data := it.Value()
+ hasher := s.hashfunc()
+ hasher.Write(data)
+ hash := hasher.Sum(nil)
+
+ newKey := make([]byte, 10)
+ oldCntKey := make([]byte, 2)
+ newCntKey := make([]byte, 2)
+ oldCntKey[0] = keyDistanceCnt
+ newCntKey[0] = keyDistanceCnt
+ key[0] = keyData
+ key[1] = byte(s.po(Key(key[1:])))
+ oldCntKey[1] = key[1]
+ newCntKey[1] = byte(s.po(Key(newKey[1:])))
+ copy(newKey[2:], key[1:])
+ newValue := append(hash, data...)
+
+ batch := new(leveldb.Batch)
+ batch.Delete(key)
+ s.bucketCnt[oldCntKey[1]]--
+ batch.Put(oldCntKey, U64ToBytes(s.bucketCnt[oldCntKey[1]]))
+ batch.Put(newKey, newValue)
+ s.bucketCnt[newCntKey[1]]++
+ batch.Put(newCntKey, U64ToBytes(s.bucketCnt[newCntKey[1]]))
+ s.db.Write(batch)
+ it.Next()
+ }
+ it.Release()
+ log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total))
+}
+
+func (s *DbStore) delete(idx uint64, idxKey []byte, po uint8) {
batch := new(leveldb.Batch)
batch.Delete(idxKey)
- batch.Delete(getDataKey(idx))
+ batch.Delete(getDataKey(idx, po))
s.entryCnt--
+ s.bucketCnt[po]--
+ cntKey := make([]byte, 2)
+ cntKey[0] = keyDistanceCnt
+ cntKey[1] = po
batch.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
+ batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
s.db.Write(batch)
}
-func (s *DbStore) Counter() uint64 {
+func (s *DbStore) Size() uint64 {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ return s.entryCnt
+}
+
+func (s *DbStore) CurrentStorageIndex() uint64 {
s.lock.Lock()
defer s.lock.Unlock()
return s.dataIdx
@@ -328,7 +439,6 @@ func (s *DbStore) Put(chunk *Chunk) {
}
data := encodeData(chunk)
- //data := ethutil.Encode([]interface{}{entry})
if s.entryCnt >= s.capacity {
s.collectGarbage(gcArrayFreeRatio)
@@ -336,7 +446,10 @@ func (s *DbStore) Put(chunk *Chunk) {
batch := new(leveldb.Batch)
- batch.Put(getDataKey(s.dataIdx), data)
+ po := s.po(chunk.Key)
+ t_datakey := getDataKey(s.dataIdx, po)
+ batch.Put(t_datakey, data)
+ log.Trace(fmt.Sprintf("batch put: datai dx %v prox %v chunkkey %v datakey %v data %v", s.dataIdx, s.po(chunk.Key), hex.EncodeToString(chunk.Key), t_datakey, hex.EncodeToString(data[0:64])))
index.Idx = s.dataIdx
s.updateIndexAccess(&index)
@@ -348,9 +461,17 @@ func (s *DbStore) Put(chunk *Chunk) {
s.entryCnt++
batch.Put(keyDataIdx, U64ToBytes(s.dataIdx))
s.dataIdx++
- batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
+ accesscnt := make([]byte, 8)
+ binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
+ batch.Put(keyAccessCnt, accesscnt)
s.accessCnt++
+ s.bucketCnt[po]++
+ cntKey := make([]byte, 2)
+ cntKey[0] = keyDistanceCnt
+ cntKey[1] = po
+ batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
+
s.db.Write(batch)
if chunk.dbStored != nil {
close(chunk.dbStored)
@@ -368,7 +489,10 @@ func (s *DbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
batch := new(leveldb.Batch)
- batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
+ accesscnt := make([]byte, 8)
+ binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
+ batch.Put(keyAccessCnt, accesscnt)
+
s.accessCnt++
s.updateIndexAccess(index)
idata = encodeIndex(index)
@@ -382,21 +506,32 @@ func (s *DbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
func (s *DbStore) Get(key Key) (chunk *Chunk, err error) {
s.lock.Lock()
defer s.lock.Unlock()
+ return s.get(key)
+}
- var index dpaDBIndex
+func (s *DbStore) get(key Key) (chunk *Chunk, err error) {
+ var indx dpaDBIndex
- if s.tryAccessIdx(getIndexKey(key), &index) {
+ if s.tryAccessIdx(getIndexKey(key), &indx) {
var data []byte
- data, err = s.db.Get(getDataKey(index.Idx))
+
+ proximity := s.po(key)
+ datakey := getDataKey(indx.Idx, proximity)
+ data, err = s.db.Get(datakey)
+ log.Trace(fmt.Sprintf("DBStore: Chunk %v indexkey %x datakey %x proximity %d", key.Log(), indx.Idx, datakey, proximity))
if err != nil {
log.Trace(fmt.Sprintf("DBStore: Chunk %v found but could not be accessed: %v", key.Log(), err))
- s.delete(index.Idx, getIndexKey(key))
+ s.delete(indx.Idx, getIndexKey(key), s.po(key))
return
}
+ //
+ data_mod := data[32:]
+
hasher := s.hashfunc()
- hasher.Write(data)
+ hasher.Write(data_mod)
hash := hasher.Sum(nil)
+
if !bytes.Equal(hash, key) {
s.delete(index.Idx, getIndexKey(key))
log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
@@ -454,62 +589,117 @@ func (s *DbStore) Close() {
s.db.Close()
}
-// describes a section of the DbStore representing the unsynced
-// domain relevant to a peer
-// Start - Stop designate a continuous area Keys in an address space
-// typically the addresses closer to us than to the peer but not closer
-// another closer peer in between
-// From - To designates a time interval typically from the last disconnect
-// till the latest connection (real time traffic is relayed)
-type DbSyncState struct {
- Start, Stop Key
- First, Last uint64
-}
-
-// implements the syncer iterator interface
-// iterates by storage index (~ time of storage = first entry to db)
-type dbSyncIterator struct {
- it iterator.Iterator
- DbSyncState
-}
-
// initialises a sync iterator from a syncToken (passed in with the handshake)
-func (self *DbStore) NewSyncIterator(state DbSyncState) (si *dbSyncIterator, err error) {
- if state.First > state.Last {
- return nil, fmt.Errorf("no entries found")
- }
- si = &dbSyncIterator{
- it: self.db.NewIterator(),
- DbSyncState: state,
- }
- si.it.Seek(getIndexKey(state.Start))
- return si, nil
-}
+func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error {
+ s.lock.Lock()
+ defer s.lock.Unlock()
-// walk the area from Start to Stop and returns items within time interval
-// First to Last
-func (self *dbSyncIterator) Next() (key Key) {
- for self.it.Valid() {
- dbkey := self.it.Key()
- if dbkey[0] != 0 {
+ untilkey := getDataKey(until, po)
+
+ it := s.db.NewIterator()
+ it.Seek(getDataKey(since, po))
+ defer it.Release()
+ for it.Valid() {
+ dbkey := it.Key()
+ if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 {
break
}
- key = Key(make([]byte, len(dbkey)-1))
- copy(key[:], dbkey[1:])
- if bytes.Compare(key[:], self.Start) <= 0 {
- self.it.Next()
- continue
- }
- if bytes.Compare(key[:], self.Stop) > 0 {
+
+ key := make([]byte, 32)
+ copy(key, it.Value()[:32])
+ if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) {
break
}
- var index dpaDBIndex
- decodeIndex(self.it.Value(), &index)
- self.it.Next()
- if (index.Idx >= self.First) && (index.Idx < self.Last) {
- return
- }
+ it.Next()
}
- self.it.Release()
return nil
}
+
+func Import(sourcepath string, targetpath string, sourceaccountkey string, targetaccountkey string) (uint64, error) {
+ chunkcount := uint64(0)
+ var j uint64
+ var maxcount uint64 = 0
+ var poc uint16
+ var err error
+ maxcount--
+
+ var chunks_in KeyCollection
+ var chunks_out KeyCollection
+
+ sourceaccountkeyhash := common.HexToHash(sourceaccountkey[2:])
+ targetaccountkeyhash := common.HexToHash(targetaccountkey[2:])
+
+ log.Trace(fmt.Sprintf("srckey %x targetkey %x", sourceaccountkeyhash, targetaccountkeyhash))
+
+ pofunc_source := func(k Key) (ret uint8) {
+ return uint8(Proximity(sourceaccountkeyhash[:], k[:]))
+ }
+
+ pofunc_target := func(k Key) (ret uint8) {
+ return uint8(Proximity(targetaccountkeyhash[:], k[:]))
+ }
+
+ if !databaseExists(sourcepath) {
+ return 0, fmt.Errorf("sourcepath '%s' does not exist or is unavailable (someone else using it?)", sourcepath)
+ }
+ if !databaseExists(targetpath) {
+ return 0, fmt.Errorf("targetpath '%s' does not exist or is unavailable (someone else using it?)", targetpath)
+ }
+
+ store_source, err := NewDbStore(sourcepath, MakeHashFunc(defaultHash), defaultDbCapacity, pofunc_source)
+ if err != nil {
+ return 0, err
+ }
+ store_target, err := NewDbStore(targetpath, MakeHashFunc(defaultHash), defaultDbCapacity, pofunc_target)
+ if err != nil {
+ return 0, err
+ }
+
+ // why does this have to be +1? should not be necessary
+ // if not +1, the arrays in the iterator overflow
+ chunks_in = NewKeyCollection(int(store_source.Size()) + 1)
+ chunks_out = NewKeyCollection(int(store_source.Size()) + 1)
+ bins := make([]int8, int(store_source.Size())+1)
+
+ log.Trace(fmt.Sprintf("Source db count: %v, Target db count: %v ", store_source.Size(), store_target.Size()))
+
+ for poc = 0; poc <= 255; poc++ {
+ err := store_source.SyncIterator(0, store_source.CurrentStorageIndex(), uint8(poc), func(k Key, n uint64) bool {
+ chunks_in[n] = make(Key, 32)
+ copy(chunks_in[n], k)
+ bins[n] = int8(poc)
+ log.Trace(fmt.Sprintf("Iterator sc #%d '%v' (array stored: '%v')", n, k, chunks_in[n]))
+ chunkcount++
+ return true
+ })
+ if err != nil {
+ return 0, fmt.Errorf("Iterator error, import aborted: %v", err)
+ }
+ }
+
+ for j = 0; j < chunkcount; j++ {
+ chunk, err := store_source.Get(chunks_in[j])
+ if err != nil {
+ log.Trace(fmt.Sprintf("Chunk get sc %d bin %d key '%v' FAIL: %v", j, bins[j], chunks_in[j], err))
+ } else {
+ log.Trace(fmt.Sprintf("Chunk get sc %d bin %d key '%v' OK", j, bins[j], chunks_in[j]))
+ store_target.Put(chunk)
+ chunks_out[j] = make(Key, 32)
+ copy(chunks_out[j], chunk.Key)
+
+ }
+ }
+ return chunkcount, nil
+}
+
+func databaseExists(path string) bool {
+ o := &opt.Options{
+ ErrorIfMissing: true,
+ }
+ tdb, err := leveldb.OpenFile(path, o)
+ if err != nil {
+ return false
+ }
+ defer tdb.Close()
+ return true
+}
diff --git a/swarm/storage/dbstore_test.go b/swarm/storage/dbstore_test.go
index ddce7ccfea..3678391ca8 100644
--- a/swarm/storage/dbstore_test.go
+++ b/swarm/storage/dbstore_test.go
@@ -18,10 +18,14 @@ package storage
import (
"bytes"
+ "crypto/rand"
+ "fmt"
+ "io"
"io/ioutil"
"testing"
- "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/crypto/sha3"
+ "github.com/ethereum/go-ethereum/log"
)
func initDbStore(t *testing.T) *DbStore {
@@ -29,37 +33,71 @@ func initDbStore(t *testing.T) *DbStore {
if err != nil {
t.Fatal(err)
}
- m, err := NewDbStore(dir, MakeHashFunc(defaultHash), defaultDbCapacity, defaultRadius)
+ basekey := sha3.NewKeccak256().Sum([]byte("random"))
+ m, err := NewDbStore(dir, MakeHashFunc(defaultHash), defaultDbCapacity, func(k Key) (ret uint8) { return uint8(proximity(basekey[:], k[:])) })
if err != nil {
t.Fatal("can't create store:", err)
}
return m
}
-func testDbStore(l int64, branches int64, t *testing.T) {
+func testDbStore(indata io.Reader, l int64, branches int64, t *testing.T) {
+ t.Skip()
+ if indata == nil {
+ indata = rand.Reader
+ }
m := initDbStore(t)
defer m.Close()
- testStore(m, l, branches, t)
+ testStore(m, indata, l, branches, t)
}
func TestDbStore128_0x1000000(t *testing.T) {
- testDbStore(0x1000000, 128, t)
+ testDbStore(nil, 0x1000000, 128, t)
}
func TestDbStore128_10000_(t *testing.T) {
- testDbStore(10000, 128, t)
+ testDbStore(nil, 10000, 128, t)
}
func TestDbStore128_1000_(t *testing.T) {
- testDbStore(1000, 128, t)
+ testDbStore(nil, 1000, 128, t)
}
func TestDbStore128_100_(t *testing.T) {
- testDbStore(100, 128, t)
+ testDbStore(nil, 100, 128, t)
}
func TestDbStore2_100_(t *testing.T) {
- testDbStore(100, 2, t)
+ testDbStore(nil, 100, 2, t)
+}
+
+func TestDbStore128_1000000_fixed_(t *testing.T) {
+ b := []byte{}
+ br := getFixedData(b, 1000000, 254)
+ testDbStore(br, 1000000, 2, t)
+}
+
+func TestDbStore2_100_fixed_(t *testing.T) {
+ b := []byte{}
+ br := getFixedData(b, 100, 0)
+ testDbStore(br, 100, 2, t)
+}
+
+func getFixedData(b []byte, l uint32, p uint8) io.Reader {
+ var i byte // it will wrap and still fit byte but not be of much use >255 cos its will only generate more of the same chunks
+ var c uint32
+ if p == 0 {
+ p = 255
+ }
+ for c = 0; c < l; c++ {
+ b = append(b, byte(i))
+ if i == p {
+ i = 0
+ } else {
+ i++
+ }
+ }
+ return bytes.NewReader(b)
}
func TestDbStoreNotFound(t *testing.T) {
@@ -71,121 +109,166 @@ func TestDbStoreNotFound(t *testing.T) {
}
}
-func TestDbStoreSyncIterator(t *testing.T) {
+// func TestDbStoreSyncIterator(t *testing.T) {
+// m := initDbStore(t)
+// defer m.Close()
+// keys := []Key{
+// Key(common.Hex2Bytes("0000000000000000000000000000000000000000000000000000000000000000")),
+// Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
+// Key(common.Hex2Bytes("5000000000000000000000000000000000000000000000000000000000000000")),
+// Key(common.Hex2Bytes("3000000000000000000000000000000000000000000000000000000000000000")),
+// Key(common.Hex2Bytes("2000000000000000000000000000000000000000000000000000000000000000")),
+// Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
+// }
+// for _, key := range keys {
+// m.Put(NewChunk(key, nil))
+// }
+// it, err := m.NewSyncIterator(DbSyncState{
+// Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
+// Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
+// First: 2,
+// Last: 4,
+// })
+// if err != nil {
+// t.Fatalf("unexpected error creating NewSyncIterator")
+// }
+
+// var chunk Key
+// var res []Key
+// for {
+// chunk = it.Next()
+// if chunk == nil {
+// break
+// }
+// res = append(res, chunk)
+// }
+// if len(res) != 1 {
+// t.Fatalf("Expected 1 chunk, got %v: %v", len(res), res)
+// }
+// if !bytes.Equal(res[0][:], keys[3]) {
+// t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
+// }
+
+// if err != nil {
+// t.Fatalf("unexpected error creating NewSyncIterator")
+// }
+
+// it, err = m.NewSyncIterator(DbSyncState{
+// Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
+// Stop: Key(common.Hex2Bytes("5000000000000000000000000000000000000000000000000000000000000000")),
+// First: 2,
+// Last: 4,
+// })
+
+// res = nil
+// for {
+// chunk = it.Next()
+// if chunk == nil {
+// break
+// }
+// res = append(res, chunk)
+// }
+// if len(res) != 2 {
+// t.Fatalf("Expected 2 chunk, got %v: %v", len(res), res)
+// }
+// if !bytes.Equal(res[0][:], keys[3]) {
+// t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
+// }
+// if !bytes.Equal(res[1][:], keys[2]) {
+// t.Fatalf("Expected %v chunk, got %v", keys[2], res[1])
+// }
+
+// if err != nil {
+// t.Fatalf("unexpected error creating NewSyncIterator")
+// }
+
+// it, _ = m.NewSyncIterator(DbSyncState{
+// Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
+// Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
+// First: 2,
+// Last: 5,
+// })
+// res = nil
+// for {
+// chunk = it.Next()
+// if chunk == nil {
+// break
+// }
+// res = append(res, chunk)
+// }
+// if len(res) != 2 {
+// t.Fatalf("Expected 2 chunk, got %v", len(res))
+// }
+// if !bytes.Equal(res[0][:], keys[4]) {
+// t.Fatalf("Expected %v chunk, got %v", keys[4], res[0])
+// }
+// if !bytes.Equal(res[1][:], keys[3]) {
+// t.Fatalf("Expected %v chunk, got %v", keys[3], res[1])
+// }
+
+// it, _ = m.NewSyncIterator(DbSyncState{
+// Start: Key(common.Hex2Bytes("2000000000000000000000000000000000000000000000000000000000000000")),
+// Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
+// First: 2,
+// Last: 5,
+// })
+// res = brokenLimitReader(data, size, errAt)
+// for {
+// chunk = it.Next()
+// if chunk == nil {
+// break
+// }
+// res = append(res, chunk)
+// }
+// if len(res) != 1 {
+// t.Fatalf("Expected 1 chunk, got %v", len(res))
+// }
+// if !bytes.Equal(res[0][:], keys[3]) {
+// t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
+// }
+// }
+
+func TestIterator(t *testing.T) {
+ var chunkcount int = 32
+ var i int
+ var poc uint
+ chunkkeys := NewKeyCollection(chunkcount)
+ chunkkeys_results := NewKeyCollection(chunkcount)
+ chunks := make([]Chunk, chunkcount)
+
m := initDbStore(t)
defer m.Close()
- keys := []Key{
- Key(common.Hex2Bytes("0000000000000000000000000000000000000000000000000000000000000000")),
- Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
- Key(common.Hex2Bytes("5000000000000000000000000000000000000000000000000000000000000000")),
- Key(common.Hex2Bytes("3000000000000000000000000000000000000000000000000000000000000000")),
- Key(common.Hex2Bytes("2000000000000000000000000000000000000000000000000000000000000000")),
- Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
- }
- for _, key := range keys {
- m.Put(NewChunk(key, nil))
- }
- it, err := m.NewSyncIterator(DbSyncState{
- Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
- Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
- First: 2,
- Last: 4,
- })
- if err != nil {
- t.Fatalf("unexpected error creating NewSyncIterator")
+
+ FakeChunk(getDefaultChunkSize(), chunkcount, chunks)
+
+ for i = 0; i < len(chunks); i++ {
+ m.Put(&chunks[i])
+ chunkkeys[i] = chunks[i].Key
}
- var chunk Key
- var res []Key
- for {
- chunk = it.Next()
- if chunk == nil {
- break
+ //testSplit(m, l, 128, chunkkeys, t)
+
+ for i = 0; i < len(chunkkeys); i++ {
+ log.Trace(fmt.Sprintf("Chunk array pos %d/%d: '%v'", i, chunkcount, chunkkeys[i]))
+ }
+
+ i = 0
+ for poc = 0; poc <= 255; poc++ {
+ err := m.SyncIterator(0, uint64(chunkkeys.Len()), uint8(poc), func(k Key, n uint64) bool {
+ log.Trace(fmt.Sprintf("Got key %v number %d poc %d", k, n, uint8(poc)))
+ chunkkeys_results[n] = k
+ i++
+ return true
+ })
+ if err != nil {
+ t.Fatalf("Iterator call failed: %v", err)
}
- res = append(res, chunk)
- }
- if len(res) != 1 {
- t.Fatalf("Expected 1 chunk, got %v: %v", len(res), res)
- }
- if !bytes.Equal(res[0][:], keys[3]) {
- t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
}
- if err != nil {
- t.Fatalf("unexpected error creating NewSyncIterator")
- }
-
- it, err = m.NewSyncIterator(DbSyncState{
- Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
- Stop: Key(common.Hex2Bytes("5000000000000000000000000000000000000000000000000000000000000000")),
- First: 2,
- Last: 4,
- })
-
- res = nil
- for {
- chunk = it.Next()
- if chunk == nil {
- break
+ for i = 0; i < chunkcount; i++ {
+ if bytes.Compare(chunkkeys[i], chunkkeys_results[i]) != 0 {
+ t.Fatalf("Chunk put #%d key '%v' does not match iterator's key '%v'", i, chunkkeys[i], chunkkeys_results[i])
}
- res = append(res, chunk)
- }
- if len(res) != 2 {
- t.Fatalf("Expected 2 chunk, got %v: %v", len(res), res)
- }
- if !bytes.Equal(res[0][:], keys[3]) {
- t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
- }
- if !bytes.Equal(res[1][:], keys[2]) {
- t.Fatalf("Expected %v chunk, got %v", keys[2], res[1])
}
- if err != nil {
- t.Fatalf("unexpected error creating NewSyncIterator")
- }
-
- it, _ = m.NewSyncIterator(DbSyncState{
- Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
- Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
- First: 2,
- Last: 5,
- })
- res = nil
- for {
- chunk = it.Next()
- if chunk == nil {
- break
- }
- res = append(res, chunk)
- }
- if len(res) != 2 {
- t.Fatalf("Expected 2 chunk, got %v", len(res))
- }
- if !bytes.Equal(res[0][:], keys[4]) {
- t.Fatalf("Expected %v chunk, got %v", keys[4], res[0])
- }
- if !bytes.Equal(res[1][:], keys[3]) {
- t.Fatalf("Expected %v chunk, got %v", keys[3], res[1])
- }
-
- it, _ = m.NewSyncIterator(DbSyncState{
- Start: Key(common.Hex2Bytes("2000000000000000000000000000000000000000000000000000000000000000")),
- Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
- First: 2,
- Last: 5,
- })
- res = nil
- for {
- chunk = it.Next()
- if chunk == nil {
- break
- }
- res = append(res, chunk)
- }
- if len(res) != 1 {
- t.Fatalf("Expected 1 chunk, got %v", len(res))
- }
- if !bytes.Equal(res[0][:], keys[3]) {
- t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
- }
}
diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go
index e16e4aacb4..a8dbc5aab2 100644
--- a/swarm/storage/dpa.go
+++ b/swarm/storage/dpa.go
@@ -64,11 +64,11 @@ type DPA struct {
}
// for testing locally
-func NewLocalDPA(datadir string) (*DPA, error) {
+func NewLocalDPA(datadir string, basekey []byte) (*DPA, error) {
hash := MakeHashFunc("SHA256")
- dbStore, err := NewDbStore(datadir, hash, singletonSwarmDbCapacity, 0)
+ dbStore, err := NewDbStore(datadir, hash, singletonSwarmDbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
if err != nil {
return nil, err
}
diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go
index 14827e3618..97446acaad 100644
--- a/swarm/storage/localstore.go
+++ b/swarm/storage/localstore.go
@@ -28,8 +28,8 @@ type LocalStore struct {
}
// This constructor uses MemStore and DbStore as components
-func NewLocalStore(hash Hasher, params *StoreParams) (*LocalStore, error) {
- dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, params.Radius)
+func NewLocalStore(hash Hasher, params *StoreParams, basekey []byte) (*LocalStore, error) {
+ dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
if err != nil {
return nil, err
}
diff --git a/swarm/storage/memstore_test.go b/swarm/storage/memstore_test.go
index 2e0ab535af..b06e68cac1 100644
--- a/swarm/storage/memstore_test.go
+++ b/swarm/storage/memstore_test.go
@@ -17,28 +17,44 @@
package storage
import (
+ "bytes"
+ "crypto/rand"
+ "io"
"testing"
)
-func testMemStore(l int64, branches int64, t *testing.T) {
+func testMemStore(indata io.Reader, l int64, branches int64, t *testing.T) {
+ if indata == nil {
+ indata = rand.Reader
+ }
m := NewMemStore(nil, defaultCacheCapacity)
- testStore(m, l, branches, t)
+ testStore(m, indata, l, branches, t)
}
func TestMemStore128_10000(t *testing.T) {
- testMemStore(10000, 128, t)
+ testMemStore(nil, 10000, 128, t)
}
func TestMemStore128_1000(t *testing.T) {
- testMemStore(1000, 128, t)
+ testMemStore(nil, 1000, 128, t)
}
func TestMemStore128_100(t *testing.T) {
- testMemStore(100, 128, t)
+ testMemStore(nil, 100, 128, t)
}
func TestMemStore2_100(t *testing.T) {
- testMemStore(100, 2, t)
+ testMemStore(nil, 100, 2, t)
+}
+
+func TestMemStore2_100_fixed_(t *testing.T) {
+ b := []byte{}
+ for i := 0; i < 100; i++ {
+ b = append(b, byte(i))
+ }
+
+ br := bytes.NewReader(b)
+ testMemStore(br, 100, 2, t)
}
func TestMemStoreNotFound(t *testing.T) {
diff --git a/swarm/storage/types.go b/swarm/storage/types.go
index cc5ded931d..f8993f4b83 100644
--- a/swarm/storage/types.go
+++ b/swarm/storage/types.go
@@ -19,6 +19,8 @@ package storage
import (
"bytes"
"crypto"
+ "crypto/rand"
+ "encoding/binary"
"fmt"
"hash"
"io"
@@ -71,6 +73,24 @@ func (h Key) bits(i, j uint) uint {
return res
}
+/*
+func proximity(one, other []byte) (ret int) {
+ retbig, _ := binary.Varint(other)
+ ret = int(int8(retbig))
+ return
+}*/
+func Proximity(one, other []byte) (ret int) {
+ for i := 0; i < len(one); i++ {
+ oxo := one[i] ^ other[i]
+ for j := 0; j < 8; j++ {
+ if (uint8(oxo)>>uint8(7-j))&0x01 != 0 {
+ return i*8 + j
+ }
+ }
+ }
+ return len(one) * 8
+}
+
func IsZeroKey(key Key) bool {
return len(key) == 0 || bytes.Equal(key, ZeroKey)
}
@@ -114,6 +134,27 @@ func (key *Key) UnmarshalJSON(value []byte) error {
return nil
}
+type KeyCollection []Key
+
+func NewKeyCollection(l int) KeyCollection {
+ return make(KeyCollection, l)
+}
+
+func (c KeyCollection) Len() int {
+ return len(c)
+}
+
+func (c KeyCollection) Less(i, j int) bool {
+ if bytes.Compare(c[i], c[j]) == -1 {
+ return true
+ }
+ return false
+}
+
+func (c KeyCollection) Swap(i, j int) {
+ c[i], c[j] = c[j], c[i]
+}
+
// 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
@@ -155,6 +196,41 @@ func NewChunk(key Key, rs *RequestStatus) *Chunk {
return &Chunk{Key: key, Req: rs}
}
+func FakeChunk(size int64, count int, chunks []Chunk) int {
+ var i int
+ hasher := MakeHashFunc(defaultHash)()
+ chunksize := getDefaultChunkSize()
+ if size > chunksize {
+ size = chunksize
+ }
+
+ for i = 0; i < count; i++ {
+ /*
+ hasher.Reset()
+ data := make([]byte, size)
+ rand.Read(data)
+ binary.LittleEndian.PutUint64(data[8:], uint64(size))
+ hasher.Write(data)
+ chunks[i].SData = make([]byte, chunksize)
+ copy(chunks[i].SData, hasher.Sum(nil))
+ */
+ hasher.Reset()
+ chunks[i].SData = make([]byte, size)
+ rand.Read(chunks[i].SData)
+ binary.LittleEndian.PutUint64(chunks[i].SData[:8], uint64(size))
+ hasher.Write(chunks[i].SData)
+ chunks[i].Key = make([]byte, 32)
+ copy(chunks[i].Key, hasher.Sum(nil))
+ }
+
+ return i
+}
+
+func getDefaultChunkSize() int64 {
+ return defaultBranches * int64(MakeHashFunc(defaultHash)().Size())
+
+}
+
/*
The ChunkStore interface is implemented by :
diff --git a/swarm/storage/types_test.go b/swarm/storage/types_test.go
new file mode 100644
index 0000000000..ceb7a635c5
--- /dev/null
+++ b/swarm/storage/types_test.go
@@ -0,0 +1,17 @@
+// 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 .
+
+package storage
diff --git a/swarm/swarm.go b/swarm/swarm.go
index 8304908433..255749107f 100644
--- a/swarm/swarm.go
+++ b/swarm/swarm.go
@@ -94,7 +94,8 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
log.Debug(fmt.Sprintf("Setting up Swarm service components"))
hash := storage.MakeHashFunc(config.ChunkerParams.Hash)
- self.lstore, err = storage.NewLocalStore(hash, config.StoreParams)
+ basehash := common.HexToHash(self.config.BzzKey)
+ self.lstore, err = storage.NewLocalStore(hash, config.StoreParams, basehash[:])
if err != nil {
return
}
@@ -320,7 +321,7 @@ func NewLocalSwarm(datadir, port string) (self *Swarm, err error) {
}
config.Port = port
- dpa, err := storage.NewLocalDPA(datadir)
+ dpa, err := storage.NewLocalDPA(datadir, storage.ZeroKey)
if err != nil {
return
}