<commit_before>#include "hecl/ClientProcess.hpp"
#include "hecl/Database.hpp"
#include "athena/FileReader.hpp"
#include "BlenderConnection.hpp"

#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#else
#include <sys/wait.h>
#endif

namespace hecl
{
static logvisor::Module Log("hecl::ClientProcess");

static int GetCPUCount()
{
#if _WIN32
    SYSTEM_INFO sysinfo;
    GetSystemInfo(&sysinfo);
    return sysinfo.dwNumberOfProcessors;
#else
    return sysconf(_SC_NPROCESSORS_ONLN);
#endif
}

void ClientProcess::BufferTransaction::run(BlenderToken& btok)
{
    athena::io::FileReader r(m_path.getAbsolutePath(), 32 * 1024, false);
    if (r.hasError())
    {
        Log.report(logvisor::Fatal, _S("unable to background-buffer '%s'"),
                   m_path.getAbsolutePath().c_str());
        return;
    }
    if (m_offset)
        r.seek(m_offset, athena::Begin);
    r.readBytesToBuf(m_targetBuf, m_maxLen);
    m_complete = true;
}

void ClientProcess::CookTransaction::run(BlenderToken& btok)
{
    m_returnResult = m_parent.syncCook(m_path, m_dataSpec, btok);
    m_complete = true;
}

ClientProcess::Worker::Worker(ClientProcess& proc)
: m_proc(proc)
{
    m_thr = std::thread(std::bind(&Worker::proc, this));
}

void ClientProcess::Worker::proc()
{
    while (m_proc.m_running)
    {
        std::unique_lock<std::mutex> lk(m_proc.m_mutex);
        if (m_proc.m_pendingQueue.size())
        {
            std::unique_ptr<Transaction> trans = std::move(m_proc.m_pendingQueue.front());
            m_proc.m_pendingQueue.pop_front();
            lk.unlock();
            trans->run(m_blendTok);
            lk.lock();
            m_proc.m_completedQueue.push_back(std::move(trans));
        }
        m_proc.m_cv.wait(lk);
    }
}

ClientProcess::ClientProcess(int verbosityLevel)
: m_verbosity(verbosityLevel)
{
    int cpuCount = GetCPUCount();
    m_workers.reserve(cpuCount);
    for (int i=0 ; i<cpuCount ; ++i)
        m_workers.emplace_back(*this);
}

const ClientProcess::BufferTransaction*
ClientProcess::addBufferTransaction(const ProjectPath& path, void* target,
                                    size_t maxLen, size_t offset)
{
    std::unique_lock<std::mutex> lk(m_mutex);
    BufferTransaction* ret = new BufferTransaction(*this, path, target, maxLen, offset);<commit_msg>Client process worker loop tweak<commit_after>#include "hecl/ClientProcess.hpp"
#include "hecl/Database.hpp"
#include "athena/FileReader.hpp"
#include "BlenderConnection.hpp"

#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#else
#include <sys/wait.h>
#endif

namespace hecl
{
static logvisor::Module Log("hecl::ClientProcess");

static int GetCPUCount()
{
#if _WIN32
    SYSTEM_INFO sysinfo;
    GetSystemInfo(&sysinfo);
    return sysinfo.dwNumberOfProcessors;
#else
    return sysconf(_SC_NPROCESSORS_ONLN);
#endif
}

void ClientProcess::BufferTransaction::run(BlenderToken& btok)
{
    athena::io::FileReader r(m_path.getAbsolutePath(), 32 * 1024, false);
    if (r.hasError())
    {
        Log.report(logvisor::Fatal, _S("unable to background-buffer '%s'"),
                   m_path.getAbsolutePath().c_str());
        return;
    }
    if (m_offset)
        r.seek(m_offset, athena::Begin);
    r.readBytesToBuf(m_targetBuf, m_maxLen);
    m_complete = true;
}

void ClientProcess::CookTransaction::run(BlenderToken& btok)
{
    m_returnResult = m_parent.syncCook(m_path, m_dataSpec, btok);
    m_complete = true;
}

ClientProcess::Worker::Worker(ClientProcess& proc)
: m_proc(proc)
{
    m_thr = std::thread(std::bind(&Worker::proc, this));
}

void ClientProcess::Worker::proc()
{
    while (m_proc.m_running)
    {
        std::unique_lock<std::mutex> lk(m_proc.m_mutex);
        while (m_proc.m_pendingQueue.size())
        {
            std::unique_ptr<Transaction> trans = std::move(m_proc.m_pendingQueue.front());
            m_proc.m_pendingQueue.pop_front();
            lk.unlock();
            trans->run(m_blendTok);
            lk.lock();
            m_proc.m_completedQueue.push_back(std::move(trans));
        }
        m_proc.m_cv.wait(lk);
    }
}

ClientProcess::ClientProcess(int verbosityLevel)
: m_verbosity(verbosityLevel)
{
    int cpuCount = GetCPUCount();
    m_workers.reserve(cpuCount);
    for (int i=0 ; i<cpuCount ; ++i)
        m_workers.emplace_back(*this);
}

const ClientProcess::BufferTransaction*
ClientProcess::addBufferTransaction(const ProjectPath& path, void* target,
                                    size_t maxLen, size_t offset)
{
    std::unique_lock<std::mutex> lk(m_mutex);
    BufferTransaction* ret = new BufferTransaction(*this, path, target, maxLen, offset);
{'fraction_non_alphanumeric': 0.13485566566135287, 'fraction_numerical': 0.005601034037052995, 'mean_word_length': 3.6946410515672397, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5869773', 'n_tokens_mistral': 1618, 'n_tokens_neox': 1578, 'n_words': 259}