diff --git a/p2p/protocols/reporter.go b/p2p/protocols/reporter.go index 28f64b1814..b96e2d0d77 100644 --- a/p2p/protocols/reporter.go +++ b/p2p/protocols/reporter.go @@ -29,15 +29,14 @@ import ( //AccountMetrics abstracts away the metrics DB and //the reporter to persist metrics type AccountingMetrics struct { - metricsStore *leveldb.DB - reporter *reporter + reporter *reporter } //Close will be called when the node is being shutdown //for a graceful cleanup func (am *AccountingMetrics) Close() { - am.reporter.quit <- struct{}{} - am.metricsStore.Close() + close(am.reporter.quit) + am.reporter.db.Close() } //reporter is an internal structure used to write p2p accounting related @@ -99,14 +98,13 @@ func NewAccountingMetrics(r metrics.Registry, d time.Duration, path string) *Acc go rep.run() m := &AccountingMetrics{ - metricsStore: db, - reporter: rep, + reporter: rep, } return m } -//run is the go routine which periodically sends the metrics to the configued LevelDB +//run is the goroutine which periodically sends the metrics to the configured LevelDB func (r *reporter) run() { intervalTicker := time.NewTicker(r.interval) @@ -114,7 +112,7 @@ func (r *reporter) run() { select { case <-intervalTicker.C: //at each tick send the metrics - if err := r.send(); err != nil { + if err := r.save(); err != nil { log.Error("unable to send metrics to LevelDB", "err", err) //If there is an error in writing, exit the routine; we assume here that the error is //severe and don't attempt to write again. @@ -129,21 +127,30 @@ func (r *reporter) run() { } //send the metrics to the DB -func (r *reporter) send() error { +func (r *reporter) save() error { var err error + //create a LevelDB Batch + batch := leveldb.Batch{} //for each metric in the registry (which is independent)... r.reg.Each(func(name string, i interface{}) { - switch metric := i.(type) { - //assuming every metric here to be a Counter (separate registry) - case metrics.Counter: + metric, ok := i.(metrics.Counter) + if ok { + //assuming every metric here to be a Counter (separate registry) //...create a snapshot... ms := metric.Snapshot() byteVal := make([]byte, 8) binary.BigEndian.PutUint64(byteVal, uint64(ms.Count())) //...and save the value to the DB - err = r.db.Put([]byte(name), byteVal, nil) - default: + batch.Put([]byte(name), byteVal) } }) + if batch.Len() > 0 { + err = r.db.Write(&batch, nil) + if err != nil { + return err + } + batch.Reset() + } + return err } diff --git a/swarm/swarm.go b/swarm/swarm.go index 28e88d5ca0..d19ce89d2d 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -136,7 +136,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e LightNode: config.LightNodeEnabled, } - stateStore, err := state.NewDBStore(filepath.Join(config.Path, "state-store.db")) + self.stateStore, err = state.NewDBStore(filepath.Join(config.Path, "state-store.db")) if err != nil { return } @@ -206,7 +206,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e SyncUpdateDelay: config.SyncUpdateDelay, MaxPeerServers: config.MaxStreamPeerServers, } - self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, registryOptions, self.swap) + self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, self.stateStore, registryOptions, self.swap) // Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage self.fileStore = storage.NewFileStore(self.netStore, self.config.FileStoreParams) @@ -229,7 +229,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e log.Debug("Setup local storage") - self.bzz = network.NewBzz(bzzconfig, to, stateStore, self.streamer.GetSpec(), self.streamer.Run) + self.bzz = network.NewBzz(bzzconfig, to, self.stateStore, self.streamer.GetSpec(), self.streamer.Run) // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss)