go - 如何并行化递归函数

Question

我正在尝试在 Go 中并行化一个递归问题，但我不确定最好的方法是什么。

我有一个递归函数，它是这样工作的:

func recFunc(input string) (result []string) {
    for subInput := range getSubInputs(input) {
        subOutput := recFunc(subInput)
        result = result.append(result, subOutput...)
    }
    result = result.append(result, getOutput(input)...)
}

func main() {
    output := recFunc("some_input")
    ...
}

因此函数调用自身 N 次(其中 N 在某个级别为 0)，生成自己的输出并返回列表中的所有内容。

现在我想让这个函数并行运行。但我不确定最干净的方法是什么。我的想法:

• 有一个“结果” channel ，所有函数调用都将结果发送到该 channel 。
• 在主函数中收集结果。
• 有一个 WaitGroup ，它决定何时收集所有结果。

问题:我需要等待 WaitGroup 并并行收集所有结果。我可以为此启动一个单独的 go 函数，但是我该如何退出这个单独的 go 函数？

func recFunc(input string) (result []string, outputChannel chan []string, waitGroup &sync.WaitGroup) {
    defer waitGroup.Done()
    waitGroup.Add(len(getSubInputs(input))
    for subInput := range getSubInputs(input) {
        go recFunc(subInput)
    }
    outputChannel <-getOutput(input)
}

func main() {
    outputChannel := make(chan []string)
    waitGroup := sync.WaitGroup{}

    waitGroup.Add(1)
    go recFunc("some_input", outputChannel, &waitGroup)

    result := []string{}
    go func() {
       nextResult := <- outputChannel
       result = append(result, nextResult ...)
    }
    waitGroup.Wait()
}

也许有更好的方法来做到这一点？或者我如何确保收集结果的匿名 go 函数在完成后退出？

Answer 1

tl;博士;

• 递归算法应该对昂贵的资源(网络连接、goroutines、堆栈空间等)有限制
• 应支持取消 - 以确保在不再需要某个结果时可以快速清理昂贵的操作
• 分支遍历应该支持错误报告；这允许错误在堆栈中冒泡并返回部分结果，而不会导致整个递归遍历失败。

---

对于异步结果 - 无论是否使用递归 - 建议使用 channel 。此外，对于具有许多 goroutine 的长时间运行的作业，提供一种取消方法 ( context.Context ) 以帮助清理。

由于递归会导致资源的指数级消耗，因此设置限制很重要(请参阅 bounded parallelism)。

下面是我在异步任务中经常使用的设计模式:

• 始终支持context.Context取消
• 任务所需的 worker 数量
• 返回一个chan结果和一个chan错误(将只返回一个错误或nil)

---

var (
    workers = 10
    ctx     = context.TODO() // use request context here - otherwise context.Background()
    input   = "abc"
)

resultC, errC := recJob(ctx, workers, input) // returns results & `error` channels

// asynchronous results - so read that channel first in the event of partial results ...
for r := range resultC {
    fmt.Println(r)
}

// ... then check for any errors
if err := <-errC; err != nil {
    log.Fatal(err)
}

递归:

由于递归可以快速横向扩展，因此需要一种一致的方式来用工作填充有限的工作人员列表，同时还要确保工作人员有空时，他们会迅速从其他(过度工作的)工作人员那里接手工作。

与其创建一个管理层，不如使用一个协作的 worker 对等系统:

• 每个工作人员共享一个输入 channel
• 在递归输入 (subIinputs) 之前检查是否有其他 worker 空闲
  • 如果是，委托(delegate)给那个 worker
  • 如果不是，则当前工作人员继续递归该分支

使用此算法，有限数量的工作人员很快就会因工作而饱和。任何提前完成其分支机构的工作人员 - 将很快被另一名工作人员委派一个子分支机构。最终所有 worker 都将用完子分支，此时所有 worker 都将空闲(阻塞)并且递归任务可以完成。

要实现这一目标，需要进行一些仔细的协调。允许工作人员写入到输入 channel 有助于通过委托(delegate)进行这种对等协调。 “递归深度”WaitGroup 用于跟踪所有工作人员的所有分支何时耗尽。

(为了包括上下文支持和错误链接 - 我更新了您的 getSubInputs 函数以获取 ctx 并返回可选的 error):

func recFunc(ctx context.Context, input string, in chan string, out chan<- string, rwg *sync.WaitGroup) error {

    defer rwg.Done() // decrement recursion count when a depth of recursion has completed

    subInputs, err := getSubInputs(ctx, input)
    if err != nil {
        return err
    }

    for subInput := range subInputs { 
        rwg.Add(1) // about to recurse (or delegate recursion)

        select {
        case in <- subInput:
            // delegated - to another goroutine

        case <-ctx.Done():
            // context canceled...

            // but first we need to undo the earlier `rwg.Add(1)`
            // as this work item was never delegated or handled by this worker
            rwg.Done()
            return ctx.Err()

        default:
            // noone available to delegate - so this worker will need to recurse this item themselves
            err = recFunc(ctx, subInput, in, out, rwg)
            if err != nil {
                return err
            }
        }

        select {
        case <-ctx.Done():
            // always check context when doing anything potentially blocking (in this case writing to `out`)
            // context canceled
            return ctx.Err()

        case out <- subInput:
        }
    }

    return nil
}

连接件:

recJob 创建:

• 输入和输出 channel - 由所有 worker 共享
• “递归” WaitGroup 检测所有工作人员何时空闲
  • 然后可以安全地关闭“输出” channel
• 所有 worker 的错误 channel
• 通过将初始输入写入输入 channel 来启动递归工作负载

---

func recJob(ctx context.Context, workers int, input string) (resultsC <-chan string, errC <-chan error) {

    // RW channels
    out := make(chan string)
    eC := make(chan error, 1)

    // R-only channels returned to caller
    resultsC, errC = out, eC

    // create workers + waitgroup logic
    go func() {

        var err error // error that will be returned to call via error channel

        defer func() {
            close(out)
            eC <- err
            close(eC)
        }()

        var wg sync.WaitGroup
        wg.Add(1)
        in := make(chan string) // input channel: shared by all workers (to read from and also to write to when they need to delegate)

        workerErrC := createWorkers(ctx, workers, in, out, &wg)

        // get the ball rolling, pass input job to one of the workers
        // Note: must be done *after* workers are created - otherwise deadlock
        in <- input

        errCount := 0

        // wait for all worker error codes to return
        for err2 := range workerErrC {
            if err2 != nil {
                log.Println("worker error:", err2)
                errCount++
            }
        }

        // all workers have completed
        if errCount > 0 {
            err = fmt.Errorf("PARTIAL RESULT: %d of %d workers encountered errors", errCount, workers)
            return
        }

        log.Printf("All %d workers have FINISHED\n", workers)
    }()

    return
}

最后，创建 worker :

func createWorkers(ctx context.Context, workers int, in chan string, out chan<- string, rwg *sync.WaitGroup) (errC <-chan error) {

    eC := make(chan error) // RW-version
    errC = eC              // RO-version (returned to caller)

    // track the completeness of the workers - so we know when to wrap up
    var wg sync.WaitGroup
    wg.Add(workers)

    for i := 0; i < workers; i++ {
        i := i
        go func() {
            defer wg.Done()

            var err error

            // ensure the current worker's return code gets returned
            // via the common workers' error-channel
            defer func() {
                if err != nil {
                    log.Printf("worker #%3d ERRORED: %s\n", i+1, err)
                } else {
                    log.Printf("worker #%3d FINISHED.\n", i+1)
                }
                eC <- err
            }()

            log.Printf("worker #%3d STARTED successfully\n", i+1)

            // worker scans for input
            for input := range in {

                err = recFunc(ctx, input, in, out, rwg)
                if err != nil {
                    log.Printf("worker #%3d recurseManagers ERROR: %s\n", i+1, err)
                    return
                }
            }

        }()
    }

    go func() {
        rwg.Wait() // wait for all recursion to finish
        close(in)  // safe to close input channel as all workers are blocked (i.e. no new inputs)
        wg.Wait()  // now wait for all workers to return
        close(eC)  // finally, signal to caller we're truly done by closing workers' error-channel
    }()

    return
}