Goで学ぶRedisプロトコル（RESP）の読み書き実装入門

この記事では、Go言語でRedisクライアントを構成する2つのコンポーネント——コマンドの書き込みと読み取り——を、シンプルで理解しやすい形で実装しながら、Redisプロトコル（RESP）の仕組みと、その設計がなぜ優れているのかを解説します。

本番環境で使えるフル機能のGo向けRedisクライアントをお探しの場合は、Gary Burd氏のredigoライブラリをチェックすることをおすすめします。

始める前に、Redisプロトコルの入門記事に目を通しておいてください。このガイドで必要となるプロトコルの基礎知識がそこでカバーされています。

GoでのRESPコマンドライター

仮想的なRedisクライアントにおいて、書き込む必要があるオブジェクトは1種類だけです。それは、Redisへコマンドを送信するための「バルク文字列の配列」です。以下は、コマンドをRESP形式に変換して書き出すシンプルな実装例です。

package redis

import (
  "bufio"
  "io"
  "strconv"     // 整数を文字列に変換するために使用
)

var (
  arrayPrefixSlice      = []byte{'*'}
  bulkStringPrefixSlice = []byte{'$'}
  lineEndingSlice       = []byte{'\r', '\n'}
)

type RESPWriter struct {
  *bufio.Writer
}

func NewRESPWriter(writer io.Writer) *RESPWriter {
  return &RESPWriter{
    Writer: bufio.NewWriter(writer),
  }
}

func (w *RESPWriter) WriteCommand(args ...string) (err error) {
  // 配列プレフィックスと引数の数を書き込む。
  w.Write(arrayPrefixSlice)
  w.WriteString(strconv.Itoa(len(args)))
  w.Write(lineEndingSlice)

  // 各引数をバルク文字列として書き込む。
  for _, arg := range args {
    w.Write(bulkStringPrefixSlice)
    w.WriteString(strconv.Itoa(len(arg)))
    w.Write(lineEndingSlice)
    w.WriteString(arg)
    w.Write(lineEndingSlice)
  }

  return w.Flush()
}

RESPWriterはnet.Connオブジェクトに直接書き込むのではなく、io.Writerインターフェースに対して書き込みを行います。これにより、netスタックに密結合することなくパーサーをテストでき、他のio処理とまったく同じ要領でネットワークプロトコルを検証できます。

たとえば、bytes.Bufferを渡せば、最終的なRESPの出力内容を簡単に確認できます。

var buf bytes.Buffer
writer := NewRESPWriter(&buf)
writer.WriteCommand("GET", "foo")
buf.Bytes() // *2\r\n$3\r\nGET\r\n$3\r\nfoo\r\n

GoでのシンプルなRESPリーダー

RESPWriterでRedisにコマンドを送信した後、クライアントはRESPReaderを使ってTCP接続からデータを読み込み、完全なRESPレスポンスを受け取るまで待機します。まずは、受信データのバッファリングと解析に必要なパッケージをいくつかインポートします。

package redis

import (
  "bufio"
  "bytes"
  "errors"
  "io"
  "strconv"
)

さらに、コードを読みやすくするために、いくつかの定数と変数を定義します。

const (
  SIMPLE_STRING = '+'
  BULK_STRING   = '$'
  INTEGER       = ':'
  ARRAY         = '*'
  ERROR         = '-'
)

var (
  ErrInvalidSyntax = errors.New("resp: invalid syntax")
)

RESPWriterと同様に、RESPReaderも読み込み元オブジェクトの実装詳細には関心がありません。必要なのは、完全なRESPオブジェクトを読み終えるまでバイト列を読み続けられることだけです。ここではio.Readerを使用し、それをbufio.Readerでラップして受信データのバッファリングを行います。

構造体とイニシャライザは非常にシンプルです。

type RESPReader struct {
  *bufio.Reader
}

func NewReader(reader io.Reader) *RESPReader {
  return &RESPReader{
    Reader: bufio.NewReaderSize(reader, 32*1024),
  }
}

bufio.Readerのバッファサイズは、開発段階での暫定的な値です。実際のクライアントでは、サイズを設定可能にし、最適値を見つけるためのテストを行うのが望ましいでしょう。開発中は32KBで十分に機能します。

RESPReaderにはReadObject()というメソッドが1つだけあります。このメソッドは呼び出されるたびに、完全なRESPオブジェクトを含むバイトスライスを返します。io.Readerから発生したエラーはそのまま返し、無効なRESP構文を検出した場合にもエラーを返します。

RESPはプレフィックス（先頭バイト）ベースのプロトコルなので、最初の1バイトを読むだけで、後続のバイトをどう処理すべきか判断できます。ただし、常に最低でも最初の1行（つまり最初の\r\nまで）を読む必要があるため、まず最初の行全体を読み込むところから始めます。

func (r *RESPReader) ReadObject() ([]byte, error) {
  line, err := r.readLine()
  if err != nil {
    return nil, err
  }

  switch line[0] {
  case SIMPLE_STRING, INTEGER, ERROR:
    return line, nil
  case BULK_STRING:
    return r.readBulkString(line)
  case ARRAY:
    return r.readArray(line)
  default:
    return nil, ErrInvalidSyntax
  }
}

読み取った行が単純文字列・整数・エラーのいずれかのプレフィックスを持つ場合、それらの型は1行内に完結しているため、行全体をそのままRESPオブジェクトとして返します。

readLine()では、最初の\nまでを読み込み、その直前が\rであることを確認してから、行をバイトスライスとして返します。

func (r *RESPReader) readLine() (line []byte, err error) {
  line, err = r.ReadBytes('\n')
  if err != nil {
    return nil, err
  }

  if len(line) > 1 && line[len(line)-2] == '\r' {
    return line, nil
  } else {
    // 行が短すぎる、または \n の直前が \r ではない。
    return nil, ErrInvalidSyntax
  }
}

readBulkString()では、バルク文字列の長さ指定を解析し、何バイト読み込むべきかを把握します。その後、指定されたバイト数と\r\nの行終端を読み込みます。

func (r *RESPReader) readBulkString(line []byte) ([]byte, error) {
  count, err := r.getCount(line)
  if err != nil {
    return nil, err
  }
  if count == -1 {
    return line, nil
  }

  buf := make([]byte, len(line)+count+2)
  copy(buf, line)
  _, err = io.ReadFull(r, buf[len(line):])
  if err != nil {
    return nil, err
  }

  return buf, nil
}

getCount()を独立したメソッドとして切り出したのは、この長さ指定が配列の処理でも使われるからです。

func (r *RESPReader) getCount(line []byte) (int, error) {
  end := bytes.IndexByte(line, '\r')
  return strconv.Atoi(string(line[1:end]))
}

配列を処理する際は、まず配列の要素数を取得し、その後ReadObject()を再帰的に呼び出して、得られたオブジェクトを現在のRESPバッファに追加していきます。

func (r *RESPReader) readArray(line []byte) ([]byte, error) {
  // 配列の要素数を取得する。
  count, err := r.getCount(line)
  if err != nil {
    return nil, err
  }

  // 配列内の `count` 個のRESPオブジェクトを読み込む。
  for i := 0; i < count; i++ {
    buf, err := r.ReadObject()
    if err != nil {
      return nil, err
    }
    line = append(line, buf...)
  }

  return line, nil
}

まとめ

上記のわずか100行程度のコードで、RedisからあらゆるRESPオブジェクトを読み取ることができます。ただし、このライブラリを本番環境で使用する前に実装しておくべき要素もいくつか残っています。

• RESPから実際の値を抽出する機能。現在のRESPReaderは完全なRESPレスポンスを返すだけで、たとえばバルク文字列レスポンスから文字列本体を取り出すことはできません。とはいえ、この実装自体は簡単です。
• RESPReaderには、より堅牢な構文エラー処理が必要です。

また、このコードはまったく最適化されておらず、必要以上のメモリ確保（アロケーション）とコピーが発生しています。たとえばreadArray()メソッドでは、配列内の各オブジェクトを読み込んだ後、それをローカルバッファへコピーしています。

これらの実装方法について詳しく知りたい方は、hiredisやredigoといった人気ライブラリがどのように実装しているかを調べてみることをおすすめします。

最後に、この記事のコードに潜んでいたバグの発見に協力してくださったNiel Smith氏に感謝いたします。