// BroadcastSub substitute matrix as bitwise.
184523
func (m *Matrix) BroadcastSub(v float32) *Matrix {
	return m.BroadcastFunc(func(e float32, a ...interface{}) float32 {
		return e - a[0].(float32)
	}, v)
}
823512
func (m *Melody) BroadcastBinary(msg []byte) error {
	if m.hub.closed() {
		return errors.New("melody instance is closed")
	}

	message := &envelope{t: websocket.BinaryMessage, msg: msg}
	m.hub.broadcast <- message

	return nil
}
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
		}
	}
}
661512
func NewSubscriptions(conn *Conn) *Subscriptions {
	s := &Subscriptions{
		channels:       make(map[string]struct{}),
		pchannels:      make(map[string]struct{}),
		conn:           conn,
		messageChannel: make(chan *Message, 100),
		ready:          true,
		readyChannel:   make(chan bool, 1),
		throwError:     false,
	}
	go s.receive()
	return s
}
513019
func (c *Ring) PSubscribe(channels ...string) *PubSub {
	if len(channels) == 0 {
		panic("at least one channel is required")
	}

	shard, err := c.shards.GetByKey(channels[0])
	if err != nil {
		// TODO: return PubSub with sticky error
		panic(err)
	}
	return shard.Client.PSubscribe(channels...)
}
547464
func shiftLower(bit byte, b []byte) byte {
	bit = bit << 7
	for i := len(b) - 1; i >= 0; i-- {
		newByte := b[i] >> 1
		newByte |= bit
		bit = (b[i] & 1) << 7
		b[i] = newByte
	}
	return bit >> 7
}
215617
func getBroadcastToEmit() *Broadcast {
	// Get all broadcast messages.
	values := make([]*Broadcast, 0, 0)
	broadcasts.RLock()
	for _, v := range broadcasts.m {
		values = append(values, v)
	}
	broadcasts.RUnlock()

	// Remove all overly-emitted messages from the list
	broadcastSlice := make([]*Broadcast, 0, 0)
	broadcasts.Lock()
	for _, b := range values {
		if b.emitCounter <= broadcastRemoveValue {
			logDebug("Removing", b.Label(), "from recently updated list")
			delete(broadcasts.m, b.Label())
		} else {
			broadcastSlice = append(broadcastSlice, b)
		}
	}
	broadcasts.Unlock()

	if len(broadcastSlice) > 0 {
		// Put the newest nodes on top.
		sort.Sort(byBroadcastEmitCounter(broadcastSlice))
		return broadcastSlice[0]
	}

	return nil
}
198906
func countSubs(subs []*Subscriber, channel string) int {
	n := 0
	for _, p := range subs {
		for c := range p.channels {
			if c == channel {
				n++
				break
			}
		}
	}
	return n
}
302366
func (c *Client) PubSub(subcommand string, arguments ...interface{}) ([]string, error) {
	var ret []string

	args := make([][]byte, len(arguments)+2)
	args[0] = []byte("PUBSUB")
	args[1] = []byte(subcommand)

	for i := range arguments {
		args[2+i] = to.Bytes(arguments[i])
	}

	err := c.command(&ret, args...)

	return ret, err
}
98187
func (mat *MatMxN) Sub(dst *MatMxN, subtrahend *MatMxN) *MatMxN {
	if mat == nil || subtrahend == nil || mat.m != subtrahend.m || mat.n != subtrahend.n {
		return nil
	}

	dst = dst.Reshape(mat.m, mat.n)

	// No need to care about rows and columns
	// since it's element-wise anyway
	for i, el := range mat.dat {
		dst.dat[i] = el - subtrahend.dat[i]
	}

	return dst
}
730710
func (m *Miniredis) PubSubNumPat() int {
	m.Lock()
	defer m.Unlock()

	return countPsubs(m.allSubscribers())
}
1164047
func unaryOpCubed(rate int8, in Input, numOutputs int) Input {
	CheckRate(rate)
	return NewInput(UnaryOpUgenName, rate, UnaryOpCubed, numOutputs, in)
}
353989
func (server *RateServer) ReallySubscribeExchanges(hello *dcrrates.ExchangeSubscription, stream GRPCStream) (err error) {
	// For now, require the ExchangeBot clients to have the same base currency.
	// ToDo: Allow any index.
	if hello.BtcIndex != server.btcIndex {
		return fmt.Errorf("Exchange subscription has wrong BTC index. Given: %s, Required: %s", hello.BtcIndex, server.btcIndex)
	}
	// Save the client for use in the main loop.
	client, sid := server.addClient(stream, hello)

	// Get the address for an Infof. Seems like peerInfo is always found, but
	// am checking anyway.
	peerInfo, peerFound := grpcPeer.FromContext(client.Stream().Context())
	var clientAddr string
	if peerFound {
		clientAddr = peerInfo.Addr.String()
		log.Infof("Client has connected from %s", clientAddr)
	}

	state := server.xcBot.State()
	// Send Decred exchanges.
	err = sendStateList(client, state.DcrBtc)
	if err != nil {
		return err
	}
	// Send Bitcoin-fiat indices.
	for token := range state.FiatIndices {
		err = client.SendExchangeUpdate(&dcrrates.ExchangeRateUpdate{
			Token:   token,
			Indices: server.xcBot.Indices(token),
		})
		if err != nil {
			log.Errorf("Error encountered while sending fiat indices to client at %s: %v", clientAddr, err)
			// Assuming the Done channel will be closed on error, no further iteration
			// is necessary.
			break
		}
	}

	// Keep stream alive.
	<-client.Stream().Context().Done()

	server.deleteClient(sid)
	log.Infof("Client at %s has disconnected", clientAddr)
	return
}
1188967
func NewSpatialTransformer(samplerType SamplerType, dataType DataType, nbDims int, dimA []int) (retVal *SpatialTransformer, err error) {
	var internal C.cudnnSpatialTransformerDescriptor_t
	if err := result(C.cudnnCreateSpatialTransformerDescriptor(&internal)); err != nil {
		return nil, err
	}

	dimAC, dimACManaged := ints2CIntPtr(dimA)
	defer returnManaged(dimACManaged)
	if err := result(C.cudnnSetSpatialTransformerNdDescriptor(internal, samplerType.C(), dataType.C(), C.int(nbDims), dimAC)); err != nil {
		return nil, err
	}

	retVal = &SpatialTransformer{
		internal:    internal,
		samplerType: samplerType,
		dataType:    dataType,
		nbDims:      nbDims,
		dimA:        dimA,
	}
	runtime.SetFinalizer(retVal, destroySpatialTransformer)
	return retVal, nil
}
466051
func (b *Broadcast) Bytes() []byte {
	length := len(b.bytes)
	bytesCopy := make([]byte, length, length)
	copy(bytesCopy, b.bytes)

	return bytesCopy
}
804991
func BroadcastLobbyStart(lob *lobby.Lobby) {
	for _, slot := range lob.GetAllSlots() {
		player, _ := player.GetPlayerByID(slot.PlayerID)

		connectInfo := lobby.DecorateLobbyConnect(lob, player, slot.Slot)
		broadcaster.SendMessage(player.SteamID, "lobbyStart", connectInfo)
	}
}
632494
func (b *EventBus) SubscribeUnbuffered(ctx context.Context, subscriber string, query tmpubsub.Query) (Subscription, error) {
	return b.pubsub.SubscribeUnbuffered(ctx, subscriber, query)
}