libaio(Linuxの非同期I/Oライブラリ)の使い方

Linuxで非同期I/Oを行うためのライブラリ「libaio」の使い方を書いてみる事にする。少し昔の話になるが、lighttpdが使用し、スループットを80%も上げたらしい。

TOEFLに向けて転置ファイルについての論文(Inverted files for text search engine [moffat 06])でReading対策をしていたところ、意外とスニペット(検索にヒットした箇所の前後の文章)を作るところが時間がかかるという事を教えてもらったので、適当にそれを例題にしてみる。具体的には以下のようなコードを非同期I/Oを使用して速くなるかどうか見てみる。

for (unsigned int i = 0; i < files.size(); i++) {
  FILE* fp = fopen(files[i].c_str(), "rb");
  if (fp == NULL) continue;
  fseek(fp, offsets[i], SEEK_SET);
  char buf[64];
  size_t nread = fread(buf, 1, 64, fp);
  fclose(fp);
}

ファイルリストとそれに付随するオフセットのリストが与えられた時、各ファイルの指定位置から64 byteづつ読み込むようなプログラムだ。
色々高速化する手段は有るが、日本語でlibaioの使い方が解説してあるところがなかったので、布教も兼ねて書いてみる。

まず、このコードのどこが遅いかというとfreadする時にI/Oが完了するのを待ってしまう点が挙げられる。

その点、非同期I/Oを使えばファイル分だけI/O要求を一気に出しておけば、それをI/O Schedulerが賢くスケジューリングし、全体としての読み込み時間をminimizeする事が出来る。読み込みが完了したものについては、指定したコールバック内で処理を行う事が出来る。LinuxのI/O Schedulerについてはもわせんぱいの記事に詳しい。

では実際にlibaioの使い方を見ていく。使い方の流れとしては、大体以下の様になる。

• io_context_tの初期化
• I/O要求タスクの準備
• I/O要求をOSに投げる
• I/O要求の完了を待ち受ける

では、順番に具体的な関数名と共に流れを追っていく。
(1) io_context_tの初期化
io_setup関数を用いて以下の様に初期化する。

io_context_t ctx;
memset(&ctx, 0, sizeof(io_context_t));
r = io_setup(size, &ctx);
assert(r == 0);

(2) I/O要求タスクの準備
I/O要求タスクを作成する。具体的にはstruct iocb*の配列を自分でallocateし、それに対してディスクリプタ、バッファ、読み込みサイズ、オフセットの指定を行う。これはよく行われる処理であるので、io_prep_pread関数が用意されている。終了時に呼ばれるコールバックはio_set_callback関数でセットする事ができる。

struct iocb **iocbs = new struct iocb*[size];
char **bufs = new char*[size];
for (int i = 0; i < size; i++) {
  int fd = open(files[i].c_str(), O_RDONLY);
  assert(fd >= 0);
  iocbs[i] = new struct iocb();
  bufs[i] = (char*)malloc(64);
  io_prep_pread(iocbs[i], fd, bufs[i], 64, offset);
  io_set_callback(iocbs[i], read_done);
}

io_prep_pread関数は非常に簡単で、libaio.hで以下の様に定義されている。

static inline void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t co
unt, long long offset)
{
  memset(iocb, 0, sizeof(*iocb));
  iocb->aio_fildes = fd;
  iocb->aio_lio_opcode = IO_CMD_PREAD;
  iocb->aio_reqprio = 0;
  iocb->u.c.buf = buf;
  iocb->u.c.nbytes = count;
  iocb->u.c.offset = offset;
}

iocb構造体のメンバに値をセットしているだけだ。
(3) I/O要求をOSに投げる
さて、次は準備したI/O要求をOSに投げる。これにはio_submit関数を使う。

r = io_submit(ctx, size, iocbs);
assert(r == size);

(4) I/O要求の完了を待ち受ける
あとは、OSに投げた要求が終わるのを待つだけである。これにはio_getevents関数を用いる。ここでは同時に受け付けるイベントの最大数は32個とした。io_event配列の中身に、完了したI/Oについての情報が入っている。ネットワークプログラミングでいうとselect(2)やpoll(2)のようなものにあたる。

int cnt = 0;
while (true) {
  struct io_event events[32];
  int n = io_getevents(ctx, 1, 32, events, NULL);
  if (n > 0)
    cnt += n;
  for (int i = 0; i < n; i++) {
    struct io_event *ev = events + i;
    io_callback_t callback = (io_callback_t)ev->data;
    struct iocb *iocb = ev->obj;
    callback(ctx, iocb, ev->res, ev->res2);
  }
  if (n == 0 || cnt == size)
    break;
}

コールバック関数read_doneは次のようになる(io_set_callback関数に引数として渡した)。

static void
read_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
{
  close(iocb->aio_fildes);
  return;
}

以上がlibaioを使ったプログラムになる。libaio.hを見ると他にもAPIがあるが、ここまで分かっていれば見れば分かるものばかりである(書き込み、poll、sync等)。

最後に、非同期I/Oを使うと本当に速くなるのか測定を行ってみた。以下の環境で計測を行った。

Linux core 2.6.20-15-generic #2 SMP Sun Apr 15 06:17:24 UTC 2007 x86_64 GNU/Linux
g++ version: 4.1.2
libaio version: 0.3.106-3ubuntu2
SATA disk

多少長くなるが、プログラム全体を掲載する。C++。

#include <fstream>
#include <cassert>
#include <iostream>
#include <vector>
#include <string>
#include <map>
#include <time.h>;
#include <sys/time.h>;
#include <sys/types.h>;
#include <sys/stat.h>;
#include <fcntl.h>;
#include <libaio.h>;

using namespace std;

static double
gettimeofday_sec()
{
  struct timeval tv;
  gettimeofday(&tv, NULL);
  return tv.tv_sec + (double)tv.tv_usec*1e-6;
}

//-----------------------------------------------------------------------------
// Sync I/O
//
void
snippet1(const vector<string>& files, size_t offset)
{
  int size = files.size();
  char **bufs = new char*[size];

  for (int i = 0; i < size; i++) {
    bufs[i] = (char*)malloc(64);

    FILE *fp = fopen(files[i].c_str(), "rb");
    assert(fp != NULL);
    fseek(fp, offset, SEEK_SET);
    fread(bufs[i], 1, 64, fp);
    fclose(fp);
  }

  for (int i = 0; i < size; i++) {
    /*
    cout << "----------" << endl
         << buf << endl;
    */
    free(bufs[i]);
  }
  delete[] bufs;

  return;
}

//-----------------------------------------------------------------------------
// Async I/O
//
static void
read_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
{
  close(iocb->aio_fildes);
  return;
}

void
snippet2(const vector<string>& files, size_t offset)
{
  int r;
  int size = files.size();

  // (1) io_context_tの初期化
  io_context_t ctx;
  memset(&ctx, 0, sizeof(io_context_t));
  r = io_setup(size, &ctx);
  assert(r == 0);

  // (2) iocbs(I/O要求)の構築
  struct iocb **iocbs = new struct iocb*[size];
  char **bufs = new char*[size];
  for (int i = 0; i < size; i++) {
    int fd = open(files[i].c_str(), O_RDONLY);
    assert(fd >= 0);

    iocbs[i] = new struct iocb();
    bufs[i] = (char*)malloc(64);

    io_prep_pread(iocbs[i], fd, bufs[i], 64, offset);
    io_set_callback(iocbs[i], read_done);
  }

  // (3) I/O要求を投げる
  r = io_submit(ctx, size, iocbs);
  assert(r == size);

  // (4) 完了したI/O要求を待ち、終わったものについてはcallbackを呼び出す
  int cnt = 0;
  while (true) {
    struct io_event events[32];
    int n = io_getevents(ctx, 1, 32, events, NULL);
    if (n > 0)
      cnt += n;

    for (int i = 0; i < n; i++) {
      struct io_event *ev = events + i;
      io_callback_t callback = (io_callback_t)ev->data;
      struct iocb *iocb = ev->obj;
      callback(ctx, iocb, ev->res, ev->res2);
    }

    if (n == 0 || cnt == size)
      break;
  }

  for (int i = 0; i < size; i++) {
    /*
    cout << "----------" << endl
         << bufs[i] << endl;
    */

    delete iocbs[i];
    free(bufs[i]);
  }
  delete[] iocbs;
  delete[] bufs;
}

int
main(int argc, char **argv)
{
  if (argc != 4) {
    cout << argv[0] << " filelist maxnum offset" << endl;
    return 0;
  }

  ifstream ifs(argv[1]);
  int maxnum = atoi(argv[2]);
  assert(maxnum > 0);
  size_t offset = atoi(argv[3]);

  vector<string> files;
  string line;
  while (getline(ifs, line)) {
    files.push_back(line);
    if (files.size() == maxnum)
      break;
  }

  cout << "Num: " << files.size() << endl;

  double t1 = gettimeofday_sec();
  snippet1(files, offset);
  double t2 = gettimeofday_sec();
  cout << "Sync  I/O Time: " << t2 - t1 << endl;

  double t3 = gettimeofday_sec();
  snippet2(files, offset);
  double t4 = gettimeofday_sec();
  cout << "ASync I/O Time: " << t4 - t3 << endl;

  return 0;
}

以下のようにして実行する。第一引数にファイルリスト、第二引数にそのファイルリストからいくつ使うか(文章数)、第三引数に読み込んでくる offset を指定する。ビルド方法及び実行結果は以下の様になる。

core% g++ snippet.cpp -laio; ./a.out /data/jawiki-070427/fileList 1000 100
Num: 1000
Sync  I/O Time: 0.895428
ASync I/O Time: 0.00889802
pficore%
core% g++ snippet.cpp -laio; ./a.out /data/jawiki-070427/fileList 1000 100
Num: 1000
Sync  I/O Time: 0.022953
ASync I/O Time: 0.00649595
core% g++ snippet.cpp -laio; ./a.out /data/jawiki-070427/fileList 1000 100
Num: 1000
Sync  I/O Time: 0.023381
ASync I/O Time: 0.00715804

DiskCacheに乗った後でも、大体3倍ぐらいは速くなっている事が分かる。これで一応非同期I/Oの優位性は示せた事になったので一安心。「DBとか使うんだったら独自インデックス作って自分で最適化した方が全然速いよね」とかいう人には是非お勧め。

何か間違った事とか書いてたら教えてください。

Similar Posts:

• エンディアンフリーコード
• spliceシステムコールの使い方
• gflags + glogでdeadlock?