<commit_before>//-----------------------------------
// Copyright <NAME> 2013-2017
//-----------------------------------

#include "BpfTrace.h"
#include "LinuxUtils.h"
#include "CoreApp.h"
#include "Capture.h"
#include "OrbitModule.h"
#include "OrbitProcess.h"
#include "Params.h"
#include "Utils.h"
#include <fstream>
#include <sstream>

#if __linux__
#include "BpfTraceVisitor.h"
#include "LinuxPerfEvent.h"
#include "LinuxPerfEventProcessor.h"
#include "LinuxPerfRingBuffer.h"
#include "LinuxPerfUtils.h"
#include "OrbitFunction.h"
#include "Pdb.h"
#include <linux/perf_event.h>
#include <map>
#include <vector>

using namespace LinuxPerfUtils;
#endif

//-----------------------------------------------------------------------------
BpfTrace::BpfTrace(Callback a_Callback)
{
    m_Callback = a_Callback ? a_Callback : [this](const std::string& a_Buffer)
    {
        if( GParams.m_BpftraceCallstacks )
            CommandCallbackWithCallstacks(a_Buffer);
        else
            CommandCallback(a_Buffer);
    };

    m_ScriptFileName = ws2s(Path::GetBasePath()) + "orbit.bt";
}

//-----------------------------------------------------------------------------
void BpfTrace::Start()
{
#if __linux__
    m_ExitRequested = false;
    m_TimerStacks.clear();
    if( !WriteBpfScript() )
        return;

    m_BpfCommand = std::string("bpftrace ") + m_ScriptFileName;
    if ( GParams.m_UseBpftrace || GParams.m_BpftraceCallstacks ) {    
        m_Thread = std::make_shared<std::thread>
            ( &LinuxUtils::StreamCommandOutput
            , m_BpfCommand.c_str()
            , m_Callback
            , &m_ExitRequested );
    }
    else
    {
        m_Thread = std::make_shared<std::thread>
            ( BpfTrace::RunPerfEventOpen
            , &m_ExitRequested );
    }
    
    m_Thread->detach();
#endif
}

//-----------------------------------------------------------------------------
void BpfTrace::Stop()
{
    m_ExitRequested = true;
}

//-----------------------------------------------------------------------------
std::string BpfTrace::GetBpfScript()
{
    if (!m_Script.empty())
    {
        return m_Script;
    }

    std::stringstream ss;

    for (Function *func : Capture::GTargetProcess->GetFunctions())
    {
        if (func->IsSelected())
        {
            uint64_t virtual_address = (uint64_t)func->GetVirtualAddress();
            Capture::GSelectedFunctionsMap[func->m_Address] = func;

            if( GParams.m_BpftraceCallstacks )
            {
                ss << "   uprobe:" << func->m_Probe << R"({ printf("b )" << std::to_string(virtual_address) << R"( %u %lld\n%s\n\nd\n\n", tid, nsecs, ustack(perf)); })" << std::endl;
            }
            else
            {
                ss << "   uprobe:" << func->m_Probe << R"({ printf("b )" << std::to_string(virtual_address) << R"( %u %lld\n", tid, nsecs); })" << std::endl;
            }

            ss << "uretprobe:" << func->m_Probe << R"({ printf("e )" << std::to_string(virtual_address) << R"( %u %lld\n", tid, nsecs); })" << std::endl;
        }
    }

    return ss.str();
}

//-----------------------------------------------------------------------------
bool BpfTrace::WriteBpfScript()
{
    std::string script = GetBpfScript();
    if (script.empty())
        return false;

    std::ofstream outFile;
    outFile.open(m_ScriptFileName);
    if (outFile.fail())
        return false;

    outFile << script;
    outFile.close();
    return true;
}

//-----------------------------------------------------------------------------
uint64_t BpfTrace::ProcessString(const std::string& a_String)
{
    auto hash = StringHash(a_String);
    if (m_StringMap.find(hash) == m_StringMap.end())
    {
        m_StringMap[hash] = a_String;
    }

    return hash;
}

//-----------------------------------------------------------------------------
void BpfTrace::CommandCallback(const std::string& a_Line)
{
    auto tokens = Tokenize(a_Line);
    const std::string& mode             = tokens[0];
    const std::string& functionAddress  = tokens[1];
    const std::string& threadName       = tokens[2];
    const std::string& timestamp        = tokens[3];

    bool isBegin = mode == "b";
    bool isEnd   = !isBegin;

    if (isBegin)
    {
        Timer timer;
        timer.m_TID = atoi(threadName.c_str());
        uint64_t nanos = std::stoull(timestamp);
        timer.m_Start = nanos;
        timer.m_Depth = (uint8_t)m_TimerStacks[threadName].size();
        timer.m_FunctionAddress = std::stoull(functionAddress);
        m_TimerStacks[threadName].push_back(timer);
    }

    if (isEnd)
    {
        std::vector<Timer>& timers = m_TimerStacks[threadName];
        if (timers.size())
        {
            Timer& timer = timers.back();
            uint64_t nanos = std::stoull(timestamp);
            timer.m_End = nanos;
            GCoreApp->ProcessTimer(&timer, functionAddress);
            timers.pop_back();
        }
    }
}

//-----------------------------------------------------------------------------
void BpfTrace::CommandCallbackWithCallstacks(const std::string& a_Line)
{
    if (a_Line.empty() || a_Line == "\n")
        return;
    
    auto tokens = Tokenize(a_Line);
    
    const std::string& mode = tokens[0];
    bool isBegin = mode == "b";
    bool isEnd   = mode == "e";

    bool isStackLine = StartsWith(a_Line, "\t");
    bool isEndOfStack = a_Line == "d\n";

    if (!isBegin && !isEnd && !isStackLine && !isEndOfStack)
    {  
        if (StartsWith(a_Line, "Lost"))
        {
            PRINT(a_Line.c_str());
            return;
        }

        if (StartsWith(a_Line, "Attaching")) 
            return;

        // if the line does not start with one of the above,
        // we might have a broken line, e.g. due to a small buffer
        PRINT("read unexpected line:%s\nthe buffer might be to small.", a_Line.c_str());
        return;
    }

    if (isStackLine) {
        const std::string& addressStr = LTrim(tokens[0]);
        uint64_t address = std::stoull(addressStr, nullptr, 16);

        std::string function = tokens[1];

        std::string moduleRaw = tokens[2];
        std::string module = Replace(moduleRaw.substr(1), ")\n", "");

        // TODO: this is copy&paste from LinuxPerf.cpp
        std::wstring moduleName = ToLower(Path::GetFileName(s2ws(module)));
        std::shared_ptr<Module> moduleFromName = Capture::GTargetProcess->GetModuleFromName( ws2s(moduleName) );
       
        if( moduleFromName )
        {
            uint64_t new_address = moduleFromName->ValidateAddress(address);
            address = new_address;
        }

        m_CallStack.m_Data.push_back(address);
        if( Capture::GTargetProcess && !Capture::GTargetProcess->HasSymbol(address))
        {
            GCoreApp->AddSymbol(address, module, function);
        }

        return;
    }

    if (isEndOfStack)
    {
        if ( m_CallStack.m_Data.size() ) {
            m_CallStack.m_Depth = (uint32_t)m_CallStack.m_Data.size();
            m_CallStack.m_ThreadId = atoi(m_LastThreadName.c_str());
            std::vector<Timer>& timers = m_TimerStacks[m_LastThreadName];
            if (timers.size())
            {
                Timer& timer = timers.back();
                timer.m_CallstackHash = m_CallStack.Hash();
                GCoreApp->ProcessCallStack( m_CallStack );
            }
        }

        m_CallStack.Clear();
        m_LastThreadName = "";
        return;
    }

    const std::string& functionAddress  = tokens[1];
    const std::string& threadName       = tokens[2];
    const std::string& timestamp        = tokens[3];

    m_LastThreadName = threadName;

    if (isBegin)
    {
        Timer timer;
        timer.m_TID = atoi(threadName.c_str());
        uint64_t nanos = std::stoull(timestamp);
        timer.m_Start = nanos;
        timer.m_Depth = (uint8_t)m_TimerStacks[threadName].size();
        timer.m_FunctionAddress = std::stoull(functionAddress);
        m_TimerStacks[threadName].push_back(timer);
        return;
    }

    if (isEnd)
    {
        std::vector<Timer>& timers = m_TimerStacks[threadName];
        if (timers.size())
        {
            Timer& timer = timers.back();
            uint64_t nanos = std::stoull(timestamp);
            timer.m_End = nanos;
            GCoreApp->ProcessTimer(&timer, functionAddress);
            timers.pop_back();
        }
        return;
    }
}


//-----------------------------------------------------------------------------
void BpfTrace::RunPerfEventOpen(bool* a_ExitRequested)
{
    #if __linux__
    int ROUND_ROBIN_BATCH_SIZE = 5;
    std::vector<uint64_t> fds;
    std::map<Function*, LinuxPerfRingBuffer> uprobe_ring_buffers;
    std::map<Function*, LinuxPerfRingBuffer> uretprobe_ring_buffers;
    
    for (const auto& pair : Capture::GSelectedFunctionsMap)
    {
        // gather function information
        Function* function = pair.second;
        uint64_t offset = function->m_Address;
        uint64_t virtual_address = function->GetVirtualAddress();
        std::string module = ws2s(function->m_Pdb->GetFileName());

        // create uprobe for that function on that PID profiling all cpus
        uint64_t uprobe_fd = 0;
        if (GParams.m_BpftraceCallstacks)
            uprobe_fd = LinuxPerfUtils::uprobe_event_open(
                module.c_str(), offset, Capture::GTargetProcess->GetID(), -1, 
                PERF_SAMPLE_STACK_USER | PERF_SAMPLE_REGS_USER
            );
        else
            uprobe_fd = LinuxPerfUtils::uprobe_event_open(
                module.c_str(), offset, Capture::GTargetProcess->GetID(), -1
            );

        uprobe_ring_buffers.emplace(function, uprobe_fd);
        fds.push_back(uprobe_fd);

        // create uretprobe for that function on that PID profiling all cpus
        uint64_t uretprobe_fd = LinuxPerfUtils::uretprobe_event_open(
            module.c_str(), offset, Capture::GTargetProcess->GetID(), -1
        );
        uretprobe_ring_buffers.emplace(function, uretprobe_fd);
        fds.push_back(uretprobe_fd);

        // start capturing
        LinuxPerfUtils::start_capturing(uprobe_fd);
        LinuxPerfUtils::start_capturing(uretprobe_fd);
    }

    LinuxPerfEventProcessor event_buffer(std::make_unique<BpfTraceVisitor>());

    bool new_events = false;

    while( !(*a_ExitRequested) )
    {
        // Lets sleep a bit, such that we are not constantly reading from the buffers
        // and thus wasting cpu time. 1000 microseconds are still small enought to 
        // not have our buffers overflown and therefore loosing events.
        if (!new_events)
        {
            usleep(10000);
        }

        new_events = false;

        // read from all ring buffers, create events and store them in the event_queue
        // TODO: implement a better scheduling strategy.
        for (auto& pair : uprobe_ring_buffers)
        {
            auto& ring_buffer = pair.second;
            int i = 0;
            // read everything that is new
            while (ring_buffer.HasNewData() && i < ROUND_ROBIN_BATCH_SIZE)
            {
                perf_event_header header{};
                ring_buffer.ReadHeader(&header);

                // perf_event_header::type contains the type of record,
                // defined in enum perf_event_type in perf_event.h.
                switch (header.type)
                {
                case PERF_RECORD_SAMPLE:
                    {
                        if (GParams.m_BpftraceCallstacks)
                        {
                            auto record = ring_buffer.ConsumeRecord<LinuxUprobeEventWithStack>(header);
                            record.SetFunction(pair.first);
                            event_buffer.Push(std::make_unique<LinuxUprobeEventWithStack>(std::move(record)));
                        }
                        else
                        {
                            auto record = ring_buffer.ConsumeRecord<LinuxUprobeEvent>(header);
                            record.SetFunction(pair.first);
                            event_buffer.Push(std::make_unique<LinuxUprobeEvent>(std::move(record)));
                        }
                        
                    }
                    break;
                 case PERF_RECORD_LOST:
                    {
                        auto lost = ring_buffer.ConsumeRecord<LinuxPerfLostEvent>(header);
                        PRINT("Lost %u Events\n", lost.Lost());
                    }
                    break;
                default:
                    PRINT("Unexpected Perf Sample Type: %u", header.type);
                    ring_buffer.SkipRecord(header);
                    break;
                }
            }
        }

        for (auto& pair : uretprobe_ring_buffers)
        {
            auto& ring_buffer = pair.second;
            int i = 0;
            // read everything that is new
            while (ring_buffer.HasNewData() && i < ROUND_ROBIN_BATCH_SIZE)
            {
                perf_event_header header{};
                ring_buffer.ReadHeader(&header);

                // perf_event_header::type contains the type of record,
                // defined in enum perf_event_type in perf_event.h.
                switch (header.type)
                {
                case PERF_RECORD_SAMPLE:
                    {
                        auto record = ring_buffer.ConsumeRecord<LinuxUretprobeEvent>(header);
                        record.SetFunction(pair.first);<commit_msg>fixed round robin<commit_after>//-----------------------------------
// Copyright <NAME> 2013-2017
//-----------------------------------

#include "BpfTrace.h"
#include "LinuxUtils.h"
#include "CoreApp.h"
#include "Capture.h"
#include "OrbitModule.h"
#include "OrbitProcess.h"
#include "Params.h"
#include "Utils.h"
#include <fstream>
#include <sstream>

#if __linux__
#include "BpfTraceVisitor.h"
#include "LinuxPerfEvent.h"
#include "LinuxPerfEventProcessor.h"
#include "LinuxPerfRingBuffer.h"
#include "LinuxPerfUtils.h"
#include "OrbitFunction.h"
#include "Pdb.h"
#include <linux/perf_event.h>
#include <map>
#include <vector>

using namespace LinuxPerfUtils;
#endif

//-----------------------------------------------------------------------------
BpfTrace::BpfTrace(Callback a_Callback)
{
    m_Callback = a_Callback ? a_Callback : [this](const std::string& a_Buffer)
    {
        if( GParams.m_BpftraceCallstacks )
            CommandCallbackWithCallstacks(a_Buffer);
        else
            CommandCallback(a_Buffer);
    };

    m_ScriptFileName = ws2s(Path::GetBasePath()) + "orbit.bt";
}

//-----------------------------------------------------------------------------
void BpfTrace::Start()
{
#if __linux__
    m_ExitRequested = false;
    m_TimerStacks.clear();
    if( !WriteBpfScript() )
        return;

    m_BpfCommand = std::string("bpftrace ") + m_ScriptFileName;
    if ( GParams.m_UseBpftrace || GParams.m_BpftraceCallstacks ) {    
        m_Thread = std::make_shared<std::thread>
            ( &LinuxUtils::StreamCommandOutput
            , m_BpfCommand.c_str()
            , m_Callback
            , &m_ExitRequested );
    }
    else
    {
        m_Thread = std::make_shared<std::thread>
            ( BpfTrace::RunPerfEventOpen
            , &m_ExitRequested );
    }
    
    m_Thread->detach();
#endif
}

//-----------------------------------------------------------------------------
void BpfTrace::Stop()
{
    m_ExitRequested = true;
}

//-----------------------------------------------------------------------------
std::string BpfTrace::GetBpfScript()
{
    if (!m_Script.empty())
    {
        return m_Script;
    }

    std::stringstream ss;

    for (Function *func : Capture::GTargetProcess->GetFunctions())
    {
        if (func->IsSelected())
        {
            uint64_t virtual_address = (uint64_t)func->GetVirtualAddress();
            Capture::GSelectedFunctionsMap[func->m_Address] = func;

            if( GParams.m_BpftraceCallstacks )
            {
                ss << "   uprobe:" << func->m_Probe << R"({ printf("b )" << std::to_string(virtual_address) << R"( %u %lld\n%s\n\nd\n\n", tid, nsecs, ustack(perf)); })" << std::endl;
            }
            else
            {
                ss << "   uprobe:" << func->m_Probe << R"({ printf("b )" << std::to_string(virtual_address) << R"( %u %lld\n", tid, nsecs); })" << std::endl;
            }

            ss << "uretprobe:" << func->m_Probe << R"({ printf("e )" << std::to_string(virtual_address) << R"( %u %lld\n", tid, nsecs); })" << std::endl;
        }
    }

    return ss.str();
}

//-----------------------------------------------------------------------------
bool BpfTrace::WriteBpfScript()
{
    std::string script = GetBpfScript();
    if (script.empty())
        return false;

    std::ofstream outFile;
    outFile.open(m_ScriptFileName);
    if (outFile.fail())
        return false;

    outFile << script;
    outFile.close();
    return true;
}

//-----------------------------------------------------------------------------
uint64_t BpfTrace::ProcessString(const std::string& a_String)
{
    auto hash = StringHash(a_String);
    if (m_StringMap.find(hash) == m_StringMap.end())
    {
        m_StringMap[hash] = a_String;
    }

    return hash;
}

//-----------------------------------------------------------------------------
void BpfTrace::CommandCallback(const std::string& a_Line)
{
    auto tokens = Tokenize(a_Line);
    const std::string& mode             = tokens[0];
    const std::string& functionAddress  = tokens[1];
    const std::string& threadName       = tokens[2];
    const std::string& timestamp        = tokens[3];

    bool isBegin = mode == "b";
    bool isEnd   = !isBegin;

    if (isBegin)
    {
        Timer timer;
        timer.m_TID = atoi(threadName.c_str());
        uint64_t nanos = std::stoull(timestamp);
        timer.m_Start = nanos;
        timer.m_Depth = (uint8_t)m_TimerStacks[threadName].size();
        timer.m_FunctionAddress = std::stoull(functionAddress);
        m_TimerStacks[threadName].push_back(timer);
    }

    if (isEnd)
    {
        std::vector<Timer>& timers = m_TimerStacks[threadName];
        if (timers.size())
        {
            Timer& timer = timers.back();
            uint64_t nanos = std::stoull(timestamp);
            timer.m_End = nanos;
            GCoreApp->ProcessTimer(&timer, functionAddress);
            timers.pop_back();
        }
    }
}

//-----------------------------------------------------------------------------
void BpfTrace::CommandCallbackWithCallstacks(const std::string& a_Line)
{
    if (a_Line.empty() || a_Line == "\n")
        return;
    
    auto tokens = Tokenize(a_Line);
    
    const std::string& mode = tokens[0];
    bool isBegin = mode == "b";
    bool isEnd   = mode == "e";

    bool isStackLine = StartsWith(a_Line, "\t");
    bool isEndOfStack = a_Line == "d\n";

    if (!isBegin && !isEnd && !isStackLine && !isEndOfStack)
    {  
        if (StartsWith(a_Line, "Lost"))
        {
            PRINT(a_Line.c_str());
            return;
        }

        if (StartsWith(a_Line, "Attaching")) 
            return;

        // if the line does not start with one of the above,
        // we might have a broken line, e.g. due to a small buffer
        PRINT("read unexpected line:%s\nthe buffer might be to small.", a_Line.c_str());
        return;
    }

    if (isStackLine) {
        const std::string& addressStr = LTrim(tokens[0]);
        uint64_t address = std::stoull(addressStr, nullptr, 16);

        std::string function = tokens[1];

        std::string moduleRaw = tokens[2];
        std::string module = Replace(moduleRaw.substr(1), ")\n", "");

        // TODO: this is copy&paste from LinuxPerf.cpp
        std::wstring moduleName = ToLower(Path::GetFileName(s2ws(module)));
        std::shared_ptr<Module> moduleFromName = Capture::GTargetProcess->GetModuleFromName( ws2s(moduleName) );
       
        if( moduleFromName )
        {
            uint64_t new_address = moduleFromName->ValidateAddress(address);
            address = new_address;
        }

        m_CallStack.m_Data.push_back(address);
        if( Capture::GTargetProcess && !Capture::GTargetProcess->HasSymbol(address))
        {
            GCoreApp->AddSymbol(address, module, function);
        }

        return;
    }

    if (isEndOfStack)
    {
        if ( m_CallStack.m_Data.size() ) {
            m_CallStack.m_Depth = (uint32_t)m_CallStack.m_Data.size();
            m_CallStack.m_ThreadId = atoi(m_LastThreadName.c_str());
            std::vector<Timer>& timers = m_TimerStacks[m_LastThreadName];
            if (timers.size())
            {
                Timer& timer = timers.back();
                timer.m_CallstackHash = m_CallStack.Hash();
                GCoreApp->ProcessCallStack( m_CallStack );
            }
        }

        m_CallStack.Clear();
        m_LastThreadName = "";
        return;
    }

    const std::string& functionAddress  = tokens[1];
    const std::string& threadName       = tokens[2];
    const std::string& timestamp        = tokens[3];

    m_LastThreadName = threadName;

    if (isBegin)
    {
        Timer timer;
        timer.m_TID = atoi(threadName.c_str());
        uint64_t nanos = std::stoull(timestamp);
        timer.m_Start = nanos;
        timer.m_Depth = (uint8_t)m_TimerStacks[threadName].size();
        timer.m_FunctionAddress = std::stoull(functionAddress);
        m_TimerStacks[threadName].push_back(timer);
        return;
    }

    if (isEnd)
    {
        std::vector<Timer>& timers = m_TimerStacks[threadName];
        if (timers.size())
        {
            Timer& timer = timers.back();
            uint64_t nanos = std::stoull(timestamp);
            timer.m_End = nanos;
            GCoreApp->ProcessTimer(&timer, functionAddress);
            timers.pop_back();
        }
        return;
    }
}


//-----------------------------------------------------------------------------
void BpfTrace::RunPerfEventOpen(bool* a_ExitRequested)
{
    #if __linux__
    int ROUND_ROBIN_BATCH_SIZE = 5;
    std::vector<uint64_t> fds;
    std::map<Function*, LinuxPerfRingBuffer> uprobe_ring_buffers;
    std::map<Function*, LinuxPerfRingBuffer> uretprobe_ring_buffers;
    
    for (const auto& pair : Capture::GSelectedFunctionsMap)
    {
        // gather function information
        Function* function = pair.second;
        uint64_t offset = function->m_Address;
        uint64_t virtual_address = function->GetVirtualAddress();
        std::string module = ws2s(function->m_Pdb->GetFileName());

        // create uprobe for that function on that PID profiling all cpus
        uint64_t uprobe_fd = 0;
        if (GParams.m_BpftraceCallstacks)
            uprobe_fd = LinuxPerfUtils::uprobe_event_open(
                module.c_str(), offset, Capture::GTargetProcess->GetID(), -1, 
                PERF_SAMPLE_STACK_USER | PERF_SAMPLE_REGS_USER
            );
        else
            uprobe_fd = LinuxPerfUtils::uprobe_event_open(
                module.c_str(), offset, Capture::GTargetProcess->GetID(), -1
            );

        uprobe_ring_buffers.emplace(function, uprobe_fd);
        fds.push_back(uprobe_fd);

        // create uretprobe for that function on that PID profiling all cpus
        uint64_t uretprobe_fd = LinuxPerfUtils::uretprobe_event_open(
            module.c_str(), offset, Capture::GTargetProcess->GetID(), -1
        );
        uretprobe_ring_buffers.emplace(function, uretprobe_fd);
        fds.push_back(uretprobe_fd);

        // start capturing
        LinuxPerfUtils::start_capturing(uprobe_fd);
        LinuxPerfUtils::start_capturing(uretprobe_fd);
    }

    LinuxPerfEventProcessor event_buffer(std::make_unique<BpfTraceVisitor>());

    bool new_events = false;

    while( !(*a_ExitRequested) )
    {
        // Lets sleep a bit, such that we are not constantly reading from the buffers
        // and thus wasting cpu time. 1000 microseconds are still small enought to 
        // not have our buffers overflown and therefore loosing events.
        if (!new_events)
        {
            usleep(10000);
        }

        new_events = false;

        // read from all ring buffers, create events and store them in the event_queue
        // TODO: implement a better scheduling strategy.
        for (auto& pair : uprobe_ring_buffers)
        {
            auto& ring_buffer = pair.second;
            int i = 0;
            // read everything that is new
            while (ring_buffer.HasNewData() && i < ROUND_ROBIN_BATCH_SIZE)
            {
                i++;
                new_events = true;
                perf_event_header header{};
                ring_buffer.ReadHeader(&header);

                // perf_event_header::type contains the type of record,
                // defined in enum perf_event_type in perf_event.h.
                switch (header.type)
                {
                case PERF_RECORD_SAMPLE:
                    {
                        if (GParams.m_BpftraceCallstacks)
                        {
                            auto record = ring_buffer.ConsumeRecord<LinuxUprobeEventWithStack>(header);
                            record.SetFunction(pair.first);
                            event_buffer.Push(std::make_unique<LinuxUprobeEventWithStack>(std::move(record)));
                        }
                        else
                        {
                            auto record = ring_buffer.ConsumeRecord<LinuxUprobeEvent>(header);
                            record.SetFunction(pair.first);
                            event_buffer.Push(std::make_unique<LinuxUprobeEvent>(std::move(record)));
                        }
                        
                    }
                    break;
                 case PERF_RECORD_LOST:
                    {
                        auto lost = ring_buffer.ConsumeRecord<LinuxPerfLostEvent>(header);
                        PRINT("Lost %u Events\n", lost.Lost());
                    }
                    break;
                default:
                    PRINT("Unexpected Perf Sample Type: %u", header.type);
                    ring_buffer.SkipRecord(header);
                    break;
                }
            }
        }

        for (auto& pair : uretprobe_ring_buffers)
        {
            auto& ring_buffer = pair.second;
            int i = 0;
            // read everything that is new
            while (ring_buffer.HasNewData() && i < ROUND_ROBIN_BATCH_SIZE)
            {
                i++;
                new_events = true;
                perf_event_header header{};
                ring_buffer.ReadHeader(&header);

                // perf_event_header::type contains the type of record,
                // defined in enum perf_event_type in perf_event.h.
                switch (header.type)
                {
                case PERF_RECORD_SAMPLE:
                    {
                        auto record = ring_buffer.ConsumeRecord<LinuxUretprobeEvent>(header);
                        record.SetFunction(pair.first);
{'fraction_non_alphanumeric': 0.16672819906963007, 'fraction_numerical': 0.005464077383149967, 'mean_word_length': 1.7073463268365816, 'pattern_counts': {'":': 0, '<': 127, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2187846', 'n_tokens_mistral': 8096, 'n_tokens_neox': 7767, 'n_words': 1475}