Switch to a thread pool for aio operations
Creating a thread for each operation was
slightly wasteful. This also makes traces
easier to read.
Test: Transfer files, view traces
Bug: 76154677
Change-Id: I6028f113b9e5a84f01fb020a209a4dccadfdece4
diff --git a/media/mtp/PosixAsyncIO.cpp b/media/mtp/PosixAsyncIO.cpp
index e67c568..72c07cc 100644
--- a/media/mtp/PosixAsyncIO.cpp
+++ b/media/mtp/PosixAsyncIO.cpp
@@ -15,42 +15,109 @@
*/
#include <android-base/logging.h>
-#include <condition_variable>
#include <memory>
-#include <mutex>
+#include <pthread.h>
#include <queue>
+#include <thread>
#include <unistd.h>
#include "PosixAsyncIO.h"
namespace {
-void read_func(struct aiocb *aiocbp) {
- aiocbp->ret = TEMP_FAILURE_RETRY(pread(aiocbp->aio_fildes,
- aiocbp->aio_buf, aiocbp->aio_nbytes, aiocbp->aio_offset));
- if (aiocbp->ret == -1) aiocbp->error = errno;
+std::thread gWorkerThread;
+std::deque<struct aiocb*> gWorkQueue;
+bool gSuspended = true;
+int gAiocbRefcount = 0;
+std::mutex gLock;
+std::condition_variable gWait;
+
+void work_func(void *) {
+ pthread_setname_np(pthread_self(), "AsyncIO work");
+ while (true) {
+ struct aiocb *aiocbp;
+ {
+ std::unique_lock<std::mutex> lk(gLock);
+ gWait.wait(lk, []{return gWorkQueue.size() > 0 || gSuspended;});
+ if (gSuspended)
+ return;
+ aiocbp = gWorkQueue.back();
+ gWorkQueue.pop_back();
+ }
+ CHECK(aiocbp->queued);
+ int ret;
+ if (aiocbp->read) {
+ ret = TEMP_FAILURE_RETRY(pread(aiocbp->aio_fildes,
+ aiocbp->aio_buf, aiocbp->aio_nbytes, aiocbp->aio_offset));
+ } else {
+ ret = TEMP_FAILURE_RETRY(pwrite(aiocbp->aio_fildes,
+ aiocbp->aio_buf, aiocbp->aio_nbytes, aiocbp->aio_offset));
+ }
+ {
+ std::unique_lock<std::mutex> lk(aiocbp->lock);
+ aiocbp->ret = ret;
+ if (aiocbp->ret == -1) {
+ aiocbp->error = errno;
+ }
+ aiocbp->queued = false;
+ }
+ aiocbp->cv.notify_all();
+ }
}
-void write_func(struct aiocb *aiocbp) {
- aiocbp->ret = TEMP_FAILURE_RETRY(pwrite(aiocbp->aio_fildes,
- aiocbp->aio_buf, aiocbp->aio_nbytes, aiocbp->aio_offset));
- if (aiocbp->ret == -1) aiocbp->error = errno;
+int aio_add(struct aiocb *aiocbp) {
+ CHECK(!aiocbp->queued);
+ aiocbp->queued = true;
+ {
+ std::unique_lock<std::mutex> lk(gLock);
+ gWorkQueue.push_front(aiocbp);
+ }
+ gWait.notify_one();
+ return 0;
}
} // end anonymous namespace
+aiocb::aiocb() {
+ this->ret = 0;
+ this->queued = false;
+ {
+ std::unique_lock<std::mutex> lk(gLock);
+ if (gAiocbRefcount == 0) {
+ CHECK(gWorkQueue.size() == 0);
+ CHECK(gSuspended);
+ gSuspended = false;
+ gWorkerThread = std::thread(work_func, nullptr);
+ }
+ gAiocbRefcount++;
+ }
+}
+
aiocb::~aiocb() {
- CHECK(!thread.joinable());
+ CHECK(!this->queued);
+ {
+ std::unique_lock<std::mutex> lk(gLock);
+ CHECK(!gSuspended);
+ if (gAiocbRefcount == 1) {
+ CHECK(gWorkQueue.size() == 0);
+ gSuspended = true;
+ lk.unlock();
+ gWait.notify_one();
+ gWorkerThread.join();
+ lk.lock();
+ }
+ gAiocbRefcount--;
+ }
}
int aio_read(struct aiocb *aiocbp) {
- aiocbp->thread = std::thread(read_func, aiocbp);
- return 0;
+ aiocbp->read = true;
+ return aio_add(aiocbp);
}
int aio_write(struct aiocb *aiocbp) {
- aiocbp->thread = std::thread(write_func, aiocbp);
- return 0;
+ aiocbp->read = false;
+ return aio_add(aiocbp);
}
int aio_error(const struct aiocb *aiocbp) {
@@ -64,7 +131,10 @@
int aio_suspend(struct aiocb *aiocbp[], int n,
const struct timespec *) {
for (int i = 0; i < n; i++) {
- aiocbp[i]->thread.join();
+ {
+ std::unique_lock<std::mutex> lk(aiocbp[i]->lock);
+ aiocbp[i]->cv.wait(lk, [aiocbp, i]{return !aiocbp[i]->queued;});
+ }
}
return 0;
}
diff --git a/media/mtp/PosixAsyncIO.h b/media/mtp/PosixAsyncIO.h
index 590aaef..2bb5735 100644
--- a/media/mtp/PosixAsyncIO.h
+++ b/media/mtp/PosixAsyncIO.h
@@ -17,10 +17,11 @@
#ifndef _POSIXASYNCIO_H
#define _POSIXASYNCIO_H
+#include <condition_variable>
+#include <mutex>
#include <sys/cdefs.h>
#include <sys/types.h>
#include <time.h>
-#include <thread>
#include <unistd.h>
/**
@@ -35,10 +36,15 @@
size_t aio_nbytes;
// Used internally
- std::thread thread;
+ bool read;
+ bool queued;
ssize_t ret;
int error;
+ std::mutex lock;
+ std::condition_variable cv;
+
+ aiocb();
~aiocb();
};