/********************************************************************
* This software is "commercial computer software" as defined in the *
* Federal Acquisition Regulations and is subject to BlackLynx       *
* Inc's standard End User License Agreement.                        *
*                                                                   *
* CONFIDENTIAL - All source code is the propriety and confidential  *
* information of BlackLynx Inc.                                     *
*                                                                   *
* Copyright 2022 BlackLynx Inc.                                     *
* Unpublished -- all rights reserved under the copyright laws       *
* of the United States.                                             *
********************************************************************/
#include <cerrno>
#include <sstream>
#include <stdexcept>

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include "URingExecDriver.h"
#include "Utilities/ChunkConstants.h"
#include "Utilities/FileUtils.h"
#include "Utilities/SearchLynxConfiguration.h"
#include "Utilities/BufferAllocator.h"

    
BlackLynx::SearchLynx::URingExecDriver::URingExecDriver(
    const std::string& inputFile,
    uint32_t numReaderThreads)
    : ExecDriver(inputFile, numReaderThreads),
      m_ioCounter(0)
{
    // Init the ring
    int rc = io_uring_queue_init(QUEUE_DEPTH, &m_ring, 0);
    if (rc < 0) 
    {
        std::ostringstream msg;
        msg << "FATAL - io_uring_queue_init returned: " << rc;
        throw std::runtime_error(msg.str());
    }
    
//    BlnxLog(info) << "RING SPACE: " << io_uring_sq_space_left(&m_ring);
}
    
BlackLynx::SearchLynx::URingExecDriver::~URingExecDriver()
{
    // Clean up the ring
    io_uring_queue_exit(&m_ring);
}

void BlackLynx::SearchLynx::URingExecDriver::start()
{
    // Start threads in reverse order staring with cleanup thread    
    p_cleanupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody, 
                                      this,
                                      &m_cleanupQueue);
    
    // Start specified number of reader threads
    for (uint32_t idx = 0; idx < m_numReaderThreads; ++idx)
    {
        std::thread* thisThread = 
                new std::thread(&BlackLynx::SearchLynx::URingExecDriver::readerThreadBody, 
                                this,
                                &m_readerQueue,
                                &m_cleanupQueue);
        m_readerThreads.push_back(thisThread);
    }
    
    // Start setup thread body
    p_setupThread = new std::thread(&BlackLynx::SearchLynx::ExecDriver::setupThreadBody, 
                                    this,
                                    &m_readerQueue,
                                    m_numReaderThreads);
}

void BlackLynx::SearchLynx::URingExecDriver::readerThreadBody(
    chunk_queue_t* inQueue,
    chunk_queue_t* outQueue)
{
    BlnxLog(info) << "URING READER starting";
    
    const uint32_t MIN_SIZE_ALIGN = 512;
    // Setup IO context
    
    int fd = -1;
    struct io_uring_sqe* sqe = nullptr;
    Chunk* chunk = nullptr;
    std::string prevFilename("");
    int rc = 0;
    
//    std::ofstream output("chunk.bin");
    
    while (true)
    {
        chunk = inQueue->pop_front();
        
        // Check for terminator
        if (chunk == nullptr)
        {
            break;
        }
        
        if (prevFilename != chunk->m_filename)
        {
            // Open the file
            fd = ::open(r_inputFile.c_str(), O_RDONLY | O_DIRECT);
            if (fd < 0)
            {
                BlnxLog(error) << "Unable to open file: "  << r_inputFile;
                return;
            }
        }
        
        // Grab a buffer to hold the chunk data
        chunk->allocateBuffer();
        
        // Wait until a submission queue entry is ready
        sqe = nullptr;
        while (sqe == nullptr)
        {
            sqe = io_uring_get_sqe(&m_ring);
            if (! sqe) 
            {
                waitForCompletion(outQueue);
            }
        }
               
        uint32_t totalSize = chunk->m_chunkSize + chunk->m_extraSize;
        uint32_t leftOverSize = (totalSize & (MIN_SIZE_ALIGN - 1));
        uint32_t readSize = totalSize - leftOverSize;
        
        // Prep the read request and set the chunk object as its associated 
        // data
        io_uring_prep_read(sqe, fd, chunk->p_chunkData, readSize, 
                           chunk->m_fileOffset);
        
        io_uring_sqe_set_data(sqe, reinterpret_cast<void*>(chunk));
        
        BlnxLog(debug) << "Reader Thread reading:"
                       << "   Filename: " << chunk->m_filename
                       << "   ChunkID: " << chunk->m_chunkId
                       << "   Chunk Size: " << chunk->m_chunkSize
                       << "   Extra Size: " << chunk->m_extraSize
                       << "   File Offset: " << chunk->m_fileOffset
                       << "   IOPs in flight: " << m_ioCounter
                       << "   SPACE LEFT: " << io_uring_sq_space_left(&m_ring)
                       << "   READY: " <<  io_uring_cq_ready(&m_ring);
        
        // Update the counter and submit the request
        ++m_ioCounter;
        rc = io_uring_submit(&m_ring);
        if (rc < 0)
        {
            BlnxLog(error) << "io_uring_submit returned: " << rc;
        }
        
        // NOTE: when using O_DIRECT for high performance the read size *must* 
        // be 512 byte aligned. Manually read any unaligned bytes, which should 
        // only happen for the last "tail" chunk.
        if (leftOverSize > 0)
        {
            BlnxLog(info) << "CHUNK: " << chunk->m_chunkId << " -- " 
                          << leftOverSize << " unaligned bytes to read";
            
            std::ifstream inFileStream(chunk->m_filename, std::ios::in | std::ios::binary);
            if (! inFileStream)
            {
                BlnxLog(error) << "Unable to open file: " << chunk->m_filename;
                break;
            }
            
            inFileStream.seekg(chunk->m_fileOffset + (totalSize - leftOverSize));
            inFileStream.read(chunk->p_chunkData + (totalSize - leftOverSize), 
                              leftOverSize);
                        
            inFileStream.close();
        }
        
        if (m_ioCounter >= QUEUE_DEPTH)
        {
            waitForCompletion(outQueue);    
        }
    }
    
    // Wait for any lingering IO completions
    while (m_ioCounter > 0)
    {
        waitForCompletion(outQueue);
    }
    
//    output.close();
    
    // Release file descriptor
    ::close(fd);
    
    // Send terminator to clean up thread
    m_cleanupQueue.push_back(nullptr);
}

void BlackLynx::SearchLynx::URingExecDriver::waitForCompletion(
    chunk_queue_t* outQueue)
{
    int rc = 0;
    struct io_uring_cqe *cqe = nullptr;
    bool gotOne = false;
    Chunk* chunk = nullptr;
    
    while (m_ioCounter > 0)
    {
        // Wait for at least one completion but handle any that are ready
        if (! gotOne) 
        {
            rc = io_uring_wait_cqe(&m_ring, &cqe);
            gotOne = true;
        } 
        else 
        {
            rc = io_uring_peek_cqe(&m_ring, &cqe);
            if (rc == -EAGAIN) 
            {
                cqe = nullptr;
                rc = 0;
            }
        }
        
        // Break out if no cqe was present
        if (cqe == nullptr)
        {
            break;
        }
        
        // Grab the data (chunk) associated with this request and then pass it
        // along
        chunk = reinterpret_cast<Chunk*>(io_uring_cqe_get_data(cqe));
        
        if (cqe->res < 0)
        {
            BlnxLog(error) << "Chunk ID: " << chunk->m_chunkId 
                           << " -- completion returned: " << cqe->res;
        }
        
        BlnxLog(debug) << "Got completion for chunk ID: " << chunk->m_chunkId;
        
        outQueue->push_back(chunk);
        --m_ioCounter;
        io_uring_cqe_seen(&m_ring, cqe);
    }
}