<commit_before>/*
 * Copyright 2014-2019 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#if defined(__linux__)
#define _BSD_SOURCE
#define _GNU_SOURCE
#endif

#include <errno.h>
#include <string.h>
#include "aeron_socket.h"
#include <inttypes.h>
#include "concurrent/aeron_term_scanner.h"
#include "util/aeron_netutil.h"
#include "util/aeron_error.h"
#include "aeron_network_publication.h"
#include "aeron_alloc.h"
#include "media/aeron_send_channel_endpoint.h"
#include "aeron_driver_conductor.h"
#include "concurrent/aeron_logbuffer_unblocker.h"

#if !defined(HAVE_STRUCT_MMSGHDR)
struct mmsghdr
{
    struct msghdr msg_hdr;
    unsigned int msg_len;
};
#endif

int aeron_network_publication_create(
    aeron_network_publication_t **publication,
    aeron_send_channel_endpoint_t *endpoint,
    aeron_driver_context_t *context,
    int64_t registration_id,
    int32_t session_id,
    int32_t stream_id,
    int32_t initial_term_id,
    aeron_position_t *pub_pos_position,
    aeron_position_t *pub_lmt_position,
    aeron_position_t *snd_pos_position,
    aeron_position_t *snd_lmt_position,
    aeron_counter_t *snd_bpe_counter,
    aeron_flow_control_strategy_t *flow_control_strategy,
    aeron_uri_publication_params_t *params,
    bool is_exclusive,
    bool spies_simulate_connection,
    aeron_system_counters_t *system_counters)
{
    char path[AERON_MAX_PATH];
    int path_length = aeron_network_publication_location(
        path,
        sizeof(path),
        context->aeron_dir,
        endpoint->conductor_fields.udp_channel->canonical_form,
        session_id,
        stream_id,
        registration_id);

    aeron_network_publication_t *_pub = NULL;
    const uint64_t usable_fs_space = context->usable_fs_space_func(context->aeron_dir);
    const uint64_t log_length = aeron_logbuffer_compute_log_length(params->term_length, context->file_page_size);
    const int64_t now_ns = context->nano_clock();

    *publication = NULL;

    if (usable_fs_space < log_length)
    {
        aeron_set_err(
            ENOSPC,
            "Insufficient usable storage for new log of length=%" PRId64 " in %s", log_length, context->aeron_dir);
        return -1;
    }

    if (aeron_alloc((void **)&_pub, sizeof(aeron_network_publication_t)) < 0)
    {
        aeron_set_err(ENOMEM, "%s", "Could not allocate network publication");
        return -1;
    }

    _pub->log_file_name = NULL;
    if (aeron_alloc((void **)(&_pub->log_file_name), (size_t)path_length + 1) < 0)
    {
        aeron_free(_pub);
        aeron_set_err(ENOMEM, "%s", "Could not allocate network publication log_file_name");
        return -1;
    }

    if (aeron_retransmit_handler_init(
        &_pub->retransmit_handler,
        aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_INVALID_PACKETS),
        context->retransmit_unicast_delay_ns,
        context->retransmit_unicast_linger_ns) < 0)
    {
        aeron_free(_pub->log_file_name);
        aeron_free(_pub);
        aeron_set_err(aeron_errcode(), "Could not init network publication retransmit handler: %s", aeron_errmsg());
        return -1;
    }

    if (context->map_raw_log_func(
        &_pub->mapped_raw_log, path, params->is_sparse, params->term_length, context->file_page_size) < 0)
    {
        aeron_free(_pub->log_file_name);
        aeron_free(_pub);
        aeron_set_err(aeron_errcode(), "error mapping network raw log %s: %s", path, aeron_errmsg());
        return -1;
    }
    _pub->map_raw_log_close_func = context->map_raw_log_close_func;

    strncpy(_pub->log_file_name, path, (size_t)path_length);
    _pub->log_file_name[path_length] = '\0';
    _pub->log_file_name_length = (size_t)path_length;
    _pub->log_meta_data = (aeron_logbuffer_metadata_t *)(_pub->mapped_raw_log.log_meta_data.addr);

    if (params->is_replay)
    {
        int64_t term_id = params->term_id;
        int32_t term_count = params->term_id - initial_term_id;
        size_t active_index = aeron_logbuffer_index_by_term_count(term_count);

        _pub->log_meta_data->term_tail_counters[active_index] =
            (term_id * ((int64_t)1 << 32)) | params->term_offset;
        
        for (int i = 1; i < AERON_LOGBUFFER_PARTITION_COUNT; i++)
        {
            int64_t expected_term_id = (term_id + i) - AERON_LOGBUFFER_PARTITION_COUNT;
            active_index = (active_index + 1) % AERON_LOGBUFFER_PARTITION_COUNT;
            _pub->log_meta_data->term_tail_counters[active_index] = expected_term_id * ((int64_t)1 << 32);
        }

        _pub->log_meta_data->active_term_count = term_count;
    }
    else
    {
        _pub->log_meta_data->term_tail_counters[0] = initial_term_id * ((int64_t)1 << 32);
        for (int i = 1; i < AERON_LOGBUFFER_PARTITION_COUNT; i++)
        {
            int64_t expected_term_id = (initial_term_id + i) - AERON_LOGBUFFER_PARTITION_COUNT;
            _pub->log_meta_data->term_tail_counters[i] = expected_term_id * ((int64_t)1 << 32);
        }

        _pub->log_meta_data->active_term_count = 0;
    }

    _pub->log_meta_data->initial_term_id = initial_term_id;
    _pub->log_meta_data->mtu_length = (int32_t)params->mtu_length;
    _pub->log_meta_data->term_length = (int32_t)params->term_length;
    _pub->log_meta_data->page_size = (int32_t)context->file_page_size;
    _pub->log_meta_data->correlation_id = registration_id;
    _pub->log_meta_data->is_connected = 0;
    _pub->log_meta_data->end_of_stream_position = INT64_MAX;
    aeron_logbuffer_fill_default_header(
        _pub->mapped_raw_log.log_meta_data.addr, session_id, stream_id, initial_term_id);

    _pub->endpoint = endpoint;
    _pub->flow_control = flow_control_strategy;
    _pub->nano_clock = context->nano_clock;
    _pub->conductor_fields.subscribable.array = NULL;
    _pub->conductor_fields.subscribable.length = 0;
    _pub->conductor_fields.subscribable.capacity = 0;
    _pub->conductor_fields.subscribable.add_position_hook_func = aeron_network_publication_add_subscriber_hook;
    _pub->conductor_fields.subscribable.remove_position_hook_func = aeron_network_publication_remove_subscriber_hook;
    _pub->conductor_fields.subscribable.clientd = _pub;
    _pub->conductor_fields.managed_resource.registration_id = registration_id;
    _pub->conductor_fields.managed_resource.clientd = _pub;
    _pub->conductor_fields.managed_resource.incref = aeron_network_publication_incref;
    _pub->conductor_fields.managed_resource.decref = aeron_network_publication_decref;
    _pub->conductor_fields.has_reached_end_of_life = false;
    _pub->conductor_fields.clean_position = 0;
    _pub->conductor_fields.status = AERON_NETWORK_PUBLICATION_STATUS_ACTIVE;
    _pub->conductor_fields.refcnt = 1;
    _pub->conductor_fields.time_of_last_activity_ns = now_ns;
    _pub->conductor_fields.last_snd_pos = 0;
    _pub->session_id = session_id;
    _pub->stream_id = stream_id;
    _pub->pub_lmt_position.counter_id = pub_lmt_position->counter_id;
    _pub->pub_lmt_position.value_addr = pub_lmt_position->value_addr;
    _pub->pub_pos_position.counter_id = pub_pos_position->counter_id;
    _pub->pub_pos_position.value_addr = pub_pos_position->value_addr;
    _pub->snd_pos_position.counter_id = snd_pos_position->counter_id;
    _pub->snd_pos_position.value_addr = snd_pos_position->value_addr;
    _pub->snd_lmt_position.counter_id = snd_lmt_position->counter_id;
    _pub->snd_lmt_position.value_addr = snd_lmt_position->value_addr;
    _pub->snd_bpe_counter.counter_id = snd_bpe_counter->counter_id;
    _pub->snd_bpe_counter.value_addr = snd_bpe_counter->value_addr;
    _pub->initial_term_id = initial_term_id;
    _pub->term_length_mask = (int32_t)params->term_length - 1;
    _pub->position_bits_to_shift = (size_t)aeron_number_of_trailing_zeroes((int32_t)params->term_length);
    _pub->mtu_length = params->mtu_length;
    _pub->term_window_length = (int64_t)aeron_producer_window_length(
        context->publication_window_length, params->term_length);
    _pub->linger_timeout_ns = (int64_t)params->linger_timeout_ns;
    _pub->unblock_timeout_ns = (int64_t)context->publication_unblock_timeout_ns;
    _pub->connection_timeout_ns = (int64_t)context->publication_connection_timeout_ns;
    _pub->time_of_last_send_or_heartbeat_ns = now_ns - AERON_NETWORK_PUBLICATION_HEARTBEAT_TIMEOUT_NS - 1;
    _pub->time_of_last_setup_ns = now_ns - AERON_NETWORK_PUBLICATION_SETUP_TIMEOUT_NS - 1;
    _pub->status_message_deadline_ns = spies_simulate_connection ?
        now_ns : now_ns + (int64_t)context->publication_connection_timeout_ns;
    _pub->is_exclusive = is_exclusive;
    _pub->spies_simulate_connection = spies_simulate_connection;
    _pub->signal_eos = params->signal_eos;
    _pub->should_send_setup_frame = true;
    _pub->has_receivers = false;
    _pub->has_spies = false;
    _pub->is_connected = false;
    _pub->is_end_of_stream = false;
    _pub->track_sender_limits = true;
    _pub->has_sender_released = false;

    _pub->short_sends_counter = aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_SHORT_SENDS);
    _pub->heartbeats_sent_counter = aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_HEARTBEATS_SENT);
    _pub->sender_flow_control_limits_counter = aeron_system_counter_addr(
        system_counters, AERON_SYSTEM_COUNTER_SENDER_FLOW_CONTROL_LIMITS);
    _pub->retransmits_sent_counter = aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_RETRANSMITS_SENT);
    _pub->unblocked_publications_counter = aeron_system_counter_addr(
        system_counters, AERON_SYSTEM_COUNTER_UNBLOCKED_PUBLICATIONS);

    *publication = _pub;<commit_msg>[C] Avoid initial false positive on back pressure events for network publications.<commit_after>/*
 * Copyright 2014-2019 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#if defined(__linux__)
#define _BSD_SOURCE
#define _GNU_SOURCE
#endif

#include <errno.h>
#include <string.h>
#include "aeron_socket.h"
#include <inttypes.h>
#include "concurrent/aeron_term_scanner.h"
#include "util/aeron_netutil.h"
#include "util/aeron_error.h"
#include "aeron_network_publication.h"
#include "aeron_alloc.h"
#include "media/aeron_send_channel_endpoint.h"
#include "aeron_driver_conductor.h"
#include "concurrent/aeron_logbuffer_unblocker.h"

#if !defined(HAVE_STRUCT_MMSGHDR)
struct mmsghdr
{
    struct msghdr msg_hdr;
    unsigned int msg_len;
};
#endif

int aeron_network_publication_create(
    aeron_network_publication_t **publication,
    aeron_send_channel_endpoint_t *endpoint,
    aeron_driver_context_t *context,
    int64_t registration_id,
    int32_t session_id,
    int32_t stream_id,
    int32_t initial_term_id,
    aeron_position_t *pub_pos_position,
    aeron_position_t *pub_lmt_position,
    aeron_position_t *snd_pos_position,
    aeron_position_t *snd_lmt_position,
    aeron_counter_t *snd_bpe_counter,
    aeron_flow_control_strategy_t *flow_control_strategy,
    aeron_uri_publication_params_t *params,
    bool is_exclusive,
    bool spies_simulate_connection,
    aeron_system_counters_t *system_counters)
{
    char path[AERON_MAX_PATH];
    int path_length = aeron_network_publication_location(
        path,
        sizeof(path),
        context->aeron_dir,
        endpoint->conductor_fields.udp_channel->canonical_form,
        session_id,
        stream_id,
        registration_id);

    aeron_network_publication_t *_pub = NULL;
    const uint64_t usable_fs_space = context->usable_fs_space_func(context->aeron_dir);
    const uint64_t log_length = aeron_logbuffer_compute_log_length(params->term_length, context->file_page_size);
    const int64_t now_ns = context->nano_clock();

    *publication = NULL;

    if (usable_fs_space < log_length)
    {
        aeron_set_err(
            ENOSPC,
            "Insufficient usable storage for new log of length=%" PRId64 " in %s", log_length, context->aeron_dir);
        return -1;
    }

    if (aeron_alloc((void **)&_pub, sizeof(aeron_network_publication_t)) < 0)
    {
        aeron_set_err(ENOMEM, "%s", "Could not allocate network publication");
        return -1;
    }

    _pub->log_file_name = NULL;
    if (aeron_alloc((void **)(&_pub->log_file_name), (size_t)path_length + 1) < 0)
    {
        aeron_free(_pub);
        aeron_set_err(ENOMEM, "%s", "Could not allocate network publication log_file_name");
        return -1;
    }

    if (aeron_retransmit_handler_init(
        &_pub->retransmit_handler,
        aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_INVALID_PACKETS),
        context->retransmit_unicast_delay_ns,
        context->retransmit_unicast_linger_ns) < 0)
    {
        aeron_free(_pub->log_file_name);
        aeron_free(_pub);
        aeron_set_err(aeron_errcode(), "Could not init network publication retransmit handler: %s", aeron_errmsg());
        return -1;
    }

    if (context->map_raw_log_func(
        &_pub->mapped_raw_log, path, params->is_sparse, params->term_length, context->file_page_size) < 0)
    {
        aeron_free(_pub->log_file_name);
        aeron_free(_pub);
        aeron_set_err(aeron_errcode(), "error mapping network raw log %s: %s", path, aeron_errmsg());
        return -1;
    }
    _pub->map_raw_log_close_func = context->map_raw_log_close_func;

    strncpy(_pub->log_file_name, path, (size_t)path_length);
    _pub->log_file_name[path_length] = '\0';
    _pub->log_file_name_length = (size_t)path_length;
    _pub->log_meta_data = (aeron_logbuffer_metadata_t *)(_pub->mapped_raw_log.log_meta_data.addr);

    if (params->is_replay)
    {
        int64_t term_id = params->term_id;
        int32_t term_count = params->term_id - initial_term_id;
        size_t active_index = aeron_logbuffer_index_by_term_count(term_count);

        _pub->log_meta_data->term_tail_counters[active_index] =
            (term_id * ((int64_t)1 << 32)) | params->term_offset;
        
        for (int i = 1; i < AERON_LOGBUFFER_PARTITION_COUNT; i++)
        {
            int64_t expected_term_id = (term_id + i) - AERON_LOGBUFFER_PARTITION_COUNT;
            active_index = (active_index + 1) % AERON_LOGBUFFER_PARTITION_COUNT;
            _pub->log_meta_data->term_tail_counters[active_index] = expected_term_id * ((int64_t)1 << 32);
        }

        _pub->log_meta_data->active_term_count = term_count;
    }
    else
    {
        _pub->log_meta_data->term_tail_counters[0] = initial_term_id * ((int64_t)1 << 32);
        for (int i = 1; i < AERON_LOGBUFFER_PARTITION_COUNT; i++)
        {
            int64_t expected_term_id = (initial_term_id + i) - AERON_LOGBUFFER_PARTITION_COUNT;
            _pub->log_meta_data->term_tail_counters[i] = expected_term_id * ((int64_t)1 << 32);
        }

        _pub->log_meta_data->active_term_count = 0;
    }

    _pub->log_meta_data->initial_term_id = initial_term_id;
    _pub->log_meta_data->mtu_length = (int32_t)params->mtu_length;
    _pub->log_meta_data->term_length = (int32_t)params->term_length;
    _pub->log_meta_data->page_size = (int32_t)context->file_page_size;
    _pub->log_meta_data->correlation_id = registration_id;
    _pub->log_meta_data->is_connected = 0;
    _pub->log_meta_data->end_of_stream_position = INT64_MAX;
    aeron_logbuffer_fill_default_header(
        _pub->mapped_raw_log.log_meta_data.addr, session_id, stream_id, initial_term_id);

    _pub->endpoint = endpoint;
    _pub->flow_control = flow_control_strategy;
    _pub->nano_clock = context->nano_clock;
    _pub->conductor_fields.subscribable.array = NULL;
    _pub->conductor_fields.subscribable.length = 0;
    _pub->conductor_fields.subscribable.capacity = 0;
    _pub->conductor_fields.subscribable.add_position_hook_func = aeron_network_publication_add_subscriber_hook;
    _pub->conductor_fields.subscribable.remove_position_hook_func = aeron_network_publication_remove_subscriber_hook;
    _pub->conductor_fields.subscribable.clientd = _pub;
    _pub->conductor_fields.managed_resource.registration_id = registration_id;
    _pub->conductor_fields.managed_resource.clientd = _pub;
    _pub->conductor_fields.managed_resource.incref = aeron_network_publication_incref;
    _pub->conductor_fields.managed_resource.decref = aeron_network_publication_decref;
    _pub->conductor_fields.has_reached_end_of_life = false;
    _pub->conductor_fields.clean_position = 0;
    _pub->conductor_fields.status = AERON_NETWORK_PUBLICATION_STATUS_ACTIVE;
    _pub->conductor_fields.refcnt = 1;
    _pub->conductor_fields.time_of_last_activity_ns = now_ns;
    _pub->conductor_fields.last_snd_pos = 0;
    _pub->session_id = session_id;
    _pub->stream_id = stream_id;
    _pub->pub_lmt_position.counter_id = pub_lmt_position->counter_id;
    _pub->pub_lmt_position.value_addr = pub_lmt_position->value_addr;
    _pub->pub_pos_position.counter_id = pub_pos_position->counter_id;
    _pub->pub_pos_position.value_addr = pub_pos_position->value_addr;
    _pub->snd_pos_position.counter_id = snd_pos_position->counter_id;
    _pub->snd_pos_position.value_addr = snd_pos_position->value_addr;
    _pub->snd_lmt_position.counter_id = snd_lmt_position->counter_id;
    _pub->snd_lmt_position.value_addr = snd_lmt_position->value_addr;
    _pub->snd_bpe_counter.counter_id = snd_bpe_counter->counter_id;
    _pub->snd_bpe_counter.value_addr = snd_bpe_counter->value_addr;
    _pub->initial_term_id = initial_term_id;
    _pub->term_length_mask = (int32_t)params->term_length - 1;
    _pub->position_bits_to_shift = (size_t)aeron_number_of_trailing_zeroes((int32_t)params->term_length);
    _pub->mtu_length = params->mtu_length;
    _pub->term_window_length = (int64_t)aeron_producer_window_length(
        context->publication_window_length, params->term_length);
    _pub->linger_timeout_ns = (int64_t)params->linger_timeout_ns;
    _pub->unblock_timeout_ns = (int64_t)context->publication_unblock_timeout_ns;
    _pub->connection_timeout_ns = (int64_t)context->publication_connection_timeout_ns;
    _pub->time_of_last_send_or_heartbeat_ns = now_ns - AERON_NETWORK_PUBLICATION_HEARTBEAT_TIMEOUT_NS - 1;
    _pub->time_of_last_setup_ns = now_ns - AERON_NETWORK_PUBLICATION_SETUP_TIMEOUT_NS - 1;
    _pub->status_message_deadline_ns = spies_simulate_connection ?
        now_ns : now_ns + (int64_t)context->publication_connection_timeout_ns;
    _pub->is_exclusive = is_exclusive;
    _pub->spies_simulate_connection = spies_simulate_connection;
    _pub->signal_eos = params->signal_eos;
    _pub->should_send_setup_frame = true;
    _pub->has_receivers = false;
    _pub->has_spies = false;
    _pub->is_connected = false;
    _pub->is_end_of_stream = false;
    _pub->track_sender_limits = false;
    _pub->has_sender_released = false;

    _pub->short_sends_counter = aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_SHORT_SENDS);
    _pub->heartbeats_sent_counter = aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_HEARTBEATS_SENT);
    _pub->sender_flow_control_limits_counter = aeron_system_counter_addr(
        system_counters, AERON_SYSTEM_COUNTER_SENDER_FLOW_CONTROL_LIMITS);
    _pub->retransmits_sent_counter = aeron_system_counter_addr(system_counters, AERON_SYSTEM_COUNTER_RETRANSMITS_SENT);
    _pub->unblocked_publications_counter = aeron_system_counter_addr(
        system_counters, AERON_SYSTEM_COUNTER_UNBLOCKED_PUBLICATIONS);

    *publication = _pub;
{'fraction_non_alphanumeric': 0.09570908375159926, 'fraction_numerical': 0.010136797559295346, 'mean_word_length': 4.4007440871644965, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 289, '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': '1455606', 'n_tokens_mistral': 7428, 'n_tokens_neox': 7167, 'n_words': 1040}