如何并行化递归函数

博士;递归算法应该对昂贵的资源（网络连接、goroutines、堆栈空间等）有限制。应支持取消 - 以确保在不再需要结果时可以快速清理昂贵的操作分支遍历应支持错误报告;这允许错误冒泡堆栈并返回部分结果，而不会使整个递归遍历失败。对于同步结果（无论是否使用递归 - 建议使用通道。此外，对于具有许多 goroutine 的长时间运行的作业，请提供取消方法（上下文。上下文）以帮助清理。由于递归可能导致资源呈指数级消耗，因此设置限制非常重要（请参阅有界并行性）。以下是我经常用于异步任务的设计模式：始终支持采取上下文。取消的上下文任务所需的工作人员数返回 a 的结果和 a（将只返回一个错误或chanchan errornil)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 errorsif err := <-errC; err != nil {    log.Fatal(err)}递归：由于递归可以快速水平扩展，因此需要一种一致的方式来填充有限的工人列表，同时还要确保当工人被释放时，他们能够快速从其他（过度工作）工人那里接手工作。与其创建经理层，不如采用合作的同事对等系统：每个工作线程共享一个输入通道在输入上递归之前（）检查是否有任何其他工作线程处于空闲状态subIinputs如果是这样，请委派给该工作人员如果不是，则当前工作线程继续递归该分支有了这个算法，有限的工人数量很快就会被工作所淹没。任何提前完成分支的工人 - 将很快从另一个工人那里获得子分支。最终，所有工作线程都将用完子分支，此时所有工作线程都将空闲（阻止），递归任务可以完成。要实现这一目标，需要进行一些认真的协调。允许工作线程写入输入通道有助于通过委派进行这种对等协调。“递归深度”用于跟踪所有工作线程的所有分支何时耗尽。WaitGroup（为了包括上下文支持和错误链接 - 我更新了您的函数以采用a并返回可选）：getSubInputsctxerrorfunc 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创建：输入和输出通道 - 由所有工人共享“递归”检测所有工作线程何时空闲WaitGroup然后可以安全地关闭“输出”通道所有工作线程的错误通道通过将初始输入写入输入通道来启动递归工作负载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}最后，创建工作线程：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}

如何并行化递归函数

2回答