Linux下异步I/O运行正常但无法终止的问题咨询

问题背景 #

之前咨询过异步I/O实现方法，当前代码可在Ubuntu 20.04 LTS上编译运行，能成功读取1-5块（每块512字节），但读取完文件后无法终止，会在第4、5块间循环往复。测试文件包含5块内容，每块512字节，内容分别为1、2、3、4、5。

代码示例 #

#include <aio.h>
#include <fcntl.h>
#include <signal.h>
#include <unistd.h>

#include <condition_variable>
#include <cstring>
#include <iostream>
#include <thread>

using namespace std;
using namespace std::chrono_literals;

constexpr uint32_t blockSize = 512;

mutex readMutex;
bool readReady = false;
condition_variable cv;
bool operation_completed = false;

int fh;
int bytesRead;

void process(char* buf, uint32_t bytesRead) {
  cout << "processing..." << endl;
  usleep(100000);
}

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

  // check whether asynch operation is complete
  int status;
  if ((status = aio_error(req)) != 0) {
    cout << "Error: " << status << '\n';
    return;
  }
  int ret = aio_return(req);
    bytesRead = req->aio_nbytes;
  cout << "ret == " << ret << endl;
  cout << (char*)req->aio_buf << endl;
  unique_lock<mutex> readLock(readMutex);
  operation_completed = true;
  cv.notify_one();
}

void thready() {
  char* buf1 = new char[blockSize];
  char* buf2 = new char[blockSize];
  aiocb cb;
  char* processbuf = buf1;
  char* readbuf = buf2;
  fh = open("smallfile.dat", O_RDONLY);
  if (fh < 0) {
    throw std::runtime_error("cannot open file!");
  }

  memset(&cb, 0, sizeof(aiocb));
  cb.aio_fildes = fh;
  cb.aio_nbytes = blockSize;
  cb.aio_offset = 0;

  // Fill in callback information
  /*
  Using SIGEV_THREAD to request a thread callback function as a notification
  method
  */
  cb.aio_sigevent.sigev_notify_attributes = nullptr;
  cb.aio_sigevent.sigev_notify = SIGEV_THREAD;
  cb.aio_sigevent.sigev_notify_function = aio_completion_handler;
  /*
  The context to be transmitted is loaded into the handler (in this case, a
  reference to the aiocb request itself). In this handler, we simply refer to
  the arrived sigval pointer and use the AIO function to verify that the request
  has been completed.
  */
  cb.aio_sigevent.sigev_value.sival_ptr = &cb;

  int cursor = 0;
  int currentBytesRead = read(fh, buf1, blockSize);  // read the 1st block

  while (true) {
    cb.aio_buf = readbuf;
    operation_completed = false; // set predicate to true and wait until asynch changes it
    cb.aio_offset = cursor;
    aio_read(&cb);  // each next block is read asynchronously
    process(processbuf, currentBytesRead);  // process while waiting
    {
      unique_lock<mutex> readLock(readMutex);
      cv.wait( readLock, []{ return operation_completed; } );
    }
    if (!operation_completed)
      break;
    currentBytesRead = bytesRead; // make local copy of global modified by the asynch code
    cursor += bytesRead;
    if (currentBytesRead < blockSize) {
      break;  // last time, get out
    }
    cout << "back from wait" << endl;
    swap(processbuf, readbuf);     // switch to other buffer for next time
    currentBytesRead = bytesRead;  // create local copy
  }

  delete[] buf1;
  delete[] buf2;
}

int main() {
  try {
    thready();
  } catch (std::exception& e) {
    cerr << e.what() << '\n';
  }
  return 0;
}

咨询问题 #

• 上述代码是否适合用于获取文件长度并确定准确的读取次数？
• 为何调用aio_read读取超出文件末尾时仍返回成功（错误状态始终为0）？

问题解答 #

1. 代码是否适合获取文件长度并确定准确读取次数？ #

不适合。当前代码存在两处关键问题，导致无法可靠获取文件长度或确定读取次数：

• 文件指针错位：初始同步read调用会移动文件描述符的当前指针，而异步aio_read是基于aio_offset指定的绝对偏移操作，两者不共享指针状态。代码中cursor从0开始累加，但初始read已经读取了第一块，后续异步读取的偏移会重复读取已读内容，导致逻辑混乱。
• 无预获取文件长度逻辑：代码依赖循环中判断读取字节数是否小于块大小来终止循环，但这种方式无法预先计算准确读取次数，且在文件长度恰好是块大小整数倍时，会多发起一次读取（返回0字节），而当前代码的错误处理会导致循环无法终止。

如果要准确控制读取次数，建议先用stat或lseek获取文件总长度，再计算总块数（如(file_size + blockSize - 1) / blockSize向上取整），然后按块数发起异步读取。

2. 读取超出文件末尾时aio_read返回成功的原因 #

这是符合POSIX标准的正常行为：

• 当aio_read请求读取的位置超出文件末尾时，操作本身不属于错误，aio_error会返回0表示操作完成。
• 实际读取的字节数由aio_return返回：如果已完全到达文件末尾，返回0；如果读取位置部分覆盖文件内容，返回剩余的字节数。

你的代码中存在逻辑错误：在aio_completion_handler里，错误地将bytesRead赋值为req->aio_nbytes（即请求的512字节），而非aio_return返回的实际读取字节数ret。这导致即使读取到文件末尾，bytesRead仍为512，循环判断currentBytesRead < blockSize永远不成立，从而陷入循环。

关键修复点 #

• 在aio_completion_handler中，将bytesRead = req->aio_nbytes;改为bytesRead = ret;，确保获取实际读取的字节数。
• 初始同步read后，将cursor设置为blockSize，而非0，保证后续异步读取的偏移正确。

内容的提问来源于stack exchange，提问作者Dov