// BroadcastTxs will propagate a batch of transactions to all peers which are not known to
// already have the given transaction.
518483
func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) {
	var txset = make(map[*peer]types.Transactions)

	// Broadcast transactions to a batch of peers not knowing about it
	for _, tx := range txs {
		peers := pm.peers.PeersWithoutTx(tx.Hash())
		for _, peer := range peers {
			txset[peer] = append(txset[peer], tx)
		}
		log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers))
	}
	// FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))]
	for peer, txs := range txset {
		peer.AsyncSendTransactions(txs)
	}
}
1338482
func (wsh *WebsocketHub) MaybeSendTxns(tx *exptypes.MempoolTx) (someReadyToSend bool) {
	// addTxToBuffer adds the transaction to each client's tx buffer, and
	// indicates if at least one client has a buffer at or above the send limit.
	someReadyToSend = wsh.addTxToBuffer(tx)
	if someReadyToSend {
		// Reset the "time to send" ticker since the event loop is about send.
		wsh.bufferTickerChan <- tickerSigReset
	}
	return
}
388742
func (bcR *BlockchainReactor) BroadcastStatusRequest() error {
	msgBytes := cdc.MustMarshalBinaryBare(&bcStatusRequestMessage{bcR.store.Height()})
	bcR.Switch.Broadcast(BlockchainChannel, msgBytes)
	return nil
}
111889
func PeersToTxnProcessors(peers []fab.Peer) []fab.ProposalProcessor {
	tpp := make([]fab.ProposalProcessor, len(peers))

	for i := range peers {
		tpp[i] = peers[i]
	}
	return tpp
}
489126
func (tx *VTGateTx) ExecuteBatchShards(ctx context.Context, queries []*vtgatepb.BoundShardQuery, tabletType topodatapb.TabletType, options *querypb.ExecuteOptions) ([]sqltypes.Result, error) {
	if tx.session == nil {
		return nil, fmt.Errorf("executeBatchShards: not in transaction")
	}
	session, res, err := tx.conn.impl.ExecuteBatchShards(ctx, queries, tabletType, false /* asTransaction */, tx.session, options)
	tx.session = session
	return res, err
}
739990
func (c *eventConnections) Broadcast(msg Message) {
	for conn := range c.connections {
		if err := conn.Send(msg); err != nil {
			// Bad send, assume connection is dead and delete
			delete(c.connections, conn)
		}
	}
}
1362954
func (p *peer) SendTxs(reqID, cost uint64, txs rlp.RawValue) error {
	p.Log().Debug("Sending batch of transactions", "size", len(txs))
	return sendRequest(p.rw, SendTxV2Msg, reqID, cost, txs)
}
411493
func (s *server) AnnounceNewTransactions(txns []*mempool.TxDesc) {
	// Generate and relay inventory vectors for all newly accepted
	// transactions.
	s.relayTransactions(txns)

	// Notify both websocket and getblocktemplate long poll clients of all
	// newly accepted transactions.
	if s.rpcServer != nil {
		s.rpcServer.NotifyNewTransactions(txns)
	}
}
217500
func (pgb *ChainDB) storeBlockTxnTree(msgBlock *MsgBlockPG, txTree int8,
	chainParams *chaincfg.Params, isValid, isMainchain bool,
	updateExistingRecords, updateAddressesSpendingInfo,
	updateTicketsSpendingInfo bool) storeTxnsResult {
	// For the given block and transaction tree, extract the transactions, vins,
	// and vouts.
	dbTransactions, dbTxVouts, dbTxVins := dbtypes.ExtractBlockTransactions(
		msgBlock.MsgBlock, txTree, chainParams, isValid, isMainchain)

	// Store the transactions, vins, and vouts. This sets the VoutDbIds,
	// VinDbIds, and Vouts fields of each Tx in the dbTransactions slice.
	dbAddressRows, txDbIDs, totalAddressRows, numOuts, numIns, err :=
		pgb.storeTxns(dbTransactions, dbTxVouts, dbTxVins, updateExistingRecords)
	if err != nil {
		return storeTxnsResult{err: err}
	}

	// The return value, containing counts of inserted vins/vouts/txns, and an
	// error value.
	txRes := storeTxnsResult{
		numVins:  int64(numIns),
		numVouts: int64(numOuts),
		txDbIDs:  txDbIDs,
	}

	// Flatten the address rows into a single slice, and update the utxoCache.
	var dbAddressRowsFlat []*dbtypes.AddressRow
	var wg sync.WaitGroup
	processAddressRows := func() {
		dbAddressRowsFlat = pgb.flattenAddressRows(dbAddressRows, dbTransactions,
			isMainchain && isValid)
		wg.Done()
	}
	// Do this in parallel with stake tree tree data insertion.
	wg.Add(1)
	go processAddressRows()

	// For a side chain block, set Validators to an empty slice so that there
	// will be no misses even if there are less than 5 votes. Any Validators
	// that do not match a spent ticket hash in InsertVotes are considered
	// misses. By listing no required validators, there are no misses. For side
	// chain blocks, this is acceptable and necessary because the misses table
	// does not record the block hash or main/side chain status.
	if !isMainchain {
		msgBlock.Validators = []string{}
	}

	// If processing stake tree transactions, insert tickets, votes, and misses.
	// Also update pool status and spending information in tickets table
	// pertaining to the new votes, revokes, misses, and expires.
	if txTree == wire.TxTreeStake {
		// Tickets: Insert new (unspent) tickets
		newTicketDbIDs, newTicketTx, err := InsertTickets(pgb.db, dbTransactions, txDbIDs,
			pgb.dupChecks, updateExistingRecords)
		if err != nil && err != sql.ErrNoRows {
			log.Error("InsertTickets:", err)
			txRes.err = err
			return txRes
		}

		// Cache the unspent ticket DB row IDs and and their hashes. Needed do
		// efficiently update their spend status later.
		var unspentTicketCache *TicketTxnIDGetter
		if updateTicketsSpendingInfo {
			for it, tdbid := range newTicketDbIDs {
				pgb.unspentTicketCache.Set(newTicketTx[it].TxID, tdbid)
			}
			unspentTicketCache = pgb.unspentTicketCache
		}

		// Votes: insert votes and misses (tickets that did not vote when
		// called). Return the ticket hash of all misses, which may include
		// revokes at this point. Unrevoked misses are identified when updating
		// ticket spend info below.

		// voteDbIDs, voteTxns, spentTicketHashes, ticketDbIDs, missDbIDs, err := ...
		var missesHashIDs map[string]uint64
		_, _, _, _, missesHashIDs, err = InsertVotes(pgb.db, dbTransactions, txDbIDs,
			unspentTicketCache, msgBlock, pgb.dupChecks, updateExistingRecords,
			pgb.chainParams, pgb.ChainInfo())
		if err != nil && err != sql.ErrNoRows {
			log.Error("InsertVotes:", err)
			txRes.err = err
			return txRes
		}

		if updateTicketsSpendingInfo {
			// Get information for transactions spending tickets (votes and
			// revokes), and the ticket DB row IDs themselves. Also return
			// tickets table row IDs for newly spent tickets, if we are updating
			// them as we go (SetSpendingForTickets). CollectTicketSpendDBInfo
			// uses ChainDB's ticket DB row ID cache (unspentTicketCache), and
			// immediately expires any found entries for a main chain block.
			spendingTxDbIDs, spendTypes, spentTicketHashes, ticketDbIDs, err :=
				pgb.CollectTicketSpendDBInfo(dbTransactions, txDbIDs,
					msgBlock.MsgBlock, isMainchain)
			if err != nil {
				log.Error("CollectTicketSpendDBInfo:", err)
				txRes.err = err
				return txRes
			}

			// Get a consistent view of the stake node at its present height.
			pgb.stakeDB.LockStakeNode()

			// Classify and record the height of each ticket spend (vote or
			// revoke). For revokes, further distinguish miss or expire.
			revokes := make(map[string]uint64)
			blockHeights := make([]int64, len(spentTicketHashes))
			poolStatuses := make([]dbtypes.TicketPoolStatus, len(spentTicketHashes))
			for iv := range spentTicketHashes {
				blockHeights[iv] = int64(msgBlock.Header.Height) /* voteDbTxns[iv].BlockHeight */

				// Vote or revoke
				switch spendTypes[iv] {
				case dbtypes.TicketVoted:
					poolStatuses[iv] = dbtypes.PoolStatusVoted
				case dbtypes.TicketRevoked:
					revokes[spentTicketHashes[iv]] = ticketDbIDs[iv]
					// Revoke reason
					h, err0 := chainhash.NewHashFromStr(spentTicketHashes[iv])
					if err0 != nil {
						log.Errorf("Invalid hash %v", spentTicketHashes[iv])
						continue // no info about spent ticket!
					}
					expired := pgb.stakeDB.BestNode.ExistsExpiredTicket(*h)
					if !expired {
						poolStatuses[iv] = dbtypes.PoolStatusMissed
					} else {
						poolStatuses[iv] = dbtypes.PoolStatusExpired
					}
				}
			}

			// Update tickets table with spending info.
			_, err = SetSpendingForTickets(pgb.db, ticketDbIDs, spendingTxDbIDs,
				blockHeights, spendTypes, poolStatuses)
			if err != nil {
				log.Error("SetSpendingForTickets:", err)
			}

			// Unspent not-live tickets are also either expired or missed.

			// Missed but not revoked
			var unspentMissedTicketHashes []string
			var missStatuses []dbtypes.TicketPoolStatus
			unspentMisses := make(map[string]struct{})
			// missesHashIDs refers to lottery winners that did not vote.
			for miss := range missesHashIDs {
				if _, ok := revokes[miss]; !ok {
					// unrevoked miss
					unspentMissedTicketHashes = append(unspentMissedTicketHashes, miss)
					unspentMisses[miss] = struct{}{}
					missStatuses = append(missStatuses, dbtypes.PoolStatusMissed)
				}
			}

			// Expired but not revoked
			unspentEnM := make([]string, len(unspentMissedTicketHashes))
			// Start with the unspent misses and append unspent expires to get
			// "unspent expired and missed".
			copy(unspentEnM, unspentMissedTicketHashes)
			unspentExpiresAndMisses := pgb.stakeDB.BestNode.MissedByBlock()
			for _, missHash := range unspentExpiresAndMisses {
				// MissedByBlock includes tickets that missed votes or expired
				// (and which may be revoked in this block); we just want the
				// expires, and not the revoked ones. Screen each ticket from
				// MissedByBlock for the actual unspent expires.
				if pgb.stakeDB.BestNode.ExistsExpiredTicket(missHash) {
					emHash := missHash.String()
					// Next check should not be unnecessary. Make sure not in
					// unspent misses from above, and not just revoked.
					_, justMissed := unspentMisses[emHash] // should be redundant
					_, justRevoked := revokes[emHash]      // exclude if revoked
					if !justMissed && !justRevoked {
						unspentEnM = append(unspentEnM, emHash)
						missStatuses = append(missStatuses, dbtypes.PoolStatusExpired)
					}
				}
			}

			// Release the stake node.
			pgb.stakeDB.UnlockStakeNode()

			// Locate the row IDs of the unspent expired and missed tickets. Do
			// not expire the cache entry.
			unspentEnMRowIDs := make([]uint64, len(unspentEnM))
			for iu := range unspentEnM {
				t, err0 := unspentTicketCache.TxnDbID(unspentEnM[iu], false)
				if err0 != nil {
					txRes.err = fmt.Errorf("failed to retrieve ticket %s DB ID: %v",
						unspentEnM[iu], err0)
					return txRes
				}
				unspentEnMRowIDs[iu] = t
			}

			// Update status of the unspent expired and missed tickets.
			numUnrevokedMisses, err := SetPoolStatusForTickets(pgb.db,
				unspentEnMRowIDs, missStatuses)
			if err != nil {
				log.Errorf("SetPoolStatusForTicketsByHash: %v", err)
			} else if numUnrevokedMisses > 0 {
				log.Tracef("Noted %d unrevoked newly-missed tickets.", numUnrevokedMisses)
			}
		} // updateTicketsSpendingInfo
	} // txTree == wire.TxTreeStake

	wg.Wait()

	// Begin a database transaction to insert spending address rows, and (if
	// updateAddressesSpendingInfo) update matching_tx_hash in corresponding
	// funding rows.
	dbTx, err := pgb.db.Begin()
	if err != nil {
		txRes.err = fmt.Errorf(`unable to begin database transaction: %v`, err)
		return txRes
	}

	// Insert each new AddressRow, absent MatchingTxHash (spending txn since
	// these new address rows are *funding*).
	_, err = InsertAddressRowsDbTx(dbTx, dbAddressRowsFlat, pgb.dupChecks, updateExistingRecords)
	if err != nil {
		_ = dbTx.Rollback()
		log.Error("InsertAddressRows:", err)
		txRes.err = err
		return txRes
	}
	txRes.numAddresses = int64(totalAddressRows)
	txRes.addresses = make(map[string]struct{})
	for _, ad := range dbAddressRowsFlat {
		txRes.addresses[ad.Address] = struct{}{}
	}

	for it, tx := range dbTransactions {
		// vins array for this transaction
		txVins := dbTxVins[it]
		for iv := range txVins {
			// Transaction that spends an outpoint paying to >=0 addresses
			vin := &txVins[iv]

			// Skip coinbase inputs (they are new coins and thus have no
			// previous outpoint funding them).
			if bytes.Equal(zeroHashStringBytes, []byte(vin.PrevTxHash)) {
				continue
			}

			// Insert spending txn data in addresses table, and updated spend
			// status for the previous outpoints' rows in the same table.
			vinDbID := tx.VinDbIds[iv]
			spendingTxHash := vin.TxID
			spendingTxIndex := vin.TxIndex
			validMainchain := tx.IsValidBlock && tx.IsMainchainBlock
			// Attempt to retrieve cached data for this now-spent TXO. A
			// successful get will delete the entry from the cache.
			utxoData, ok := pgb.utxoCache.Get(vin.PrevTxHash, vin.PrevTxIndex)
			if !ok {
				log.Tracef("Data for that utxo (%s:%d) wasn't cached!", vin.PrevTxHash, vin.PrevTxIndex)
			}
			numAddressRowsSet, err := insertSpendingAddressRow(dbTx,
				vin.PrevTxHash, vin.PrevTxIndex, int8(vin.PrevTxTree),
				spendingTxHash, spendingTxIndex, vinDbID, utxoData, pgb.dupChecks,
				updateExistingRecords, validMainchain, vin.TxType, updateAddressesSpendingInfo,
				tx.BlockTime)
			if err != nil {
				_ = dbTx.Rollback()
				txRes.err = fmt.Errorf(`insertSpendingAddressRow: %v + %v (rollback)`,
					err, dbTx.Rollback())
				return txRes
			}
			txRes.numAddresses += numAddressRowsSet
		}
	}

	txRes.err = dbTx.Commit()

	return txRes
}
1072351
func (s *PublicTransactionPoolAPI) SendTransaction(ctx context.Context, args SendTxArgs) (common.Hash, error) {
	// Look up the wallet containing the requested signer
	account := accounts.Account{Address: args.From}

	wallet, err := s.b.AccountManager().Find(account)
	if err != nil {
		return common.Hash{}, err
	}

	if args.Nonce == nil {
		// Hold the addresse's mutex around signing to prevent concurrent assignment of
		// the same nonce to multiple accounts.
		s.nonceLock.LockAddr(args.From)
		defer s.nonceLock.UnlockAddr(args.From)
	}

	// Set some sanity defaults and terminate on failure
	if err := args.setDefaults(ctx, s.b); err != nil {
		return common.Hash{}, err
	}
	// Assemble the transaction and sign with the wallet
	tx := args.toTransaction()

	signed, err := wallet.SignTx(account, tx, s.b.ChainConfig().ChainID)
	if err != nil {
		return common.Hash{}, err
	}
	return SubmitTransaction(ctx, s.b, signed)
}
185675
func (w *Wallet) resendUnminedTxs() {
	var txs []*wire.MsgTx
	err := walletdb.View(w.db, func(tx walletdb.ReadTx) error {
		txmgrNs := tx.ReadBucket(wtxmgrNamespaceKey)
		var err error
		txs, err = w.TxStore.UnminedTxs(txmgrNs)
		return err
	})
	if err != nil {
		log.Errorf("Unable to retrieve unconfirmed transactions to "+
			"resend: %v", err)
		return
	}

	for _, tx := range txs {
		txHash, err := w.publishTransaction(tx)
		if err != nil {
			log.Debugf("Unable to rebroadcast transaction %v: %v",
				tx.TxHash(), err)
			continue
		}

		log.Debugf("Successfully rebroadcast unconfirmed transaction %v",
			txHash)
	}
}
829371
func (m *MockBroadcastServer) Broadcast(server po.AtomicBroadcast_BroadcastServer) error {
	res, err := server.Recv()
	if err == io.EOF {
		return nil
	}

	if err != nil {
		return err
	}

	if m.BroadcastError != nil {
		return m.BroadcastError
	}

	if m.BroadcastInternalServerError {
		return server.Send(broadcastResponseError)
	}

	if m.BroadcastCustomResponse != nil {
		return server.Send(m.BroadcastCustomResponse)
	}

	err = server.Send(broadcastResponseSuccess)
	if err != nil {
		return err
	}

	return m.mockBlockDelivery(res.Payload)
}
1295781
func (l *LeaderListener) broadcast() { // TODO: we may want to pass in a type
	l.mu.Lock()
	defer l.mu.Unlock()
	for _, w := range l.watchers {
		w.c <- struct{}{}
	}
	l.watchers = make([]watcher, 0)
}
377794
func (client *Client) Broadcast(message string) {
	if client.ctx.Err() != nil {
		return
	}

	client.push <- ("[ " + message)
}
90458
func (s *grpcAPIService) Broadcast(ctx context.Context, req *BroadcastRequest) (*BroadcastResponse, error) {
	return s.api.Broadcast(ctx, req), nil
}
1201874
func (trans *Transactor) BroadcastTxAsync(ctx context.Context, txEnv *txs.Envelope) (*txs.Receipt, error) {
	return trans.CheckTxSync(ctx, txEnv)
}
323951
func (pm *ProtocolManager) syncTransactions(p *peer) {
	var txs types.Transactions
	pending, _ := pm.txpool.Pending()
	for _, batch := range pending {
		txs = append(txs, batch...)
	}
	if len(txs) == 0 {
		return
	}
	select {
	case pm.txsyncCh <- &txsync{p, txs}:
	case <-pm.quitSync:
	}
}