Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit febd7fe9 authored by Jerry Zhang's avatar Jerry Zhang
Browse files

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
parent 4d9abd93

media/mtp/PosixAsyncIO.cpp

+86 −16
Original line number Diff line number Diff line
@@ -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,

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));

    if (aiocbp->ret == -1) aiocbp->error = errno;

        }

        {

            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 @@ ssize_t aio_return(struct aiocb *aiocbp) { 
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;

}

media/mtp/PosixAsyncIO.h

+8 −2
Original line number Diff line number Diff line
@@ -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 @@ struct aiocb { 
    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();

};