moratorium

まず、このコードのどこが遅いかというと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とか使うんだったら独自インデックス作って自分で最適化した方が全然速いよね」とかいう人には是非お勧め。

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