Merge pull request #870 from devlights/add-gate-example

Author: devlights

examples/singleapp/gate/Taskfile.yml

@@ -0,0 +1,8 @@
+# https://taskfile.dev
+
+version: '3'
+
+tasks:
+  default:
+    cmds:
+      - go run .

examples/singleapp/gate/countdownlatch.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+// CountdownLatch は、C#のCountdownEventやJavaのCountDownLatchと同様の機能を提供する構造体です.
+type CountdownLatch struct {
+	count atomic.Int32
+	mutex sync.Mutex
+	cond  *sync.Cond
+}
+
+// NewCountdownLatch は、指定されたカウント数でCountdownLatchを初期化します.
+func NewCountdownLatch(initialCount int) *CountdownLatch {
+	if initialCount < 0 {
+		panic("初期カウントは0以上である必要があります")
+	}
+
+	var (
+		latch CountdownLatch
+	)
+	latch.count.Store(int32(initialCount))
+	latch.cond = sync.NewCond(&latch.mutex)
+
+	return &latch
+}
+
+// Signal は、カウントを1減らします.
+// 戻り値として、カウントダウンが満了したかどうかを返します.
+func (me *CountdownLatch) Signal() bool {
+	return me.SignalCount(1)
+}
+
+// SignalCount は、指定された数だけカウントを減らします.
+// 戻り値として、カウントダウンが満了したかどうかを返します.
+func (me *CountdownLatch) SignalCount(count int) bool {
+	if count <= 0 {
+		return false
+	}
+
+	me.mutex.Lock()
+	defer me.mutex.Unlock()
+
+	newCount := me.count.Add(-int32(count))
+	if newCount <= 0 {
+		me.cond.Broadcast()
+		return true
+	}
+
+	return false
+}
+
+// Wait は、カウントが0になるまでブロックします.
+func (me *CountdownLatch) Wait() {
+	me.mutex.Lock()
+	defer me.mutex.Unlock()
+
+	for me.count.Load() > 0 {
+		me.cond.Wait()
+	}
+}
+
+// CurrentCount は、現在のカウント値を返します.
+func (me *CountdownLatch) CurrentCount() int {
+	me.mutex.Lock()
+	defer me.mutex.Unlock()
+
+	return int(me.count.Load())
+}
+
+// Reset は、カウントを指定された値にリセットします.
+// リセットすることになるため、強制的にカウント満了したことになり、待機している非同期処理が存在する場合は解除されます.
+func (me *CountdownLatch) Reset(count int) {
+	if count < 0 {
+		panic("リセットカウントは0以上である必要があります")
+	}
+
+	me.mutex.Lock()
+	defer me.mutex.Unlock()
+
+	me.cond.Broadcast()
+
+	me.count.Store(int32(count))
+}

examples/singleapp/gate/gate.go

@@ -0,0 +1,32 @@
+package main
+
+type (
+	Gate struct {
+		latch *CountdownLatch
+	}
+)
+
+func NewGate() *Gate {
+	var (
+		latch = NewCountdownLatch(1)
+		gate  = Gate{latch}
+	)
+
+	return &gate
+}
+
+func (me *Gate) Await() {
+	if me.latch.CurrentCount() < 1 {
+		return
+	}
+
+	me.latch.Wait()
+}
+
+func (me *Gate) Open() {
+	if me.latch.CurrentCount() < 1 {
+		return
+	}
+
+	me.latch.Signal()
+}

examples/singleapp/gate/main.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log"
+	"sync"
+	"time"
+)
+
+const (
+	MainTimeout = 20 * time.Second
+	ProcTimeout = 10 * time.Second
+)
+
+var (
+	ErrMainTooSlow = errors.New("(MAIN) TOO SLOW")
+	ErrProcTooSlow = errors.New("(PROC) TOO SLOW")
+)
+
+func init() {
+	log.SetFlags(0)
+}
+
+func main() {
+	var (
+		rootCtx          = context.Background()
+		mainCtx, mainCxl = context.WithTimeoutCause(rootCtx, MainTimeout, ErrMainTooSlow)
+		procCtx          = run(mainCtx)
+		err              error
+	)
+	defer mainCxl()
+
+	select {
+	case <-mainCtx.Done():
+		err = context.Cause(mainCtx)
+	case <-procCtx.Done():
+		if err = context.Cause(procCtx); errors.Is(err, context.Canceled) {
+			err = nil
+		}
+	}
+
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func run(pCtx context.Context) context.Context {
+	var (
+		ctx, cxl = context.WithCancelCause(pCtx)
+	)
+
+	go func() {
+		cxl(proc(ctx))
+	}()
+	go func() {
+		<-time.After(ProcTimeout)
+		cxl(ErrProcTooSlow)
+	}()
+
+	return ctx
+}
+
+func proc(_ context.Context) error {
+	var (
+		gate = NewGate()
+		wg   sync.WaitGroup
+	)
+
+	// 10個のゴルーチンがゲート前に待機する
+	for i := range 10 {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+
+			log.Printf("[%2d] 待機開始", i)
+			gate.Await()
+			log.Printf("[%2d] 待機解除", i)
+		}(i)
+	}
+
+	// 何か準備処理などを行っているとする
+	<-time.After(time.Second)
+	log.Println("-------------------------------------")
+
+	// ゲートを開き、待機解除したゴルーチン達が全完了するのを待つ
+	gate.Open()
+	wg.Wait()
+
+	// 一度開いたゲートは開きっぱなしになるため、開いた後のAwait呼び出しは即座に返る.
+	gate.Await()
+	gate.Await()
+
+	return nil
+}