1
\$\begingroup\$

This is a simple implementation of IO completion ports in user mode for Linux.

In Windows IO completion ports work like this: There is a global queue for issuing IO requests. Some worker kernel threads pop from this queue and operate on each request. There is a queue associated with each port you create. When you associate a file handle to a port and perform IO on it it is associated with the same queue. When a worker thread in the Kernel finishes your request it pushes the result into your queue. Then you pop from the port queue using GetQueuedCompletionStatus.

So to sum up: You make a request and bind your queue with it and submit it to a kernel queue and worker threads will do the IO for you then send you the result in your queue.

However in Linux the system doesn't have those worker threads. So to simulate IO completion ports I had to use worker threads in user mode and queues in user mode with epoll.

It is implemented like this: There is a global thread queue that you push your requests to and workers pop from. There is also a global epoll file descriptor. You submit the request either using epoll (sockets) or directly pushing into the queue for files that don't support epoll. When you create a port you are actually creating a queue to pop from. When you bind a file to the port you bind it to the queue it has. When you issue read or write request you submit a request as described above. When you wait on the port handle you are popping from the queue.

The implementation is at a very early stage and I tested it only with files on disk.

I left out things like handling errors or joining the IO threads for now as I am more concerned with the design at first. Also I used a C-style approach in dealing with structures that could be avoided with C++ constructors and methods.

And this is my code:

#include <queue>
#include <condition_variable>
#include <unistd.h>
#include <sys/epoll.h>
#include <thread>
#include <iostream>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

using namespace std;

template <class T>
class SafeQueue
{
public:

    SafeQueue()
    {}
    void Push(const T& val);
    void Push(T&& val);
    void Pop(T& val);
    T Pop();
    void Clear();
    bool IsEmpty();
    size_t Size();
private :
    std::queue<T> q;
    std::mutex mtx;
    std::condition_variable cv; 
};

template<class T>
inline void SafeQueue<T>::Push(const T& val)
{
    std::unique_lock<std::mutex> lock(mtx);
    q.emplace(val);
    cv.notify_one();
}

template<class T>
inline void SafeQueue<T>::Push(T&& val)
{
    std::unique_lock<std::mutex> lock(mtx);
    q.emplace(std::move(val));
    cv.notify_one();
}

template<class T>
inline void SafeQueue<T>::Pop(T& val)
{
    std::unique_lock<std::mutex> lock(mtx);
    while (q.empty())
    {
        cv.wait(lock);
    }
    val = q.front();
    q.pop();
}

template <class T>
inline T SafeQueue<T>::Pop()
{
    T val;
    std::unique_lock<std::mutex> lock(mtx);
    while (q.empty())
    {
        cv.wait(lock);
    }
    val = q.front();
    q.pop();
    return val;
}

template <class T>
inline void SafeQueue<T>::Clear()
{
    std::unique_lock<std::mutex> lock(mtx);
    while(!q.empty())
        q.pop();
}

template <class T>
inline bool SafeQueue<T>::IsEmpty()
{
    std::unique_lock<std::mutex> lock(mtx);
    return q.empty();
}

template <class T>
inline size_t SafeQueue<T>::Size()
{
    std::unique_lock<std::mutex> lock(mtx);
    return q.size();
}

#define Io_Chunk_Size size_t(1024)

enum IoOpCode
{
    IoWriteCode,
    IoReadCode,
    IoSpecialCode,
};

enum class IoStatusCode
{
    IoStatusOk,
    IoStatusPending,
    IoStatusFailed,
};

struct IoContext
{
    IoStatusCode code;
    unsigned int bytes;
    void *buffer;
    unsigned int key;
};

struct IoOperationStruct
{
    IoOperationStruct()
    {}
    IoOperationStruct(const epoll_event& ev);
    std::shared_ptr<SafeQueue<IoContext*>> rport;
    int fd;
    char *buffer;
    IoOpCode code;
    IoContext *context;
    size_t rest;
    size_t offset;
};

static int io_epoll_instance;

static SafeQueue<IoOperationStruct*> io_requests;

static bool IoStopFlag;

void IoMainThread()
{
    epoll_event evs[50];
    while(!IoStopFlag)
    {
        int num = epoll_wait(io_epoll_instance, evs, 50, 10000);
        if (num < 0)
            return;
        for (int i = 0; i < num; ++i)
        {
            io_requests.Push((IoOperationStruct*)evs[i].data.ptr);
        }
    }
}

bool AddIoToQueue(IoOperationStruct *IoStruct, bool modify = false)
{
    epoll_event ev;
    ev.events = EPOLLONESHOT;
    switch(IoStruct->code)
    {
        case IoWriteCode :
            ev.events |= EPOLLOUT;
            break;
        case IoReadCode :
            ev.events |= EPOLLIN;
            break;
        default:
            break;
    }
    ev.data.ptr = IoStruct;
    bool res = epoll_ctl(io_epoll_instance, modify ? EPOLL_CTL_MOD : EPOLL_CTL_ADD,IoStruct->fd,  &ev) != -1;
    if (!res)
    {
        if (errno == EPERM)
        {
            res = true;
            io_requests.Push(IoStruct);
        }
    }
    if (!res)
        cout << "failed with error : " << strerror(errno) << endl;
    return res;
}

struct HANDLE
{
    shared_ptr<SafeQueue<IoContext*>> q;
    int fd;
    unsigned int key;
};

HANDLE CreateIoCompletionPort()
{
    HANDLE port;
    port.q = make_shared<SafeQueue<IoContext*>>();
    port.fd = 0;
    return port;
}

void CloseHandle(HANDLE Handle)
{
    Handle.q.reset();
    if (Handle.fd)
        close(Handle.fd);
    Handle.fd = 0;
}

HANDLE BindToPort(HANDLE port, int fd, unsigned int key)
{
    HANDLE handle;
    handle.fd = fd;
    handle.q = port.q;
    handle.key = key;
    return handle;
}

bool GetQueuedCompletionStatus(HANDLE port, unsigned int *num_of_bytes, IoContext **ctx, unsigned int *key, unsigned int wait_time)
{
    port.q->Pop(*ctx);
    *num_of_bytes = (*ctx)->bytes;
    *key = (*ctx)->key;
    if (*num_of_bytes)
        (*ctx)->code = IoStatusCode::IoStatusOk;
    else
        (*ctx)->code = IoStatusCode::IoStatusFailed;
    return true;
}

bool WriteFile(HANDLE Handle, const void *data, size_t size, IoContext * Ctx)
{
    if (!data || !size || !Ctx)
        return false;
    IoOperationStruct *IoStruct = new IoOperationStruct();
    IoStruct->fd = Handle.fd;
    IoStruct->rport = Handle.q;
    IoStruct->buffer = (char*)data;
    IoStruct->context = Ctx;
    IoStruct->offset = 0;
    IoStruct->rest = size;
    IoStruct->code = IoOpCode::IoWriteCode;
    IoStruct->context->bytes = 0;
    IoStruct->context->code = IoStatusCode::IoStatusPending;
    return AddIoToQueue(IoStruct);
}

bool ReadFile(HANDLE Handle, void *data, size_t size, IoContext *Ctx)
{
    if (!data || !size || !Ctx)
        return false;
    IoOperationStruct *IoStruct = new IoOperationStruct();
    IoStruct->fd = Handle.fd;
    IoStruct->rport = Handle.q;
    IoStruct->buffer = (char*)data;
    IoStruct->context = Ctx;
    IoStruct->offset = 0;
    IoStruct->rest = size;
    IoStruct->code = IoOpCode::IoReadCode;
    IoStruct->context->bytes = 0;
    IoStruct->context->code = IoStatusCode::IoStatusPending;
    return AddIoToQueue(IoStruct);
}

bool IoHandleWrite(IoOperationStruct & IoStruct)
{
    if (!IoStruct.rest)
        return false;
    size_t to_write = std::min(IoStruct.rest, Io_Chunk_Size);
    int written = write(IoStruct.fd, IoStruct.buffer + IoStruct.offset, to_write);
    if (written < 0)
    {
        if (errno == EWOULDBLOCK || errno == EAGAIN)
            return true;
        return false;
    }
    IoStruct.offset += written;
    IoStruct.rest -= written;
    IoStruct.context->bytes += written;
    if (!IoStruct.rest)
        return false;
    return true;    
}

bool IoHandleRead(IoOperationStruct& IoStruct)
{
    if (!IoStruct.rest)
        return false;
    size_t to_read = std::min(IoStruct.rest, Io_Chunk_Size);
    int was_read = read(IoStruct.fd, IoStruct.buffer + IoStruct.offset, to_read);
    if (was_read < 0)
    {
        if (errno == EWOULDBLOCK || errno == EAGAIN)
            return true;
        return false;
    }
    IoStruct.offset += was_read;
    IoStruct.context->bytes += was_read;
    IoStruct.rest -= was_read;
    if (!IoStruct.rest)
        return false;
    return true;
}

void IoHandleResult(bool result, IoOperationStruct *IoStruct)
{
    if (!result)
    {
        shared_ptr<SafeQueue<IoContext*>> rport = IoStruct->rport;
        IoContext *Ctx = IoStruct->context;
        delete IoStruct;
        rport->Push(Ctx);
    }
    else
    {
        if (!AddIoToQueue(IoStruct, true))
            return IoHandleResult(false, IoStruct);
    }   
}

void IoHandlerThread()
{
    IoOperationStruct *IoStruct;
    while(!IoStopFlag)
    {
        io_requests.Pop(IoStruct);

        if (IoStruct->code == IoOpCode::IoWriteCode)
        {
            bool result = IoHandleWrite(*IoStruct);
            IoHandleResult(result, IoStruct);       
        }
        else if (IoStruct->code == IoOpCode::IoReadCode)
        {
            bool result = IoHandleRead(*IoStruct);
            IoHandleResult(result, IoStruct);
        }
        else if (IoStruct->code == IoOpCode::IoSpecialCode)
        {
            IoHandleResult(false, IoStruct);
        }
    }
}

void start_io_ports()
{
    io_epoll_instance = epoll_create(10);
    if (io_epoll_instance < 0)
        cout << "failed to open epoll with error : " << strerror(errno) << endl;
    IoStopFlag = false;
    std::thread(IoMainThread).detach();
    for (size_t i = 0; i < std::thread::hardware_concurrency(); ++i)
    {
        std::thread(IoHandlerThread).detach();
    }
}

HANDLE port;

std::condition_variable cv;
std::mutex mtx;

void worker_thread()
{
    cout << "id : " << this_thread::get_id() << endl;
    while (1)
    {
        unsigned int bytes;
        unsigned int key;
        IoContext *ctx;
        GetQueuedCompletionStatus(port, &bytes, &ctx, &key, 1000);
        cout << " ======== " << endl;
        cout << "io was performed " << endl;
        cout << "bytes transferred : " << bytes << endl;
        cout << " ========= " << endl;
        delete (char*)ctx->buffer;
        delete ctx;
    }
}

int main()
{
    start_io_ports();
    port = CreateIoCompletionPort();
    int fd = open("/sdcard/Ayat/test.txt", O_WRONLY | O_CREAT);
    if (fd < 0)
    {
        cout << "error : " << strerror(errno) << endl;
        return 0;
    }
    HANDLE file = BindToPort(port, fd, 5);
    std::thread(worker_thread).detach();
    sleep(1);
    while (1)
    {
        cout << "submitting a job" << endl;
        char *buff = new char[10010];
        IoContext *ctx = new IoContext();
        ctx->buffer = buff;
        size_t to_write = 0;
        for (size_t i = 0; i < 10000; i += 9)
        {
            memcpy(buff + i, "ttttttttt", 9);
            to_write += 9;
        }
        if(!WriteFile(file, buff,to_write,  ctx))
            cout << "failed !" << endl;
        sleep(2);
    }
    CloseHandle(file);
    sleep(10000000000000000);
}
\$\endgroup\$
0

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.