この記事は Go Advent Calendar 2020 の10日目の記事です。

こんにちは、辻です。最近はクラウドからMCプロトコルというプロトコルを用いてスマート工場のPLCを操作するTCPクライアントを実装しています。TCPクライアントの実装は future-architect/go-mcprotocol として公開しています。

本記事では高速で信頼できるTCPクライアントをGoで実装するための知見を共有します。TCPとソケットプログラミングはざっと概要だけ触れておきます。

TCPとは？

TCPとはRFC793で仕様が定められているプロトコルです。特徴を簡単におさらいしておくと、

非構造化ストリーム
全二重通信
コネクション管理
高信頼性
- シーケンス番号
- 再送制御
- 順序制御
- 輻輳制御
- チェックサム

といった特徴があります。TCPのヘッダーフォーマットは以下のようになっていました。TCPのヘッダーは32ビット(4バイト)の列が5つで20バイトです。data がTCPのペイロードに該当します。TCPクライアントは data にアプリケーションのプロトコルに従ったペイロードを格納して、データをやりとりします。TCPは非構造化ストリームのプロトコルです。TCPのペイロードは単なるビット列です。ビット列に含まれるデータの解釈は上位層の仕事です。

「TCPヘッダのフォーマット」からの引用

TCPの状態遷移

またTCPはいくつかの状態を持つプロトコルです。それぞれの状態の遷移図は以下のようになります。

「TCP（Transmission Control Protocol）」からの引用

TCPサーバを起動して netstat などのコマンドを使って、TCP通信の状態を確認することができます。たとえば、これから紹介するEchoサーバをローカルホストの5000番ポートで起動した状態で netstat を使うと以下のように LISTEN していることがわかります。(Windowsの例)

> netstat -ap tcp | grep 5000
  TCP         0.0.0.0:5000           0.0.0.0:0      LISTENING

TCPクライアントから SYN フラグを含む接続リクエストを受け取るとソケットの状態が SYN_RECEIVED に遷移し、さらに ACK を受信すると ESTABLISHED になる、という仕組みです。

TCPそのものの仕様詳細については TCP 詳説などの資料を参考にしてください。

TCPソケットプログラミング

TCPのクライアントとサーバで通信するときの基本的なソケットシステムコールをおさらいしておきます。TCPクライアントとTCPサーバが通信するときは以下のようなソケット関数を用いて通信します。

例えばソケットを生成するときは以下の socket システムコールを呼び出します。

int socket(int domain, int type, int protocol);

domain 引数は通信を行なうドメインを指定するもので、以下の定数を用いることできます。IPv4で通信する場合は AF_INET を用いる、という具合です。
type は以下の定数が定義されています。TCP通信をしたい場合、通常は SOCK_STREAM を指定することになります。UDPの場合は SOCK_DGRAM です。

名前	説明
SOCK_STREAM	順序性と信頼性があり、双方向の、接続されたバイトストリーム (byte stream) を提供する。帯域外 (out-of-band) データ転送メカニズムもサポートされる。
SOCK_DGRAM	データグラム (コネクションレス、信頼性無し、固定最大長メッセージ) をサポートする。
...	...

protocol はソケットによって使用される固有のプロトコルを指定します。通常は 0 です。

socket はドメインやソケットの型を指定して呼び出しますが、ローカルアドレスやリモートアドレスは指定していません。3Wayハンドシェイクを行ってTCP接続するのは connect 関数の役割です。

socket 以外にも上記の図に登場したシステムコールを用いてTCP通信をしますが、GoでTCP通信するときは、直接開発者がシステムコールを発行する必要はありません。Goはこのようなソケットに関するシステムコールを抽象化して、開発者が扱いやすいようなAPIを提供しています。その代わりにGoのAPIが内部的にシステムコールを発行します。

`net` パッケージでTCP通信するときの基本

GoでTCP通信するときは net パッケージを使います。net パッケージには Dial や DialTCP といった関数が備わっており、これらの関数を使うとTCPサーバとクライアント間で1つのコネクションを確立します。

Dial

func Dial(network, address string) (Conn, error)

DialTCP

func DialTCP(network string, laddr, raddr *TCPAddr) (*TCPConn, error)

Conn はコネクションを示すインターフェースです。以下のメソッドが定義されています。

type Conn interface {
	Read(b []byte) (n int, err error)
	Write(b []byte) (n int, err error)
	Close() error
	LocalAddr() Addr
	RemoteAddr() Addr
	SetDeadline(t time.Time) error
	SetReadDeadline(t time.Time) error
	SetWriteDeadline(t time.Time) error
}

Conn を満たす構造体の例としては IPConn, TCPConn, UDPConn, UnixConn などがあります。実体はファイルディスクリプタである *netFD を保持する conn 構造体を埋め込んでいる構造体です。

type TCPConn struct {
	conn
}

conn 構造体は以下のようになっています。

type conn struct {
	fd *netFD
}

Dial("tcp", c.addr) としてコネクションを確立しようとすると内部的には DialTCP が呼び出されます。第一引数の文字列によって処理を切り替える、デザインパターンでいうところの、いわゆるStrategyパターンです。dial.go の実装の一部ですが、以下のようになっています。

	switch ra := ra.(type) {
	case *TCPAddr:
		la, _ := la.(*TCPAddr)
		c, err = sd.dialTCP(ctx, la, ra)
	case *UDPAddr:
		la, _ := la.(*UDPAddr)
		c, err = sd.dialUDP(ctx, la, ra)
	case *IPAddr:
		la, _ := la.(*IPAddr)
		c, err = sd.dialIP(ctx, la, ra)
	case *UnixAddr:
		la, _ := la.(*UnixAddr)
		c, err = sd.dialUnix(ctx, la, ra)
	default:
		return nil, &OpError{Op: "dial", Net: sd.network, Source: la, Addr: ra, Err: &AddrError{Err: "unexpected address type", Addr: sd.address}}
	}

DialTCP はシステムコールでいうところの socket によるソケットの作成や bind によるプロトコルアドレスのソケットへの割り当て connect による3Wayハンドシェイクを実施してコネクションを確立する、といったことを行ってくれます。

Conn インターフェースには Read や Write のメソッドが定義されています。コネクションへの読み書きはこの Read や Write メソッドを用いてバイト列を読み書きします。

GoでTCPクライアントを作る

実際にGoでTCPクライアントを作ってみましょう。リクエスト先のTCPサーバは methane/echoserver にある以下のTCPサーバのGo実装を拝借しましょう。io.Copy を使ったEchoサーバです。

package main

import (
	"io"
	"log"
	"net"
)

func echo_handler(conn net.Conn) {
	defer conn.Close()
	io.Copy(conn, conn)
}

func main() {
	psock, e := net.Listen("tcp", ":5000")
	if e != nil {
		log.Fatal(e)
		return
	}
	for {
		conn, e := psock.Accept()
		if e != nil {
			log.Fatal(e)
			return
		}
		go echo_handler(conn)
	}
}

本記事ではGoによる実務的なTCPサーバの実装方法について詳しくは述べませんが、信頼できるTCPサーバの作り方は「堅牢なTCPサーバを作るために - katsubushiの知見から/kamakura.go#5」の資料が参考になります。

TCPクライアントの実装

Echoサーバに対して hello を書き込んで、結果を受け取るTCPクライアントを実装します。まずはともあれ動作させてみましょう。上記のEchoサーバを起動させた状態で、以下の main.go を実行します。

client.go

package app

import "net"

type Client struct {
	addr string
}

func NewClient(addr string) *Client {
	return &Client{addr: addr}
}

func (c *Client) Hello(b []byte) (string, error) {
	conn, err := net.Dial("tcp", c.addr)
	if err != nil {
		return "", err
	}
	defer conn.Close()

	_, err = conn.Write(b)
	if err != nil {
		return "", err
	}

	buf := make([]byte, 1024)
	n, err := conn.Read(buf)
	if err != nil {
		return "", err
	}

	return string(buf[:n]), nil
}

cmd/main.go

package main

import (
	"fmt"

	"github.com/d-tsuji/go-sandbox/tcpsample/app"
)

func main() {
	c := app.NewClient(":5000")

	resp, err := c.Hello([]byte("hello"))
	if err != nil {
		panic(err)
	}

	fmt.Println(resp)
}

実行してみると、以下のように hello という文字列がサーバから返ってきて表示されることがわかります。

$ go run .
hello

Goではとても簡単にTCPクライアントを作ることができます。また生成したコネクションへの読み書きも Read, Write を用いて io.Reader, io.Write への読み書きと同じ感覚で扱うことができます。(Conn インターフェースが io.Reader, io.Write インターフェースを満たしているため)

なお Read は引数に []byte 型であるバイト配列のバッファを受け取りますが、バッファの長さがサーバからのレスポンス未満の場合で、まだサーバからレスポンスが残っている場合は複数回 Read を実行する必要があります。先程の cleint.go の buf を以下のように変えてみると挙動がわかるでしょう。

-	buf := make([]byte, 1024)
+	buf := make([]byte, 1)
	n, err := conn.Read(buf)
	if err != nil {
		panic(err)
	}

実行してみると、バッファの長さ分のレスポンスしか取得できないことがわかります。

$ go run .
h

より実用的なTCPクライアント

1.タイムアウト

先程のTCPクライアントはタイムアウトなしでEchoサーバと通信しています。 Read メソッドでデータを読み取ろうとしたときにレスポンスが返ってこなかった場合は無限にブロッキングします。実際にはタイムアウトを設定したいケースが多いでしょう。net パッケージには Dial, Write, Read のタイムアウトを設定するAPIが用意されています。

Dial 時のタイムアウト

net.DialTimeout で引数にタイムアウトを設定する方法や net.Dialer 構造体の Timeout フィールドにタイムアウト値を設定して Dialer.Dial でDialする方法があります。なお net.DialTimeout は以下のような実装になっているため、2つの方法は実体は同じです。

func DialTimeout(network, address string, timeout time.Duration) (Conn, error) {
	d := Dialer{Timeout: timeout}
	return d.Dial(network, address)
}

Write/Read のタイムアウト

SetWriteDeadline メソッドを使って書き込みのタイムアウトを設定できます。同様に SetReadDeadline メソッドで読み込みのタイムアウトを設定できます。SetDeadline は読み込みと書き込みの両方のタイムアウトを設定できます。

それでは、先程の client.go にタイムアウトを設定できるようにします。SetDeadline は読み書きのデッドラインの時刻を指定します。デッドラインはそれぞれの操作ごとにリセットされるわけではなく、固定の値を取ります。それぞれの Read や Write の操作ごとのタイムアウトを設定するには、それぞれデッドラインを設定する必要があります。

func (c *Client) Hello(b []byte) (string, error) {
-	conn, err := net.Dial("tcp", c.addr)
+	conn, err := net.DialTimeout("tcp", c.addr, c.Timeout)
	if err != nil {
		return "", err
	}
	defer conn.Close()

+	if c.Timeout > 0 {
+		if err = conn.SetDeadline(time.Now().Add(c.Timeout)); err != nil {
+			return "", err
+		}
+	}
	_, err = conn.Write(b)
	if err != nil {
		return "", err
	}

	buf := make([]byte, 1024)
+	if c.Timeout > 0 {
+		if err = conn.SetDeadline(time.Now().Add(c.Timeout)); err != nil {
+			return "", err
+		}
+	}
	n, err := conn.Read(buf)
	if err != nil {
		panic(err)
	}

	return string(buf[:n]), nil
}

デッドラインを設定すると、例えば main.go で異様に短いタイムアウトを設定した場合にエラーが返ってくることがわかります。

func main() {
	c := app.NewClient(":5000")
+	c.Timeout = 1 * time.Nanosecond

	resp, err := c.Hello([]byte("hello"))
	if err != nil {
		panic(err)
	}

	fmt.Println(resp)
}

実行すると以下のエラーが出力されるでしょう。タイムアウトを設定することによって、コネクションの確立時や読み書きのタイミングで無限にブロッキングすることはなくなりました。

panic: dial tcp :5000: i/o timeout

2.コネクションプーリング

続いてTCPクライアントのコネクションプーリングを考えましょう。現状の client.go は Hello メソッドを呼び出すたびにTCPサーバとコネクションを確立し、TCPサーバからEchoされた文字列を読み込んだらコネクションをクローズしています。net.Dial によるTCPのコネクション確立は3ウェイハンドシェイクによって確立されるため、TCPクライアントからTCPサーバへデータを送受信する場合と比較して、コストの高い処理になります。TCPクライアントが1回だけTCPサーバと通信する場合は問題ありませんが、複数回の呼び出しを高速に行いたい場合は net.Dial がボトルネックになる可能性があります。

このような場合、一度確立したコネクションを保持しておき、複数回TCPクライアントからTCPサーバへデータを送受信する場合に、確立済のコネクションが存在すればそのコネクションを使い回すといった、コネクションプーリングを行うことが有効になります。実装的には net.Conn を Client 構造体のフィールドとして保持することがポイントです。

TCPクライアントの実装は以下のような感じです。TCPクライアントが複数のゴルーチンで共有されても安全に読み書きが行えるように、コネクションの確立やクローズ、読み書きを sync.Mutex で排他制御をする必要があります。また、エラーが発生した場合は速やかにプールしているコネクションを開放するのが良いでしょう。リトライなどを考慮して、サーバ側からなんらかの理由で切断された不都合のあるコネクションに対して行った操作でエラーが発生した場合はそのコネクションを開放して、再度新しいコネクションを接続できるようにするためです。

client.go

package app

import (
	"net"
	"sync"
	"time"
)

type Client struct {
	addr string

	Timeout time.Duration

	// TCP connection
	mu   sync.Mutex
	conn net.Conn
}

func NewClient(addr string) *Client {
	return &Client{addr: addr}
}

func (c *Client) Hello(b []byte) (string, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if err := c.connect(); err != nil {
		return "", err
	}

	if c.Timeout > 0 {
		if err := c.conn.SetDeadline(time.Now().Add(c.Timeout)); err != nil {
			_ = c.conn.Close()
			return "", err
		}
	}
	_, err := c.conn.Write(b)
	if err != nil {
		_ = c.conn.Close()
		return "", err
	}

	buf := make([]byte, 1024)
	n, err := c.conn.Read(buf)
	if err != nil {
		_ = c.conn.Close()
		return "", err
	}

	return string(buf[:n]), nil
}

func (c *Client) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.close()
}

func (c *Client) Connect() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.connect()
}

// connectはコネクションがnilの場合はコネクションを試みます。
// すでにコネクションが確立されている場合は、そのコネクションを使い回します。
func (c *Client) connect() error {
	if c.conn == nil {
		conn, err := net.DialTimeout("tcp", c.addr, c.Timeout)
		if err != nil {
			return err
		}
		c.conn = conn
	}
	return nil
}

// closeはコネクションがnilでない場合にコネクションをCloseします。
func (c *Client) close() error {
	var err error
	if c.conn != nil {
		err = c.conn.Close()
		c.conn = nil
	}
	return err
}

cmd/main.go

package main

import (
	"fmt"

	"github.com/d-tsuji/go-sandbox/tcpsample/app"
)

func main() {
	c := app.NewClient(":5000")
	defer c.Close()

	// コネクションを確立してRead/Writeを実施
	resp, err := c.Hello([]byte("hello"))
	if err != nil {
		panic(err)
	}

	// 2回目以降は確立済のコネクションを使ってRead/Writeを実施
	resp, err = c.Hello([]byte("hello"))
	if err != nil {
		panic(err)
	}

	fmt.Println(resp)
}

3.リトライ

何らかの理由で一時的なエラーになった場合を考慮して、一定回数リトライできるようにしておくとより信頼性が高まるでしょう。リトライのアルゴリズムといえば、AWSのSDKでも採用されている ¹ ようなExponential Backoffが有名です。Goのライブラリだと cenkalti/backoff が有名です。しかし高速な通信が必要なTCPクライアントの場合はExponential Backoffではなく、単にエラーが発生した場合はシンプルに再試行あるいは、短い一定時間だけ待機して再試行するだけで十分な場合も多いでしょう。Exponential Backoffが有効なポイントの一つは、サーバ負荷を軽減する、ということがありますが、今回はループによるリトライでもサーバ側が問題ないような状況を仮定します。Exponential Backoffを使わずにシンプルにリトライすることにします。

またリトライをTCPクライアント側(client.go)に組み込むか、呼び出し側(main.go)に任せるかは選択の余地があります。今回は呼び出し側である main.go でリトライすることにします。

retry.go

一定回数ループでリトライの実装はこんな感じ

func Retry(ctx context.Context, attempts uint, interval time.Duration, fn func() error) error {
	var err error
	for attempts > 0 {
		attempts--
		if err = fn(); err == nil {
			break
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(interval):
		}
	}
	return err
}

cmd/main.go

呼び出し側の main.go でリトライします。以下はエラーがあった場合は最大3回試行、待機時間0秒でリトライする実装例です。

func main() {
	ctx := context.Background()

	c := app.NewClient(":5000")
	defer c.Close()

	var resp string
	err := app.Retry(ctx, 3, 0, func() error {
		var ierr error
		resp, ierr = c.Hello([]byte("hello"))
		return ierr
	})
	if err != nil {
		panic(err)
	}

	fmt.Println(resp)
}

まとめ

TCPの概要とGoによるTCPクライアントの作り方を紹介しました。Goはネットワーク周りの標準ライブラリが充実していて、サードパーティのライブラリを使わずとも高速で信頼できるTCPクライアントを実装できます。もちろんTCPクライアントで重要なのはTCPの data に書き込むペイロードです。TCPクライアントを作成して memchached やMCプロトコルなどのプロトコル実装に挑戦してみてください。本記事がTCPクライアントを作成する際に参考になれば幸いです。

AWS でのエラー再試行とエクスポネンシャルバックオフ ↩