<commit_before>#include <fcntl.h>
#include <inttypes.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#ifdef __APPLE__
#  include <libkern/OSByteOrder.h>
#  define be32toh(x) OSSwapBigToHostInt32(x)
#  define be64toh(x) OSSwapBigToHostInt64(x)
#else
#  include <endian.h>
#endif

#include "buffer.h"
#include "receiver.h"
#include "stream.h"
#include "timer.h"

typedef unsigned char uchar;

struct Stream* stream_create(unsigned short int port, int stream_id,
        struct Receiver* receiver)
{
    struct Stream* cls = (struct Stream*) calloc(1, sizeof(struct Stream));
    const int requested_buffer_len = 16*1024*1024;
    cls->port = port;
    cls->buffer_len = requested_buffer_len;
    cls->stream_id = stream_id;
    cls->receiver = receiver;
    cls->tmr_memcpy = tmr_create();
    if ((cls->socket_handle = socket(AF_INET, SOCK_DGRAM, 0)) < 0)
    {
        fprintf(stderr, "Cannot create socket.\n");
        return cls;
    }
    fcntl(cls->socket_handle, F_SETFL, O_NONBLOCK);
    setsockopt(cls->socket_handle, SOL_SOCKET, SO_RCVBUF,
            &cls->buffer_len, sizeof(int));
    uint32_t int_size = (uint32_t) sizeof(int);
    getsockopt(cls->socket_handle, SOL_SOCKET, SO_RCVBUF,
            &cls->buffer_len, &int_size);
    if (int_size != (uint32_t) sizeof(int))
    {
        fprintf(stderr, "Error at line %d\n", __LINE__);
        exit(1);
    }
    if ((cls->buffer_len / 2) < requested_buffer_len)
    {
        printf("Requested socket buffer of %d bytes; actual size is %d bytes\n",
                requested_buffer_len, cls->buffer_len / 2);
    }
    struct sockaddr_in myaddr;
    myaddr.sin_family = AF_INET;
    myaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    myaddr.sin_port = htons(port);
    if (bind(cls->socket_handle, (struct sockaddr*)&myaddr, sizeof(myaddr)) < 0)
    {
        fprintf(stderr, "Bind failed.\n");
        return cls;
    }
    cls->socket_buffer = (uchar*) malloc(cls->buffer_len);
    memset(cls->socket_buffer, 0, cls->buffer_len);
    return cls;
}

void stream_decode(struct Stream* self, const uchar* buf, int depth)
{
    /* Extract SPEAD packet headers. */
    const uchar magic = buf[0];
    const uchar version = buf[1];
    if (magic != 'S' || version != (uchar)4) return;
    const uchar item_id_bits = buf[2] * 8 - 1;
    const uchar heap_address_bits = buf[3] * 8;
    const uchar num_items = buf[7];

    /* Get pointers to items and payload start. */
    const uint64_t* items = (const uint64_t*) &buf[8];
    const uint64_t mask_addr = (1ull << heap_address_bits) - 1;
    const uint64_t mask_id   = (1ull << item_id_bits) - 1;
    const uchar* payload_start = (const uchar*) &items[num_items];

    /* Heap data. */
    int packet_has_header_data = 0, packet_has_stream_control = 0;
    uint32_t timestamp_count = 0, timestamp_fraction = 0;
    uint64_t scan_id = 0;
    size_t packet_payload_length = 0;
    size_t heap_offset = 0, heap_size = 0;
    size_t vis_data_start = 0;

    /* Iterate ItemPointers. */
    for (uchar i = 0; i < num_items; ++i)
    {
        const uint64_t item = be64toh(items[i]);
        /*const uint64_t immediate = item & (1ull << 63);*/
        const uint64_t item_addr = item & mask_addr;
        const uint64_t item_id   = (item >> heap_address_bits) & mask_id;
        switch (item_id)
        {
        case 0x0:
            /* NULL - ignore. */
            break;
        case 0x1:
            /* Heap counter (immediate addressing, big endian). */
            if (depth == 0) self->heap_count = (int) item_addr - 2;
            break;
        case 0x2:
            /* Heap size (immediate addressing, big endian). */
            heap_size = item_addr;
            break;
        case 0x3:
            /* Heap offset (immediate addressing, big endian). */
            heap_offset = (size_t) item_addr;
            break;
        case 0x4:
            /* Packet payload length (immediate addressing, big endian). */
            packet_payload_length = (size_t) item_addr;
            break;
        case 0x5:
            /* Nested item descriptor (recursive call). */
            /*stream_decode(self, &payload_start[item_addr], depth + 1);*/
            break;
        case 0x6:
            /* Stream control messages (immediate addressing, big endian). */
            packet_has_stream_control = 1;
            if (item_addr == 2) self->done = 1;
            break;
        case 0x10: /* Item descriptor name     (absolute addressing). */
        case 0x11: /* Item descriptor desc.    (absolute addressing). */
        case 0x12: /* Item descriptor shape    (absolute addressing). */
        case 0x13: /* Item descriptor type     (absolute addressing). */
        case 0x14: /* Item descriptor ID       (immediate addressing). */
        case 0x15: /* Item descriptor DataType (absolute addressing). */
            break;
        case 0x6000:
            /* Visibility timestamp count (immediate addressing). */
            timestamp_count = be32toh((uint32_t) item_addr);
            packet_has_header_data = 1;
            break;
        case 0x6001:
            /* Visibility timestamp fraction (immediate addressing). */
            timestamp_fraction = be32toh((uint32_t) item_addr);
            packet_has_header_data = 1;
            break;
        case 0x6005:
            /* Visibility baseline count (immediate addressing). */
            self->receiver->num_baselines = (uint32_t) item_addr;
            packet_has_header_data = 1;
            break;
        case 0x6008:
            /* Scan ID (absolute addressing). */
            scan_id = *( (uint64_t*)(payload_start + item_addr) );
            packet_has_header_data = 1;
            break;
        case 0x600A:
            /* Visibility data (absolute addressing). */
            self->vis_data_heap_offset = (size_t) item_addr;
            vis_data_start = (size_t) item_addr;
            break;
        default:
            /*printf("Heap %3d  ID: %#6llx, %s: %llu\n", self->heap_count,
                    item_id, immediate ? "VAL" : "ptr", item_addr);*/
            break;
        }
    }
    if (0 && !packet_has_stream_control)
        printf("==== Packet in heap %3d "
                "(heap offset %zu/%zu, payload length %zu)\n", self->heap_count,
                heap_offset, heap_size, packet_payload_length);
    if (0 && packet_has_header_data)
    {
        printf("     heap               : %d\n", self->heap_count);
        printf("     timestamp_count    : %" PRIu32 "\n", timestamp_count);
        printf("     timestamp_fraction : %" PRIu32 "\n", timestamp_fraction);
        printf("     scan_id            : %" PRIu64 "\n", scan_id);
        printf("     num_baselines      : %d\n", self->receiver->num_baselines);
    }


    if (!packet_has_stream_control && self->vis_data_heap_offset > 0 &&
            self->receiver->num_baselines > 0)
    {
        const double timestamp = tmr_get_timestamp();<commit_msg>UKSW-9 Travis updated with coveralls v3<commit_after>#include <fcntl.h>
#include <inttypes.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#ifdef __APPLE__
#  include <libkern/OSByteOrder.h>
#  define be32toh(x) OSSwapBigToHostInt32(x)
#  define be64toh(x) OSSwapBigToHostInt64(x)
#else
#  include <endian.h>
#endif

#include "buffer.h"
#include "receiver.h"
#include "stream.h"
#include "timer.h"

typedef unsigned char uchar;

struct Stream* stream_create(unsigned short int port, int stream_id,
        struct Receiver* receiver)
{
    struct Stream* cls = (struct Stream*) calloc(1, sizeof(struct Stream));
    const int requested_buffer_len = 16*1024*1024;
    cls->port = port;
    cls->buffer_len = requested_buffer_len;
    cls->stream_id = stream_id;
    cls->receiver = receiver;
    cls->tmr_memcpy = tmr_create();
    if ((cls->socket_handle = socket(AF_INET, SOCK_DGRAM, 0)) < 0)
    {
        fprintf(stderr, "Cannot create socket.\n");
        return cls;
    }
    fcntl(cls->socket_handle, F_SETFL, O_NONBLOCK);
    setsockopt(cls->socket_handle, SOL_SOCKET, SO_RCVBUF,
            &cls->buffer_len, sizeof(int));
    uint32_t int_size = (uint32_t) sizeof(int);
    getsockopt(cls->socket_handle, SOL_SOCKET, SO_RCVBUF,
            &cls->buffer_len, &int_size);
    if (int_size != (uint32_t) sizeof(int))
    {
        fprintf(stderr, "Error at line %d\n", __LINE__);
        exit(1);
    }
    if ((cls->buffer_len / 2) < requested_buffer_len)
    {
        printf("Requested socket buffer of %d bytes; actual size is %d bytes\n",
                requested_buffer_len, cls->buffer_len / 2);
    }
    struct sockaddr_in myaddr;
    myaddr.sin_family = AF_INET;
    myaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    myaddr.sin_port = htons(port);
    if (bind(cls->socket_handle, (struct sockaddr*)&myaddr, sizeof(myaddr)) < 0)
    {
        fprintf(stderr, "Bind failed.\n");
        return cls;
    }
    cls->socket_buffer = (uchar*) malloc(cls->buffer_len);
    memset(cls->socket_buffer, 0, cls->buffer_len);
    return cls;
}

void stream_decode(struct Stream* self, const uchar* buf, int depth)
{
    /* Extract SPEAD packet headers. */
    const uchar magic = buf[0];
    const uchar version = buf[1];
    if (magic != 'S' || version != (uchar)4) return;
    const uchar item_id_bits = buf[2] * 8 - 1;
    const uchar heap_address_bits = buf[3] * 8;
    const uchar num_items = buf[7];

    /* Get pointers to items and payload start. */
    const uint64_t* items = (const uint64_t*) &buf[8];
    const uint64_t mask_addr = (1ull << heap_address_bits) - 1;
    const uint64_t mask_id   = (1ull << item_id_bits) - 1;
    const uchar* payload_start = (const uchar*) &items[num_items];

    /* Heap data. */
    int packet_has_header_data = 0, packet_has_stream_control = 0;
    uint32_t timestamp_count = 0, timestamp_fraction = 0;
    uint64_t scan_id = 0;
    size_t packet_payload_length = 0;
    size_t heap_offset = 0, heap_size = 0;
    size_t vis_data_start = 0;

    /* Iterate ItemPointers. */
    for (uchar i = 0; i < num_items; ++i)
    {
        const uint64_t item = be64toh(items[i]);
        /*const uint64_t immediate = item & (1ull << 63);*/
        const uint64_t item_addr = item & mask_addr;
        const uint64_t item_id   = (item >> heap_address_bits) & mask_id;
        switch (item_id)
        {
        case 0x0:
            /* NULL - ignore. */
            break;
        case 0x1:
            /* Heap counter (immediate addressing, big endian). */
            if (depth == 0) self->heap_count = (int) item_addr - 2;
            break;
        case 0x2:
            /* Heap size (immediate addressing, big endian). */
            heap_size = item_addr;
            break;
        case 0x3:
            /* Heap offset (immediate addressing, big endian). */
            heap_offset = (size_t) item_addr;
            break;
        case 0x4:
            /* Packet payload length (immediate addressing, big endian). */
            packet_payload_length = (size_t) item_addr;
            break;
        case 0x5:
            /* Nested item descriptor (recursive call). */
            /*stream_decode(self, &payload_start[item_addr], depth + 1);*/
            break;
        case 0x6:
            /* Stream control messages (immediate addressing, big endian). */
            packet_has_stream_control = 1;
            if (item_addr == 2) self->done = 1;
            break;
        case 0x10: /* Item descriptor name     (absolute addressing). */
        case 0x11: /* Item descriptor desc.    (absolute addressing). */
        case 0x12: /* Item descriptor shape    (absolute addressing). */
        case 0x13: /* Item descriptor type     (absolute addressing). */
        case 0x14: /* Item descriptor ID       (immediate addressing). */
        case 0x15: /* Item descriptor DataType (absolute addressing). */
            break;
        case 0x6000:
            /* Visibility timestamp count (immediate addressing). */
            timestamp_count = be32toh((uint32_t) item_addr);
            packet_has_header_data = 1;
            break;
        case 0x6001:
            /* Visibility timestamp fraction (immediate addressing). */
            timestamp_fraction = be32toh((uint32_t) item_addr);
            packet_has_header_data = 1;
            break;
        case 0x6005:
            /* Visibility baseline count (immediate addressing). */
            self->receiver->num_baselines = be32toh((uint32_t) item_addr);
            packet_has_header_data = 1;
            break;
        case 0x6008:
            /* Scan ID (absolute addressing). */
            scan_id = *( (uint64_t*)(payload_start + item_addr) );
            packet_has_header_data = 1;
            break;
        case 0x600A:
            /* Visibility data (absolute addressing). */
            self->vis_data_heap_offset = (size_t) item_addr;
            vis_data_start = (size_t) item_addr;
            break;
        default:
            /*printf("Heap %3d  ID: %#6llx, %s: %llu\n", self->heap_count,
                    item_id, immediate ? "VAL" : "ptr", item_addr);*/
            break;
        }
    }
    if (0 && !packet_has_stream_control)
        printf("==== Packet in heap %3d "
                "(heap offset %zu/%zu, payload length %zu)\n", self->heap_count,
                heap_offset, heap_size, packet_payload_length);
    if (0 && packet_has_header_data)
    {
        printf("     heap               : %d\n", self->heap_count);
        printf("     timestamp_count    : %" PRIu32 "\n", timestamp_count);
        printf("     timestamp_fraction : %" PRIu32 "\n", timestamp_fraction);
        printf("     scan_id            : %" PRIu64 "\n", scan_id);
        printf("     num_baselines      : %d\n", self->receiver->num_baselines);
    }
    if (!packet_has_stream_control && self->vis_data_heap_offset > 0 &&
            self->receiver->num_baselines > 0)
    {
        const double timestamp = tmr_get_timestamp();
{'fraction_non_alphanumeric': 0.12092923822418584, 'fraction_numerical': 0.024798688804959737, 'mean_word_length': 2.258416531228233, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 93, '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': '6615935', 'n_tokens_mistral': 4747, 'n_tokens_neox': 4502, 'n_words': 1112}