<commit_before>/*
 * The MIT License (MIT)
 *
 * Copyright (c) 2015 Microsoft Corporation
 * 
 * -=- Robust Distributed System Nucleus (rDSN) -=- 
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */
#include "replica.h"
#include "mutation.h"
#include "mutation_log.h"
#include "replica_stub.h"

# ifdef __TITLE__
# undef __TITLE__
# endif
# define __TITLE__ "replica"

namespace dsn { namespace replication {

// for replica::load(..) only
replica::replica(replica_stub* stub, replication_options& options)
: serverlet<replica>("replica")
{
    dassert (stub, "");
    _stub = stub;
    _app = nullptr;
        
    _options = options;

    init_state();
}

// for create new replica only used in replica_stub::on_config_proposal
replica::replica(replica_stub* stub, global_partition_id gpid, replication_options& options)
: serverlet<replica>("replica")
{
    dassert (stub, "");
    _stub = stub;
    _app = nullptr;    
    _options = options;

    init_state();
    _config.gpid = gpid;
}

void replica::init_state()
{
    _inactive_is_transient = false;
    _log = nullptr;
    _prepare_list = new prepare_list(
        0, 
        _options.staleness_for_start_prepare_for_potential_secondary,
        std::bind(
            &replica::execute_mutation,
            this,
            std::placeholders::_1
            ),
        _options.prepare_ack_on_secondary_before_logging_allowed
    );

    _config.ballot = 0;
    _config.gpid.pidx = 0;
    _config.gpid.app_id = 0;
    _config.status = PS_INACTIVE;
    _primary_states.membership.ballot = 0;
    _last_config_change_time_ms = now_ms();
}

replica::~replica(void)
{
    close();

    if (nullptr != _prepare_list)
    {
        delete _prepare_list;
        _prepare_list = nullptr;
    }

    if (nullptr != _app)
    {
        delete _app;
        _app = nullptr;
    }
}

void replica::on_client_read(const read_request_header& meta, message_ptr& request)
{
    if (status() == PS_INACTIVE || status() == PS_POTENTIAL_SECONDARY)
    {
        response_client_message(request, ERR_INVALID_STATE);
        return;
    }

    if (meta.semantic == read_semantic_t::ReadLastUpdate)
    {
        if (status() != PS_PRIMARY || 
            last_committed_decree() < _primary_states.last_prepare_decree_on_new_primary)
        {
            response_client_message(request, ERR_INVALID_STATE);
            return;
        }
    }

    dassert (_app != nullptr, "");
    _app->dispatch_rpc_call(meta.code, request, true);
}

void replica::response_client_message(message_ptr& request, error_code error, decree d/* = invalid_decree*/)
{
    if (nullptr == request)
        return;

    message_ptr resp = request->create_response();
    int err = error.get();
    resp->writer().write(err);

    dassert(error != ERR_OK, "");
    dinfo("handle replication request with rpc_id = %016llx failed, err = %s",
        request->header().rpc_id, error.to_string());

    rpc::reply(resp);
}

void replica::execute_mutation(mutation_ptr& mu)
{
    dassert (nullptr != _app, "");

    error_code err = ERR_OK;
    switch (status())
    {
    case PS_INACTIVE:
        if (_app->last_committed_decree() + 1 == mu->data.header.decree)
            err = _app->write_internal(mu, false);
        break;
    case PS_PRIMARY:<commit_msg>enable error code "defined but not used" runtime checking - bugs to be fixed next<commit_after>/*
 * The MIT License (MIT)
 *
 * Copyright (c) 2015 Microsoft Corporation
 * 
 * -=- Robust Distributed System Nucleus (rDSN) -=- 
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */
#include "replica.h"
#include "mutation.h"
#include "mutation_log.h"
#include "replica_stub.h"

# ifdef __TITLE__
# undef __TITLE__
# endif
# define __TITLE__ "replica"

namespace dsn { namespace replication {

// for replica::load(..) only
replica::replica(replica_stub* stub, replication_options& options)
: serverlet<replica>("replica")
{
    dassert (stub, "");
    _stub = stub;
    _app = nullptr;
        
    _options = options;

    init_state();
}

// for create new replica only used in replica_stub::on_config_proposal
replica::replica(replica_stub* stub, global_partition_id gpid, replication_options& options)
: serverlet<replica>("replica")
{
    dassert (stub, "");
    _stub = stub;
    _app = nullptr;    
    _options = options;

    init_state();
    _config.gpid = gpid;
}

void replica::init_state()
{
    _inactive_is_transient = false;
    _log = nullptr;
    _prepare_list = new prepare_list(
        0, 
        _options.staleness_for_start_prepare_for_potential_secondary,
        std::bind(
            &replica::execute_mutation,
            this,
            std::placeholders::_1
            ),
        _options.prepare_ack_on_secondary_before_logging_allowed
    );

    _config.ballot = 0;
    _config.gpid.pidx = 0;
    _config.gpid.app_id = 0;
    _config.status = PS_INACTIVE;
    _primary_states.membership.ballot = 0;
    _last_config_change_time_ms = now_ms();
}

replica::~replica(void)
{
    close();

    if (nullptr != _prepare_list)
    {
        delete _prepare_list;
        _prepare_list = nullptr;
    }

    if (nullptr != _app)
    {
        delete _app;
        _app = nullptr;
    }
}

void replica::on_client_read(const read_request_header& meta, message_ptr& request)
{
    if (status() == PS_INACTIVE || status() == PS_POTENTIAL_SECONDARY)
    {
        response_client_message(request, ERR_INVALID_STATE);
        return;
    }

    if (meta.semantic == read_semantic_t::ReadLastUpdate)
    {
        if (status() != PS_PRIMARY || 
            last_committed_decree() < _primary_states.last_prepare_decree_on_new_primary)
        {
            response_client_message(request, ERR_INVALID_STATE);
            return;
        }
    }

    dassert (_app != nullptr, "");
    _app->dispatch_rpc_call(meta.code, request, true);
}

void replica::response_client_message(message_ptr& request, error_code error, decree d/* = invalid_decree*/)
{
    if (nullptr == request)
        return;

    message_ptr resp = request->create_response();
    int err = error;
    resp->writer().write(err);

    dassert(error != ERR_OK, "");
    dinfo("handle replication request with rpc_id = %016llx failed, err = %s",
        request->header().rpc_id, error.to_string());

    rpc::reply(resp);
}

void replica::execute_mutation(mutation_ptr& mu)
{
    dassert (nullptr != _app, "");

    error_code err = ERR_OK;
    switch (status())
    {
    case PS_INACTIVE:
        if (_app->last_committed_decree() + 1 == mu->data.header.decree)
            err = _app->write_internal(mu, false);
        break;
    case PS_PRIMARY:
{'fraction_non_alphanumeric': 0.10285977859778597, 'fraction_numerical': 0.0032287822878228783, 'mean_word_length': 3.2410757946210267, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, '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': '1724949', 'n_tokens_mistral': 2739, 'n_tokens_neox': 2690, 'n_words': 754}