func (m *Manager) updateAllocs(keep map[ids.BroadcastID]bool, new map[ids.BroadcastID]*bInfo) []*bInfo {
	m.l.Lock()
	defer m.l.Unlock()
	var ret []*bInfo
	for _, info := range m.infos {
		if keep[info.bID] {
			continue
		}
		if new[info.bID] == nil {
			ret = append(ret, info)
			delete(m.infos, info.bID)
			continue
		}
	}
	for id, info := range new {
		if m.infos[id] != nil {
			ret = append(ret, m.infos[id])
		}
		m.infos[id] = info
	}
	return ret
}
updateAllocs updates the map m.info.

"keep" identifies records which should be preserved while "new" identifies
records which should be updated. Any other record will be removed.  The
return value lists those structs which were removed or replaced.
package broadcasts
import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	log "github.com/golang/glog"
	"github.com/google/fleetspeak/fleetspeak/src/common"
	"github.com/google/fleetspeak/fleetspeak/src/server/db"
	"github.com/google/fleetspeak/fleetspeak/src/server/ids"
	"github.com/google/fleetspeak/fleetspeak/src/server/internal/cache"
	"github.com/google/fleetspeak/fleetspeak/src/server/internal/ftime"
	"github.com/google/fleetspeak/fleetspeak/src/server/internal/notifications"
	fspb "github.com/google/fleetspeak/fleetspeak/src/common/proto/fleetspeak"
	spb "github.com/google/fleetspeak/fleetspeak/src/server/proto/fleetspeak_server"
)
const (
	allocFrac     = 0.2
	allocDuration = 10 * time.Minute
)
type Manager struct {
	bs           db.BroadcastStore
	infos        map[ids.BroadcastID]*bInfo
	l            sync.RWMutex    
	done         chan bool       
	working      sync.WaitGroup  
	basePollWait time.Duration
	clientCache *cache.Clients
	dispatcher  *notifications.Dispatcher
}
func MakeManager(ctx context.Context, bs db.BroadcastStore, pw time.Duration, clientCache *cache.Clients, dispatcher *notifications.Dispatcher) (*Manager, error) {
	r := &Manager{
		bs:           bs,
		infos:        make(map[ids.BroadcastID]*bInfo),
		done:         make(chan bool),
		basePollWait: pw,
		clientCache:  clientCache,
		dispatcher:   dispatcher,
	}
	if err := r.refreshInfo(ctx); err != nil {
		return nil, err
	}
	r.working.Add(1)
	go r.refreshLoop()
	return r, nil
}
type bInfo struct {
	bID      ids.BroadcastID
	b        *spb.Broadcast
	useCount sync.WaitGroup  
	aID    ids.AllocationID
	limit  uint64
	sent   uint64  
	expiry time.Time
	lock   sync.Mutex  
}
func limitedAtomicIncrement(addr *uint64, limit uint64) bool {
	for {
		c := atomic.LoadUint64(addr)
		if c >= limit {
			return false
		}
		if atomic.CompareAndSwapUint64(addr, c, c+1) {
			return true
		}
	}
}
func (m *Manager) pollWait() time.Duration {
	return m.basePollWait + time.Duration(float64(m.basePollWait)*rand.Float64())
}
type label struct {
	serviceName string
	label       string
}
func labelFromProto(l *fspb.Label) label {
	return label{serviceName: l.ServiceName, label: l.Label}
}
func (m *Manager) MakeBroadcastMessagesForClient(ctx context.Context, id common.ClientID, labels []*fspb.Label) ([]*fspb.Message, error) {
	labelSet := make(map[label]bool)
	for _, l := range labels {
		labelSet[labelFromProto(l)] = true
	}
	sent, err := m.bs.ListSentBroadcasts(ctx, id)
	if err != nil {
		return nil, err
	}
	sentSet := make(map[ids.BroadcastID]bool)
	for _, s := range sent {
		sentSet[s] = true
	}
	var is []*bInfo
	m.l.RLock()
Infos:
	for _, info := range m.infos {
		if sentSet[info.bID] {
			continue
		}
		if !info.expiry.IsZero() && info.expiry.Before(ftime.Now()) {
			continue
		}
		for _, kw := range info.b.RequiredLabels {
			if !labelSet[labelFromProto(kw)] {
				continue Infos
			}
		}
		if info.limit == db.BroadcastUnlimited {
			atomic.AddUint64(&info.sent, 1)
		} else {
			if !limitedAtomicIncrement(&info.sent, info.limit) {
				continue
			}
		}
		info.useCount.Add(1)
		is = append(is, info)
	}
	m.l.RUnlock()
	msgs := make([]*fspb.Message, 0, len(is))
	for _, i := range is {
		mid, err := common.RandomMessageID()
		if err != nil {
			log.Errorf("unable to create message id: %v", err)
			if i.limit != db.BroadcastUnlimited {
				atomic.AddUint64(&i.sent, ^uint64(0))
			}
			i.useCount.Done()
			continue
		}
		msg := &fspb.Message{
			MessageId: mid.Bytes(),
			Source:    i.b.Source,
			Destination: &fspb.Address{
				ClientId:    id.Bytes(),
				ServiceName: i.b.Source.ServiceName,
			},
			MessageType:  i.b.MessageType,
			Data:         i.b.Data,
			CreationTime: db.NowProto(),
		}
		i.lock.Lock()
		err = m.bs.SaveBroadcastMessage(ctx, msg, i.bID, id, i.aID)
		i.lock.Unlock()
		if err != nil {
			log.Errorf("SaveBroadcastMessage of instance of broadcast %v failed. Not sending. [%v]", i.bID, err)
			if i.limit != db.BroadcastUnlimited {
				atomic.AddUint64(&i.sent, ^uint64(0))
			}
			i.useCount.Done()
			continue
		}
		msgs = append(msgs, msg)
		i.useCount.Done()
	}
	return msgs, nil
}
func (m *Manager) refreshLoop() {
	defer m.working.Done()
	ctx := context.Background()
	for {
		select {
		case _, ok := <-m.done:
			if !ok {
				return
			}
		case <-time.After(m.pollWait()):
		}
		if err := m.refreshInfo(ctx); err != nil {
			log.Errorf("Error refreshing broadcast infos: %v", err)
		}
	}
}
func (m *Manager) refreshInfo(ctx context.Context) error {
	curr := m.findCurrentAllocs()
	bs, err := m.bs.ListActiveBroadcasts(ctx)
	if err != nil {
		return fmt.Errorf("unable to list active broadcasts: %v", err)
	}
	newAllocs := make(map[ids.BroadcastID]*bInfo)
	activeIds := make(map[ids.BroadcastID]bool)
	for _, b := range bs {
		id, err := ids.BytesToBroadcastID(b.Broadcast.BroadcastId)
		if err != nil {
			log.Errorf("Broadcast [%v] has bad id, skipping: %v", b.Broadcast, err)
			continue
		}
		activeIds[id] = true
		if !curr[id] {
			if b.Sent == b.Limit {
				continue
			}
			a, err := m.bs.CreateAllocation(ctx, id, allocFrac, ftime.Now().Add(allocDuration))
			if err != nil {
				log.Errorf("Unable to create alloc for broadcast %v, skipping: %v", id, err)
				continue
			}
			if a != nil {
				newAllocs[id] = &bInfo{
					bID: id,
					b:   b.Broadcast,
					aID:    a.ID,
					limit:  a.Limit,
					sent:   0,
					expiry: a.Expiry,
				}
			}
		}
	}
	for id := range curr {
		if !activeIds[id] {
			delete(curr, id)
		}
	}
	c := m.updateAllocs(curr, newAllocs)
	if len(newAllocs) > 0 {
		m.clientCache.Clear()
		go func() {
			ctx, fin := context.WithTimeout(context.Background(), m.basePollWait)
			m.dispatcher.NotifyAll(ctx)
			fin()
		}()
	}
	var errMsgs []string
	for _, a := range c {
		a.useCount.Wait()
		if err := m.bs.CleanupAllocation(ctx, a.bID, a.aID); err != nil {
			errMsgs = append(errMsgs, fmt.Sprintf("[%v,%v]:\"%v\"", a.bID, a.aID, err))
		}
	}
	if len(errMsgs) > 0 {
		return errors.New("errors cleaning up allocations - " + strings.Join(errMsgs, " "))
	}
	return nil
}
func (m *Manager) findCurrentAllocs() map[ids.BroadcastID]bool {
	r := make(map[ids.BroadcastID]bool)
	for _, info := range m.infos {
		if (info.limit == db.BroadcastUnlimited || atomic.LoadUint64(&info.sent) < info.limit) &&
			(info.expiry.IsZero() || info.expiry.After(ftime.Now().Add(2*m.basePollWait))) {
			r[info.bID] = true
		}
	}
	return r
}