txsyncLoop takes care of the initial transaction sync for each new connection. When a new peer appears, we relay all currently pending transactions. In order to minimise egress bandwidth usage, we send the transactions in small packs to one peer at a time.
()
| 70 | // transactions. In order to minimise egress bandwidth usage, we send |
| 71 | // the transactions in small packs to one peer at a time. |
| 72 | func (pm *ProtocolManager) txsyncLoop() { |
| 73 | var ( |
| 74 | pending = make(map[discover.NodeID]*txsync) |
| 75 | sending = false // whether a send is active |
| 76 | pack = new(txsync) // the pack that is being sent |
| 77 | done = make(chan error, 1) // result of the send |
| 78 | ) |
| 79 | |
| 80 | // send starts a sending a pack of transactions from the sync. |
| 81 | send := func(s *txsync) { |
| 82 | // Fill pack with transactions up to the target size. |
| 83 | size := common.StorageSize(0) |
| 84 | pack.p = s.p |
| 85 | pack.txs = pack.txs[:0] |
| 86 | for i := 0; i < len(s.txs) && size < txsyncPackSize; i++ { |
| 87 | pack.txs = append(pack.txs, s.txs[i]) |
| 88 | size += s.txs[i].Size() |
| 89 | } |
| 90 | // Remove the transactions that will be sent. |
| 91 | s.txs = s.txs[:copy(s.txs, s.txs[len(pack.txs):])] |
| 92 | if len(s.txs) == 0 { |
| 93 | delete(pending, s.p.ID()) |
| 94 | } |
| 95 | // Send the pack in the background. |
| 96 | s.p.Log().Trace("Sending batch of transactions", "count", len(pack.txs), "bytes", size) |
| 97 | sending = true |
| 98 | go func() { done <- pack.p.SendTransactions(pack.txs) }() |
| 99 | } |
| 100 | |
| 101 | // pick chooses the next pending sync. |
| 102 | pick := func() *txsync { |
| 103 | if len(pending) == 0 { |
| 104 | return nil |
| 105 | } |
| 106 | n := rand.Intn(len(pending)) + 1 |
| 107 | for _, s := range pending { |
| 108 | if n--; n == 0 { |
| 109 | return s |
| 110 | } |
| 111 | } |
| 112 | return nil |
| 113 | } |
| 114 | |
| 115 | for { |
| 116 | select { |
| 117 | case s := <-pm.txsyncCh: |
| 118 | pending[s.p.ID()] = s |
| 119 | if !sending { |
| 120 | send(s) |
| 121 | } |
| 122 | case err := <-done: |
| 123 | sending = false |
| 124 | // Stop tracking peers that cause send failures. |
| 125 | if err != nil { |
| 126 | pack.p.Log().Debug("Transaction send failed", "err", err) |
| 127 | delete(pending, pack.p.ID()) |
| 128 | } |
| 129 | // Schedule the next send. |
no test coverage detected