main.go

package main

import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// Более оптимальное решение задачи из этого видео: 
// https://youtu.be/wZCLVt_5-4c
// Из сигнатуры функции processData убран кастомный тип + убран один селект 
// (он не был прям сильно лишним, но от него можно избавиться +- безболезненно)

// условие задачи:
// реализовать функцию processParallel
// прокинуть контекст

var errTimeout = errors.New("timed out")

func processData(ctx context.Context, v int) (int, error) {
	ch := make(chan struct{})

	go func() {
		time.Sleep(time.Duration(rand.Intn(10)) * time.Second)
		close(ch)
	}()

	select {
	case <-ch:
	case <-ctx.Done():
		return 0, errTimeout
	}

	return v * 2, nil
}

func main() {
	in := make(chan int)
	out := make(chan int)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	go func() {
		defer close(in)
		for i := range 10 {
			select {
			case in <- i + 1:
			case <-ctx.Done():
				return
			}
		}
	}()

	start := time.Now()
	processParallel(ctx, in, out, 5)

	for v := range out {
		fmt.Println("v =", v)
	}
	fmt.Println("main duration:", time.Since(start))
}

func processParallel(ctx context.Context, in <-chan int, out chan<- int, numWorkers int) {
	wg := &sync.WaitGroup{}

	for range numWorkers {
		wg.Add(1)
		go worker(ctx, in, out, wg)
	}

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

func worker(ctx context.Context, in <-chan int, out chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		select {
		case v, ok := <-in:
			if !ok {
				return
			}

			val, err := processData(ctx, v)
			if errors.Is(err, errTimeout) {
				return
			}
			//...handle other error types
			select {
			case <-ctx.Done():
				return
			case out <- val:
			}
		case <-ctx.Done():
			return
		}
	}
}

здравствуйте! только изучаю голанг, у вас крутые видосы, спасибо за них! пытаюсь прикинуть можно ли сделать поменьше кода и по-возможности поменьше менять написанного от изначального в задаче

многие в комментах к видео говорили о том, что неочевидно (что надо) и непрактично менять сигнатуру processData(), потому что она может использоваться где-то еще

не будет оптимальнее применить паттерн генератор в обертке над функцией, куда прокинуть сам контекст?

func processDataWithTimeout(val int, ctx context.Context) (int, bool) {
	valChan := make(chan int)

	go func() {
		valChan <- processData(val)
	}()

	select {
	case <-ctx.Done():
		return 0, false
	case val := <-valChan:
		return val, true
	}
}

и тогда воркер, по идее, будет станет просто

func worker(in <-chan int, out chan<- int, ctx context.Context) {
	for {
		select {
		case val, ok := <-in:
			if !ok {
				return
			}

			processedVal, ok := processDataWithTimeout(val, ctx)
			if !ok {
				return
			}

			select {
			case <-ctx.Done():
				return
			default:
				out <- processedVal
			}

		case <-ctx.Done():
			return
		}
	}
}

или я что-то упустил?

нужен ли в таком варианте в воркере, если мы попали в processedDataWithTimeout вот этот кусок?

			select {
			case <-ctx.Done():
				return
			default:
				out <- processedVal
			}

ведь если контекст истек внутри processDataWithTimeout или одновременно случился и кейс чтения из in и истечения контекста, то из processDataWithTimeout вернется false и мы просто выйдем, следовательно запись в канал out дальше относительно безопасна и как будто можно сократить до

func worker(in <-chan int, out chan<- int, ctx context.Context) {
	for {
		select {
		case val, ok := <-in:
			if !ok {
				return
			}

			processedVal, ok := processDataWithTimeout(val, ctx)
			if !ok {
				return
			}

			out <- processedVal
		case <-ctx.Done():
			return
		}
	}
}

@irenkovich у автора так case out <- val, у вас через default: out <- val.
На практике это значит, что вы проваливаетесь в default быстро, а внутри можете долго делать out <- val, если читатель out занят прошлыми сообщениями. Надо как у автора через case, чтобы select выбирал то, что наступит быстрее: успешная запись в out, или отмена контекста

Кмк в processData не надо городить каналы и запускать тяжелую функцию в отдельной горутине (паттерн "генератор").
Потому, что такая стратегия ведет к утечке горутин (представьте, что происходит частый старт/стоп воркер пула, тогда успевшие стартовать тяжелые процессы забьют всю память).
Думаю, что нужно этим "2м чувакам" сказать, что можно запустить процесс в фоновой горутине, но раз уж мы пробросили контекст в processData (изменили сигнатуру функции), то пробрасываем его до конца:

func processData (ctx context.Context, v int) (int, error) {
	select {
	case <-ctx.Done():
		return 0, errors.New("cancel")
	case <-time.After(time.Duration(rand.Intn(10))*time.Second):
	}
	return v * 2, nil
}

А требуют на собесах еще Graceful shutdown воткнуть?
Т.е. после прихода сигнала ctx.Done() подождать некоторое время закрытия входящего канала in?