Function newQuerier

enterprise/tailnet/pgcoord.go:893–940

(ctx context.Context,
	logger slog.Logger,
	coordinatorID uuid.UUID,
	ps pubsub.Pubsub,
	store database.Store,
	self uuid.UUID,
	newConnections chan *connIO,
	closeConnections chan *connIO,
	numWorkers int,
	firstHeartbeat chan struct{},
	clk quartz.Clock,
)

}

func newQuerier(ctx context.Context,
	logger slog.Logger,
	coordinatorID uuid.UUID,
	ps pubsub.Pubsub,
	store database.Store,
	self uuid.UUID,
	newConnections chan *connIO,
	closeConnections chan *connIO,
	numWorkers int,
	firstHeartbeat chan struct{},
	clk quartz.Clock,
) *querier {
	updates := make(chan hbUpdate)
	q := &querier{
		ctx:              ctx,
		logger:           logger.Named("querier"),
		coordinatorID:    coordinatorID,
		pubsub:           ps,
		store:            store,
		newConnections:   newConnections,
		closeConnections: closeConnections,
		peerUpdateQ:      newWorkQ[uuid.UUID](ctx),
		mappingQ:         newWorkQ[mKey](ctx),
		heartbeats:       newHeartbeats(ctx, logger, ps, store, self, updates, firstHeartbeat, clk),
		mappers:          make(map[mKey]*mapper),
		updates:          updates,
		healthy:          true, // assume we start healthy
	}
	q.subscribe()

	// For an odd number of workers we allocate more to the mapping workers since they're busier.
	mappingWorkers := int(math.Ceil(float64(numWorkers) / 2))
	peerWorkers := numWorkers - mappingWorkers

	q.wg.Add(2 + mappingWorkers + peerWorkers)
	go func() {
		<-firstHeartbeat
		go q.handleIncoming()
		go q.handleUpdates()
		for range mappingWorkers {
			go q.mappingWorker()
		}
		for range peerWorkers {
			go q.peerUpdateWorker()
		}
	}()
	return q
}

func (q *querier) wait() {
	q.wg.Wait()