void replica::assign_primary(configuration_update_request& proposal)
{
    dassert(proposal.node == primary_address(), "");

    if (status() == PS_PRIMARY)
    {
        dwarn(
            "%s: invalid assgin primary proposal as the node is in %s",
            name(),
            enum_to_string(status()));
        return;
    }

    if (proposal.type == CT_UPGRADE_TO_PRIMARY 
        && (status() != PS_SECONDARY || _secondary_states.checkpoint_task != nullptr))
    {
        dwarn(
            "%s: invalid upgrade to primary proposal as the node is in %s or during checkpointing",
            name(),
            enum_to_string(status()));

        // TODO: tell meta server so new primary is built more quickly
        return;
    }

    proposal.config.primary = primary_address();
    replica_helper::remove_node(primary_address(), proposal.config.secondaries);

    update_configuration_on_meta_server(proposal.type, proposal.node, proposal.config);
}

static gpointer
sv_to_callback_data (SV * sv,
                     GPerlI11nInvocationInfo * invocation_info)
{
	GSList *l;
	if (!invocation_info)
		return NULL;
	for (l = invocation_info->callback_infos; l != NULL; l = l->next) {
		GPerlI11nPerlCallbackInfo *callback_info = l->data;
		if (callback_info->data_pos == ((gint) invocation_info->current_pos)) {
			dwarn ("      user data for Perl callback %p\n",
			       callback_info);
			attach_perl_callback_data (callback_info, sv);
			/* If the user did not specify any code and data and if
			 * there is no destroy notify function, then there is
			 * no need for us to pass on our callback info struct
			 * as C user data.  Some libraries (e.g., vte) even
			 * assert that the C user data be NULL if the C
			 * function pointer is NULL. */
			if (!gperl_sv_is_defined (callback_info->code) &&
			    !gperl_sv_is_defined (callback_info->data) &&
			    -1 == callback_info->destroy_pos)
			{
				dwarn ("        handing over NULL");
				return NULL;
			}
			return callback_info;
		}
	}
	if (invocation_info->is_callback) {
		GPerlI11nCCallbackInfo *wrapper = INT2PTR (GPerlI11nCCallbackInfo*, SvIV (sv));
		dwarn ("      user data for C callback %p\n", wrapper);
		return wrapper->data;
	}

TEST(tools_hpc, tail_logger_cb)
{
    std::string output;
    bool ret = dsn::command_manager::instance().run_command("tail-log", output);
    if(!ret)
        return;

    EXPECT_TRUE(strcmp(output.c_str(), "invalid arguments for tail-log command") == 0 );

    std::ostringstream in;
    in << "tail-log 12345 4 1 " << dsn::utils::get_current_tid();
    std::this_thread::sleep_for(std::chrono::seconds(3));
    dwarn("key:12345");
    std::this_thread::sleep_for(std::chrono::seconds(2));
    dwarn("key:12345");
    output.clear();
    dsn::command_manager::instance().run_command(in.str().c_str(), output);
    EXPECT_TRUE(strstr(output.c_str(), "In total (1) log entries are found between") != nullptr);
    dsn::command_manager::instance().run_command("tail-log-dump", output);

    ::dsn::logging_provider* logger = ::dsn::service_engine::fast_instance().logging();
    if (logger != nullptr)
    {
        logger->flush();
    }
}

bool failure_detector::end_ping_internal(::dsn::error_code err, const beacon_ack& ack)
{
    /*
     * the caller of the end_ping_internal should lock necessarily!!!
     */
    uint64_t beacon_send_time = ack.time;
    auto node = ack.this_node;
    uint64_t now = now_ms();

    master_map::iterator itr = _masters.find(node);

    if (itr == _masters.end())
    {
        dwarn("received beacon ack without corresponding master, ignore it, "
            "remote_master[%s], local_worker[%s]",
            node.to_string(), primary_address().to_string());
        return false;
    }

    master_record& record = itr->second;
    if (!ack.allowed)
    {
        dwarn("worker rejected, stop sending beacon message, "
            "remote_master[%s], local_worker[%s]",
            node.to_string(), primary_address().to_string());
        record.rejected = true;
        record.send_beacon_timer->cancel(true);
        return false;
    }

    if (!is_time_greater_than(beacon_send_time, record.last_send_time_for_beacon_with_ack))
    {
        // out-dated beacon acks, do nothing
        dinfo("ignore out dated beacon acks");
        return false;
    }

    // now the ack is applicable

    if (err != ERR_OK)
    {
        dwarn("ping master failed, err=%s", err.to_string());
        return true;
    }

    // update last_send_time_for_beacon_with_ack
    record.last_send_time_for_beacon_with_ack = beacon_send_time;
    record.rejected = false;

    if (record.is_alive == false
        && now - record.last_send_time_for_beacon_with_ack <= _lease_milliseconds)
    {
        // report master connected
        report(node, true, true);
        itr->second.is_alive = true;
        on_master_connected(node);
    }
    return true;
}

void hpc_rpc_session::do_read(int sz)
{
    add_ref();
    _read_event.callback = [this](int err, uint32_t length, uintptr_t lolp)
    {
        //dinfo("WSARecv completed, err = %d, size = %u", err, length);
        dassert((LPOVERLAPPED)lolp == &_read_event.olp, "must be exact this overlapped");
        if (err != ERROR_SUCCESS)
        {
            dwarn("WSARecv failed, err = %d", err);
            on_failure();
        }
        else
        {
            int read_next;
            message_ex* msg = _parser->get_message_on_receive((int)length, read_next);

            while (msg != nullptr)
            {
                this->on_read_completed(msg);
                msg = _parser->get_message_on_receive(0, read_next);
            }

            do_read(read_next);
        }

        release_ref();
    };
    memset(&_read_event.olp, 0, sizeof(_read_event.olp));

    WSABUF buf[1];

    void* ptr = _parser->read_buffer_ptr((int)sz);
    int remaining = _parser->read_buffer_capacity();
    buf[0].buf = (char*)ptr;
    buf[0].len = remaining;

    DWORD bytes = 0;
    DWORD flag = 0;
    int rt = WSARecv(
                 _socket,
                 buf,
                 1,
                 &bytes,
                 &flag,
                 &_read_event.olp,
                 NULL
             );

    if (SOCKET_ERROR == rt && (WSAGetLastError() != ERROR_IO_PENDING))
    {
        dwarn("WSARecv failed, err = %d", ::WSAGetLastError());
        release_ref();
        on_failure();
    }

    //dinfo("WSARecv called, err = %d", rt);
}

void task_queue::enqueue_internal(task* task)
{
    auto& sp = task->spec();
    auto throttle_mode = sp.rpc_request_throttling_mode;
    if (throttle_mode != TM_NONE)
    {        
        int ac_value = 0;
        if (_spec->enable_virtual_queue_throttling)
        {
            ac_value = _virtual_queue_length;
        }
        else
        {
            ac_value = count();
        }
               
        if (throttle_mode == TM_DELAY)
        {
            int delay_ms = sp.rpc_request_delayer.delay(ac_value, _spec->queue_length_throttling_threshold);
            if (delay_ms > 0)
            {
                auto rtask = static_cast<rpc_request_task*>(task);
                rtask->get_request()->io_session->delay_recv(delay_ms);

                dwarn("too many pending tasks (%d), delay traffic from %s for %d milliseconds",
                    ac_value,
                    rtask->get_request()->header->from_address.to_string(),
                    delay_ms
                    );
            }
        }
        else
        {
            dbg_dassert(TM_REJECT == throttle_mode, "unknow mode %d", (int)throttle_mode);

            if (ac_value > _spec->queue_length_throttling_threshold)
            {
                auto rtask = static_cast<rpc_request_task*>(task);
                auto resp = rtask->get_request()->create_response();
                task::get_current_rpc()->reply(resp, ERR_BUSY);

                dwarn("too many pending tasks (%d), reject message from %s with trace_id = %016" PRIx64,
                    ac_value,
                    rtask->get_request()->header->from_address.to_string(),
                    rtask->get_request()->header->trace_id
                    );

                task->release_ref(); // added in task::enqueue(pool)
                return;
            }
        }
    }

    tls_dsn.last_worker_queue_size = increase_count();
    enqueue(task);
}

        error_code hpc_network_provider::start(rpc_channel channel, int port, bool client_only, io_modifer& ctx)
        {
            if (_listen_fd != -1)
                return ERR_SERVICE_ALREADY_RUNNING;

            _looper = get_io_looper(node(), ctx.queue, ctx.mode);

            dassert(channel == RPC_CHANNEL_TCP || channel == RPC_CHANNEL_UDP,
                "invalid given channel %s", channel.to_string());

            char hostname[128];
            gethostname(hostname, sizeof(hostname));
            _address = ::dsn::rpc_address(HOST_TYPE_IPV4, hostname, port);

            if (!client_only)
            {
                struct sockaddr_in addr;
                addr.sin_family = AF_INET;
                addr.sin_addr.s_addr = INADDR_ANY;
                addr.sin_port = htons(port);

                _listen_fd = create_tcp_socket(&addr);
                if (_listen_fd == -1)
                {
                    dassert(false, "cannot create listen socket");
                }

                int forcereuse = 1;
                if (setsockopt(_listen_fd, SOL_SOCKET, SO_REUSEADDR,
                    (char*)&forcereuse, sizeof(forcereuse)) != 0)
                {
                    dwarn("setsockopt SO_REUSEDADDR failed, err = %s", strerror(errno));
                }

                if (listen(_listen_fd, SOMAXCONN) != 0)
                {
                    dwarn("listen failed, err = %s", strerror(errno));
                    return ERR_NETWORK_START_FAILED;
                }

                _accept_event.callback = [this](int err, uint32_t size, uintptr_t lpolp)
                {
                    this->do_accept();
                };

                // bind for accept
                _looper->bind_io_handle((dsn_handle_t)(intptr_t)_listen_fd, &_accept_event.callback,
                    EVFILT_READ,
                    nullptr // network_provider is a global object
                    );
            }

            return ERR_OK;
        }

void hpc_rpc_session::connect()
{
    if (!try_connecting())
        return;

    _connect_event.callback = [this](int err, uint32_t io_size, uintptr_t lpolp)
    {
        //dinfo("ConnectEx completed, err = %d, size = %u", err, io_size);
        if (err != ERROR_SUCCESS)
        {
            dwarn("ConnectEx failed, err = %d", err);
            this->on_failure();
        }
        else
        {
            dinfo("client session %s:%hu connected",
                  _remote_addr.name(),
                  _remote_addr.port()
                 );

            set_connected();
            on_send_completed(nullptr);
            do_read();
        }
        this->release_ref(); // added before ConnectEx
    };
    memset(&_connect_event.olp, 0, sizeof(_connect_event.olp));

    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(_remote_addr.ip());
    addr.sin_port = htons(_remote_addr.port());

    this->add_ref(); // released in _connect_event.callback
    BOOL rt = s_lpfnConnectEx(
                  _socket,
                  (struct sockaddr*)&addr,
                  (int)sizeof(addr),
                  0,
                  0,
                  0,
                  &_connect_event.olp
              );

    if (!rt && (WSAGetLastError() != ERROR_IO_PENDING))
    {
        dwarn("ConnectEx failed, err = %d", ::WSAGetLastError());
        this->release_ref();

        on_failure();
    }
}

static void
prepare_invocation_info (GPerlI11nInvocationInfo *iinfo,
                         GICallableInfo *info)
{
	gint orig_n_args;
	guint i;

	dwarn ("invoke: %s\n"
	       "  n_args: %d\n",
	       g_base_info_get_name (info),
	       g_callable_info_get_n_args (info));

	iinfo->interface = info;

	iinfo->is_function = GI_IS_FUNCTION_INFO (info);
	iinfo->is_vfunc = GI_IS_VFUNC_INFO (info);
	iinfo->is_callback = (g_base_info_get_type (info) == GI_INFO_TYPE_CALLBACK);
	iinfo->is_signal = GI_IS_SIGNAL_INFO (info);
	dwarn ("  is_function = %d, is_vfunc = %d, is_callback = %d\n",
	       iinfo->is_function, iinfo->is_vfunc, iinfo->is_callback);

	orig_n_args = g_callable_info_get_n_args (info);
	g_assert (orig_n_args >= 0);
	iinfo->n_args = (guint) orig_n_args;

	if (iinfo->n_args) {
		iinfo->arg_infos = gperl_alloc_temp (sizeof (GITypeInfo*) * iinfo->n_args);
		iinfo->arg_types = gperl_alloc_temp (sizeof (GITypeInfo*) * iinfo->n_args);
		iinfo->aux_args = gperl_alloc_temp (sizeof (GIArgument) * iinfo->n_args);
	} else {
		iinfo->arg_infos = NULL;
		iinfo->arg_types = NULL;
		iinfo->aux_args = NULL;
	}

	for (i = 0 ; i < iinfo->n_args ; i++) {
		iinfo->arg_infos[i] = g_callable_info_get_arg (info, (gint) i);
		iinfo->arg_types[i] = g_arg_info_get_type (iinfo->arg_infos[i]);
	}

	iinfo->return_type_info = g_callable_info_get_return_type (info);
	iinfo->has_return_value =
		GI_TYPE_TAG_VOID != g_type_info_get_tag (iinfo->return_type_info);
	iinfo->return_type_ffi = g_type_info_get_ffi_type (iinfo->return_type_info);
	iinfo->return_type_transfer = g_callable_info_get_caller_owns (info);

	iinfo->callback_infos = NULL;
	iinfo->array_infos = NULL;

	iinfo->free_after_call = NULL;
}

        void asio_rpc_session::set_options()
        {
            if (_socket->is_open())
            {
                try {
                    boost::asio::socket_base::send_buffer_size option, option2(16 * 1024 * 1024);
                    _socket->get_option(option);
                    int old = option.value();
                    _socket->set_option(option2);
                    _socket->get_option(option);

                    dinfo("boost asio send buffer size is %u, set as 16MB, now is %u",
                            old, option.value());

                    boost::asio::socket_base::receive_buffer_size option3, option4(16 * 1024 * 1024);
                    _socket->get_option(option3);
                    old = option3.value();
                    _socket->set_option(option4);
                    _socket->get_option(option3);

                    dinfo("boost asio recv buffer size is %u, set as 16MB, now is %u",
                            old, option.value());
                }
                catch (std::exception& ex)
                {
                    dwarn("network session 0x%x:%hu set socket option failed, err = %s",
                        remote_address().ip(),
                        remote_address().port(),
                        ex.what()
                        );
                }
            }
        }

    std::shared_ptr<uri_resolver> uri_resolver_manager::get(rpc_uri_address* uri) const
    {
        std::shared_ptr<uri_resolver> ret = nullptr;
        auto pr = uri->get_uri_components();
        if (pr.second.length() == 0)
            return ret;
        
        {
            utils::auto_read_lock l(_lock);
            auto it = _resolvers.find(pr.first);
            if (it != _resolvers.end())
                ret = it->second;
        }
        
        if (ret == nullptr)
        {
            dwarn(
                "cannot find uri resolver for uri '%s' with resolver address as '%s', "
                "please fix it by setting up a uri resolver section in config file, as follows:\r\n"
                "[uri-resolver.%s]\r\n"
                "factory = partition-resolver-factory (e.g., partition_resolver_simple)\r\n"
                "arguments = uri-resolver-arguments (e.g., localhost:34601,localhost:34602)\r\n",
                uri->uri(),
                pr.first.c_str(),
                pr.first.c_str()
                );
        }

        return ret;
    }

void meta_server_failure_detector::on_worker_disconnected(const std::vector<end_point>& nodes)
{
    if (!is_primary())
    {
        return;
    }

    node_states states;
    for (auto& n : nodes)
    {
        states.push_back(std::make_pair(n, false));

        dwarn("client expired: %s:%hu", n.name.c_str(), n.port);
    }
    
    machine_fail_updates pris;
    _state->set_node_state(states, &pris);
    
    for (auto& pri : pris)
    {
        dinfo("%d.%d primary node for %s:%hu is gone, update configuration on meta server", 
            pri.first.app_id,
            pri.first.pidx,
            pri.second->node.name.c_str(),
            pri.second->node.port
            );
        _svc->update_configuration(pri.second);
    }
}

        void hpc_rpc_session::connect()
        {
            if (!try_connecting())
                return;
            
            dassert(_socket != -1, "invalid given socket handle");

            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_addr.s_addr = htonl(_remote_addr.ip());
            addr.sin_port = htons(_remote_addr.port());

            int rt = ::connect(_socket, (struct sockaddr*)&addr, (int)sizeof(addr));
            int err = errno;
            dinfo("(s = %d) call connect to %s:%hu, return %d, err = %s",
                _socket,
                _remote_addr.name(),
                _remote_addr.port(),
                rt,
                strerror(err)
                );

            if (rt == -1 && err != EINPROGRESS)
            {
                dwarn("(s = %d) connect failed, err = %s", _socket, strerror(err));
                on_failure();
                return;
            }

            // bind for connect
            _looper->bind_io_handle((dsn_handle_t)(intptr_t)_socket, &_ready_event,
                EVFILT_WRITE,
                this
                );
        }

        void native_linux_aio_provider::get_event()
        {
            struct io_event events[1];
            int ret;

            const char* name = ::dsn::tools::get_service_node_name(node());
            char buffer[128];
            sprintf(buffer, "%s.aio", name);
            task_worker::set_name(buffer);

            while (true)
            {
                ret = io_getevents(_ctx, 1, 1, events, NULL);
                if (ret > 0) // should be 1
                {
                    dassert(ret == 1, "");
                    struct iocb *io = events[0].obj;
                    complete_aio(io, static_cast<int>(events[0].res), static_cast<int>(events[0].res2));
                }
                else
                {
                    dwarn("io_getevents returns %d, you probably want to try on another machine:-(", ret);
                }
            }
        }

static void
release_perl_callback (gpointer data)
{
	GPerlI11nPerlCallbackInfo *info = data;
	dwarn ("info = %p\n", info);

	/* g_callable_info_free_closure reaches into info->cif, so it needs to
	 * be called before we free it.  See
	 * <https://bugzilla.gnome.org/show_bug.cgi?id=652954>. */
	if (info->closure)
		g_callable_info_free_closure (info->interface, info->closure);
	if (info->cif)
		g_free (info->cif);

	if (info->interface)
		g_base_info_unref ((GIBaseInfo*) info->interface);

	if (info->code)
		SvREFCNT_dec (info->code);
	if (info->data)
		SvREFCNT_dec (info->data);
	if (info->sub_name)
		g_free (info->sub_name);

	if (info->args_converter)
		SvREFCNT_dec (info->args_converter);

	g_free (info);
}

    void sim_client_session::send(message_ptr& msg)
    {
        sim_network_provider* rnet = nullptr;
        if (!s_switch[task_spec::get(msg->header().local_rpc_code)->rpc_call_channel].get(msg->header().to_address, rnet))
        {
            dwarn("cannot find destination node %s:%d in simulator", 
                msg->header().to_address.name.c_str(), 
                static_cast<int>(msg->header().to_address.port)
                );
            return;
        }
        
        auto server_session = rnet->get_server_session(_net.address());
        if (nullptr == server_session)
        {
            rpc_client_session_ptr cptr = this;
            server_session.reset(new sim_server_session(*rnet, _net.address(), cptr));
            rnet->on_server_session_accepted(server_session);
        }

        message_ptr recv_msg(new message(msg->writer().get_buffer()));
        recv_msg->header().from_address = msg->header().from_address;
        recv_msg->header().to_address = msg->header().to_address;

        server_session->on_recv_request(recv_msg, 
            recv_msg->header().from_address == recv_msg->header().to_address ?
            0 : rnet->net_delay_milliseconds()
            );
    }

void meta_server_failure_detector::on_worker_disconnected(const std::vector< ::dsn::rpc_address>& nodes)
{
    if (!is_primary())
    {
        return;
    }

    if (!_svc->_started)
    {
        return;
    }

    node_states states;
    for (auto& n : nodes)
    {
        states.push_back(std::make_pair(n, false));

        dwarn("client expired: %s", n.to_string());
    }
    
    machine_fail_updates pris;
    _state->set_node_state(states, &pris);
    
    for (auto& pri : pris)
    {
        dinfo("%d.%d primary node for %s is gone, update configuration on meta server", 
            pri.first.app_id,
            pri.first.pidx,
            pri.second->node.to_string()
            );
        _svc->update_configuration_on_machine_failure(pri.second);
    }
}

static void check_wait_task(task *waitee)
{
    lock_checker::check_wait_safety();

    // not in worker thread
    if (task::get_current_worker() == nullptr)
        return;

    // caller and callee don't share the same thread pool,
    if (waitee->spec().type != TASK_TYPE_RPC_RESPONSE &&
        (waitee->spec().pool_code != task::get_current_worker()->pool_spec().pool_code))
        return;

    // callee is empty
    if (waitee->is_empty())
        return;

    // there are enough concurrency
    if (!task::get_current_worker()->pool_spec().partitioned &&
        task::get_current_worker()->pool_spec().worker_count > 1)
        return;

    dwarn("task %s waits for another task %s sharing the same thread pool "
          "- will lead to deadlocks easily (e.g., when worker_count = 1 or when the pool "
          "is partitioned)",
          task::get_current_task()->spec().code.to_string(),
          waitee->spec().code.to_string());
}

        void net_io::set_options()
        {
            if (_socket.is_open())
            {
                try {
                    boost::asio::socket_base::send_buffer_size option, option2(16 * 1024 * 1024);
                    _socket.get_option(option);
                    int old = option.value();
                    _socket.set_option(option2);
                    _socket.get_option(option);

                    /*ddebug("boost asio send buffer size is %u, set as 16MB, now is %u",
                    old, option.value());*/

                    boost::asio::socket_base::receive_buffer_size option3, option4(16 * 1024 * 1024);
                    _socket.get_option(option3);
                    old = option3.value();
                    _socket.set_option(option4);
                    _socket.get_option(option3);
                    /*ddebug("boost asio recv buffer size is %u, set as 16MB, now is %u",
                    old, option.value());*/
                }
                catch (std::exception& ex)
                {
                    dwarn("network session %s:%d set socket option failed, err = %s",
                        _remote_addr.to_ip_string().c_str(),
                        static_cast<int>(_remote_addr.port),
                        ex.what()
                        );
                }
            }
        }

error_code hpc_network_provider::start(rpc_channel channel, int port, bool client_only, io_modifer& ctx)
{
    if (_listen_fd != INVALID_SOCKET)
        return ERR_SERVICE_ALREADY_RUNNING;

    _looper = get_io_looper(node(), ctx.queue, ctx.mode);

    dassert(channel == RPC_CHANNEL_TCP || channel == RPC_CHANNEL_UDP,
            "invalid given channel %s", channel.to_string());

    char name[128];
    gethostname(name, sizeof(name));
    _address = ::dsn::rpc_address(HOST_TYPE_IPV4, name, port);

    if (!client_only)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = INADDR_ANY;
        addr.sin_port = htons(port);

        _listen_fd = create_tcp_socket(&addr);
        if (_listen_fd == INVALID_SOCKET)
        {
            dassert(false, "");
        }

        int forcereuse = 1;
        if (setsockopt(_listen_fd, SOL_SOCKET, SO_REUSEADDR,
                       (char*)&forcereuse, sizeof(forcereuse)) != 0)
        {
            dwarn("setsockopt SO_REUSEDADDR failed, err = %d", ::GetLastError());
        }

        _looper->bind_io_handle((dsn_handle_t)_listen_fd, &_accept_event.callback);

        if (listen(_listen_fd, SOMAXCONN) != 0)
        {
            dwarn("listen failed, err = %d", ::GetLastError());
            return ERR_NETWORK_START_FAILED;
        }

        do_accept();
    }

    return ERR_OK;
}