AsyncIOについて(その2) - 最速配信研究会(@yamaz)

AsyncIOについて(その1)の続き.

NONBlockでIO処理をする方法としてselectとシグナルを使う方法があるというのが前回の話だったが,
selectはselectよりkqueue,epollで述べたとおり,
ビジーループがかかるためあまり効率はよくなく,シグナル方式は制約があるためあまり使い勝手がよくない.
というわけで新しく出てきたのがPOSIX Asynchronous I/O(AIO)という機構だ.

これはIOのwaitをイベントドリブン形式にしてビジーループをなくそうというものだ.
プログラムの流れとしては下記のようになる.

1. 対象となるファイルディスクリプタにシグナルハンドラもしくはイベントハンドラを登録しておく
2. aio_read/aio_writeを呼び出すと制御はすぐにユーザに戻る.
3．対象のファイルディスクリプタの処理が終わると登録されていたハンドラが非同期に呼ばれる.

実際のコードなどに関しては
http://www-128.ibm.com/developerworks/linux/library/l-async/
が詳しい.上記ページには3つの実装が載っていてそれぞれは下記のようになっている.

1．ハンドラを登録しない場合

#include <aio.h>

...

  int fd, ret;
  struct aiocb my_aiocb;

  fd = open( "file.txt", O_RDONLY );
  if (fd < 0) perror("open");

  /* Zero out the aiocb structure (recommended) */
  bzero( (char *)&my_aiocb, sizeof(struct aiocb) );

  /* Allocate a data buffer for the aiocb request */
  my_aiocb.aio_buf = malloc(BUFSIZE+1);
  if (!my_aiocb.aio_buf) perror("malloc");

  /* Initialize the necessary fields in the aiocb */
  my_aiocb.aio_fildes = fd;
  my_aiocb.aio_nbytes = BUFSIZE;
  my_aiocb.aio_offset = 0;

  ret = aio_read( &my_aiocb );
  if (ret < 0) perror("aio_read");

  while ( aio_error( &my_aiocb ) == EINPROGRESS ) ;

  if ((ret = aio_return( &my_iocb )) > 0) {
    /* got ret bytes on the read */
  } else {
    /* read failed, consult errno */
  }

while文の部分がビジーループになっているため,NonBlockのselectを呼ぶのとほとんど変わらない.

2．シグナルハンドラでcallbackする場合

void setup_io( ... )
{
  int fd;
  struct sigaction sig_act;
  struct aiocb my_aiocb;

  ...

  /* Set up the signal handler */
  sigemptyset(&sig_act.sa_mask);
  sig_act.sa_flags = SA_SIGINFO;
  sig_act.sa_sigaction = aio_completion_handler;


  /* Set up the AIO request */
  bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
  my_aiocb.aio_fildes = fd;
  my_aiocb.aio_buf = malloc(BUF_SIZE+1);
  my_aiocb.aio_nbytes = BUF_SIZE;
  my_aiocb.aio_offset = next_offset;

  /* Link the AIO request with the Signal Handler */
  my_aiocb.aio_sigevent.sigev_notify = SIGEV_SIGNAL;
  my_aiocb.aio_sigevent.sigev_signo = SIGIO;
  my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;

  /* Map the Signal to the Signal Handler */
  ret = sigaction( SIGIO, &sig_act, NULL );

  ...

  ret = aio_read( &my_aiocb );

}


void aio_completion_handler( int signo, siginfo_t *info, void *context )
{
  struct aiocb *req;


  /* Ensure it's our signal */
  if (info->si_signo == SIGIO) {

    req = (struct aiocb *)info->si_value.sival_ptr;

    /* Did the request complete? */
    if (aio_error( req ) == 0) {

      /* Request completed successfully, get the return status */
      ret = aio_return( req );

    }

  }

  return;
}

3．スレッドベースのハンドラでcallbackする場合

void setup_io( ... )
{
  int fd;
  struct aiocb my_aiocb;

  ...

  /* Set up the AIO request */
  bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
  my_aiocb.aio_fildes = fd;
  my_aiocb.aio_buf = malloc(BUF_SIZE+1);
  my_aiocb.aio_nbytes = BUF_SIZE;
  my_aiocb.aio_offset = next_offset;

  /* Link the AIO request with a thread callback */
  my_aiocb.aio_sigevent.sigev_notify = SIGEV_THREAD;
  my_aiocb.aio_sigevent.notify_function = aio_completion_handler;
  my_aiocb.aio_sigevent.notify_attributes = NULL;
  my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;

  ...

  ret = aio_read( &my_aiocb );

}

void aio_completion_handler( sigval_t sigval )
{
  struct aiocb *req;

  req = (struct aiocb *)sigval.sival_ptr;

  /* Did the request complete? */
  if (aio_error( req ) == 0) {

    /* Request completed successfully, get the return status */
    ret = aio_return( req );

  }

  return;
}

双方とも自分のファイルディスクリプタをaio_completion_handlerハンドラ内で取得できているので,スムーズにデータ処理ができていることがわかると思う.

AIOはIOに対するBlockとNonBlock時のビジーループを避けられ,イベントハンドリングというわかりやすいモデルで作られているので,今後はやっていくと思われる.小さなファイルを3,4個だけreadするようなケースではあまり恩恵が受けられないが,システムにIOWaitがかかっている状況の時は改善する可能性があるので試す価値はあると思う.

(おしまい)