ledger/recv: reduce mutex contention and remove unnecessary closures
@@ -15,18 +15,16 @@ func recv(ledger *Ledger) func(ctx context.Context) error {
case <-ctx.Done():
return nil
case evt := <-ledger.gossipIn:
- return func() error {
err := ledger.addTransaction(evt.TX)
evt.Vote <- err
return err
- }()
case evt := <-ledger.queryIn:
r := evt.Round
f := func() (*Round, error) {
- ledger.mu.Lock()
- defer ledger.mu.Unlock()
+ ledger.mu.RLock()
+ defer ledger.mu.RUnlock()
if r.Index < ledger.round { // Respond with the round we decided beforehand.
round, available := ledger.rounds[r.Index]
@@ -62,7 +60,6 @@ func recv(ledger *Ledger) func(ctx context.Context) error {
return nil
case evt := <-ledger.outOfSyncIn:
- return func() error {
ledger.mu.RLock()
defer ledger.mu.RUnlock()
@@ -73,9 +70,7 @@ func recv(ledger *Ledger) func(ctx context.Context) error {
}
return nil
- }()
case evt := <-ledger.syncInitIn:
- return func() error {
ledger.mu.RLock()
data := SyncInitMetadata{RoundID: ledger.round}
ledger.mu.RUnlock()
@@ -98,9 +93,7 @@ func recv(ledger *Ledger) func(ctx context.Context) error {
evt.Response <- data
return nil
- }()
case evt := <-ledger.syncDiffIn:
- return func() error {
if chunk, found := diffs.load(evt.ChunkHash); found {
chunk := chunk.([]byte)
@@ -118,9 +111,7 @@ func recv(ledger *Ledger) func(ctx context.Context) error {
}
return nil
- }()
case evt := <-ledger.downloadTxIn:
- return func() error {
var txs []Transaction
for _, id := range evt.IDs {
@@ -131,7 +122,6 @@ func recv(ledger *Ledger) func(ctx context.Context) error {
evt.Response <- txs
return nil
- }()
case evt := <-ledger.forwardTxIn:
return ledger.addTransaction(evt.TX)
}
