// SyncQueueLen will set the maximum queue len for the internal
// channel used for SubscribeSync().
9889
func SyncQueueLen(max int) Option {
	return func(o *Options) error {
		o.SubChanLen = max
		return nil
	}
}
1321835
func (f *Facade) syncService(ctx datastore.Context, tenantID, serviceID string, setLockOnCreate, setLockOnUpdate bool) error {
	defer ctx.Metrics().Stop(ctx.Metrics().Start("Facade.syncService"))
	svc, err := f.GetService(ctx, serviceID)
	if err != nil {
		glog.Errorf("Could not get service %s to sync: %s", serviceID, err)
		return err
	}
	if err := f.zzk.UpdateService(ctx, tenantID, svc, setLockOnCreate, setLockOnUpdate); err != nil {
		glog.Errorf("Could not sync service %s to the coordinator: %s", serviceID, err)
		return err
	}
	return nil
}
1136566
func (w *Watcher) Sync() {
	done := make(chan bool)
	w.sendReq(reqSync{done})
	select {
	case <-done:
	case <-w.tomb.Dying():
	}
}
1057997
func NewLimitedQueue(max int) Queue {
	return Queue{queue: make([]interface{}, 0, max), maxLen: max, mode: modeNormal}
}
141101
func (q *Queue) CallbackWatch(matcher events.Matcher) (eventq chan events.Event, cancel func()) {
	chanSink, ch := q.sinkGen.NewChannelSink()
	lq := queue.NewLimitQueue(chanSink, q.limit)
	sink := events.Sink(lq)

	if matcher != nil {
		sink = events.NewFilter(sink, matcher)
	}

	q.broadcast.Add(sink)

	cancelFunc := func() {
		q.broadcast.Remove(sink)
		ch.Close()
		sink.Close()
	}

	externalCancelFunc := func() {
		q.mu.Lock()
		cancelFunc := q.cancelFuncs[sink]
		delete(q.cancelFuncs, sink)
		q.mu.Unlock()

		if cancelFunc != nil {
			cancelFunc()
		}
	}

	q.mu.Lock()
	q.cancelFuncs[sink] = cancelFunc
	q.mu.Unlock()

	// If the output channel shouldn't be closed and the queue is limitless,
	// there's no need for an additional goroutine.
	if !q.closeOutChan && q.limit == 0 {
		return ch.C, externalCancelFunc
	}

	outChan := make(chan events.Event)
	go func() {
		for {
			select {
			case <-ch.Done():
				// Close the output channel if the ChannelSink is Done for any
				// reason. This can happen if the cancelFunc is called
				// externally or if it has been closed by a wrapper sink, such
				// as the TimeoutSink.
				if q.closeOutChan {
					close(outChan)
				}
				externalCancelFunc()
				return
			case <-lq.Full():
				// Close the output channel and tear down the Queue if the
				// LimitQueue becomes full.
				if q.closeOutChan {
					close(outChan)
				}
				externalCancelFunc()
				return
			case event := <-ch.C:
				outChan <- event
			}
		}
	}()

	return outChan, externalCancelFunc
}
270460
func (g *GossipSyncer) setSyncType(syncType SyncerType) {
	atomic.StoreUint32(&g.syncType, uint32(syncType))
}
441181
func (p *NFPacket) SetRequeueVerdict(newQueueId uint16) {
	v := uint(NF_QUEUE)
	q := (uint(newQueueId) << 16)
	v = v | q
	p.verdictChannel <- VerdictContainer{Verdict: Verdict(v), Packet: nil}
}
414197
func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, queue, cb, nil, false)
}
1214157
func (q *TransmitLimitedQueue) Reset() {
	q.mu.Lock()
	defer q.mu.Unlock()

	q.walkReadOnlyLocked(false, func(cur *limitedBroadcast) bool {
		cur.b.Finished()
		return true
	})

	q.tq = nil
	q.tm = nil
	q.idGen = 0
}
153522
func NewSyncProtocolServerMiddleware(bizDecoder Decoder, bizEncoder Encoder, writer func(IOSession, interface{}) error) Middleware {
	return &syncServerMiddleware{
		bizDecoder:     bizDecoder,
		bizEncoder:     bizEncoder,
		writer:         writer,
		offsetQueueMap: make(map[interface{}]*OffsetQueue),
	}
}
353770
func (m *Mirror) ScheduleNextSync() {
	m.NextSync = time.Now().Add(time.Duration(m.Interval) * time.Hour)
}
786193
func (rkq *RekeyQueueStandard) Enqueue(id tlf.ID) {
	rkq.mu.Lock()
	defer rkq.mu.Unlock()
	rkq.pendings[id] = true

	select {
	case rkq.queue <- id:
	default:
		// The queue is full; drop this one for now until the next
		// request to the server for more rekeys.
		rkq.log.Debug("Rekey queue is full; dropping %s", id)
	}
}
1246258
func (sm *SyncManager) QueueHeaders(headers *wire.MsgHeaders, peer *peerpkg.Peer) {
	// No channel handling here because peers do not need to block on
	// headers messages.
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}

	sm.msgChan <- &headersMsg{headers: headers, peer: peer}
}
234998
func (q *CtxQueue) Stop() {
	atomic.StoreUint32(&q.hasStarted, 0)
	q.o = &sync.Once{}
}
744197
func (c *ConfigLocal) SetSyncBlockCacheFraction(fraction float64) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.syncBlockCacheFraction = fraction
}
131417
func (c *Client) allocSync() {
	staggered := false
	syncTicker := time.NewTicker(allocSyncIntv)
	updates := make(map[string]*structs.Allocation)
	for {
		select {
		case <-c.shutdownCh:
			syncTicker.Stop()
			return
		case alloc := <-c.allocUpdates:
			// Batch the allocation updates until the timer triggers.
			updates[alloc.ID] = alloc
		case <-syncTicker.C:
			// Fast path if there are no updates
			if len(updates) == 0 {
				continue
			}

			sync := make([]*structs.Allocation, 0, len(updates))
			for _, alloc := range updates {
				sync = append(sync, alloc)
			}

			// Send to server.
			args := structs.AllocUpdateRequest{
				Alloc:        sync,
				WriteRequest: structs.WriteRequest{Region: c.Region()},
			}

			var resp structs.GenericResponse
			if err := c.RPC("Node.UpdateAlloc", &args, &resp); err != nil {
				c.logger.Error("error updating allocations", "error", err)
				syncTicker.Stop()
				syncTicker = time.NewTicker(c.retryIntv(allocSyncRetryIntv))
				staggered = true
			} else {
				updates = make(map[string]*structs.Allocation)
				if staggered {
					syncTicker.Stop()
					syncTicker = time.NewTicker(allocSyncIntv)
					staggered = false
				}
			}
		}
	}
}
579856
func NewChannelQueue(capacity int) (Queue, error) {
	cap := int32(capacity)
	if capacity <= 0 {
		return nil, fmt.Errorf("Invalid size for queue: %d", capacity)
	}
	qChan := make(chan interface{}, cap)

	return &chanQueue{capacity: cap, qChan: qChan}, nil
}