// broadcast is a write loop that multiplexes block propagations, announcements
// and transaction broadcasts into the remote peer. The goal is to have an async
// writer that does not lock up node internals.
558101
func (p *peer) broadcast() {
	for {
		select {
		case txs := <-p.queuedTxs:
			if err := p.SendTransactions(txs); err != nil {
				return
			}
			p.Log().Trace("Broadcast transactions", "count", len(txs))

		case prop := <-p.queuedProps:
			if err := p.SendNewBlock(prop.block, prop.td); err != nil {
				return
			}
			p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)

		case block := <-p.queuedAnns:
			if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil {
				return
			}
			p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())

		case <-p.term:
			return
		}
	}
}
1261826
func (b *Broadcaster) Write(events ...Event) error {
	select {
	case b.events <- events:
	case <-b.closed:
		return ErrSinkClosed
	}
	return nil
}
766320
func (hub *Hub) Broadcast(event string, data interface{}) {
	hub.broadcast <- &message{
		event: event,
		data:  data,
	}
}
604638
func (pm *ProtocolManager) minedBroadcastLoop() {
	// automatically stops if unsubscribe
	for obj := range pm.minedBlockSub.Chan() {
		if ev, ok := obj.Data.(core.NewMinedBlockEvent); ok {
			pm.BroadcastBlock(ev.Block, true)  // First propagate block to peers
			pm.BroadcastBlock(ev.Block, false) // Only then announce to the rest
		}
	}
}
355639
func (dev *etherDev) Broadcast(data []byte) (err error) {
    d := bytesToUchar(data)
    res := C.ether_broadcast(dev.fd, C.int(dev.iface.Index), &dev.hwaddr[0], &d[0], C.size_t(len(data)))
    if res == -1 {
        err = errors.New("failed to broadcast ethernet frame")
    }
    return
}
1197795
func (rs *Result) Broadcast() {
	rs.consolidator.mu.Lock()
	defer rs.consolidator.mu.Unlock()
	delete(rs.consolidator.queries, rs.query)
	rs.executing.Unlock()
}
393127
func (t *tribe) broadcast(mt msgType, msg interface{}, notify chan<- struct{}) error {
	raw, err := encodeMessage(mt, msg)
	if err != nil {
		return err
	}

	t.broadcasts.QueueBroadcast(&broadcast{
		msg:    raw,
		notify: notify,
	})
	return nil
}
1342241
func (c *ChannelCond) Broadcast() {
	var old chan struct{}
	c.Lock()
	defer c.Unlock()
	c.c, old = make(chan struct{}), c.c
	close(old)
}
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)
}
857877
func (worker *Worker) broadcast(outpack *outPack) {
	for _, v := range worker.agents {
		v.write(outpack)
	}
}
204072
func (s *Server) BroadcastJSON(room string, v interface{}) {
	s.roomsMu.RLock()
	for _, client := range s.rooms[room] {
		//log.Printf("sending to %s %s\n", client.id, room)
		go func(c *Client) {
			err := c.EmitJSON(v)
			if err != nil {
				log.Println(err)
			}
		}(client)
	}
	s.roomsMu.RUnlock()
}
973853
func (b *broadcast) Broadcast() {
	b.mutex.Lock()
	for _, channel := range b.clients {
		channel <- struct{}{}
	}
	b.mutex.Unlock()
}
224856
func (s *LocalSupervisor) BroadcastEvent(event Event) {
	s.Lock()
	defer s.Unlock()

	switch event.Name {
	case TeleportExitEvent:
		s.signalExit()
	case TeleportReloadEvent:
		s.signalReload()
	}

	s.events[event.Name] = event

	// Log all events other than recovered events to prevent the logs from
	// being flooded.
	if event.String() != TeleportOKEvent {
		log.WithFields(logrus.Fields{"event": event.String(), trace.Component: teleport.Component(teleport.ComponentProcess, s.id)}).Debugf("Broadcasting event.")
	}

	go func() {
		select {
		case s.eventsC <- event:
		case <-s.closeContext.Done():
			return
		}
	}()

	for _, m := range s.eventMappings {
		if m.matches(event.Name, s.events) {
			mappedEvent := Event{Name: m.Out}
			s.events[mappedEvent.Name] = mappedEvent
			go func(e Event) {
				select {
				case s.eventsC <- e:
				case <-s.closeContext.Done():
					return
				}
			}(mappedEvent)
			log.WithFields(logrus.Fields{"in": event.String(), "out": m.String(), trace.Component: teleport.Component(teleport.ComponentProcess, s.id)}).Debugf("Broadcasting mapped event.")
		}
	}
}
1305463
func (ipv4 IPv4Addr) Broadcast() IPAddr {
	// Nothing should listen on a broadcast address.
	return IPv4Addr{
		Address: IPv4Address(ipv4.BroadcastAddress()),
		Mask:    IPv4HostMask,
	}
}
312356
func (sp *serverPeer) OnBlock(_ *peer.Peer, msg *wire.MsgBlock, buf []byte) {
	// Convert the raw MsgBlock to a btcutil.Block which provides some
	// convenience methods and things such as hash caching.
	block := btcutil.NewBlockFromBlockAndBytes(msg, buf)

	// Add the block to the known inventory for the peer.
	iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
	sp.AddKnownInventory(iv)

	// Queue the block up to be handled by the block
	// manager and intentionally block further receives
	// until the bitcoin block is fully processed and known
	// good or bad.  This helps prevent a malicious peer
	// from queuing up a bunch of bad blocks before
	// disconnecting (or being disconnected) and wasting
	// memory.  Additionally, this behavior is depended on
	// by at least the block acceptance test tool as the
	// reference implementation processes blocks in the same
	// thread and therefore blocks further messages until
	// the bitcoin block has been fully processed.
	sp.server.syncManager.QueueBlock(block, sp.Peer, sp.blockProcessed)
	<-sp.blockProcessed
}
305131
func (b *Broadcaster) Start() error {
	var err error
	b.start.Do(func() {
		sub, err := b.cfg.SubscribeBlocks()
		if err != nil {
			err = fmt.Errorf("unable to subscribe for block "+
				"notifications: %v", err)
			return
		}

		b.wg.Add(1)
		go b.broadcastHandler(sub)
	})
	return err
}
915620
func (b *Broadcaster) broadcastHandler(sub *blockntfns.Subscription) {
	defer b.wg.Done()
	defer sub.Cancel()

	log.Infof("Broadcaster now active")

	reBroadcastTicker := time.NewTicker(b.cfg.RebroadcastInterval)
	defer reBroadcastTicker.Stop()

	for {
		select {
		// A new broadcast request was submitted by an external caller.
		case req := <-b.broadcastReqs:
			req.errChan <- b.handleBroadcastReq(req)

		// A new block notification has arrived, so we'll rebroadcast
		// all of our pending transactions.
		case block, ok := <-sub.Notifications:
			if !ok {
				log.Warn("Unable to rebroadcast transactions: " +
					"block subscription was canceled")
				continue
			}

			blockHeader := block.Header()

			if len(b.transactions) != 0 {
				log.Debugf("Re-broadcasting transaction at "+
					"height=%v, hash=%v", block.Height(),
					blockHeader.BlockHash())
			}

			b.rebroadcast()

		// Between blocks, we'll also try to attempt additional
		// re-broadcasts to ensure a timely confirmation.
		case <-reBroadcastTicker.C:
			if len(b.transactions) == 0 {
				continue
			}

			b.rebroadcast()

		case <-b.quit:
			return
		}
	}
}