src/worker.cc
// Do an asynchronous disk I/O operation.
// Return false if the IO is not set up.
Do an asynchronous disk I/O operation.
Return false if the IO is not set up.
Do an asynchronous disk I / O operation . Return false if the IO is not set up .
bool DiskThread::AsyncDiskIO(IoOp op, int fd, void *buf, int64 size,
                            int64 offset, int64 timeout) {
#ifdef HAVE_LIBAIO_H
  struct {
    const int opcode;
    const char *op_str;
    const char *error_str;
  } operations[2] = {
    { IO_CMD_PREAD, "read", "disk-read-error" },
    { IO_CMD_PWRITE, "write", "disk-write-error" }
  };
  struct iocb cb;
  memset(&cb, 0, sizeof(cb));
  cb.aio_fildes = fd;
  cb.aio_lio_opcode = operations[op].opcode;
  cb.u.c.buf = buf;
  cb.u.c.nbytes = size;
  cb.u.c.offset = offset;
  struct iocb *cbs[] = { &cb };
  if (io_submit(aio_ctx_, 1, cbs) != 1) {
    int error = errno;
    char buf[256];
    sat_strerror(error, buf, sizeof(buf));
    logprintf(0, "Process Error: Unable to submit async %s "
                 "on disk %s (thread %d). Error %d, %s\n",
              operations[op].op_str, device_name_.c_str(),
              thread_num_, error, buf);
    return false;
  }
  struct io_event event;
  memset(&event, 0, sizeof(event));
  struct timespec tv;
  tv.tv_sec = timeout / 1000000;
  tv.tv_nsec = (timeout % 1000000) * 1000;
  if (io_getevents(aio_ctx_, 1, 1, &event, &tv) != 1) {
    int error = errno;
    if (error == EINTR) {
      logprintf(5, "Log: %s interrupted on disk %s (thread %d).\n",
                operations[op].op_str, device_name_.c_str(),
                thread_num_);
    } else {
      os_->ErrorReport(device_name_.c_str(), operations[op].error_str, 1);
      errorcount_ += 1;
      logprintf(0, "Hardware Error: Timeout doing async %s to sectors "
                   "starting at %lld on disk %s (thread %d).\n",
                operations[op].op_str, offset / kSectorSize,
                device_name_.c_str(), thread_num_);
    }
    io_cancel(aio_ctx_, &cb, &event);
    io_destroy(aio_ctx_);
    aio_ctx_ = 0;
    if (io_setup(5, &aio_ctx_)) {
      int error = errno;
      char buf[256];
      sat_strerror(error, buf, sizeof(buf));
      logprintf(0, "Process Error: Unable to create aio context on disk %s"
                " (thread %d) Error %d, %s\n",
                device_name_.c_str(), thread_num_, error, buf);
    }
    return false;
  }
  if (event.res != static_cast<uint64>(size)) {
    errorcount_++;
    os_->ErrorReport(device_name_.c_str(), operations[op].error_str, 1);
    if (event.res < 0) {
      switch (event.res) {
        case -EIO:
          logprintf(0, "Hardware Error: Low-level I/O error while doing %s to "
                       "sectors starting at %lld on disk %s (thread %d).\n",
                    operations[op].op_str, offset / kSectorSize,
                    device_name_.c_str(), thread_num_);
          break;
        default:
          logprintf(0, "Hardware Error: Unknown error while doing %s to "
                       "sectors starting at %lld on disk %s (thread %d).\n",
                    operations[op].op_str, offset / kSectorSize,
                    device_name_.c_str(), thread_num_);
      }
    } else {
      logprintf(0, "Hardware Error: Unable to %s to sectors starting at "
                   "%lld on disk %s (thread %d).\n",
                operations[op].op_str, offset / kSectorSize,
                device_name_.c_str(), thread_num_);
    }
    return false;
  }
  return true;
#else 
  return false;
#endif
}
bool DiskThread :: AsyncDiskIO ( IoOp op , int fd , void * buf , int64 size , int64 offset , int64 timeout ) { #ifdef HAVE_LIBAIO_H struct { const int opcode ; const char * op_str ; const char * error_str ; } operations [ 2 ] = { { IO_CMD_PREAD , " " , " " } , { IO_CMD_PWRITE , " " , " " } } ; struct iocb cb ; memset ( & cb , 0 , sizeof ( cb ) ) ; cb . aio_fildes = fd ; cb . aio_lio_opcode = operations [ op ] . opcode ; cb . u . c . buf = buf ; cb . u . c . nbytes = size ; cb . u . c . offset = offset ; struct iocb * cbs [ ] = { & cb } ; if ( io_submit ( aio_ctx_ , 1 , cbs ) != 1 ) { int error = errno ; char buf [ 256 ] ; sat_strerror ( error , buf , sizeof ( buf ) ) ; logprintf ( 0 , " " " \n " , operations [ op ] . op_str , device_name_ . c_str ( ) , thread_num_ , error , buf ) ; return false ; } struct io_event event ; memset ( & event , 0 , sizeof ( event ) ) ; struct timespec tv ; tv . tv_sec = timeout / 1000000 ; tv . tv_nsec = ( timeout % 1000000 ) * 1000 ; if ( io_getevents ( aio_ctx_ , 1 , 1 , & event , & tv ) != 1 ) { int error = errno ; if ( error == EINTR ) { logprintf ( 5 , " \n " , operations [ op ] . op_str , device_name_ . c_str ( ) , thread_num_ ) ; } else { os_ -> ErrorReport ( device_name_ . c_str ( ) , operations [ op ] . error_str , 1 ) ; errorcount_ += 1 ; logprintf ( 0 , " " " \n " , operations [ op ] . op_str , offset / kSectorSize , device_name_ . c_str ( ) , thread_num_ ) ; } io_cancel ( aio_ctx_ , & cb , & event ) ; io_destroy ( aio_ctx_ ) ; aio_ctx_ = 0 ; if ( io_setup ( 5 , & aio_ctx_ ) ) { int error = errno ; char buf [ 256 ] ; sat_strerror ( error , buf , sizeof ( buf ) ) ; logprintf ( 0 , " " " \n " , device_name_ . c_str ( ) , thread_num_ , error , buf ) ; } return false ; } if ( event . res != static_cast < uint64 > ( size ) ) { errorcount_ ++ ; os_ -> ErrorReport ( device_name_ . c_str ( ) , operations [ op ] . error_str , 1 ) ; if ( event . res < 0 ) { switch ( event . res ) { case - EIO : logprintf ( 0 , " " " \n " , operations [ op ] . op_str , offset / kSectorSize , device_name_ . c_str ( ) , thread_num_ ) ; break ; default : logprintf ( 0 , " " " \n " , operations [ op ] . op_str , offset / kSectorSize , device_name_ . c_str ( ) , thread_num_ ) ; } } else { logprintf ( 0 , " " " \n " , operations [ op ] . op_str , offset / kSectorSize , device_name_ . c_str ( ) , thread_num_ ) ; } return false ; } return true ; #else return false ; #endif }
Do an asynchronous disk I/O operation.
Do an asynchronous disk I / O operation .
// Use the Linux native asynchronous I/O interface for reading/writing. // A read/write consists of three basic steps: //    1. create an io context. //    2. prepare and submit an io request to the context //    3. wait for an event on the context. // A ctrl-c from the keyboard will cause io_getevents to fail with an // EINTR error code.  This is not an error and so don't treat it as such, // but still log it. // Don't bother checking return codes since io_cancel seems to always fail. // Since io_cancel is always failing, destroying and recreating an I/O // context is a workaround for canceling an in-progress I/O operation. // TODO(amistry): Find out why io_cancel isn't working and make it work. // event.res contains the number of bytes written/read or // error if < 0, I think. // !HAVE_LIBAIO_H