无法同时使用goroutines来查找最大值，直到上下文被取消。

huangapple
go评论118阅读模式
英文:

Unable to use goroutines concurrently to find max until context is cancelled

问题

我已经成功地创建了一个没有 goroutine 的同步解决方案，用于对 compute 调用的 findMax 进行查找。

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "math/rand"
  7.     "time"
  8. )
  10. func findMax(ctx context.Context, concurrency int) uint64 {
  11.     var (
  12.         max uint64 = 0
  13.         num uint64 = 0
  14.     )
  16.     for i := 0; i < concurrency; i++ {
  17.         num = compute()
  19.         if num > max {
  20.             max = num
  21.         }
  22.     }
  24.     return max
  25. }
  27. func compute() uint64 {
  28.     // 注意：这是一个模拟阻塞操作的实现。
  29.     
  30.     time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)
  31.     return rand.Uint64()
  32. }
  34. func main() {
  35.     maxDuration := 2 * time.Second
  36.     concurrency := 10
  38.     ctx, cancel := context.WithTimeout(context.Background(), maxDuration)
  39.     defer cancel()
  41.     max := findMax(ctx, concurrency)
  42.     fmt.Println(max)
  43. }

当我尝试使用 goroutine 来重复调用 compute 函数，直到调用者 main 函数取消上下文 ctx 时，每次都得到 0，而不是预期的 goroutine compute 函数调用的最大值。我尝试了不同的方法，大部分时间都会出现死锁。

  1. package main
  3. import (
  4.     "context"
  5.     "fmt"
  6.     "math/rand"
  7.     "time"
  8. )
  10. func findMax(ctx context.Context, concurrency int) uint64 {
  11.     var (
  12.         max uint64 = 0
  13.         num uint64 = 0
  14.     )
  16.     for i := 0; i < concurrency; i++ {
  17.         select {
  18.         case <- ctx.Done():
  19.             return max
  20.         default:
  21.             go func() {
  22.                 num = compute()
  23.                 if num > max {
  24.                     max = num
  25.                 }
  26.             }()
  27.         }
  28.     }
  30.     return max
  31. }
  33. func compute() uint64 {
  34.     // 注意：这是一个模拟阻塞操作的实现。
  35.     
  36.     time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)
  37.     return rand.Uint64()
  38. }
  40. func main() {
  41.     maxDuration := 2 * time.Second
  42.     concurrency := 10
  44.     ctx, cancel := context.WithTimeout(context.Background(), maxDuration)
  45.     defer cancel()
  47.     max := findMax(ctx, concurrency)
  48.     fmt.Println(max)
  49. }

链接：https://play.golang.org/p/3fFFq2xlXAE

英文:

I have successfully made a synchronous solution without goroutines to findMax of compute calls.

  1. package main
  3. import (
  4.     &quot;context&quot;
  5.     &quot;fmt&quot;
  6.     &quot;math/rand&quot;
  7.     &quot;time&quot;
  8. )
  10. func findMax(ctx context.Context, concurrency int) uint64 {
  11.     var (
  12.         max uint64 = 0
  13.         num uint64 = 0
  14.     )
  16.     for i := 0; i &lt; concurrency; i++ {
  17.         num = compute()
  19.         if num &gt; max {
  20.             max = num
  21.         }
  22.     }
  24.     return max
  25. }
  27. func compute() uint64 {
  28.     // NOTE: This is a MOCK implementation of the blocking operation.
  29.     
  30.     time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)
  31.     return rand.Uint64()
  32. }
  34. func main() {
  35.     maxDuration := 2 * time.Second
  36.     concurrency := 10
  38.     ctx, cancel := context.WithTimeout(context.Background(), maxDuration)
  39.     defer cancel()
  41.     max := findMax(ctx, concurrency)
  42.     fmt.Println(max)
  43. }

https://play.golang.org/p/lYXRNTDtNCI

When I attempt to use goroutines to use findMax to repeatedly call compute function using as many goroutines until context ctx is canceled by the caller main function. I am getting 0 every time and not the expected max of the grouting compute function calls. I have tried different ways to do it and get deadlock most of the time.

  1. package main
  3. import (
  4.     &quot;context&quot;
  5.     &quot;fmt&quot;
  6.     &quot;math/rand&quot;
  7.     &quot;time&quot;
  8. )
  10. func findMax(ctx context.Context, concurrency int) uint64 {
  11.     var (
  12.         max uint64 = 0
  13.         num uint64 = 0
  14.     )
  16.     for i := 0; i &lt; concurrency; i++ {
  17.         select {
  18.         case &lt;- ctx.Done():
  19.             return max
  20.         default:
  21.             go func() {
  22. 		        num = compute()
  23.                 if num &gt; max {
  24.            	        max = num
  25.         	    }
  26. 	        }()
  27.         }
  28.     }
  30.     return max
  31. }
  33. func compute() uint64 {
  34.     // NOTE: This is a MOCK implementation of the blocking operation.
  35.     
  36.     time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)
  37.     return rand.Uint64()
  38. }
  40. func main() {
  41.     maxDuration := 2 * time.Second
  42.     concurrency := 10
  44.     ctx, cancel := context.WithTimeout(context.Background(), maxDuration)
  45.     defer cancel()
  47.     max := findMax(ctx, concurrency)
  48.     fmt.Println(max)
  49. }

https://play.golang.org/p/3fFFq2xlXAE

答案1

得分: 4

你的程序存在多个问题：

  1. 你正在生成多个goroutine，这些goroutine在操作共享变量max和num，由于它们没有受到保护（例如通过互斥锁），导致数据竞争。
  2. 这里的num被每个工作goroutine修改，但它应该是每个工作goroutine的局部变量，否则计算的数据可能会丢失（例如，一个工作goroutine计算并将结果存储在num中，但紧接着第二个工作goroutine计算并替换了num的值）。
  1.  num = compute // 应该是 "num := compute"
  1. 你没有等待每个goroutine完成计算，这可能导致不正确的结果，因为即使上下文没有被取消，也没有考虑到每个工作goroutine的计算结果。使用sync.WaitGroup或通道来解决这个问题。

下面是一个示例程序，解决了你代码中的大部分问题：

  1. package main
  3. import (
  4. 	"context"
  5. 	"fmt"
  6. 	"math/rand"
  7. 	"sync"
  8. 	"time"
  9. )
  11. type result struct {
  12. 	sync.RWMutex
  13. 	max uint64
  14. }
  16. func findMax(ctx context.Context, workers int) uint64 {
  17. 	var (
  18. 		res = result{}
  19. 		wg  = sync.WaitGroup{}
  20. 	)
  22. 	for i := 0; i < workers; i++ {
  23. 		select {
  24. 		case <-ctx.Done():
  25. 			// RLock读取res.max
  26. 			res.RLock()
  27. 			ret := res.max
  28. 			res.RUnlock()
  29. 			return ret
  30. 		default:
  31. 			wg.Add(1)
  32. 			go func() {
  33. 				defer wg.Done()
  34. 				num := compute()
  36. 				// Lock以确保从res.max读取和写入res.max是安全的。否则，可能会发生数据竞争。
  37. 				res.Lock()
  38. 				if num > res.max {
  39. 					res.max = num
  40. 				}
  41. 				res.Unlock()
  42. 			}()
  43. 		}
  44. 	}
  46. 	// 等待所有goroutine完成工作，即所有工作人员都完成计算并更新了最大值。
  47. 	wg.Wait()
  49. 	return res.max
  50. }
  52. func compute() uint64 {
  53. 	rnd := rand.Int63n(100)
  54. 	time.Sleep(time.Duration(rnd) * time.Millisecond)
  55. 	return rand.Uint64()
  56. }
  58. func main() {
  59. 	maxDuration := 2 * time.Second
  60. 	concurrency := 10
  62. 	ctx, cancel := context.WithTimeout(context.Background(), maxDuration)
  63. 	defer cancel()
  65. 	fmt.Println(findMax(ctx, concurrency))
  66. }

正如@Brits在评论中指出的，当上下文被取消时，请确保停止那些不再需要的工作goroutine的处理（如果可能）。

英文:

Your program has multiple problems:

  1. You are spawning multiple goroutines that are operating on shared variables i.e., max and num leading to data race as they are not protected (eg. by Mutex).
  2. Here num is modified by every worker goroutine but it should have been local to the worker otherwise the computed data could be lost (eg. one worker goroutine computed a result and stored it in num, but right after that a second worker computes and replaces the value of num).
  1.  num = compute // Should be &quot;num := compute&quot;
  1. You are not waiting for every goroutine to finish it's computation and it may result in incorrect results as every workers computation wasn't taken into account even though context wasn't cancelled. Use sync.WaitGroup or channels to fix this.

Here's a sample program that addresses most of the issues in your code:

  1. package main
  3. import (
  4. 	&quot;context&quot;
  5. 	&quot;fmt&quot;
  6. 	&quot;math/rand&quot;
  7. 	&quot;sync&quot;
  8. 	&quot;time&quot;
  9. )
  11. type result struct {
  12. 	sync.RWMutex
  13. 	max uint64
  14. }
  16. func findMax(ctx context.Context, workers int) uint64 {
  17. 	var (
  18. 		res = result{}
  19. 		wg  = sync.WaitGroup{}
  20. 	)
  22. 	for i := 0; i &lt; workers; i++ {
  23. 		select {
  24. 		case &lt;-ctx.Done():
  25. 			// RLock to read res.max
  26. 			res.RLock()
  27. 			ret := res.max
  28. 			res.RUnlock()
  29. 			return ret
  30. 		default:
  31. 			wg.Add(1)
  32. 			go func() {
  33. 				defer wg.Done()
  34. 				num := compute()
  36. 				// Lock so that read from res.max and write
  37. 				// to res.max is safe. Else, data race could
  38. 				// occur.
  39. 				res.Lock()
  40. 				if num &gt; res.max {
  41. 					res.max = num
  42. 				}
  43. 				res.Unlock()
  44. 			}()
  45. 		}
  46. 	}
  48. 	// Wait for all the goroutine to finish work i.e., all
  49. 	// workers are done computing and updating the max.
  50. 	wg.Wait()
  52. 	return res.max
  53. }
  55. func compute() uint64 {
  56. 	rnd := rand.Int63n(100)
  57. 	time.Sleep(time.Duration(rnd) * time.Millisecond)
  58. 	return rand.Uint64()
  59. }
  61. func main() {
  62. 	maxDuration := 2 * time.Second
  63. 	concurrency := 10
  65. 	ctx, cancel := context.WithTimeout(context.Background(), maxDuration)
  66. 	defer cancel()
  68. 	fmt.Println(findMax(ctx, concurrency))
  69. }

As @Brits pointed out in the comments that when context is cancelled make sure that you stop those worker goroutines to stop processing (if possible) because it is not needed anymore.

huangapple
  • 本文由 发表于 2021年6月17日 14:40:17
  • 转载请务必保留本文链接：https://go.coder-hub.com/68014184.html
  • go
  • goroutine
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定