Рабочий пул с буферизованными заданиями и фиксированным интервалом опроса

Question

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

Производитель заданий должен работать с фиксированным интервалом тикера. Результаты должны быть сброшены, прежде чем читать достаточно новых заданий, чтобы заполнить буфер. Критически важно получить результаты sh и прочитать новые задания в пакетном режиме .

См. Пример кода ниже, запустите его на детской площадке здесь .

Можно ли переписать это без счетчика c для отслеживания заданий в полете ?

// Worker pool with buffered jobs and fixed polling interval

package main

import (
    "fmt"
    "math/rand"
    "os"
    "os/signal"
    "strings"
    "sync"
    "sync/atomic"
    "syscall"
    "time"
)

func main() {
    rand.Seed(time.Now().UnixNano())

    // buf is the size of the jobs buffer
    buf := 5

    // workers is the number of workers to start
    workers := 3

    // jobs chan for workers
    jobs := make(chan int, buf)
    // results chan for workers
    results := make(chan int, buf*2)

    // jobID is incremented for each job sent on the jobs chan
    var jobID int

    // inflight is a count of the items in the jobs chan buffer
    var inflight uint64

    // pollInterval for jobs producer
    pollInterval := 500 * time.Millisecond

    // pollDone chan to stop polling
    pollDone := make(chan bool)

    // jobMultiplier on pollInterval for random job processing times
    jobMultiplier := 5

    // done chan to exit program
    done := make(chan bool)

    // Start workers
    wg := sync.WaitGroup{}
    for n := 0; n < workers; n++ {
        wg.Add(1)
        go (func(n int) {
            defer wg.Done()
            for {
                // Receive from channel or block
                jobID, more := <-jobs
                if more {
                    // To subtract a signed positive constant value...
                    // https://golang.org/pkg/sync/atomic/#AddUint64
                    c := atomic.AddUint64(&inflight, ^uint64(0))
                    fmt.Println(
                        fmt.Sprintf("worker %v processing %v - %v jobs left",
                            n, jobID, c))
                    // Processing the job...
                    m := rand.Intn(jobMultiplier)
                    time.Sleep(time.Duration(m) * pollInterval)
                    results <- jobID
                } else {
                    fmt.Println(fmt.Sprintf("worker %v exited", n))
                    return
                }
            }
        })(n)
    }

    // Signal to exit
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    fmt.Println("ctrl+c to exit")

    go (func() {
        ticker := time.NewTicker(pollInterval)
        r := make([]string, 0)
        flushResults := func() {
            fmt.Println(
                fmt.Sprintf("===> results: %v", strings.Join(r, ",")))
            r = make([]string, 0)
        }

        for {
            select {
            case <-ticker.C:
                flushResults()

                // Fetch jobs
                c := atomic.LoadUint64(&inflight)
                d := uint64(buf) - c
                for i := 0; i < int(d); i++ {
                    jobID++
                    jobs <- jobID
                    atomic.AddUint64(&inflight, 1)
                }
                fmt.Println(fmt.Sprintf("===> send %v jobs", d))

            case jobID := <-results:
                r = append(r, fmt.Sprintf("%v", jobID))

            case <-pollDone:
                // Stop polling for new jobs
                ticker.Stop()

                // Close jobs channel to stop workers
                close(jobs)

                // Wait for workers to exit
                wg.Wait()
                close(results)

                // Flush remaining results
                for {
                    jobID, more := <-results
                    if more {
                        r = append(r, fmt.Sprintf("%v", jobID))
                    } else {
                        break
                    }
                }
                flushResults()

                // Done!
                done <- true
                return
            }
        }
    })()

    // Wait for exit signal
    <-sig

    fmt.Println("---------| EXIT |---------")
    pollDone <- true
    <-done
    fmt.Println("...done")
}

Answer 1

Вот канальная версия вашего кода, функционально эквивалентная намерению примера выше. Ключевым моментом является то, что мы не используем какие-либо значения атома c для изменения логики c кода, потому что это не обеспечивает синхронизацию между программами. Все взаимодействия между программами синхронизируются с использованием каналов sync.WaitGroup или context.Context. Вероятно, есть более эффективные способы решения этой проблемы, но это демонстрирует, что нет никакой атомарности, необходимой для координации очереди и рабочих.

Единственное значение, которое все еще остается несогласованным между процедурами, здесь - это использование len(jobs) в логе вывода. Независимо от того, имеет ли смысл его использовать или нет, зависит от вас, поскольку его значение не имеет смысла в параллельном мире, но оно безопасно , поскольку оно синхронизировано для одновременного использования и отсутствует логика c, основанная на значение.

buf := 5
workers := 3
jobs := make(chan int, buf)

// results buffer must always be larger than workers + buf to prevent deadlock
results := make(chan int, buf*2)

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

// Start workers
var wg sync.WaitGroup
for n := 0; n < workers; n++ {
    wg.Add(1)
    go func(n int) {
        defer wg.Done()
        for jobID := range jobs {
            fmt.Printf("worker %v processing %v - %v jobs left\n", n, jobID, len(jobs))
            time.Sleep(time.Duration(rand.Intn(5)) * pollInterval)
            results <- jobID
        }
        fmt.Printf("worker %v exited", n)
    }(n)
}

var done sync.WaitGroup
done.Add(1)
go func() {
    defer done.Done()
    ticker := time.NewTicker(pollInterval)
    r := make([]string, 0)

    flushResults := func() {
        fmt.Printf("===> results: %v\n", strings.Join(r, ","))
        r = r[:0]
    }

    for {
        select {
        case <-ticker.C:
            flushResults()

            // send max buf jobs, or fill the queue
            for i := 0; i < buf; i++ {
                jobID++
                select {
                case jobs <- jobID:
                    continue
                }
                break
            }
            fmt.Printf("===> send %v jobs\n", i)

        case jobID := <-results:
            r = append(r, fmt.Sprintf("%v", jobID))

        case <-ctx.Done():
            // Close jobs channel to stop workers
            close(jobs)
            // Wait for workers to exit
            wg.Wait()

            // we can close results for easy iteration because we know
            // there are no more workers.
            close(results)
            // Flush remaining results
            for jobID := range results {
                r = append(r, fmt.Sprintf("%v", jobID))
            }
            flushResults()
            return
        }
    }
}()