День 1: Конкурентность и управление задачами

🎯 Цели дня

1. Понять разницу между конкурентностью и параллелизмом в Go (модель GMP).
2. Освоить паттерны Worker Pool и Fan-out/Fan-in.
3. Научиться предотвращать утечки горутин и гонки данных.
4. Подготовить ответы на типовые вопросы собеседования.

📖 Теоретический минимум (45 мин)

Планировщик Go (GMP)

• G (Goroutine) — легковесный поток с начальным стеком 2KB.
• M (Machine) — OS thread.
• P (Processor) — логический процессор (по умолчанию = числу CPU).

Work stealing: Если локальная очередь P пуста, он забирает задачи из глобальной очереди или "ворует" у другого P.

Каналы: два типа

Правило: Канал закрывает только отправитель (sender). Получатель проверяет закрытие через val, ok := <-ch.

Worker Pool с backpressure

Небуферизированный канал jobs блокирует отправителя, если все воркеры заняты. Это лучше, чем копить задачи в памяти и получить OOM.

errgroup — стандарт для fan-out

Автоматически отменяет контекст при первой ошибке. Не нужно писать ручную логику отмены.

💻 Практика (2–3 часа)

Задание 1: Worker Pool с Backpressure

Реализуй универсальный Worker Pool:

go
func WorkerPool[T any, R any](
    ctx context.Context,
    workers int,
    items []T,
    fn func(context.Context, T) (R, error),
) ([]R, error) {
    jobs := make(chan T)
    results := make(chan struct {
        val R
        err error
    })

    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                val, err := fn(ctx, job)
                results <- struct {
                    val R
                    err error
                }{val, err}
            }
        }()
    }

    go func() {
        defer close(jobs)
        for _, item := range items {
            select {
            case <-ctx.Done():
                return
            case jobs <- item:
            }
        }
    }()

    go func() {
        wg.Wait()
        close(results)
    }()

    var errs []error
    var out []R
    for r := range results {
        if r.err != nil {
            errs = append(errs, r.err)
            continue
        }
        out = append(out, r.val)
    }
    if len(errs) > 0 {
        return out, errors.Join(errs...)
    }
    return out, nil
}

Что проверить:

• Меняй workers с 1 на 10 и наблюдай за временем выполнения.
• Добавь медленный сервис (5 сек) и проверь, как контекст отменяет остальные.
• Запусти с go test -race и убедись, что гонок нет.

Задание 2: Aggregation Gateway с errgroup

Напиши сервис, который параллельно опрашивает 3 мока (User, Order, Review):

go
func (s *AggregationService) Aggregate(ctx context.Context, userID string) (*AggregatedProfile, error) {
    g, ctx := errgroup.WithContext(ctx)

    var profile UserProfile
    var orders UserOrders
    var reviews UserReviews

    g.Go(func() error {
        var err error
        profile, err = s.GetProfile(ctx, userID)
        return err
    })
    g.Go(func() error {
        var err error
        orders, err = s.GetOrders(ctx, userID)
        return err
    })
    g.Go(func() error {
        var err error
        reviews, err = s.GetReviews(ctx, userID)
        return err
    })

    if err := g.Wait(); err != nil {
        return nil, fmt.Errorf("aggregation failed: %w", err)
    }

    return &AggregatedProfile{
        Profile: profile,
        Orders:  orders,
        Reviews: reviews,
    }, nil
}

Эксперимент: Сделай один мок медленным (5 сек) и проверь, что errgroup отменяет контекст для остальных.

🏗️ Архитектурное задание для AgentForge (1 час)

Спроектируй модуль pkg/agent, который параллельно собирает данные из Jira, GitHub и внутренней документации.

Требования:

• Используй errgroup.WithContext(ctx).
• Установи общий context.WithTimeout на 3 секунды.
• Опиши, как обработаешь частичный успех (например, Jira ответила, а GitHub упал).

Вопрос для размышления: Должен ли модуль возвращать ошибку, если один из источников упал, или должен вернуть частичный результат?

🗣️ Топ-5 вопросов для самопроверки

1. Чем буферизированный канал отличается от небуферизированного и когда какой брать? Ответ: Небуферизированный даёт естественный backpressure. Буферизированный — для сглаживания пиков, но опасен утечками памяти при неправильном закрытии.

2. Как правильно закрыть канал? Ответ: Закрывает только sender. Receiver проверяет через val, ok := <-ch.

3. Что произойдет с горутиной, если она пытается писать в канал, а все получатели уже завершили работу и канал не закрыт? Ответ: Утечка горутины. Она будет висеть в памяти бесконечно.

4. Почему errgroup лучше sync.WaitGroup для параллельных вызовов? Ответ: errgroup автоматически отменяет контекст при первой ошибке и собирает ошибки. С WaitGroup нужно писать эту логику вручную.

5. Как передать контекст через границы API? Ответ: Только как первый аргумент функции. Никогда не хранить в структурах.

🧩 Проверь себя

✅ Чек-лист дня

• Могу нарисовать схему Worker Pool на листке бумаги
• Понимаю, как работает errgroup и чем он лучше sync.WaitGroup
• Могу объяснить разницу буферизированных и небуферизированных каналов
• Запустил код с go test -race и убедился, что гонок нет
• Спроектировал модуль для AgentForge