/********************************************************************
* 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 2014-2020 BlackLynx Inc.                                *
* Unpublished -- all rights reserved under the copyright laws       *
* of the United States.                                             *
********************************************************************/

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

    
BlackLynx::SearchLynx::ExecDriver::ExecDriver(
    const std::string& inputFile,
    uint32_t numReaderThreads)
    : r_inputFile(inputFile),
      m_numReaderThreads(numReaderThreads),
      DEFAULT_CHUNK_SIZE(SearchLynxConfiguration::getInstance()->getChunkSizeInKb() * 1024),
      EXTRA_BYTES_TO_READ(RAW_TEXT_EXTRA_BYTES)
{
}
    
BlackLynx::SearchLynx::ExecDriver::~ExecDriver()
{
}

void BlackLynx::SearchLynx::ExecDriver::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::ExecDriver::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::ExecDriver::waitUntilComplete()
{
    // Join threads in normal order
    
    // Join setup thread
    if (p_setupThread != nullptr)
    {
        if (p_setupThread->joinable())
        {
            p_setupThread->join();
        }
    }
    delete p_setupThread;
    p_setupThread = nullptr;
    
    // Join reader threads
    for (auto readerThread : m_readerThreads)
    {
        if (readerThread->joinable())
        {
            readerThread->join();
        }
        delete readerThread;
    }
    m_readerThreads.clear();
    
    // Join cleanup thread
    if (p_cleanupThread != nullptr)
    {
        if (p_cleanupThread->joinable())
        {
            p_cleanupThread->join();
        }
    }
    delete p_cleanupThread;
    p_cleanupThread = nullptr;
}

void BlackLynx::SearchLynx::ExecDriver::setupThreadBody(
    chunk_queue_t* outQueue, 
    uint32_t numTerminators)
{
    uint64_t chunkIdAssigner = 1;
    uint64_t fileOffset = 0;
    uint32_t chunkSize = 0;
    uint32_t extraSize = 0;
    uint64_t fileSize = FileUtils::getFileSize(r_inputFile);
        
    while (fileOffset < fileSize)
    {
        // Determine chunk size
        if ((fileSize - fileOffset) >= DEFAULT_CHUNK_SIZE)
        {
            chunkSize = DEFAULT_CHUNK_SIZE;
        }
        else
        {
            chunkSize = fileSize - fileOffset;
        }

        // Determine extra size
        if ((fileOffset + chunkSize + EXTRA_BYTES_TO_READ) <= fileSize)
        {
            extraSize = EXTRA_BYTES_TO_READ;
        }
        else
        {
            extraSize = fileSize - (fileOffset + chunkSize);
        }

//        bool firstChunkInFile = (fileOffset == 0);
//        bool lastChunkInFile = ((fileOffset + chunkSize) == fileSize);

        // Make chunk to process
        auto chunk = new Chunk(r_inputFile, chunkIdAssigner, chunkSize, 
                               extraSize, fileOffset);
        
        // Increment the chunk ID by 2 so that we match the record version 
        // of the execution driver
        chunkIdAssigner += 2;

        fileOffset += chunkSize;
        
        // Send the chunk to the reader
        outQueue->push_back(chunk);
    }
    
    // Send the specified number of terminators
    for (uint32_t idx = 0; idx < numTerminators; ++idx)
    {
        outQueue->push_back(nullptr);
    }
}
    
void BlackLynx::SearchLynx::ExecDriver::readerThreadBody(
    chunk_queue_t* inQueue,
    chunk_queue_t* outQueue)
{
    Chunk* chunk = nullptr;
    
    std::ifstream inFileStream;
    std::string prevFilename("");
    
    while (true)
    {
        chunk = inQueue->pop_front();
        
        // Check for terminator
        if (chunk == nullptr)
        {
            break;
        }
        
          // Open file for reading if the previous filename is different than the
        // current filename 
        if (chunk->m_filename != prevFilename)
        {
            if (inFileStream.is_open())
            {
                inFileStream.close();
            }
            
            inFileStream.open(chunk->m_filename, std::ios::in | std::ios::binary);
            if (! inFileStream)
            {
                BlnxLog(error) << "Unable to open file: " << chunk->m_filename;
                break;
            }
            
            // Tell the kernel this file will be read sequentially
            int rc = posix_fadvise(FileUtils::getFdHack(inFileStream), 0, 0, 
                                   POSIX_FADV_SEQUENTIAL);
            if (rc)
            {
                BlnxLog(error) << "posix_fadvise() returned: "  << rc;
            }
        }
        
        // Store filename to compare against next chunk
        prevFilename = chunk->m_filename;
        
        
        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;
        
        // Grab a buffer to hold the chunk data
        chunk->allocateBuffer();
        
        inFileStream.seekg(chunk->m_fileOffset);
        inFileStream.read(chunk->p_chunkData, 
                          chunk->m_chunkSize + chunk->m_extraSize);
            
        if (inFileStream.gcount() != (chunk->m_chunkSize + chunk->m_extraSize))
        {
            // Log a runtime error 
            std::ostringstream msg;
            msg << "Unable to read " 
                << (chunk->m_chunkSize + chunk->m_extraSize) 
                << " bytes at offset "  << chunk->m_fileOffset << " from " 
                << chunk->m_filename << " (chunk ID: " << chunk->m_chunkId 
                << ")";
            BlnxLog(error) << msg.str();
            continue;
        }
        
        outQueue->push_back(chunk);
    }
    
    inFileStream.close();
    
    // Send terminator to clean up thread. Note the cleanup thread counts the
    // number of terminators and does not terminate until it receives all 
    // expected terminators.
    m_cleanupQueue.push_back(nullptr);
}
    
void BlackLynx::SearchLynx::ExecDriver::cleanupThreadBody(
    chunk_queue_t* inQueue)
{
    uint32_t terminatorCount = 0;
    Chunk* chunk = nullptr;
    auto bufAllocator = BufferAllocator::getInstance();

    while (true)
    {
        chunk = inQueue->pop_front();
        
        // Check for terminator
        if (chunk == nullptr)
        {
            // Check if we've received one terminator from each reader 
            if (++terminatorCount == m_numReaderThreads)
            {
                break;
            }
            else
            {
                continue;
            }
        }
        
        bufAllocator->release(chunk->p_chunkData);
        delete chunk;
    }
}