<commit_before>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "rpcsend.h"
#include "rpcsend_private.h"
#include "rpcserviceaddress.h"
#include <vespa/messagebus/network/rpcnetwork.h>
#include <vespa/messagebus/tracelevel.h>
#include <vespa/messagebus/emptyreply.h>
#include <vespa/messagebus/errorcode.h>
#include <vespa/fnet/channel.h>
#include <vespa/fnet/frt/reflection.h>
#include <vespa/vespalib/util/stringfmt.h>
#include <vespa/vespalib/util/lambdatask.h>

#include <vespa/vespalib/data/slime/cursor.h>

using vespalib::make_string;
using vespalib::makeLambdaTask;

namespace mbus {

using network::internal::ReplyContext;
using network::internal::SendContext;

namespace {

class FillByCopy final : public PayLoadFiller
{
public:
    FillByCopy(BlobRef payload) : _payload(payload) { }
    void fill(FRT_Values & v) const override {
        v.AddData(_payload.data(), _payload.size());
    }
    void fill(const vespalib::Memory & name, vespalib::slime::Cursor & v) const override {
        v.setData(name, vespalib::Memory(_payload.data(), _payload.size()));
    }
private:
    BlobRef _payload;
};

class FillByHandover final : public PayLoadFiller
{
public:
    FillByHandover(Blob payload) : _payload(std::move(payload)) { }
    void fill(FRT_Values & v) const override {
        size_t sz = _payload.size();
        v.AddData(std::move(_payload.payload()), sz);
    }
    void fill(const vespalib::Memory & name, vespalib::slime::Cursor & v) const override {
        v.setData(name, vespalib::Memory(_payload.data(), _payload.size()));
    }
private:
    mutable Blob _payload;
};

}

RPCSend::RPCSend() :
    _net(nullptr),
    _clientIdent("client"),
    _serverIdent("server")
{ }

RPCSend::~RPCSend() {}

void
RPCSend::attach(RPCNetwork &net)
{
    _net = &net;
    const string &prefix = _net->getIdentity().getServicePrefix();
    if (!prefix.empty()) {
        _clientIdent = make_string("'%s'", prefix.c_str());
        _serverIdent = _clientIdent;
    }

    FRT_ReflectionBuilder builder(&_net->getSupervisor());
    build(builder);
}

void
RPCSend::replyError(FRT_RPCRequest *req, const vespalib::Version &version, uint32_t traceLevel, const Error &err)
{
    Reply::UP reply(new EmptyReply());
    reply->setContext(Context(new ReplyContext(*req, version)));
    reply->getTrace().setLevel(traceLevel);
    reply->addError(err);
    handleReply(std::move(reply));
}

void
RPCSend::handleDiscard(Context ctx)
{
    ReplyContext::UP tmp(static_cast<ReplyContext*>(ctx.value.PTR));
    FRT_RPCRequest &req = tmp->getRequest();
    FNET_Channel *chn = req.GetContext()._value.CHANNEL;
    req.SubRef();
    chn->Free();
}

void
RPCSend::sendByHandover(RoutingNode &recipient, const vespalib::Version &version, Blob payload, uint64_t timeRemaining)
{
    send(recipient, version, FillByHandover(std::move(payload)), timeRemaining);
}

void
RPCSend::send(RoutingNode &recipient, const vespalib::Version &version, BlobRef payload, uint64_t timeRemaining)
{
    send(recipient, version, FillByCopy(payload), timeRemaining);
}

void
RPCSend::send(RoutingNode &recipient, const vespalib::Version &version,
              const PayLoadFiller & payload, uint64_t timeRemaining)
{
    SendContext::UP ctx(new SendContext(recipient, timeRemaining));
    RPCServiceAddress &address = static_cast<RPCServiceAddress&>(recipient.getServiceAddress());
    const Message &msg = recipient.getMessage();
    Route route = recipient.getRoute();
    Hop hop = route.removeHop(0);

    FRT_RPCRequest *req = _net->allocRequest();
    encodeRequest(*req, version, route, address, msg, recipient.getTrace().getLevel(),  payload, timeRemaining);

    if (ctx->getTrace().shouldTrace(TraceLevel::SEND_RECEIVE)) {
        ctx->getTrace().trace(TraceLevel::SEND_RECEIVE,
                              make_string("Sending message (version %s) from %s to '%s' with %.2f seconds timeout.",
                                          version.toString().c_str(), _clientIdent.c_str(),
                                          address.getServiceName().c_str(), ctx->getTimeout()));
    }

    if (hop.getIgnoreResult()) {
        address.getTarget().getFRTTarget().InvokeVoid(req);
        if (ctx->getTrace().shouldTrace(TraceLevel::SEND_RECEIVE)) {
            ctx->getTrace().trace(TraceLevel::SEND_RECEIVE,
                                  make_string("Not waiting for a reply from '%s'.", address.getServiceName().c_str()));
        }
        Reply::UP reply(new EmptyReply());
        reply->getTrace().swap(ctx->getTrace());
        _net->getOwner().deliverReply(std::move(reply), recipient);
    } else {
        SendContext *ptr = ctx.release();
        req->SetContext(FNET_Context(ptr));
        address.getTarget().getFRTTarget().InvokeAsync(req, ptr->getTimeout(), this);
    }
}

void
RPCSend::RequestDone(FRT_RPCRequest *req)
{
    _net->getExecutor().execute(makeLambdaTask([this, req]() { doRequestDone(req);}));
}
<commit_msg>Assert that task is accepted.<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "rpcsend.h"
#include "rpcsend_private.h"
#include "rpcserviceaddress.h"
#include <vespa/messagebus/network/rpcnetwork.h>
#include <vespa/messagebus/tracelevel.h>
#include <vespa/messagebus/emptyreply.h>
#include <vespa/messagebus/errorcode.h>
#include <vespa/fnet/channel.h>
#include <vespa/fnet/frt/reflection.h>
#include <vespa/vespalib/util/stringfmt.h>
#include <vespa/vespalib/util/lambdatask.h>

#include <vespa/vespalib/data/slime/cursor.h>

using vespalib::make_string;
using vespalib::makeLambdaTask;

namespace mbus {

using network::internal::ReplyContext;
using network::internal::SendContext;

namespace {

class FillByCopy final : public PayLoadFiller
{
public:
    FillByCopy(BlobRef payload) : _payload(payload) { }
    void fill(FRT_Values & v) const override {
        v.AddData(_payload.data(), _payload.size());
    }
    void fill(const vespalib::Memory & name, vespalib::slime::Cursor & v) const override {
        v.setData(name, vespalib::Memory(_payload.data(), _payload.size()));
    }
private:
    BlobRef _payload;
};

class FillByHandover final : public PayLoadFiller
{
public:
    FillByHandover(Blob payload) : _payload(std::move(payload)) { }
    void fill(FRT_Values & v) const override {
        size_t sz = _payload.size();
        v.AddData(std::move(_payload.payload()), sz);
    }
    void fill(const vespalib::Memory & name, vespalib::slime::Cursor & v) const override {
        v.setData(name, vespalib::Memory(_payload.data(), _payload.size()));
    }
private:
    mutable Blob _payload;
};

}

RPCSend::RPCSend() :
    _net(nullptr),
    _clientIdent("client"),
    _serverIdent("server")
{ }

RPCSend::~RPCSend() {}

void
RPCSend::attach(RPCNetwork &net)
{
    _net = &net;
    const string &prefix = _net->getIdentity().getServicePrefix();
    if (!prefix.empty()) {
        _clientIdent = make_string("'%s'", prefix.c_str());
        _serverIdent = _clientIdent;
    }

    FRT_ReflectionBuilder builder(&_net->getSupervisor());
    build(builder);
}

void
RPCSend::replyError(FRT_RPCRequest *req, const vespalib::Version &version, uint32_t traceLevel, const Error &err)
{
    Reply::UP reply(new EmptyReply());
    reply->setContext(Context(new ReplyContext(*req, version)));
    reply->getTrace().setLevel(traceLevel);
    reply->addError(err);
    handleReply(std::move(reply));
}

void
RPCSend::handleDiscard(Context ctx)
{
    ReplyContext::UP tmp(static_cast<ReplyContext*>(ctx.value.PTR));
    FRT_RPCRequest &req = tmp->getRequest();
    FNET_Channel *chn = req.GetContext()._value.CHANNEL;
    req.SubRef();
    chn->Free();
}

void
RPCSend::sendByHandover(RoutingNode &recipient, const vespalib::Version &version, Blob payload, uint64_t timeRemaining)
{
    send(recipient, version, FillByHandover(std::move(payload)), timeRemaining);
}

void
RPCSend::send(RoutingNode &recipient, const vespalib::Version &version, BlobRef payload, uint64_t timeRemaining)
{
    send(recipient, version, FillByCopy(payload), timeRemaining);
}

void
RPCSend::send(RoutingNode &recipient, const vespalib::Version &version,
              const PayLoadFiller & payload, uint64_t timeRemaining)
{
    SendContext::UP ctx(new SendContext(recipient, timeRemaining));
    RPCServiceAddress &address = static_cast<RPCServiceAddress&>(recipient.getServiceAddress());
    const Message &msg = recipient.getMessage();
    Route route = recipient.getRoute();
    Hop hop = route.removeHop(0);

    FRT_RPCRequest *req = _net->allocRequest();
    encodeRequest(*req, version, route, address, msg, recipient.getTrace().getLevel(),  payload, timeRemaining);

    if (ctx->getTrace().shouldTrace(TraceLevel::SEND_RECEIVE)) {
        ctx->getTrace().trace(TraceLevel::SEND_RECEIVE,
                              make_string("Sending message (version %s) from %s to '%s' with %.2f seconds timeout.",
                                          version.toString().c_str(), _clientIdent.c_str(),
                                          address.getServiceName().c_str(), ctx->getTimeout()));
    }

    if (hop.getIgnoreResult()) {
        address.getTarget().getFRTTarget().InvokeVoid(req);
        if (ctx->getTrace().shouldTrace(TraceLevel::SEND_RECEIVE)) {
            ctx->getTrace().trace(TraceLevel::SEND_RECEIVE,
                                  make_string("Not waiting for a reply from '%s'.", address.getServiceName().c_str()));
        }
        Reply::UP reply(new EmptyReply());
        reply->getTrace().swap(ctx->getTrace());
        _net->getOwner().deliverReply(std::move(reply), recipient);
    } else {
        SendContext *ptr = ctx.release();
        req->SetContext(FNET_Context(ptr));
        address.getTarget().getFRTTarget().InvokeAsync(req, ptr->getTimeout(), this);
    }
}

void
RPCSend::RequestDone(FRT_RPCRequest *req)
{
    auto task = _net->getExecutor().execute(makeLambdaTask([this, req]() { doRequestDone(req);}));
    assert(!task);
}

{'fraction_non_alphanumeric': 0.14508286196288578, 'fraction_numerical': 0.003175548278257418, 'mean_word_length': 4.008946322067595, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5995040', 'n_tokens_mistral': 3228, 'n_tokens_neox': 3165, 'n_words': 599}