/********************************************************************
* 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 <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)
{
}
    
BlackLynx::SearchLynx::URingExecDriver::~URingExecDriver()
{
}

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(error) << "URING READER";
    
    const uint32_t QUEUE_DEPTH = 32;
    const uint32_t IO_SIZE = 4194304;
    const uint32_t MIN_SIZE_ALIGN = 512;
    // Setup IO context
    
    // Open the file
    int fd = -1;
//    int fd = ::open(r_inputFile.c_str(), O_RDONLY | O_DIRECT);
//    if (fd < 0)
//    {
//        BlnxLog(error) << "Unable to open file: "  << r_inputFile;
//        return;
//    }
    
    Chunk* chunk = nullptr;
    std::string prevFilename("");
    
//    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();
        
        // Determine how many IOs we need
//        std::string test(chunk->p_chunkData, 30);
//        BlnxLog(debug) << "CHUNK " << chunk->m_chunkId << " DATA: " << test;
        
//        output.write(chunk->p_chunkData, chunk->m_chunkSize);

        outQueue->push_back(chunk);
    }
    
//    output.close();
    
    // Release file descriptor
    ::close(fd);
    
    // Destroy the IO context
    io_destroy(context);
    
    // Send terminator to clean up thread
    m_cleanupQueue.push_back(nullptr);
}