Go 1.17の sync/atomic パッケージ更新点と CompareAndSwap

The Gopher character is based on the Go mascot designed by Renee French.

はじめに

TIGの市川です。Go 1.17連載のラストです。前回は宮崎さんによる「Go1.17で警告されるようになったerror#Is/As/Unwrap」の解説でした。

この記事では、Go 1.17で変更になった sync/atomic パッケージについて解説します。

Go 1.17の sync/atomic パッケージ更新点

sync/atomic パッケージの atomic.Value に以下のメソッドが追加されました。

• CompareAndSwap
• Swap

Go 1.16以前でもCompare And Swap(以下CAS)とSwapの関数はintとuintの32, 64型とPointer型の関数が用意されてました。今回の変更により atomic.Value でCASが利用可能になりました。

CASについて

CASは文字通り、比較と置換をアトミックに行う命令です。
処理は次の1~3です。

1. 「現在の値」と「以前取得した値」を比較
2. 異なる場合はFalseを返す
3. 一致した場合は「現在の値」を「新しい値に」入れ替えTrueを返す

CASからFalseが返った場合には、割り込みによって値が変わったと判断し、改めて取得からやり直します。

通常ロックしてから処理を行う場合と比較し、ロック時間が短くなることが利点です。(こちらの記事の図が非常に分かりやすいと感じました。)

CompareAndSwap実装例

並列処理で *big.Int の共有カウンタを回す処理のサンプルです。sync.WaitGroup は並列処理の完了制御で利用してます。

https://play.golang.org/p/7IKfoh7wYJT

(Playground上では仕様により処理時間が0秒になります。)

package main

import (
	"fmt"
	"math/big"
	"sync"
	"sync/atomic"
	"time"
)

func main() {
	counter := big.NewInt(0)
	delta := big.NewInt(1)
	var m atomic.Value
	m.Store(counter) // 値を保存
	var wg sync.WaitGroup
	start := time.Now()
	defer func() {
		fmt.Println(time.Since(start))
	}()

	wg.Add(10000)
	for i := 0; i < 10000; i++ {
		go func() {
			defer wg.Done()
			for {
				newVal := big.NewInt(0)
				oldVal := m.Load().(*big.Int) // 値を取得
				time.Sleep(time.Microsecond)
				newVal.Add(oldVal, delta)
				if m.CompareAndSwap(oldVal, newVal) {
					// oldValの値が一致し置換を成功
					break
				}
			}
		}()
	}

	wg.Wait()
	fmt.Println(m.Load().(*big.Int)) // 10000
}

int64などの場合には atomic.AddInt64 などの専用の関数が用意されており、Compare And Swapを利用しなくても実装可能です。
実装例: https://play.golang.org/p/8IeIITMaMzs

CompareAndSwapと別の処理方式との比較

CompareAndSwapPointerを使った実装との比較

既存で用意されている CompareAndSwapPointer を利用した実装です。

https://play.golang.org/p/cFZhpZIZVni
(Playground上では仕様により処理時間が0秒になります。)

func main() {
	counter := big.NewInt(0)
	delta := big.NewInt(1)
	var wg sync.WaitGroup
	start := time.Now()
	defer func() {
		fmt.Println(time.Since(start))
	}()

	wg.Add(10000)
	for i := 0; i < 10000; i++ {
		go func() {
			defer wg.Done()
			for {
				newVal := big.NewInt(0)
				oldVal := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&counter)))
				newVal.Add((*big.Int)(oldVal), delta)
				//time.Sleep(time.Microsecond)
				if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&counter)), oldVal, unsafe.Pointer(newVal)) {
					break
				}
			}
		}()
	}

	wg.Wait()
	fmt.Println((*big.Int)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&counter))))) // 10000
}

CompareAndSwapPointer での実装と比較し unsafe.Pointer を使わずに済み、CompareAndSwap のほうがシンプルに書けるようになりました。

ちなみに処理時間はあまり変わりませんでした。

排他制御との比較

前章で記載した共有カウントを排他制御を使った実装です。

https://play.golang.org/p/Lt9FYmefo5m
(Playground上では仕様により処理時間が0秒になります。)

func main() {
	counter := big.NewInt(0)
	delta := big.NewInt(1)

	var wg sync.WaitGroup
	var mt sync.Mutex
	start := time.Now()
	defer func() {
		fmt.Println(time.Since(start))
	}()

	wg.Add(10000)
	for i := 0; i < 10000; i++ {
		go func() {
			defer wg.Done()
			mt.Lock()
			defer mt.Unlock()
			//time.Sleep(time.Microsecond)
			counter.Add(counter, delta)
		}()
	}

	wg.Wait()
	fmt.Println(counter) // 10000
}

排他処理での実装はSwapの処理がいらないためシンプルな実装になります。

time.Sleep のコメントアウトを外した場合の処理時間を比較すると手元の環境でCAS: 約23ms, 排他制御: 約500msと20倍以上の処理時間がかかりました。時間がかかる処理を行う場合にはCAS方式のほうが処理が早そうです。
(Sleepがない場合には、CAS: 約3.0ms, 排他制御: 約2.5msと排他制御のほうが処理時間が短い結果となりました。)

Swapの実装サンプル

Swapのほうは良い例を思いつかなかったので、単純なサンプルを載せます。
https://play.golang.org/p/te6ewTUvomV

func main() {
	bi := big.NewInt(1)
	newVal := big.NewInt(2)
	var m atomic.Value
	m.Store(bi)

	fmt.Println("before:", m.Load().(*big.Int)) // before: 1
	oldVal := m.Swap(newVal).(*big.Int)
	fmt.Println("oldVal:", oldVal)             // oldVal: 1
	fmt.Println("after:", m.Load().(*big.Int)) // after: 2
}

上記のサンプルを既存の関数 SwapPointer で書くと以下のようになります。
https://play.golang.org/p/HcUrh3J4uNo

func main() {
	bi := big.NewInt(1)
	newVal := big.NewInt(2)

	fmt.Println("before:", (*big.Int)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&bi))))) // before: 1
	oldVal := atomic.SwapPointer((*unsafe.Pointer)(unsafe.Pointer(&bi)), unsafe.Pointer(newVal))
	fmt.Println("oldVal:", (*big.Int)(oldVal))                                                    // oldVal: 1
	fmt.Println("after:", (*big.Int)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&bi))))) // after: 2
}

Swapのケースでも既存の関数の利用と比べシンプルかつ unsafe.Pointer を使わずに記述出来るようになりました。

まとめ

• atomic.Value で CompareAndSwap, Swap のメソッドが追加になった。
• 既存の atomic.CompareAndSwapPointer, atomic.SwapPointer 関数と比較し、 unsafe.Pointer を利用せずに任意の型でCASやSwapが実装可能になった。
• 処理時間がかかる場合には排他処理よりもCASが有利になるケースを確認した。