func (s *shards) start(n int) {
	s.mtx.Lock()
	defer s.mtx.Unlock()

	newQueues := make([]chan prompb.TimeSeries, n)
	for i := 0; i < n; i++ {
		newQueues[i] = make(chan prompb.TimeSeries, s.qm.cfg.Capacity)
	}

	s.queues = newQueues

	var hardShutdownCtx context.Context
	hardShutdownCtx, s.hardShutdown = context.WithCancel(context.Background())
	s.softShutdown = make(chan struct{})
	s.running = int32(n)
	s.done = make(chan struct{})
	for i := 0; i < n; i++ {
		go s.runShard(hardShutdownCtx, i, newQueues[i])
	}
	s.qm.numShardsMetric.Set(float64(n))
}
func (s *shards) start(n int) {
	s.mtx.Lock()
	defer s.mtx.Unlock()

	newQueues := make([]chan prompb.TimeSeries, n)
	for i := 0; i < n; i++ {
		newQueues[i] = make(chan prompb.TimeSeries, s.qm.cfg.Capacity)
	}

	s.queues = newQueues

	var hardShutdownCtx context.Context
	hardShutdownCtx, s.hardShutdown = context.WithCancel(context.Background())
	s.softShutdown = make(chan struct{})
	s.running = int32(n)
	s.done = make(chan struct{})
	for i := 0; i < n; i++ {
		go s.runShard(hardShutdownCtx, i, newQueues[i])
	}
	s.qm.numShardsMetric.Set(float64(n))
}
func ( s * shards ) start ( n int ) { s . mtx . Lock ( ) defer s . mtx . Unlock ( ) newQueues := make ( [ ] chan prompb . TimeSeries , n ) for i := 0 ; i < n ; i ++ { newQueues [ i ] = make ( chan prompb . TimeSeries , s . qm . cfg . Capacity ) } s . queues = newQueues var hardShutdownCtx context . Context hardShutdownCtx , s . hardShutdown = context . WithCancel ( context . Background ( ) ) s . softShutdown = make ( chan struct { } ) s . running = int32 ( n ) s . done = make ( chan struct { } ) for i := 0 ; i < n ; i ++ { go s . runShard ( hardShutdownCtx , i , newQueues [ i ] ) } s . qm . numShardsMetric . Set ( float64 ( n ) ) }
// start the shards; must be called before any call to enqueue.
start the shards ; must be called before any call to enqueue .