// Send new mempool txs to peer.
430045
func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
	if !memR.config.Broadcast {
		return
	}

	peerID := memR.ids.GetForPeer(peer)
	var next *clist.CElement
	for {
		// In case of both next.NextWaitChan() and peer.Quit() are variable at the same time
		if !memR.IsRunning() || !peer.IsRunning() {
			return
		}
		// This happens because the CElement we were looking at got garbage
		// collected (removed). That is, .NextWait() returned nil. Go ahead and
		// start from the beginning.
		if next == nil {
			select {
			case <-memR.Mempool.TxsWaitChan(): // Wait until a tx is available
				if next = memR.Mempool.TxsFront(); next == nil {
					continue
				}
			case <-peer.Quit():
				return
			case <-memR.Quit():
				return
			}
		}

		memTx := next.Value.(*mempoolTx)

		// make sure the peer is up to date
		peerState, ok := peer.Get(types.PeerStateKey).(PeerState)
		if !ok {
			// Peer does not have a state yet. We set it in the consensus reactor, but
			// when we add peer in Switch, the order we call reactors#AddPeer is
			// different every time due to us using a map. Sometimes other reactors
			// will be initialized before the consensus reactor. We should wait a few
			// milliseconds and retry.
			time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
			continue
		}
		if peerState.GetHeight() < memTx.Height()-1 { // Allow for a lag of 1 block
			time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
			continue
		}

		// ensure peer hasn't already sent us this tx
		if _, ok := memTx.senders.Load(peerID); !ok {
			// send memTx
			msg := &TxMessage{Tx: memTx.tx}
			success := peer.Send(MempoolChannel, cdc.MustMarshalBinaryBare(msg))
			if !success {
				time.Sleep(peerCatchupSleepIntervalMS * time.Millisecond)
				continue
			}
		}

		select {
		case <-next.NextWaitChan():
			// see the start of the for loop for nil check
			next = next.Next()
		case <-peer.Quit():
			return
		case <-memR.Quit():
			return
		}
	}
}
882036
func (mock *commMock) Send(msg *proto.SignedGossipMessage, peers ...*comm.RemotePeer) {
	for _, peer := range peers {
		logger.Debug("Sending message to peer ", peer.Endpoint, "from ", mock.id)
		mock.members[peer.Endpoint].socket <- &packetMock{
			src: mock.members[mock.id],
			dst: mock.members[peer.Endpoint],
			msg: msg,
		}
	}
}
449368
func (stream *Stream) Send(request *orderer.StepRequest) error {
	if stream.Canceled() {
		return errors.New(stream.abortReason.Load().(string))
	}
	var allowDrop bool
	// We want to drop consensus transactions if the remote node cannot keep up with us,
	// otherwise we'll slow down the entire FSM.
	if request.GetConsensusRequest() != nil {
		allowDrop = true
	}

	return stream.sendOrDrop(request, allowDrop)
}
847091
func NewMemPool(N int64) Pool {
	return &memPool{
		N: N,
		Pool: NewPool(func() Buffer {
			return New(N)
		}),
	}
}
99904
func (pe *busEp) peerSender() {
	for {
		m := <-pe.q
		if m == nil {
			return
		}
		if pe.ep.SendMsg(m) != nil {
			m.Free()
			return
		}
	}
}
920308
func (r *PEXReactor) SendAddrs(p Peer, netAddrs []*p2p.NetAddress) {
	p.Send(PexChannel, cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: netAddrs}))
}
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)
	}
}
420013
func (whisper *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error {
	p, err := whisper.getPeer(peerID)
	if err != nil {
		return err
	}
	return whisper.SendP2PDirect(p, envelope)
}
111889
func PeersToTxnProcessors(peers []fab.Peer) []fab.ProposalProcessor {
	tpp := make([]fab.ProposalProcessor, len(peers))

	for i := range peers {
		tpp[i] = peers[i]
	}
	return tpp
}
549435
func (conn *UDPConn) Send(p Packet) error {
	_, err := conn.Write(p.Bytes())
	return err
}
572637
func (c *Client) SendFromAsync(fromAccount string, toAddress btcutil.Address, amount btcutil.Amount) FutureSendFromResult {
	addr := toAddress.EncodeAddress()
	cmd := btcjson.NewSendFromCmd(fromAccount, addr, amount.ToBTC(), nil,
		nil, nil)
	return c.sendCmd(cmd)
}
1158452
func (t *MempoolDataCollector) mempoolTxns() ([]exptypes.MempoolTx, txhelpers.MempoolAddressStore, txhelpers.TxnsStore, error) {
	mempooltxs, err := t.dcrdChainSvr.GetRawMempoolVerbose(dcrjson.GRMAll)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("GetRawMempoolVerbose failed: %v", err)
	}

	blockHash, _, err := t.dcrdChainSvr.GetBestBlock()
	if err != nil {
		return nil, nil, nil, err
	}
	blockhash := blockHash.String()

	txs := make([]exptypes.MempoolTx, 0, len(mempooltxs))
	addrMap := make(txhelpers.MempoolAddressStore)
	txnsStore := make(txhelpers.TxnsStore)

	for hashStr, tx := range mempooltxs {
		hash, err := chainhash.NewHashFromStr(hashStr)
		if err != nil {
			log.Warn(err)
			continue
		}
		rawtx, err := rpcutils.GetTransactionVerboseByID(t.dcrdChainSvr, hash)
		if err != nil {
			log.Warn(err)
			continue
		}

		if rawtx == nil {
			log.Errorf("Failed to get mempool transaction %s.", hash)
			continue
		}

		msgTx, err := txhelpers.MsgTxFromHex(rawtx.Hex)
		if err != nil {
			log.Errorf("Failed to decode transaction hex: %v", err)
			continue
		}

		// Set Outpoints in the addrMap.
		txhelpers.TxOutpointsByAddr(addrMap, msgTx, t.activeChain)

		// Set PrevOuts in the addrMap, and related txns data in txnsStore.
		txhelpers.TxPrevOutsByAddr(addrMap, txnsStore, msgTx, t.dcrdChainSvr, t.activeChain)

		// Store the current mempool transaction with MemPoolTime from GRM, and
		// block info zeroed.
		txnsStore[msgTx.TxHash()] = &txhelpers.TxWithBlockData{
			Tx:          msgTx,
			MemPoolTime: tx.Time,
		}

		totalOut := 0.0
		for _, v := range rawtx.Vout {
			totalOut += v.Value
		}

		var voteInfo *exptypes.VoteInfo
		if ok := stake.IsSSGen(msgTx); ok {
			validation, version, bits, choices, err := txhelpers.SSGenVoteChoices(msgTx, t.activeChain)
			if err != nil {
				log.Debugf("Cannot get vote choices for %s", hash)
			} else {
				voteInfo = &exptypes.VoteInfo{
					Validation: exptypes.BlockValidation{
						Hash:     validation.Hash.String(),
						Height:   validation.Height,
						Validity: validation.Validity,
					},
					Version:     version,
					Bits:        bits,
					Choices:     choices,
					TicketSpent: msgTx.TxIn[1].PreviousOutPoint.Hash.String(),
				}
				voteInfo.ForLastBlock = voteInfo.VotesOnBlock(blockhash)
			}
		}

		// Note that the fee computed from msgTx will be a large negative fee
		// for coinbase transactions.
		//fee, _ := txhelpers.TxFeeRate(msgTx)
		// log.Tracef("tx fee: GRM result = %f, msgTx = %f", tx.Fee, fee.ToCoin())

		txs = append(txs, exptypes.MempoolTx{
			TxID:      msgTx.TxHash().String(),
			Fees:      tx.Fee,
			VinCount:  len(msgTx.TxIn),
			VoutCount: len(msgTx.TxOut),
			Vin:       exptypes.MsgTxMempoolInputs(msgTx),
			Coinbase:  blockchain.IsCoinBaseTx(msgTx),
			Hash:      hashStr,
			Time:      tx.Time,
			Size:      tx.Size,
			TotalOut:  totalOut,
			Type:      txhelpers.DetermineTxTypeString(msgTx),
			VoteInfo:  voteInfo,
		})
	}

	return txs, addrMap, txnsStore, nil
}
1209641
func (t *tx) transmit(a, b []byte) bool {
	// Pull completions from the tx queue and add their buffers back to the
	// pool so that we can reuse them.
	for {
		id, ok := t.q.CompletedPacket()
		if !ok {
			break
		}

		if buf := t.ids.remove(id); buf != nil {
			t.bufs.free(buf)
		}
	}

	bSize := t.bufs.entrySize
	total := uint32(len(a) + len(b))
	bufCount := (total + bSize - 1) / bSize

	// Allocate enough buffers to hold all the data.
	var buf *queue.TxBuffer
	for i := bufCount; i != 0; i-- {
		b := t.bufs.alloc()
		if b == nil {
			// Failed to get all buffers. Return to the pool
			// whatever we had managed to get.
			if buf != nil {
				t.bufs.free(buf)
			}
			return false
		}
		b.Next = buf
		buf = b
	}

	// Copy data into allocated buffers.
	nBuf := buf
	var dBuf []byte
	for _, data := range [][]byte{a, b} {
		for len(data) > 0 {
			if len(dBuf) == 0 {
				dBuf = t.data[nBuf.Offset:][:nBuf.Size]
				nBuf = nBuf.Next
			}
			n := copy(dBuf, data)
			data = data[n:]
			dBuf = dBuf[n:]
		}
	}

	// Get an id for this packet and send it out.
	id := t.ids.add(buf)
	if !t.q.Enqueue(id, total, bufCount, buf) {
		t.ids.remove(id)
		t.bufs.free(buf)
		return false
	}

	return true
}
158071
func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8, syncing bool) error {
	var msg interface{}

	spanName := "send.chunk.delivery"

	//we send different types of messages if delivery is for syncing or retrievals,
	//even if handling and content of the message are the same,
	//because swap accounting decides which messages need accounting based on the message type
	if syncing {
		msg = &ChunkDeliveryMsgSyncing{
			Addr:  chunk.Address(),
			SData: chunk.Data(),
		}
		spanName += ".syncing"
	} else {
		msg = &ChunkDeliveryMsgRetrieval{
			Addr:  chunk.Address(),
			SData: chunk.Data(),
		}
		spanName += ".retrieval"
	}

	ctx = context.WithValue(ctx, "stream_send_tag", nil)
	return p.SendPriority(ctx, msg, priority)
}
975512
func (b *Batch) Send(cb func()) {
	if b.Deflate {
		b.Writer.(*zlib.Writer).Flush()
	} else {
		b.Writer.(*bufio.Writer).Flush()
	}

	b.Message.PushPayload(b.Buf.Bytes())
	cb()
}
971453
func sendPairs(w *wallet.Wallet, amounts map[string]btcutil.Amount,
	account uint32, minconf int32, feeSatPerKb btcutil.Amount) (string, error) {

	outputs, err := makeOutputs(amounts, w.ChainParams())
	if err != nil {
		return "", err
	}
	tx, err := w.SendOutputs(outputs, account, minconf, feeSatPerKb)
	if err != nil {
		if err == txrules.ErrAmountNegative {
			return "", ErrNeedPositiveAmount
		}
		if waddrmgr.IsError(err, waddrmgr.ErrLocked) {
			return "", &ErrWalletUnlockNeeded
		}
		switch err.(type) {
		case btcjson.RPCError:
			return "", err
		}

		return "", &btcjson.RPCError{
			Code:    btcjson.ErrRPCInternal.Code,
			Message: err.Error(),
		}
	}

	txHashStr := tx.TxHash().String()
	log.Infof("Successfully sent transaction %v", txHashStr)
	return txHashStr, nil
}
1152337
func (cl *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
	if conn.PeerExtensionBytes.SupportsExtended() && cl.extensionBytes.SupportsExtended() {
		conn.Post(pp.Message{
			Type:       pp.Extended,
			ExtendedID: pp.HandshakeExtendedID,
			ExtendedPayload: func() []byte {
				msg := pp.ExtendedHandshakeMessage{
					M: map[pp.ExtensionName]pp.ExtensionNumber{
						pp.ExtensionNameMetadata: metadataExtendedId,
					},
					V:            cl.config.ExtendedHandshakeClientVersion,
					Reqq:         64, // TODO: Really?
					YourIp:       pp.CompactIp(conn.remoteAddr.IP),
					Encryption:   !cl.config.DisableEncryption,
					Port:         cl.incomingPeerPort(),
					MetadataSize: torrent.metadataSize(),
					// TODO: We can figured these out specific to the socket
					// used.
					Ipv4: pp.CompactIp(cl.config.PublicIp4.To4()),
					Ipv6: cl.config.PublicIp6.To16(),
				}
				if !cl.config.DisablePEX {
					msg.M[pp.ExtensionNamePex] = pexExtendedId
				}
				return bencode.MustMarshal(msg)
			}(),
		})
	}
	func() {
		if conn.fastEnabled() {
			if torrent.haveAllPieces() {
				conn.Post(pp.Message{Type: pp.HaveAll})
				conn.sentHaves.AddRange(0, bitmap.BitIndex(conn.t.NumPieces()))
				return
			} else if !torrent.haveAnyPieces() {
				conn.Post(pp.Message{Type: pp.HaveNone})
				conn.sentHaves.Clear()
				return
			}
		}
		conn.PostBitfield()
	}()
	if conn.PeerExtensionBytes.SupportsDHT() && cl.extensionBytes.SupportsDHT() && cl.haveDhtServer() {
		conn.Post(pp.Message{
			Type: pp.Port,
			Port: cl.dhtPort(),
		})
	}
}