Использование sync/atomic в golang для float64

Привет всем гоферам! В данной статье я хочу разобрать как можно воспользоваться модулем sync/atomic для типа float64.

Постановка задачи

У нас есть канал из которого мы можем читать результаты выполнения задач. Результат задачи содержит флаг показывающий была ли ошибка при выполнении задачи и результат выполнения этой задачи (тип float64). Нужно найти сумму всех результатов и количество ошибок.

Реализация с использованием sync.Mutex

Задачу можно решить с использованием sync.Mutex:

package main  import ( "fmt" "math" "sync" "sync/atomic" )  const countWorkers = 1000 const countTasks = 10000000  type Result struct { value    float64 hasError bool }  func MakeTasks(count int) <-chan Result { ch := make(chan Result) go func() { for i := 0; i < count; i++ { ch <- Result{ value:    float64(i) * 2.42, hasError: (i % 10) == 0, }  } close(ch) }() return ch }  func ProcessUsingMutex(ch <-chan Result, countWorkers int) (float64, int64) { var wg sync.WaitGroup var errMu sync.Mutex var mu sync.Mutex var countErrors int64 var result float64  for i := 0; i < countWorkers; i++ { wg.Add(1) go func() { for item := range ch { if item.hasError { errMu.Lock() countErrors++ errMu.Unlock() } else { mu.Lock() result += item.value mu.Unlock() } } wg.Done() }() } wg.Wait() return result, countErrors }   func main() { ch := MakeTasks(countTasks) fmt.Println(ProcessUsingMutex(ch, countWorkers)) } 

Реализация с использованием sync/atomic

Но использование sync.Mutex может замедлять нашу программу. Поэтому можно переписать решение с использованием sync/atomic. Для количества ошибок это сделать довольно просто. Вместо использования sync.Mutex используем atomic.AddInt64, то есть:

errMu.Lock() countErrors++ errMu.Unlock()

заменяем на

atomic.AddInt64(&countErrors, 1)

Но для float64 нет функции AddFloat64. Но её можно реализовать. Для того, что бы лучше понять как это можно сделать напишем функцию AddInt64 с использованием CompareAndSwapInt64

func AddInt64(addr *int64, delta int64) (new int64) { for { if v := *addr; atomic.CompareAndSwapInt64(addr, v, v+delta) { return v } } }

Но для float64 нет и CompareAndSwap, но мы можем сконвертировать значение float64 в uint64 используя math.Float64bits и оперировать уже uint64.

Реализация AddFloat64 может быть такой:

func AddFloat64(addr *uint64, delta float64) uint64 { for { cur := atomic.LoadUint64(addr) curVal := math.Float64frombits(cur) nxtVal := curVal + delta nxt := math.Float64bits(nxtVal) if atomic.CompareAndSwapUint64(addr, cur, nxt) { return nxt } } }

После того, как мы реализовали AddFloat64 можно полностью переписать нашу функцию, без использования мютексов:

func ProcessUsingAtomic(ch <-chan Result, countWorkers int) (float64, int64) { var wg sync.WaitGroup var countErrors int64 var total uint64 for i := 0; i < countWorkers; i++ { wg.Add(1) go func() { for item := range ch { if item.hasError { atomic.AddInt64(&countErrors, 1) } else { AddFloat64(&total, item.value) } } wg.Done() }() } wg.Wait() return math.Float64frombits(atomic.LoadUint64(&total)), atomic.LoadInt64(&countErrors) }

Для сравнения mutex и atomic я написал небольшой бенчмарк

package main  import ( "testing" )  func BenchmarkProcessUsingAtomic(b *testing.B) {  for i := 0; i < b.N; i++ { b.StopTimer() ch := MakeTasks(countTasks) b.StartTimer() ProcessUsingAtomic(ch, countWorkers) } }  func BenchmarkProcessUsingMutex(b *testing.B) { for i := 0; i < b.N; i++ { b.StopTimer() ch := MakeTasks(countTasks) b.StartTimer() ProcessUsingMutex(ch, countWorkers) } }  

Запуск данного benchmark показывает что atomic на 5-7% быстрее.

BenchmarkProcessUsingAtomic-8 1 2468118459 ns/op
BenchmarkProcessUsingMutex-8 1 2640532917 ns/op
PASS

Заключение

Модуль sync/atomic может быть полезен, если его использовать вместо мютексов так как атомики работают немного быстрее мютексов. Если хочется использовать атомики для типов, которые не поддерживаются в sync/atomic можно попробовать использовать стандартные функции для реализации нужного функционала, как я описал в данной статье или воспользоваться структурой Value из модуля sync/atomic.