メリークリスマス！

概要

非同期にログのようなものをサービスから送信するようなclientを作るとき、「サーバーをどう落とすか」というのはとても難しい問題だと思います。
特にお金がかかわるようなログだと送り漏れがないようにする必要がありますし「もう同期でいいんじゃね」と頭を抱えてしまいます。

非同期なのでバッファのような場所にデータがたまっているので、これを掃除してから落とす必要があるのが厄介の源です。

要件

clientは、あるjson型のログを別のRESTのシステムに投げるようなユースケースを想定していますが、ここではそこは端折ってio.Writerに書いています。
clientは、Sendで送られたメッセージをキャッシュし、

キャッシュがあるサイズに達したら
定期的に
最後にサーバーをShutdownするときに

の３つの条件でflushします。

解

解１：同期的に送信する

まずシンプルな解として同期的なものを用意しておきます。非同期処理はこれと同じ事を実現する必要があります。

type V1Client struct {
	w io.Writer
}

func NewV1Clienet(w io.Writer) *V1Client {
	return &V1Client{w}
}

func (c *V1Client) Send(message string) error {
	_, err := fmt.Fprintln(c.w, message)
	return err
}

これでなぜダメだったかといいますと、メッセージを束ねる必要があったためです。

解２：戦いはここから始まった

type V2Client struct {
	w    io.Writer
	q    chan string
	logs []string
	max  int
	mu   *sync.Mutex
	itvl time.Duration
}

func NewV2Client(w io.Writer, bufsize int, itvl time.Duration) *V2Client {
	return &V2Client{
		w:    w,
		logs: make([]string, bufsize)[:0],
		mu:   &sync.Mutex{},
		max:  bufsize,
		itvl: itvl,
	}
}

func (c *V2Client) Send(ctx context.Context, message string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.logs = append(c.logs, message)
	if len(c.logs) >= c.max {
		c.flush(ctx)
	}
}

func (c *V2Client) Start(ctx context.Context) {
	go func() {
		for {
			time.Sleep(c.itvl)
			c.mu.Lock()
			if err := c.flush(ctx); err != nil {
				fmt.Printf("error: %+v\n", err)
			}
			c.mu.Unlock()
			select {
			case <-ctx.Done():
				return
			}
		}
	}()
}

func (c *V2Client) flush(ctx context.Context) error {
	defer func() {
		c.logs = c.logs[:0]
	}()
	for _, m := range c.logs {
		if _, err := fmt.Fprintln(c.w, m); err != nil {
			return err
		}

	}
	return nil
}

非同期バージョンの最初の実装はこんな感じでした。ほかの言語のイメージで実装するとこんな感じになるのではないかと思います。
この実装には様々な問題点があります。

context cancelすることでプログラムを落とす、その時にlogsの中に残ったログの掃除をしていない
logsという配列を共有メモリのようにgoroutine間のデータのやり取りに使っている
logsがオーバーフローしたとき、同期的にログを送信してしまっている

主にこの３点でしょうか。ほかにもいろいろあるとは思いますが。

解３：修正版

type V3Client struct {
	logs []*string    // log buffer
	q    chan string // channel to send log
	max  int           // max size of logs
	itvl time.Duration // log send interval
	w    io.Writer     // l2pc backup logger

	exit chan struct{}
	once *sync.Once
}


func NewV3Client(writer io.Writer, buffsize int, interval time.Duration) *V3Client {
	client := &V3Client{
		logs:         make([]string, buffsize)[:0],
		q:            make(chan string),
		max:          buffsize,
		itvl:         interval,
		w:            writer,
		once:         &sync.Once{},
	}
	return client
}

func (c *V3Client) Send(message string) {
	c.q <- message
}

func (c *V3Client) Start(ctx context.Context) {

	if c.exit != nil {
		fmt.Println("client is already started.")
		return
	}
	c.exit = make(chan struct{})

	go func() {

		t := time.NewTicker(c.itvl)
		defer t.Stop()

		for {
			select {
			case m := <-c.q:
				if err := c.tryFlush(ctx, m); err != nil {
					fmt.Println("log send error: " + err.Error())
				}
			case <-t.C:
				if err := c.flush(ctx); err != nil {
					fmt.Println("log send error: " + err.Error())
				}
			case <-c.exit:
				return
			}
		}
	}()
}

func (c *V3Client) Close(ctx context.Context) {
	c.once.Do(func() {
		c.exit <- struct{}{}
		if err := c.flushAll(ctx); err != nil {
			fmt.Println("log send error" + err.Error())
		}
	})

}

func (c *V3Client) tryFlush(ctx context.Context, msg string) error {
	c.logs = append(c.logs, msg)
	if len(c.logs) >= c.max {
		return c.flush(ctx)
	}
	return nil
}

func (c *V3Client) flushAll(ctx context.Context) error {
	close(c.q)
	all := c.logs
	for l := range c.q {
		all = append(all, l)
	}
	for {
		if len(all) < c.max {
			c.logs = all
			return c.flush(ctx)
		}
		c.logs, all = all[:c.max], all[c.max:]
		if err := c.flush(ctx); err != nil {
			return err
		}
	}
}

func (c *V3Client) flush(ctx context.Context) error {

	defer func() {
		c.logs = c.logs[:0]
	}()

	for _, l := range c.logs {
		if _, err := fmt.Fprintln(c.w, l); err != nil {
			return err
		}
	}
	return nil
}}

各メソッドはそれぞれ

client	説明
client#Start	clientの非同期送信を開始する。goroutine部。
client#Send	メッセージを送る（入力側）。業務ロジックが呼び出す。
client$Close	終了処理開始および、終了処理がおわるまで待機
client#flush	実送信部
client#tryFlush	メッセージを受け取り必要であればflushする
client#flushAll	終了処理

です。#2と比較して特に違うのは「ノンブロッキング」だという事だと思います。
ほかの言語だと、こういったプログラムには必ずといっていいほどLockが顔を出します。ですが上にはありません。

SleepではなくTickerを使っている
logsをgoroutineで共有せず、goroutine間の受け渡しにqというchanを使っている

ためです。
また、終了処理のために専用のexitというchを導入しました。なのでselect/caseで3つのchannelを待つコードになっています。

学んだこと

goroutine間を渡すデータはlockして共有メモリで渡すのではなくchannelで渡す
一定時間置きに起動する処理はSleepを使わずTickerを使う
defer cl.Close()で終了処理を開始しつつかつ終了処理の終了まで待たせる
select/caseで複数のchannelを待つとき、同時に複数のchannelにデータが入ったときどのchannelが起動するかはrandom

func main(){
   ctx := context.Background()
   cl := NewV3Client(...)
   cl.Start(ctx)
   defer cl.Close(ctx)

   // start server with client
}

最終的なmainはこのようになりました。
割と普通なインターフェースに収まったように思います。

他の言語に慣れていると、「ノンブロッキングに無限forループを回す」というのはなかなか違和感があるのですが、確かにPythonのGILのように、過剰な範囲で足止めをしてしまい、パフォーマンスの足を引っ張る要因になりやすいのはわかる気がします。

一旦ここまでで。