mtp: Update PosixAsyncIO
Signed-off-by: sekaiacg <sekaiacg@gmail.com>
Change-Id: I833f39fa7c1bfbc280c480b001c20f903aa8a47d
diff --git a/mtp/ffs/PosixAsyncIO.cpp b/mtp/ffs/PosixAsyncIO.cpp
index 435000a..8205e3b 100644
--- a/mtp/ffs/PosixAsyncIO.cpp
+++ b/mtp/ffs/PosixAsyncIO.cpp
@@ -5,7 +5,7 @@
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,62 +15,132 @@
*/
#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(pread64(aiocbp->aio_fildes,
+ aiocbp->aio_buf, aiocbp->aio_nbytes, aiocbp->aio_offset));
+ } else {
+ ret = TEMP_FAILURE_RETRY(pwrite64(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) {
- return aiocbp->error;
+ return aiocbp->error;
}
ssize_t aio_return(struct aiocb *aiocbp) {
- return aiocbp->ret;
+ return aiocbp->ret;
}
int aio_suspend(struct aiocb *aiocbp[], int n,
- const struct timespec *) {
- for (int i = 0; i < n; i++) {
- aiocbp[i]->thread.join();
- }
- return 0;
+ const struct timespec *) {
+ for (int i = 0; i < n; i++) {
+ {
+ std::unique_lock<std::mutex> lk(aiocbp[i]->lock);
+ aiocbp[i]->cv.wait(lk, [aiocbp, i]{return !aiocbp[i]->queued;});
+ }
+ }
+ return 0;
}
-void aio_prepare(struct aiocb *aiocbp, void* buf, size_t count, off_t offset) {
- aiocbp->aio_buf = buf;
- aiocbp->aio_offset = offset;
- aiocbp->aio_nbytes = count;
+void aio_prepare(struct aiocb *aiocbp, void* buf, size_t count, off64_t offset) {
+ aiocbp->aio_buf = buf;
+ aiocbp->aio_offset = offset;
+ aiocbp->aio_nbytes = count;
}