<commit_before>#include "sprot.h"

using namespace std::chrono;

namespace sprot
{
    unsigned char util::crc7(const unsigned char* buf, size_t length)
    {
        //Reworked sample code from http://www.pololu.com/docs/0J44/6.7.6
        static unsigned char crc_table[256] = { 0 };

        auto get_byte_crc = [](unsigned char byte) -> unsigned char
        {
            const unsigned char crc7_poly = 0x91;

            for (unsigned char j = 0; j < 8; ++j)
            {
                if (byte & 1)
                    byte ^= crc7_poly;
                byte >>= 1;
            }

            return byte;
        };

        auto build_crc_table = [&]() -> bool
        {
            for (int i = 0; i < 256; ++i)
                crc_table[i] = get_byte_crc(i);

            return true;
        };

        static bool one_time_function_call = build_crc_table();
        unsigned char crc = 0;

        for (size_t i = 0; i < length; ++i)
            crc = crc_table[crc ^ buf[i]];

        return crc;
    }

    bool Protocol::crc_check(const unsigned char* buf, size_t length)
    {
        if (length < 1)
            return false;

        return (util::crc7(buf, length - 1) == buf[length - 1]);
    }

    Protocol::Protocol(fplog::Transport_Interface* transport, size_t MTU, int frames_before_ack):
    transport_(transport),
    sequence_num_(0),
    MTU_(MTU),
    terminating_(false),
    evil_twin_(0),
    twin_size_(0),
    ack_after_(frames_before_ack)
    {
        std::lock_guard<std::recursive_mutex> lock(mutex_);

        if (!transport_)
            THROW(fplog::exceptions::Incorrect_Parameter);

        frame_buf_ = new unsigned char [MTU_ + Frame::overhead];
    }

    Protocol::~Protocol()
    {
        std::lock_guard<std::recursive_mutex> lock(mutex_);
        terminating_ = true;

        delete [] frame_buf_;
        frame_buf_ = 0;
    }

    size_t Protocol::read(void* buf, size_t buf_size, size_t timeout)
    {
        int frame_num = 0;

        std::lock_guard<std::recursive_mutex> lock(mutex_);
        if (terminating_)
            return 0;

        time_point<system_clock, system_clock::duration> timer_start(system_clock::now());

        int full_retries = 5;
        size_t bytes_read = 0;
        unsigned char* ptr = (unsigned char*)buf;
    
    full_read_retry:

        bytes_read = 0;
        sequence_num_ = 0;
        frame_num = 0;
        
        ptr = (unsigned char*)buf;

        auto check_time_out = [&timeout, &timer_start]()
        {
            unsigned long long int mult = 1;
            
            #ifdef _LINUX
            mult = 100;
            #endif
            
            time_point<system_clock, system_clock::duration> timer_stop(system_clock::now());
            system_clock::duration converted_timeout(static_cast<unsigned long long>(timeout) * 10000 * mult);
            if (timer_stop - timer_start >= converted_timeout)
                THROW(fplog::exceptions::Timeout);
        };

        bool multipart = false, last = false;
        int retry_count = 5;
        Frame recv_frame;

        auto frame_read = [check_time_out, &recv_frame, &full_retries, &retry_count, &frame_num, this]()
        {
            while (true)
            {
                check_time_out();

                try
                {
                    recv_frame = read_frame();
                    break;
                }
                catch (sprot::exceptions::Wrong_Sequence&)
                {
                    if (full_retries == 0)
                        THROW(fplog::exceptions::Read_Failed);

                    full_retries--;
                    return false;
                }
                catch (fplog::exceptions::Generic_Exception&)
                {
                    if (full_retries == 0)
                        THROW(fplog::exceptions::Read_Failed);

                    if (retry_count == 0)
                    {
                        full_retries--;
                        return false;
                    }

                    retry_count--;
                }
            }

            return true;
        };

        while (true)
        {
            check_time_out();

            if (!frame_read())
                goto full_read_retry;

            if ((recv_frame.type != Frame::DATA_FIRST) &&
                (recv_frame.type != Frame::DATA_SINGLE) &&
                (recv_frame.type != Frame::DATA_LAST))
            {
                if (full_retries == 0)
                    THROW(fplog::exceptions::Read_Failed);
                full_retries--;
                goto full_read_retry;
            }
            else
                frame_num++;

            if (recv_frame.type == Frame::DATA_FIRST)
                multipart = true;

            if (recv_frame.type == Frame::DATA_LAST)
                last = true;

            bytes_read += recv_frame.data.size();
            if (bytes_read > buf_size)
                THROW(fplog::exceptions::Buffer_Overflow);

            memcpy(ptr, &recv_frame.data[0], recv_frame.data.size());
            ptr += recv_frame.data.size();

            Frame last_frame(recv_frame);

            bool ack_sent = false;
            if ((frame_num % ack_after_ == 0) || (recv_frame.type == Frame::DATA_LAST) || (!multipart))<commit_msg>Fixed a data corruption bug with ACKs that are being sent not after each data packet but after N data frames.<commit_after>#include "sprot.h"

using namespace std::chrono;

namespace sprot
{
    unsigned char util::crc7(const unsigned char* buf, size_t length)
    {
        //Reworked sample code from http://www.pololu.com/docs/0J44/6.7.6
        static unsigned char crc_table[256] = { 0 };

        auto get_byte_crc = [](unsigned char byte) -> unsigned char
        {
            const unsigned char crc7_poly = 0x91;

            for (unsigned char j = 0; j < 8; ++j)
            {
                if (byte & 1)
                    byte ^= crc7_poly;
                byte >>= 1;
            }

            return byte;
        };

        auto build_crc_table = [&]() -> bool
        {
            for (int i = 0; i < 256; ++i)
                crc_table[i] = get_byte_crc(i);

            return true;
        };

        static bool one_time_function_call = build_crc_table();
        unsigned char crc = 0;

        for (size_t i = 0; i < length; ++i)
            crc = crc_table[crc ^ buf[i]];

        return crc;
    }

    bool Protocol::crc_check(const unsigned char* buf, size_t length)
    {
        if (length < 1)
            return false;

        return (util::crc7(buf, length - 1) == buf[length - 1]);
    }

    Protocol::Protocol(fplog::Transport_Interface* transport, size_t MTU, int frames_before_ack):
    transport_(transport),
    sequence_num_(0),
    MTU_(MTU),
    terminating_(false),
    evil_twin_(0),
    twin_size_(0),
    ack_after_(frames_before_ack)
    {
        std::lock_guard<std::recursive_mutex> lock(mutex_);

        if (!transport_)
            THROW(fplog::exceptions::Incorrect_Parameter);

        frame_buf_ = new unsigned char [MTU_ + Frame::overhead];
    }

    Protocol::~Protocol()
    {
        std::lock_guard<std::recursive_mutex> lock(mutex_);
        terminating_ = true;

        delete [] frame_buf_;
        frame_buf_ = 0;
    }

    size_t Protocol::read(void* buf, size_t buf_size, size_t timeout)
    {
        int frame_num = 0;

        std::lock_guard<std::recursive_mutex> lock(mutex_);
        if (terminating_)
            return 0;

        time_point<system_clock, system_clock::duration> timer_start(system_clock::now());

        int full_retries = 5;
        size_t bytes_read = 0;
        unsigned char* ptr = (unsigned char*)buf;
    
    full_read_retry:

        bytes_read = 0;
        sequence_num_ = 0;
        frame_num = 0;
        
        ptr = (unsigned char*)buf;

        auto check_time_out = [&timeout, &timer_start]()
        {
            unsigned long long int mult = 1;
            
            #ifdef _LINUX
            mult = 100;
            #endif
            
            time_point<system_clock, system_clock::duration> timer_stop(system_clock::now());
            system_clock::duration converted_timeout(static_cast<unsigned long long>(timeout) * 10000 * mult);
            if (timer_stop - timer_start >= converted_timeout)
                THROW(fplog::exceptions::Timeout);
        };

        bool multipart = false, last = false;
        int retry_count = 5;
        Frame recv_frame, prev_frame;

        auto frame_read = [check_time_out, &recv_frame, &prev_frame, &full_retries, &retry_count, &frame_num, this]()
        {
            while (true)
            {
                check_time_out();

                try
                {
                    recv_frame = read_frame();
                    break;
                }
                catch (sprot::exceptions::Wrong_Sequence&)
                {
                    if (full_retries == 0)
                        THROW(fplog::exceptions::Read_Failed);

                    full_retries--;
                    return false;
                }
                catch (fplog::exceptions::Generic_Exception&)
                {
                    if (full_retries == 0)
                        THROW(fplog::exceptions::Read_Failed);

                    if (retry_count == 0)
                    {
                        full_retries--;
                        return false;
                    }

                    retry_count--;
                }
            }

            return true;
        };

        while (true)
        {
            check_time_out();

            if (!frame_read())
                goto full_read_retry;

            if ((recv_frame.type != Frame::DATA_FIRST) &&
                (recv_frame.type != Frame::DATA_SINGLE) &&
                (recv_frame.type != Frame::DATA_LAST))
            {
                if (full_retries == 0)
                    THROW(fplog::exceptions::Read_Failed);
                full_retries--;
                goto full_read_retry;
            }
            else
            {
                frame_num++;
            }
            
            if (recv_frame.type == Frame::DATA_FIRST)
                multipart = true;

            if (recv_frame.type == Frame::DATA_LAST)
                last = true;
                
            if ((recv_frame.type == Frame::DATA_FIRST) && (prev_frame.type == recv_frame.type))
                continue;
             
            bool should_copy_data = true;
   
            if ((recv_frame.type == Frame::DATA_LAST) && (prev_frame.type == recv_frame.type))
            {
                should_copy_data = false;
            }
            
            if (should_copy_data)
            {
                bytes_read += recv_frame.data.size();
                if (bytes_read > buf_size)
                    THROW(fplog::exceptions::Buffer_Overflow);

                memcpy(ptr, &recv_frame.data[0], recv_frame.data.size());
                ptr += recv_frame.data.size();            
            }
            
            prev_frame = recv_frame;
            Frame last_frame(recv_frame);

            bool ack_sent = false;
            if ((frame_num % ack_after_ == 0) || (recv_frame.type == Frame::DATA_LAST) || (!multipart))
{'fraction_non_alphanumeric': 0.10534229612595082, 'fraction_numerical': 0.010083141694675394, 'mean_word_length': 1.4601827676240209, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4961914', 'n_tokens_mistral': 3435, 'n_tokens_neox': 3192, 'n_words': 667}