<commit_before>
#include "input_stream.h"

#include "input_command.h"
#include "mega_block.h"

#include <genomics/data/dna_sequence.h>
#include <genomics/storage/sequence_store.h>

#include <core/file_storage/file.h>

#include <core/helpers/message_helper.h>

#include <core/system/memory.h>

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include <inttypes.h>

/*
#define BSAL_INPUT_STREAM_DEBUG
*/

struct bsal_script bsal_input_stream_script = {
    .identifier = BSAL_INPUT_STREAM_SCRIPT,
    .init = bsal_input_stream_init,
    .destroy = bsal_input_stream_destroy,
    .receive = bsal_input_stream_receive,
    .size = sizeof(struct bsal_input_stream),
    .name = "input_stream"
};

void bsal_input_stream_init(struct bsal_actor *actor)
{
    struct bsal_input_stream *input;
    struct bsal_input_stream *concrete_self;

    input = (struct bsal_input_stream *)bsal_actor_concrete_actor(actor);
    concrete_self = input;
    concrete_self->proxy_ready = 0;
    concrete_self->buffer_for_sequence = NULL;
    concrete_self->maximum_sequence_length = 0;
    concrete_self->open = 0;
    concrete_self->error = BSAL_INPUT_ERROR_NO_ERROR;

    concrete_self->file_name = NULL;

    bsal_dna_codec_init(&concrete_self->codec);

    if (bsal_actor_get_node_count(actor) >= BSAL_DNA_CODEC_MINIMUM_NODE_COUNT_FOR_TWO_BIT) {
#ifdef BSAL_DNA_CODEC_USE_TWO_BIT_ENCODING_FOR_TRANSPORT
        bsal_dna_codec_enable_two_bit_encoding(&concrete_self->codec);
#endif
    }

    /*concrete_self->mega_block_size = 2097152*/

    /*
     * This is the mega block size in number of sequences.
     */
    concrete_self->mega_block_size = 2097152;
    concrete_self->granularity = 1024;

    concrete_self->last_offset = 0;
    concrete_self->last_entries = 0;

    concrete_self->starting_offset = 0;

    /*
     * Use a large ending offset.
     */
    concrete_self->ending_offset = 0;
    --concrete_self->ending_offset;

    bsal_vector_init(&concrete_self->mega_blocks, sizeof(struct bsal_mega_block));

    bsal_actor_add_route(actor, BSAL_INPUT_STREAM_SET_START_OFFSET, bsal_input_stream_set_start_offset);
    bsal_actor_add_route(actor, BSAL_INPUT_STREAM_SET_END_OFFSET, bsal_input_stream_set_end_offset);
    bsal_actor_add_route(actor, BSAL_INPUT_COUNT_IN_PARALLEL, bsal_input_stream_count_in_parallel_mock);
    bsal_actor_add_route(actor, BSAL_INPUT_COUNT_REPLY, bsal_input_stream_count_reply_mock);
    bsal_actor_add_route(actor, BSAL_ACTOR_SPAWN_REPLY, bsal_input_stream_spawn_reply);

    concrete_self->count_customer = BSAL_ACTOR_NOBODY;

#if 0
    bsal_string_init(&concrete_self->file_for_parallel_counting, NULL);
#endif

    /*
     * Parallel counting.
     */

    concrete_self->total_entries = 0;
    concrete_self->parallel_block_size = BSAL_BUFFERED_READER_BUFFER_SIZE;

    concrete_self->finished_parallel_stream_count = 0;

    bsal_vector_init(&concrete_self->spawners, sizeof(int));
    bsal_vector_init(&concrete_self->parallel_streams, sizeof(int));
    bsal_vector_init(&concrete_self->start_offsets, sizeof(uint64_t));
    bsal_vector_init(&concrete_self->end_offsets, sizeof(uint64_t));

    bsal_vector_init(&concrete_self->parallel_mega_blocks, sizeof(struct bsal_vector));
}
<commit_msg>set parallel block size to 1 GiB<commit_after>
#include "input_stream.h"

#include "input_command.h"
#include "mega_block.h"

#include <genomics/data/dna_sequence.h>
#include <genomics/storage/sequence_store.h>

#include <core/file_storage/file.h>

#include <core/helpers/message_helper.h>

#include <core/system/memory.h>

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include <inttypes.h>

/*
#define BSAL_INPUT_STREAM_DEBUG
*/

struct bsal_script bsal_input_stream_script = {
    .identifier = BSAL_INPUT_STREAM_SCRIPT,
    .init = bsal_input_stream_init,
    .destroy = bsal_input_stream_destroy,
    .receive = bsal_input_stream_receive,
    .size = sizeof(struct bsal_input_stream),
    .name = "input_stream"
};

void bsal_input_stream_init(struct bsal_actor *actor)
{
    struct bsal_input_stream *input;
    struct bsal_input_stream *concrete_self;

    input = (struct bsal_input_stream *)bsal_actor_concrete_actor(actor);
    concrete_self = input;
    concrete_self->proxy_ready = 0;
    concrete_self->buffer_for_sequence = NULL;
    concrete_self->maximum_sequence_length = 0;
    concrete_self->open = 0;
    concrete_self->error = BSAL_INPUT_ERROR_NO_ERROR;

    concrete_self->file_name = NULL;

    bsal_dna_codec_init(&concrete_self->codec);

    if (bsal_actor_get_node_count(actor) >= BSAL_DNA_CODEC_MINIMUM_NODE_COUNT_FOR_TWO_BIT) {
#ifdef BSAL_DNA_CODEC_USE_TWO_BIT_ENCODING_FOR_TRANSPORT
        bsal_dna_codec_enable_two_bit_encoding(&concrete_self->codec);
#endif
    }

    /*concrete_self->mega_block_size = 2097152*/

    /*
     * This is the mega block size in number of sequences.
     */
    concrete_self->mega_block_size = 2097152;
    concrete_self->granularity = 1024;

    concrete_self->last_offset = 0;
    concrete_self->last_entries = 0;

    concrete_self->starting_offset = 0;

    /*
     * Use a large ending offset.
     */
    concrete_self->ending_offset = 0;
    --concrete_self->ending_offset;

    bsal_vector_init(&concrete_self->mega_blocks, sizeof(struct bsal_mega_block));

    bsal_actor_add_route(actor, BSAL_INPUT_STREAM_SET_START_OFFSET, bsal_input_stream_set_start_offset);
    bsal_actor_add_route(actor, BSAL_INPUT_STREAM_SET_END_OFFSET, bsal_input_stream_set_end_offset);
    bsal_actor_add_route(actor, BSAL_INPUT_COUNT_IN_PARALLEL, bsal_input_stream_count_in_parallel_mock);
    bsal_actor_add_route(actor, BSAL_INPUT_COUNT_REPLY, bsal_input_stream_count_reply_mock);
    bsal_actor_add_route(actor, BSAL_ACTOR_SPAWN_REPLY, bsal_input_stream_spawn_reply);

    concrete_self->count_customer = BSAL_ACTOR_NOBODY;

#if 0
    bsal_string_init(&concrete_self->file_for_parallel_counting, NULL);
#endif

    /*
     * Parallel counting.
     */

    concrete_self->total_entries = 0;

    /*
     * 1 GiB.
     */
    concrete_self->parallel_block_size = 1073741824;

    concrete_self->finished_parallel_stream_count = 0;

    bsal_vector_init(&concrete_self->spawners, sizeof(int));
    bsal_vector_init(&concrete_self->parallel_streams, sizeof(int));
    bsal_vector_init(&concrete_self->start_offsets, sizeof(uint64_t));
    bsal_vector_init(&concrete_self->end_offsets, sizeof(uint64_t));

    bsal_vector_init(&concrete_self->parallel_mega_blocks, sizeof(struct bsal_vector));
}

{'fraction_non_alphanumeric': 0.09730561970746729, 'fraction_numerical': 0.01170130869899923, 'mean_word_length': 5.105263157894737, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 77, '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': '4013465', 'n_tokens_mistral': 2467, 'n_tokens_neox': 2366, 'n_words': 314}