单写多读的并发性

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

Concurrency with a SINGLE WRITE by a single writer and multiple readers

问题

注意：这个问题与其他并发性问题不同，因为不仅写入者是单个的，而且写操作仅发生一次。

在Go中，当有多个并发读取器和一个只写一次的单个写入器（例如并发环境中的setter）时，可以使用哪种同步方法？

sync.Mutex适用于这种情况，但由于只有一个写入器，sync.RWMutex比常规互斥锁稍微快一些。

然而，对于整个应用程序运行期间仅设置一次值而言，使用互斥锁进行互斥锁定似乎是一种浪费。

有更快的方法吗？

  1. package main
  3. import (
  4. 	"sync"
  5. )
  7. type RWMutexErrorNotifier struct {
  8. 	rwMutex     sync.RWMutex
  9. 	emailSender func(string)
  10. }
  12. func (a *RWMutexErrorNotifier) SetEmailSenderService(emailSender func(string)) {
  13. 	a.rwMutex.Lock()
  14. 	defer a.rwMutex.Unlock()
  16. 	a.emailSender = emailSender
  17. }
  19. func (a *RWMutexErrorNotifier) SendErrorMessage(errorMessage string) {
  20. 	a.rwMutex.RLock()
  21. 	defer a.rwMutex.RUnlock()
  23. 	if a.emailSender == nil {
  24. 		return
  25. 	}
  27. 	a.emailSender(errorMessage)
  28. }
  30. func main() {
  31. 	sender := &RWMutexErrorNotifier{}
  33. 	errorsCount := 100_000
  35. 	emailSender := func(emailMessage string) {
  36. 		// 发送电子邮件...
  37. 	}
  39. 	var wg sync.WaitGroup // 仅用于演示目的
  41. 	wg.Add(errorsCount)
  43. 	for i := 0; i < errorsCount; i++ {
  44. 		go func() {
  45. 			sender.SendErrorMessage("ALARM!")
  46. 			wg.Done()
  47. 		}()
  48. 	}
  50. 	sender.SetEmailSenderService(emailSender) // 进行一次单独的写入
  52. 	wg.Wait()
  53. }
英文:

NOTE: This question is different from other concurrency questions because not only the writer is single, but also the write operation happens strictly once.

What method of synchronization in Go to use when one has multiple concurrent readers and a single writer that writes once and only once, for example a setter in a concurrent environment?

sync.Mutex works for this case, however, since there is only one writer, sync.RWMutex is even better as it is marginally faster than a regular mutex.

Still, it feels wasteful to do the mutex locking for the duration of the application run to simply set a value once.

Is there a faster way?

  1. package main
  3. import (
  4. 	&quot;sync&quot;
  5. )
  7. type RWMutexErrorNotifier struct {
  8. 	rwMutex     sync.RWMutex
  9. 	emailSender func(string)
  10. }
  12. func (a *RWMutexErrorNotifier) SetEmailSenderService(emailSender func(string)) {
  13. 	a.rwMutex.Lock()
  14. 	defer a.rwMutex.Unlock()
  16. 	a.emailSender = emailSender
  17. }
  19. func (a *RWMutexErrorNotifier) SendErrorMessage(errorMessage string) {
  20. 	a.rwMutex.RLock()
  21. 	defer a.rwMutex.RUnlock()
  23. 	if a.emailSender == nil {
  24. 		return
  25. 	}
  27. 	a.emailSender(errorMessage)
  28. }
  30. func main() {
  31. 	sender := &amp;RWMutexErrorNotifier{}
  33. 	errorsCount := 100_000
  35. 	emailSender := func(emailMessage string) {
  36. 		// sending email...
  37. 	}
  39. 	var wg sync.WaitGroup // used only for demo purposes
  41. 	wg.Add(errorsCount)
  43. 	for i := 0; i &lt; errorsCount; i++ {
  44. 		go func() {
  45. 			sender.SendErrorMessage(&quot;ALARM!&quot;)
  46. 			wg.Done()
  47. 		}()
  48. 	}
  50. 	sender.SetEmailSenderService(emailSender) // making a single write
  52. 	wg.Wait()
  53. }

答案1

得分: 1

根据你提供的内容，这是一个关于使用不同锁机制实现错误通知器的示例代码。代码中使用了四种不同的锁机制：sync.Mutex、sync.RWMutex、atomic.Bool和无锁的解决方案（用于基准测试）。根据基准测试的结果，对于单个写入操作，使用atomic.Bool的速度比使用sync.RWMutex更快，只比无锁的解决方案稍慢一些。

代码中定义了一个ErrorNotifier接口，包含了设置邮件发送服务和发送错误消息的方法。然后分别实现了基于不同锁机制的错误通知器类型：MutexErrorNotifier、RWMutexErrorNotifier、AtomicBooleanErrorNotifier和RacyNoGoodErrorNotifier。

在代码的后面部分，定义了一个Run函数，用于模拟发送错误消息的场景。根据不同的运行类型（并发、顺序-所有核心、顺序-单个核心），调用相应的方法发送错误消息。

最后，代码中还包含了一系列基准测试函数，用于测试不同锁机制的性能。

希望以上信息对你有所帮助！如果你有任何其他问题，请随时提问。

英文:

UPDATED thanks to @peter-cordes - see comments below the answer

Apparently, reading and setting atomic.Bool is faster than sync.RWMutex for a single write by a single writer. It is only marginally slower than a no-lock racy solution (used only for benchmarking)

Benchmark results (sync.Mutex-based solution is also provided for benchmarking purposes only):

  1. $ go test -run=XXX -bench=. -benchmem -benchtime=1000000x
  2. goos: darwin
  3. goarch: arm64
  4. pkg: untitled1
  5. BenchmarkRacyNoGoodConcurrently-12                       1000000              3529 ns/op             384 B/op         12 allocs/op
  6. BenchmarkAtomicBooleanConcurrently-12                    1000000              3494 ns/op             384 B/op         12 allocs/op
  7. BenchmarkRWMutexConcurrently-12                          1000000              3909 ns/op             384 B/op         12 allocs/op
  8. BenchmarkMutexConcurrently-12                            1000000              4180 ns/op             384 B/op         12 allocs/op
  9. BenchmarkRacyNoGoodSequentiallyAllCores-12               1000000                 3.661 ns/op           0 B/op          0 allocs/op
  10. BenchmarkAtomicBooleanSequentiallyAllCores-12            1000000                 3.748 ns/op           0 B/op          0 allocs/op
  11. BenchmarkRWMutexSequentiallyAllCores-12                  1000000              1934 ns/op               0 B/op          0 allocs/op
  12. BenchmarkMutexSequentiallyAllCores-12                    1000000              1486 ns/op               0 B/op          0 allocs/op
  13. BenchmarkRacyNoGoodSequentiallySingleCore-12             1000000                28.95 ns/op            0 B/op          0 allocs/op
  14. BenchmarkAtomicBooleanSequentiallySingleCore-12          1000000                29.54 ns/op            0 B/op          0 allocs/op
  15. BenchmarkRWMutexSequentiallySingleCore-12                1000000               188.6 ns/op             0 B/op          0 allocs/op
  16. BenchmarkMutexSequentiallySingleCore-12                  1000000               187.4 ns/op             0 B/op          0 allocs/op
  17. PASS
  18. ok      untitled1       19.093s
  1. package main
  3. import (
  4. 	&quot;runtime&quot;
  5. 	&quot;sync&quot;
  6. 	&quot;sync/atomic&quot;
  7. )
  9. type ErrorNotifier interface {
  10. 	SetEmailSenderService(func(string))
  11. 	SendErrorMessage(string)
  12. }
  14. // Mutex
  16. type MutexErrorNotifier struct {
  17. 	mutex       sync.Mutex
  18. 	emailSender func(string)
  19. }
  21. var _ ErrorNotifier = (*MutexErrorNotifier)(nil)
  23. func (a *MutexErrorNotifier) SetEmailSenderService(emailSender func(string)) {
  24. 	a.mutex.Lock()
  25. 	defer a.mutex.Unlock()
  27. 	a.emailSender = emailSender
  28. }
  30. func (a *MutexErrorNotifier) SendErrorMessage(errorMessage string) {
  31. 	a.mutex.Lock()
  32. 	defer a.mutex.Unlock()
  34. 	if a.emailSender != nil {
  35. 		a.emailSender(errorMessage)
  36. 	}
  37. }
  39. // RWMutex
  41. type RWMutexErrorNotifier struct {
  42. 	rwMutex     sync.RWMutex
  43. 	emailSender func(string)
  44. }
  46. var _ ErrorNotifier = (*RWMutexErrorNotifier)(nil)
  48. func (a *RWMutexErrorNotifier) SetEmailSenderService(emailSender func(string)) {
  49. 	a.rwMutex.Lock()
  50. 	defer a.rwMutex.Unlock()
  52. 	a.emailSender = emailSender
  53. }
  55. func (a *RWMutexErrorNotifier) SendErrorMessage(errorMessage string) {
  56. 	a.rwMutex.RLock()
  57. 	defer a.rwMutex.RUnlock()
  59. 	if a.emailSender != nil {
  60. 		a.emailSender(errorMessage)
  61. 	}
  62. }
  64. // Atomic Boolean
  66. type AtomicBooleanErrorNotifier struct {
  67. 	emailerIsSet atomic.Bool
  68. 	emailSender  func(string)
  69. }
  71. var _ ErrorNotifier = (*AtomicBooleanErrorNotifier)(nil)
  73. func (a *AtomicBooleanErrorNotifier) SetEmailSenderService(emailSender func(string)) {
  74. 	defer a.emailerIsSet.Store(true)
  76. 	a.emailSender = emailSender
  77. }
  79. func (a *AtomicBooleanErrorNotifier) SendErrorMessage(errorMessage string) {
  80. 	if a.emailerIsSet.Load() {
  81. 		a.emailSender(errorMessage)
  82. 	}
  83. }
  85. // NOT A SOLUTION: racy no locking solution - just for benchmarking
  87. type RacyNoGoodErrorNotifier struct {
  88. 	emailSender func(string)
  89. }
  91. var _ ErrorNotifier = (*RacyNoGoodErrorNotifier)(nil)
  93. func (a *RacyNoGoodErrorNotifier) SetEmailSenderService(emailSender func(string)) {
  94. 	a.emailSender = emailSender
  95. }
  97. func (a *RacyNoGoodErrorNotifier) SendErrorMessage(errorMessage string) {
  98. 	if a.emailSender != nil {
  99. 		a.emailSender(errorMessage)
  100. 	}
  101. }
  103. // Demo run
  105. const allConcurrent = &quot;all concurrent&quot;
  106. const sequentialSingleCore = &quot;sequential single core&quot;
  107. const sequentialAllCores = &quot;sequential all cores&quot;
  109. func Run(n int, runner ErrorNotifier, runType string) {
  110. 	emailSender := func(emailMessage string) {
  111. 		// sending email...
  112. 	}
  114. 	var wg sync.WaitGroup
  116. 	switch runType {
  117. 	case allConcurrent:
  118. 		wg.Add(n * runtime.NumCPU())
  120. 		for i := 0; i &lt; n*runtime.NumCPU(); i++ {
  121. 			go func() {
  122. 				runner.SendErrorMessage(&quot;ALARM!&quot;)
  123. 				wg.Done()
  124. 			}()
  125. 		}
  126. 	case sequentialAllCores:
  127. 		wg.Add(runtime.NumCPU())
  129. 		for i := 0; i &lt; runtime.NumCPU(); i++ {
  130. 			go func() {
  131. 				for j := 0; j &lt; n; j++ {
  132. 					runner.SendErrorMessage(&quot;ALARM!&quot;)
  133. 				}
  134. 				wg.Done()
  135. 			}()
  136. 		}
  137. 	case sequentialSingleCore:
  138. 		wg.Add(1)
  140. 		go func() {
  141. 			for j := 0; j &lt; n*runtime.NumCPU(); j++ {
  142. 				runner.SendErrorMessage(&quot;ALARM!&quot;)
  143. 			}
  144. 			wg.Done()
  145. 		}()
  146. 	default:
  147. 		panic(&quot;unknown mode&quot;)
  148. 	}
  150. 	runner.SetEmailSenderService(emailSender)
  152. 	wg.Wait()
  153. }

Benchmarks:

  1. package main
  3. import &quot;testing&quot;
  5. func BenchmarkRacyNoGoodConcurrently(b *testing.B) {
  6. 	Run(b.N, &amp;RacyNoGoodErrorNotifier{}, allConcurrent)
  7. }
  9. func BenchmarkAtomicBooleanConcurrently(b *testing.B) {
  10. 	Run(b.N, &amp;AtomicBooleanErrorNotifier{}, allConcurrent)
  11. }
  13. func BenchmarkRWMutexConcurrently(b *testing.B) {
  14. 	Run(b.N, &amp;RWMutexErrorNotifier{}, allConcurrent)
  15. }
  17. func BenchmarkMutexConcurrently(b *testing.B) {
  18. 	Run(b.N, &amp;MutexErrorNotifier{}, allConcurrent)
  19. }
  21. func BenchmarkRacyNoGoodSequentiallyAllCores(b *testing.B) {
  22. 	Run(b.N, &amp;RacyNoGoodErrorNotifier{}, sequentialAllCores)
  23. }
  25. func BenchmarkAtomicBooleanSequentiallyAllCores(b *testing.B) {
  26. 	Run(b.N, &amp;AtomicBooleanErrorNotifier{}, sequentialAllCores)
  27. }
  29. func BenchmarkRWMutexSequentiallyAllCores(b *testing.B) {
  30. 	Run(b.N, &amp;RWMutexErrorNotifier{}, sequentialAllCores)
  31. }
  33. func BenchmarkMutexSequentiallyAllCores(b *testing.B) {
  34. 	Run(b.N, &amp;MutexErrorNotifier{}, sequentialAllCores)
  35. }
  37. func BenchmarkRacyNoGoodSequentiallySingleCore(b *testing.B) {
  38. 	Run(b.N, &amp;RacyNoGoodErrorNotifier{}, sequentialSingleCore)
  39. }
  41. func BenchmarkAtomicBooleanSequentiallySingleCore(b *testing.B) {
  42. 	Run(b.N, &amp;AtomicBooleanErrorNotifier{}, sequentialSingleCore)
  43. }
  45. func BenchmarkRWMutexSequentiallySingleCore(b *testing.B) {
  46. 	Run(b.N, &amp;RWMutexErrorNotifier{}, sequentialSingleCore)
  47. }
  49. func BenchmarkMutexSequentiallySingleCore(b *testing.B) {
  50. 	Run(b.N, &amp;MutexErrorNotifier{}, sequentialSingleCore)
  51. }

huangapple
  • 本文由 发表于 2023年4月1日 05:52:46
  • 转载请务必保留本文链接：https://go.coder-hub.com/75903010.html
  • atomic
  • concurrency
  • go
  • mutex
  • race-condition
匿名

发表评论

匿名网友

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

确定