Как оповестить пачку горутин о завершении работы?

Question

[Максим Федоров]

Запущено 20 горутин такой функции.

func (w *Writer) Work(ctx context.Context) {
	// start

	f, err := os.Create("./tmp/" + filename)
	if err != nil {
		panic(err)
	}

	defer f.Close()


	// Запись шапки в файл

	for {
		select {
		case <-ctx.Done():
			// Как  выполнить тут работу в конце работы горутины? сейчас не работает
		case it := <-*w.Channel:
			// Запись элементов в файл
		}
	}

	// Или может тут  выполнить  работу в конце работы горутины ? тоже не работает
}

В цикле я каждой из них (через канал для каждой, 20 горутин — значит 20 каналов) отправляю сообщения, а каждая пишет в свой файл

ctx := context.Background()
	ctx, cancel := context.WithCancel(ctx)

defer func() {
		fmt.Println("Finish items loading!") // это сообщеие выводится
		cancel()
	}()

// тут стартанули нашу пачку горутин
 
	for {
		res, err := esScroller.Do(ctx)
		
		for _, hit := range res.Hits.Hits {

    		// ... билд item

    		//  отправляем во все каналы элемент
    		for _, w := range f.Writers {
        		*w.Channel <- item
   		 }
	    }
}

Когда закончились сообщения — как завершить все и проделать в каждой из них полезную финальную работу?

Answer 1

[Евгений Мамонов]

Попробуйте вот так

package main

import (
    "context"
    "fmt"
    "sync"
)

func worker(ctx context.Context, worderID int, data chan int, wg *sync.WaitGroup) {
    defer wg.Done()

    fmt.Printf("worker %d started\n", worderID)
    for {
        fmt.Printf("worker %d enter for\n", worderID)
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d cancelled\n", worderID)
            return
        case v, ok := <-data:
            fmt.Printf("worker %d got data: %v, ok: %v\n", worderID, v, ok)
            if !ok {
                fmt.Printf("worker %d data channel was closed\n", worderID)
                return
            }
        }
    }
}

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    channels := make([]chan int, 10)

    for i := 0; i < 10; i++ {
        wg.Add(1)
        channels[i] = make(chan int)
        go worker(ctx, i, channels[i], &wg)
    }

    for i := 0; i < 10; i++ {
        channels[i] <- i
    }

    cancel()
    wg.Wait()
}

Comments

[Евгений Мамонов]

получилось?

[Максим Федоров]

нет, потому что перед циклом записи элементов у меня есть такая в defer() :(:(
попприавил код в вопросе

[Максим Федоров]

Евгений Мамонов, ну и напрямую написать — также не сработало

[Евгений Мамонов]

На всякий случай, вдруг просмотрели... return стоит?

case <-ctx.Done():
     // Как  выполнить тут работу в конце работы горутины? сейчас не работает
     // тут return стоит?

[Максим Федоров]

Евгений Мамонов, да, конечно

на SO нашел решение — писать после cancel в канал, который заполнится в case <-ctx.Done():

[Максим Федоров]

крч ошибка примитивная оказалась и не видна на фоне "многА кода", как удержать горутины до окончания main()

[Максим Федоров]

спасибо большое!

[Евгений Мамонов]

Это очень странно, так не должно быть.
Сделал работающий пример, посмотрите пожалуйста.

package main

import (
    "context"
    "fmt"
    "time"
)

func worker(ctx context.Context, worderID int, data chan int) {
    fmt.Printf("worker %d started\n", worderID)
    for {
        fmt.Printf("worker %d enter for\n", worderID)
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d cancelled\n", worderID)
            return
        case v, ok := <-data:
            fmt.Printf("worker %d got data: %v, ok: %v\n", worderID, v, ok)
            if !ok {
                fmt.Printf("worker %d data channel was closed\n", worderID)
                return
            }
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    channels := make([]chan int, 10)

    for i := 0; i < 10; i++ {
        channels[i] = make(chan int)
        go worker(ctx, i,  channels[i])
    }

    for i := 0; i < 10; i++ {
         channels[i] <- i
    }
    time.Sleep(100 * time.Microsecond)

    cancel()
    time.Sleep(100 * time.Microsecond)
}

[Евгений Мамонов]

Отлично, рад, что всё получилось :)

[Евгений Мамонов]

Есть еще вполне надёжный вариант, можно закрывать канал, из которого читают горутины.
Особенно хорошо, когда несколько горутин читают из одного канала

[Максим Федоров]

Евгений Мамонов, у меня каждая горутина со своим каналом :) но спасибо!

[Максим Федоров]

Евгений Мамонов,
как понимаю, ваш пример не зааработает без time.Sleep()

[Евгений Мамонов]

Без sleep выполнение программы завершится сразу и cancel не успеет отработать у всех горутин.

[Евгений Мамонов]

Точнее так, без первого Sleep не все воркеры успеют прочитать данные, а без второго - не все успеют получить cancel.

Вот работающий пример без Sleep'ов, который дожидается завершения всех горутин.

package main

import (
    "context"
    "fmt"
    "sync"
)

func worker(ctx context.Context, worderID int, data chan int, wg *sync.WaitGroup) {
    defer wg.Done()

    fmt.Printf("worker %d started\n", worderID)
    for {
        fmt.Printf("worker %d enter for\n", worderID)
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d cancelled\n", worderID)
            return
        case v, ok := <-data:
            fmt.Printf("worker %d got data: %v, ok: %v\n", worderID, v, ok)
            if !ok {
                fmt.Printf("worker %d data channel was closed\n", worderID)
                return
            }
        }
    }
}

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    channels := make([]chan int, 10)

    for i := 0; i < 10; i++ {
        wg.Add(1)
        channels[i] = make(chan int)
        go worker(ctx, i, channels[i], &wg)
    }

    for i := 0; i < 10; i++ {
        channels[i] <- i
    }

    cancel()
    wg.Wait()
}

[Евгений Мамонов]

Теперь порядок, получилось?

[Максим Федоров]

Евгений Мамонов, я сделал через waitGroup (1 в 1 как у вас :)
правда до того как прочитал ваши комменты :) ответ же уже давно даже дал

перенесите код в ответ, отмечу решением

[Евгений Мамонов]

Перенёс код :)

Answer 2

[Максим Федоров]

Сработал такой вариант со SO с применением еще канала, который не позволит завершить выполнение main()

func main() {
    forever := make(chan struct{})
    ctx, cancel := context.WithCancel(context.Background())

    go func(ctx context.Context) {
        for {
            select {
            case <-ctx.Done():  // if cancel() execute
                forever <- struct{}{}
                return
            default:
                fmt.Println("for loop")
            }

            time.Sleep(500 * time.Millisecond)
        }
    }(ctx)

    go func() {
        time.Sleep(3 * time.Second)
        cancel()
    }()

    <-forever
    fmt.Println("finish")
}

Src: https://stackoverflow.com/questions/6807590/how-to...

Вариант 2: Можно через WaitGroup завершить или через отдельную горутину с буферезированным каналом (на число горутин-воркеров)