p2p/protocols: synchronize reporter run goroutine with db close

This commit is contained in:
Janos Guljas 2019-01-29 14:37:10 +01:00
parent b0b3b0fcb3
commit 5cc0b524e3

View file

@ -36,10 +36,13 @@ type AccountingMetrics struct {
//for a graceful cleanup
func (am *AccountingMetrics) Close() {
close(am.reporter.quit)
// save account metrics on close to avoid data loss
// when more metrics are measured between the last
// reporter run ticker tick and the close
am.reporter.save()
// wait for reporter loop to finish saving metrics
// before reporter database is closed
select {
case <-time.After(10 * time.Second):
log.Error("accounting metrics reporter timeout")
case <-am.reporter.done:
}
am.reporter.db.Close()
}
@ -50,6 +53,7 @@ type reporter struct {
interval time.Duration //duration at which the reporter will persist metrics
db *leveldb.DB //the actual DB
quit chan struct{} //quit the reporter loop
done chan struct{} //signal that reporter loop is done
}
//NewMetricsDB creates a new LevelDB instance used to persist metrics defined
@ -96,6 +100,7 @@ func NewAccountingMetrics(r metrics.Registry, d time.Duration, path string) *Acc
interval: d,
db: db,
quit: make(chan struct{}),
done: make(chan struct{}),
}
//run the go routine
@ -110,6 +115,9 @@ func NewAccountingMetrics(r metrics.Registry, d time.Duration, path string) *Acc
//run is the goroutine which periodically sends the metrics to the configured LevelDB
func (r *reporter) run() {
// signal that the reporter loop is done
defer close(r.done)
intervalTicker := time.NewTicker(r.interval)
for {
@ -125,6 +133,9 @@ func (r *reporter) run() {
}
case <-r.quit:
//graceful shutdown
if err := r.save(); err != nil {
log.Error("unable to send metrics to LevelDB", "err", err)
}
return
}
}