// BroadcastBlock will either propagate a block to a subset of it's peers, or
// will only announce it's availability (depending what's requested).
770629
func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
	hash := block.Hash()
	peers := pm.peers.PeersWithoutBlock(hash)

	// If propagation is requested, send to a subset of the peer
	if propagate {
		// Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
		var td *big.Int
		if parent := pm.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
			td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1))
		} else {
			log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
			return
		}
		// Send the block to a subset of our peers
		transferLen := int(math.Sqrt(float64(len(peers))))
		if transferLen < minBroadcastPeers {
			transferLen = minBroadcastPeers
		}
		if transferLen > len(peers) {
			transferLen = len(peers)
		}
		transfer := peers[:transferLen]
		for _, peer := range transfer {
			peer.AsyncSendNewBlock(block, td)
		}
		log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
		return
	}
	// Otherwise if the block is indeed in out own chain, announce it
	if pm.blockchain.HasBlock(hash, block.NumberU64()) {
		for _, peer := range peers {
			peer.AsyncSendNewBlockHash(block)
		}
		log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
	}
}
638319
func ChannelCreationBlockToGenesisBlock(block *common.Block) (*common.Block, error) {
	if block == nil {
		return nil, errors.New("nil block")
	}
	env, err := utils.ExtractEnvelope(block, 0)
	if err != nil {
		return nil, err
	}
	payload, err := utils.ExtractPayload(env)
	if err != nil {
		return nil, err
	}
	block.Data.Data = [][]byte{payload.Data}
	block.Header.DataHash = block.Data.Hash()
	block.Header.Number = 0
	block.Header.PreviousHash = nil
	metadata := &common.BlockMetadata{
		Metadata: make([][]byte, 4),
	}
	block.Metadata = metadata
	metadata.Metadata[common.BlockMetadataIndex_LAST_CONFIG] = utils.MarshalOrPanic(&common.Metadata{
		Value: utils.MarshalOrPanic(&common.LastConfig{Index: 0}),
		// This is a genesis block, peer never verify this signature because we can't bootstrap
		// trust from an earlier block, hence there are no signatures here.
	})
	return block, nil
}
749608
func (db *StakeDatabase) block(ind int64) (*dcrutil.Block, bool) {
	block, ok := db.BlockCached(ind)
	if !ok {
		var err error
		block, _, err = rpcutils.GetBlock(ind, db.NodeClient)
		if err != nil {
			log.Error(err)
			return nil, false
		}
	}
	return block, ok
}
531836
func (tsm *TargetStatsMultiplexer) Broadcast(tse *TargetStatsEntry) {
	for _, c := range tsm.listeners {
		c <- tse
	}
}
1305143
func (service *BaseService) Broadcast(
	topic string, result interface{}, callback func([]string)) {
	service.Multicast(topic, service.IDList(topic), result, callback)
}
116340
func (m *Matrix) BroadcastDiv(v float32) *Matrix {
	return m.BroadcastFunc(func(e float32, a ...interface{}) float32 {
		return e / a[0].(float32)
	}, v)
}
377794
func (client *Client) Broadcast(message string) {
	if client.ctx.Err() != nil {
		return
	}

	client.push <- ("[ " + message)
}
909780
func (d *Dataset) Broadcast(name string, shardCount int) *Dataset {
	if shardCount == 1 && len(d.Shards) == shardCount {
		return d
	}
	ret := d.Flow.NewNextDataset(shardCount)
	step := d.Flow.AddOneToAllStep(d, ret)
	step.SetInstruction(name, instruction.NewBroadcast())
	return ret
}
262518
func (p *Peer) LastBlock() int32 {
	p.statsMtx.RLock()
	lastBlock := p.lastBlock
	p.statsMtx.RUnlock()

	return lastBlock
}
159893
func (s *Server) Broadcast(hash *uint64, msg *protocol.Message) error {
	alreadySentTo := make(map[uint64]bool)
	if msg.Gossip {
		for _, to := range msg.SentTo {
			alreadySentTo[to] = true
		}
	}
	sentTo := []uint64{murmur3.Sum64([]byte(s.LocalID()))}
	var toPeers []*Conn
	for _, peer := range s.Peers {
		peerHash := murmur3.Sum64([]byte(peer.Peer.Id))
		if (hash == nil || peer.Peer.GetKeyspace().Includes(*hash)) && !alreadySentTo[peerHash] {
			sentTo = append(sentTo, peerHash)
			toPeers = append(toPeers, peer)
		}
	}
	if msg.Gossip {
		msg.SentTo = append(msg.SentTo, sentTo...)
	}
	if len(toPeers) == 0 {
		return ErrNoRecipients
	}
	for _, peer := range toPeers {
		s.Printf("Broadcasting to %s", peer.Peer.Id)
		if err := peer.Send(msg); err != nil {
			return err
		}
	}
	return nil
}
725110
func (p *peer) AsyncSendNewBlock(block *types.Block, td *big.Int) {
	select {
	case p.queuedProps <- &propEvent{block: block, td: td}:
		p.knownBlocks.Add(block.Hash())
	default:
		p.Log().Debug("Dropping block propagation", "number", block.NumberU64(), "hash", block.Hash())
	}
}
111661
func (p *BlockPuller) popBlock(seq uint64) *common.Block {
	if len(p.blockBuff) == 0 {
		return nil
	}
	block, rest := p.blockBuff[0], p.blockBuff[1:]
	p.blockBuff = rest
	// If the requested block sequence is the wrong one, discard the buffer
	// to start fetching blocks all over again.
	if seq != block.Header.Number {
		p.blockBuff = nil
		return nil
	}
	return block
}
1081804
func (s *Sender) Broadcast(keys []string, action store.Action) {
	s.Lock()
	defer s.Unlock()
	nilCounter := 0

	if len(s.receivers) > 0 {
		log.Debugf("Sender broadcast a new event, %s %v", action, keys)
	}
	for i, receiver := range s.receivers {
		if receiver == nil { // nil receiver, ignore it
			nilCounter++
			continue
		}
		select {
		case <-receiver.ctx.Done(): // receiver was canceled, set it nil
			log.Infof("Sender find a receiver watching %s was canceled, remove it", receiver.key)
			close(receiver.ch)
			s.receivers[i] = nil
			continue
		default:
		}
		// check is there any changed keys receiver is watching
		for _, key := range keys {
			if strings.HasPrefix(key, receiver.key) {
				select {
				case receiver.ch <- &Event{
					ID:     receiver.counter,
					Action: action,
					Data:   s.Get(receiver.key),
				}:
					log.Debugf("send success, increase counter")
					receiver.counter++
				default:
					log.Warnf("Sender fail to send event to the %s receiver, may closed or blocked", receiver.key)
				}
				break // in case of noticing multi times
			}
		}
	}
	// too many nil receiver, clear it
	if nilCounter > 0 && nilCounter >= len(s.receivers)/2 {
		go s.clearNilReceivers()
	}
}
304361
func (db *PublicRecord) GetBlock(hash *wire.ShaHash) (*ombjson.Block, error) {
	row := db.selectBlock.QueryRow(hash.String())
	block, err := scanBlockHead(row)
	if err != nil {
		return &ombjson.Block{}, err
	}

	bltns, err := db.getBlockBltns(block.Head.Height)
	if err != nil {
		return &ombjson.Block{}, err
	}

	endos, err := db.getBlockEndos(block.Head.Height)
	if err != nil {
		return &ombjson.Block{}, err
	}

	block.Bulletins = bltns
	block.Endorsements = endos
	return block, nil
}
170552
func (w *worker) pendingBlock() *types.Block {
	// return a snapshot to avoid contention on currentMu mutex
	w.snapshotMu.RLock()
	defer w.snapshotMu.RUnlock()
	return w.snapshotBlock
}
847817
func (hc *Check) Blocking(block bool) {
	len := 0
	if !block {
		len = 1
	}
	hc.blocking = block
	hc.update = make(chan Config, len)
}
5888
func (v *evalVisitor) pushBlock(block *ast.BlockStatement) {
	v.blocks = append(v.blocks, block)
}