path: root/src/stateless/cp/trex_stateless_port.cpp

/*
 Itay Marom
 Cisco Systems, Inc.
*/

/*
Copyright (c) 2015-2015 Cisco Systems, Inc.

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.
*/

#include <trex_stateless.h>
#include <trex_stateless_port.h>
#include <trex_stateless_messaging.h>
#include <trex_streams_compiler.h>
#include <common/basic_utils.h>
#include <common/captureFile.h>

#include <string>

#ifndef TREX_RPC_MOCK_SERVER

// DPDK c++ issue 
#ifndef UINT8_MAX
    #define UINT8_MAX 255
#endif

#ifndef UINT16_MAX
    #define UINT16_MAX 0xFFFF
#endif

// DPDK c++ issue 
#endif

#include <os_time.h>

void
port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list);

using namespace std;



/***************************
 * trex DP events handlers
 * 
 **************************/
class AsyncStopEvent : public TrexDpPortEvent {

protected:
    /**
     * when an async event occurs (all cores have reported in)
     * 
     * @author imarom (29-Feb-16)
     */
    virtual void on_event() {
        get_port()->change_state(TrexStatelessPort::PORT_STATE_STREAMS);

        get_port()->common_port_stop_actions(true);

        assert(get_port()->m_pending_async_stop_event != TrexDpPortEvents::INVALID_ID);
        get_port()->m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
    }

    /**
     * when a DP core encountered an error
     * 
     * @author imarom (20-Apr-16)
     */
    virtual void on_error(int thread_id) {
        Json::Value data;

        data["port_id"]   = get_port()->get_port_id();
        data["thread_id"] = thread_id;

        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_ERROR, data);
    }
};

/***************************
 * trex stateless port
 * 
 **************************/
TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api) : m_dp_events(this) {
    std::vector<std::pair<uint8_t, uint8_t>> core_pair_list;

    m_port_id = port_id;
    m_port_state = PORT_STATE_IDLE;

    /* get the platform specific data */
    api->get_interface_info(port_id, m_api_info);

    /* get RX caps */
    api->get_interface_stat_info(port_id, m_rx_count_num, m_rx_caps);

    /* get the DP cores belonging to this port */
    api->port_id_to_cores(m_port_id, core_pair_list);

    for (auto core_pair : core_pair_list) {

        /* send the core id */
        m_cores_id_list.push_back(core_pair.first);
    }

    m_graph_obj = NULL;

    m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;
}

TrexStatelessPort::~TrexStatelessPort() {

    stop_traffic();
    remove_and_delete_all_streams();
}

/**
 * acquire the port
 * 
 * @author imarom (09-Nov-15)
 * 
 * @param user 
 * @param force 
 */
void 
TrexStatelessPort::acquire(const std::string &user, uint32_t session_id, bool force) {

    bool used_force = !get_owner().is_free() && force;

    if (get_owner().is_free() || force) {
        get_owner().own(user, session_id);

    } else {
        /* not same user or session id and not force - report error */
        if (get_owner().get_name() == user) {
            throw TrexException("port is already owned by another session of '" + user + "'");
        } else {
            throw TrexException("port is already taken by '" + get_owner().get_name() + "'");
        }
    }

    Json::Value data;

    data["port_id"]    = m_port_id;
    data["who"]        = user;
    data["session_id"] = session_id;
    data["force"]      = used_force;

    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_ACQUIRED, data);

}

void
TrexStatelessPort::release(void) {
    

    Json::Value data;

    data["port_id"]    = m_port_id;
    data["who"]        = get_owner().get_name();
    data["session_id"] = get_owner().get_session_id();

    get_owner().release();

    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_RELEASED, data);
}

/**
 * starts the traffic on the port
 * 
 */
void
TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration, bool force) {

    /* command allowed only on state stream */
    verify_state(PORT_STATE_STREAMS);

    /* just making sure no leftovers... */
    delete_streams_graph();

    /* on start - we can only provide absolute values */
    assert(mul.m_op == TrexPortMultiplier::OP_ABS);

    /* caclulate the effective factor for DP */
    double factor = calculate_effective_factor(mul, force);

    /* fetch all the streams from the table */
    vector<TrexStream *> streams;
    get_object_list(streams);


    /* compiler it */
    std::vector<TrexStreamsCompiledObj *> compiled_objs;
    std::string fail_msg;

    TrexStreamsCompiler compiler;
    bool rc = compiler.compile(m_port_id,
                               streams,
                               compiled_objs,
                               get_dp_core_count(),
                               factor,
                               &fail_msg);
    if (!rc) {
        throw TrexException(fail_msg);
    }

    /* generate a message to all the relevant DP cores to start transmitting */
    assert(m_pending_async_stop_event == TrexDpPortEvents::INVALID_ID);
    m_pending_async_stop_event = m_dp_events.create_event(new AsyncStopEvent());

    /* update object status */
    m_factor = factor;
    m_last_all_streams_continues = compiled_objs[0]->get_all_streams_continues();
    m_last_duration = duration;

    change_state(PORT_STATE_TX);

    /* update the DP - messages will be freed by the DP */
    int index = 0;
    for (auto core_id : m_cores_id_list) {

        /* was the core assigned a compiled object ? */
        if (compiled_objs[index]) {
            TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id,
                                                                             m_pending_async_stop_event,
                                                                             compiled_objs[index],
                                                                             duration);
            send_message_to_dp(core_id, start_msg);
        } else {

            /* mimic an end event */
            m_dp_events.on_core_reporting_in(m_pending_async_stop_event, core_id);
        }

        index++;
    }
    
    /* for debug - this can be turn on */
    //m_dp_events.barrier();

    /* update subscribers */    
    Json::Value data;
    data["port_id"] = m_port_id;
    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STARTED, data);
    
}


bool TrexStatelessPort::is_active() const {
    return   (  (m_port_state == PORT_STATE_TX) 
             || (m_port_state == PORT_STATE_PAUSE)
             || (m_port_state == PORT_STATE_PCAP_TX)
             );
}

/**
 * stop traffic on port
 * 
 * @author imarom (09-Nov-15)
 * 
 * @return TrexStatelessPort::rc_e 
 */
void
TrexStatelessPort::stop_traffic(void) {
    if (!is_active()) {
        return;
    }

    /* delete any previous graphs */
    delete_streams_graph();

    /* to avoid race, first destroy any previous stop/pause events */
    if (m_pending_async_stop_event != TrexDpPortEvents::INVALID_ID) {
        m_dp_events.destroy_event(m_pending_async_stop_event);
        m_pending_async_stop_event = TrexDpPortEvents::INVALID_ID;

    }
    
    /* generate a message to all the relevant DP cores to start transmitting */
    TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop(m_port_id);
    send_message_to_all_dp(stop_msg);

    /* a barrier - make sure all the DP cores stopped */
    m_dp_events.barrier();

    change_state(PORT_STATE_STREAMS);

    common_port_stop_actions(false);
}

/**
 * remove all RX filters from port
 * 
 * @author imarom (28-Mar-16)
 */
void
TrexStatelessPort::remove_rx_filters(void) {
    /* only valid when IDLE or with streams and not TXing */
    verify_state(PORT_STATE_STREAMS);

    for (auto entry : m_stream_table) {
        get_stateless_obj()->m_rx_flow_stat.stop_stream(entry.second);
    }

}

/**
 * when a port stops, perform various actions
 * 
 */
void
TrexStatelessPort::common_port_stop_actions(bool async) {

    Json::Value data;
    data["port_id"] = m_port_id;

    if (async) {
        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_FINISHED_TX, data);
    } else {
        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STOPPED, data);
    }

}

void
TrexStatelessPort::pause_traffic(void) {

    verify_state(PORT_STATE_TX);

    if (m_last_all_streams_continues == false) {
        throw TrexException(" pause is supported when all streams are in continues mode ");
    }

    if ( m_last_duration>0.0 ) {
        throw TrexException(" pause is supported when duration is not enable is start command ");
    }

    /* send a pause message */
    TrexStatelessCpToDpMsgBase *pause_msg = new TrexStatelessDpPause(m_port_id);
    send_message_to_all_dp(pause_msg);

    /* make sure all DP cores paused */
    m_dp_events.barrier();

    /* change state */
    change_state(PORT_STATE_PAUSE);

    Json::Value data;
    data["port_id"] = m_port_id;
    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_PAUSED, data);
}


void
TrexStatelessPort::resume_traffic(void) {

    verify_state(PORT_STATE_PAUSE);

    /* generate a message to all the relevant DP cores to start transmitting */
    TrexStatelessCpToDpMsgBase *resume_msg = new TrexStatelessDpResume(m_port_id);

    send_message_to_all_dp(resume_msg);
    change_state(PORT_STATE_TX);


    Json::Value data;
    data["port_id"] = m_port_id;
    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_RESUMED, data);
}

void
TrexStatelessPort::update_traffic(const TrexPortMultiplier &mul, bool force) {

    double factor;

    verify_state(PORT_STATE_TX | PORT_STATE_PAUSE);

    /* generate a message to all the relevant DP cores to start transmitting */
    double new_factor = calculate_effective_factor(mul, force);

    switch (mul.m_op) {
    case TrexPortMultiplier::OP_ABS:
        factor = new_factor / m_factor;
        break;

    case TrexPortMultiplier::OP_ADD:
        factor = (m_factor + new_factor) / m_factor;
        break;

    case TrexPortMultiplier::OP_SUB:
        factor = (m_factor - new_factor) / m_factor;
        if (factor <= 0) {
            throw TrexException("Update request will lower traffic to less than zero");
        }
        break;

    default:
        assert(0);
        break;
    }

    TrexStatelessCpToDpMsgBase *update_msg = new TrexStatelessDpUpdate(m_port_id, factor);
    send_message_to_all_dp(update_msg);

    m_factor *= factor;

}

void
TrexStatelessPort::push_remote(const std::string &pcap_filename,
                               double ipg_usec,
                               double speedup,
                               uint32_t count,
                               double duration) {

    /* command allowed only on state stream */
    verify_state(PORT_STATE_IDLE | PORT_STATE_STREAMS);

    /* check that file exists */
    CCapReaderBase *reader;
    std::stringstream ss;
    reader = CCapReaderFactory::CreateReader((char *)pcap_filename.c_str(), 0, ss);
    if (!reader) {
        throw TrexException(ss.str());
    }
    delete reader;

    /* only one core gets to play */
    int tx_core = m_cores_id_list[0];

    /* create async event */
    assert(m_pending_async_stop_event == TrexDpPortEvents::INVALID_ID);
    m_pending_async_stop_event = m_dp_events.create_event(new AsyncStopEvent());

    /* mark all other cores as done */
    for (int index = 1; index < m_cores_id_list.size(); index++) {
        /* mimic an end event */
        m_dp_events.on_core_reporting_in(m_pending_async_stop_event, m_cores_id_list[index]);
    }

    /* send a message to core */
    change_state(PORT_STATE_PCAP_TX);
    TrexStatelessCpToDpMsgBase *push_msg = new TrexStatelessDpPushPCAP(m_port_id,
                                                                       m_pending_async_stop_event,
                                                                       pcap_filename,
                                                                       ipg_usec,
                                                                       speedup,
                                                                       count,
                                                                       duration);
    send_message_to_dp(tx_core, push_msg);

    /* update subscribers */    
    Json::Value data;
    data["port_id"] = m_port_id;
    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STARTED, data);
}

std::string 
TrexStatelessPort::get_state_as_string() const {

    switch (get_state()) {
    case PORT_STATE_DOWN:
        return "DOWN";

    case PORT_STATE_IDLE:
        return  "IDLE";

    case PORT_STATE_STREAMS:
        return "STREAMS";

    case PORT_STATE_TX:
        return "TX";

    case PORT_STATE_PAUSE:
        return "PAUSE";

    case PORT_STATE_PCAP_TX:
        return "PCAP_TX";
    }

    return "UNKNOWN";
}

int
TrexStatelessPort::get_max_stream_id() const {
    return m_stream_table.get_max_stream_id();
}

void
TrexStatelessPort::get_properties(std::string &driver, TrexPlatformApi::driver_speed_e &speed) {

    driver = m_api_info.driver_name;
    speed  = m_api_info.speed;
}

bool
TrexStatelessPort::verify_state(int state, bool should_throw) const {
    if ( (state & m_port_state) == 0 ) {
        if (should_throw) {
            throw TrexException("command cannot be executed on current state: '" + get_state_as_string() + "'");
        } else {
            return false;
        }
    }

    return true;
}

void
TrexStatelessPort::change_state(port_state_e new_state) {

    m_port_state = new_state;
}


void
TrexStatelessPort::encode_stats(Json::Value &port) {

    const TrexPlatformApi *api = get_stateless_obj()->get_platform_api();

    TrexPlatformInterfaceStats stats;
    api->get_interface_stats(m_port_id, stats);

    port["tx_bps"]          = stats.m_stats.m_tx_bps;
    port["rx_bps"]          = stats.m_stats.m_rx_bps;

    port["tx_pps"]          = stats.m_stats.m_tx_pps;
    port["rx_pps"]          = stats.m_stats.m_rx_pps;

    port["total_tx_pkts"]   = Json::Value::UInt64(stats.m_stats.m_total_tx_pkts);
    port["total_rx_pkts"]   = Json::Value::UInt64(stats.m_stats.m_total_rx_pkts);

    port["total_tx_bytes"]  = Json::Value::UInt64(stats.m_stats.m_total_tx_bytes);
    port["total_rx_bytes"]  = Json::Value::UInt64(stats.m_stats.m_total_rx_bytes);
    
    port["tx_rx_errors"]    = Json::Value::UInt64(stats.m_stats.m_tx_rx_errors);
}

void 
TrexStatelessPort::send_message_to_all_dp(TrexStatelessCpToDpMsgBase *msg) {

    for (auto core_id : m_cores_id_list) {
        send_message_to_dp(core_id, msg->clone());
    }

    /* original was not sent - delete it */
    delete msg;
}

void 
TrexStatelessPort::send_message_to_dp(uint8_t core_id, TrexStatelessCpToDpMsgBase *msg) {

    /* send the message to the core */
    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_id);
    ring->Enqueue((CGenNode *)msg);
}

void
TrexStatelessPort::send_message_to_rx(TrexStatelessCpToRxMsgBase *msg) {

    /* send the message to the core */
    CNodeRing *ring = CMsgIns::Ins()->getCpRx()->getRingCpToDp(0);
    ring->Enqueue((CGenNode *)msg);
}

uint64_t
TrexStatelessPort::get_port_speed_bps() const {
    switch (m_api_info.speed) {
    case TrexPlatformApi::SPEED_1G:
        return (1LLU * 1000 * 1000 * 1000);

    case TrexPlatformApi::SPEED_10G:
        return (10LLU * 1000 * 1000 * 1000);

    case TrexPlatformApi::SPEED_40G:
        return (40LLU * 1000 * 1000 * 1000);

    default:
        return 0;
    }
}

static inline double
bps_to_gbps(double bps) {
    return (bps / (1000.0 * 1000 * 1000));
}

double
TrexStatelessPort::calculate_effective_factor(const TrexPortMultiplier &mul, bool force) {

    double factor = calculate_effective_factor_internal(mul);

    /* did we exceeded the max L1 line rate ? */
    double expected_l1_rate = m_graph_obj->get_max_bps_l1(factor);

    /* if not force and exceeded - throw exception */
    if ( (!force) && (expected_l1_rate > get_port_speed_bps()) ) {
        stringstream ss;
        ss << "Expected L1 B/W: '" << bps_to_gbps(expected_l1_rate) << " Gbps' exceeds port line rate: '" << bps_to_gbps(get_port_speed_bps()) << " Gbps'";
        throw TrexException(ss.str());
    }

    return factor;
}

double
TrexStatelessPort::calculate_effective_factor_internal(const TrexPortMultiplier &mul) {

    /* we now need the graph - generate it if we don't have it (happens once) */
    if (!m_graph_obj) {
        generate_streams_graph();
    }

    switch (mul.m_type) {

    case TrexPortMultiplier::MUL_FACTOR:
        return (mul.m_value);

    case TrexPortMultiplier::MUL_BPS:
        return m_graph_obj->get_factor_bps_l2(mul.m_value);

    case TrexPortMultiplier::MUL_PPS:
        return m_graph_obj->get_factor_pps(mul.m_value);

    case TrexPortMultiplier::MUL_PERCENTAGE:
        /* if abs percentage is from the line speed - otherwise its from the current speed */

        if (mul.m_op == TrexPortMultiplier::OP_ABS) {
            double required = (mul.m_value / 100.0) * get_port_speed_bps();
            return m_graph_obj->get_factor_bps_l1(required);
        } else {
            return (m_factor * (mul.m_value / 100.0));
        }

    default:
        assert(0);
    }

}


void
TrexStatelessPort::generate_streams_graph() {

    /* dispose of the old one */
    if (m_graph_obj) {
        delete_streams_graph();
    }

    /* fetch all the streams from the table */
    vector<TrexStream *> streams;
    get_object_list(streams);

    TrexStreamsGraph graph;
    m_graph_obj = graph.generate(streams);
}

void
TrexStatelessPort::delete_streams_graph() {
    if (m_graph_obj) {
        delete m_graph_obj;
        m_graph_obj = NULL;
    }
}



/***************************
 * port multiplier
 * 
 **************************/
const std::initializer_list<std::string> TrexPortMultiplier::g_types = {"raw", "bps", "pps", "percentage"};
const std::initializer_list<std::string> TrexPortMultiplier::g_ops   = {"abs", "add", "sub"};

TrexPortMultiplier::
TrexPortMultiplier(const std::string &type_str, const std::string &op_str, double value) {
    mul_type_e type;
    mul_op_e   op;

    if (type_str == "raw") {
        type = MUL_FACTOR; 

    } else if (type_str == "bps") {
        type = MUL_BPS;

    } else if (type_str == "pps") {
        type = MUL_PPS;

    } else if (type_str == "percentage") {
        type = MUL_PERCENTAGE;
    } else {
        throw TrexException("bad type str: " + type_str);
    }

    if (op_str == "abs") {
        op = OP_ABS;

    } else if (op_str == "add") {
        op = OP_ADD;

    } else if (op_str == "sub") {
        op = OP_SUB;

    } else {
        throw TrexException("bad op str: " + op_str);
    }

    m_type  = type;
    m_op    = op;
    m_value = value;

}

const TrexStreamsGraphObj *
TrexStatelessPort::validate(void) {

    /* first compile the graph */

    vector<TrexStream *> streams;
    get_object_list(streams);

    if (streams.size() == 0) {
        throw TrexException("no streams attached to port");
    }

    TrexStreamsCompiler compiler;
    std::vector<TrexStreamsCompiledObj *> compiled_objs;

    std::string fail_msg;
    bool rc = compiler.compile(m_port_id,
                               streams,
                               compiled_objs,
                               get_dp_core_count(),
                               1.0,
                               &fail_msg);
    if (!rc) {
        throw TrexException(fail_msg);
    }

    for (auto obj : compiled_objs) {
        delete obj;
    }

    /* now create a stream graph */
    if (!m_graph_obj) {
        generate_streams_graph();
    }

    return m_graph_obj;
}



void
TrexStatelessPort::get_port_effective_rate(double &pps,
                                           double &bps_L1,
                                           double &bps_L2,
                                           double &percentage) {

    if (get_stream_count() == 0) {
        return;
    }

    if (!m_graph_obj) {
        generate_streams_graph();
    }

    pps        = m_graph_obj->get_max_pps(m_factor);
    bps_L1     = m_graph_obj->get_max_bps_l1(m_factor);
    bps_L2     = m_graph_obj->get_max_bps_l2(m_factor);
    percentage = (bps_L1 / get_port_speed_bps()) * 100.0;
    
}


void
TrexStatelessPort::set_promiscuous(bool enabled) {
    get_stateless_obj()->get_platform_api()->set_promiscuous(m_port_id, enabled);
}

bool
TrexStatelessPort::get_promiscuous() {
    return get_stateless_obj()->get_platform_api()->get_promiscuous(m_port_id);
}


void
TrexStatelessPort::get_macaddr(std::string &hw_macaddr,
                               std::string &src_macaddr,
                               std::string &dst_macaddr) {

    utl_macaddr_to_str(m_api_info.mac_info.hw_macaddr, hw_macaddr);
    utl_macaddr_to_str(m_api_info.mac_info.src_macaddr, src_macaddr);
    utl_macaddr_to_str(m_api_info.mac_info.dst_macaddr, dst_macaddr);
}

void
TrexStatelessPort::get_pci_info(std::string &pci_addr, int &numa_node) {
    pci_addr  = m_api_info.pci_addr;
    numa_node = m_api_info.numa_node;
}

void
TrexStatelessPort::add_stream(TrexStream *stream) {

    verify_state(PORT_STATE_IDLE | PORT_STATE_STREAMS);

    get_stateless_obj()->m_rx_flow_stat.add_stream(stream);

    m_stream_table.add_stream(stream);
    delete_streams_graph();

    change_state(PORT_STATE_STREAMS);
}

void
TrexStatelessPort::remove_stream(TrexStream *stream) {

    verify_state(PORT_STATE_STREAMS);

    get_stateless_obj()->m_rx_flow_stat.del_stream(stream);

    m_stream_table.remove_stream(stream);
    delete_streams_graph();

    if (m_stream_table.size() == 0) {
        change_state(PORT_STATE_IDLE);
    }
}

void
TrexStatelessPort::remove_and_delete_all_streams() {
    verify_state(PORT_STATE_IDLE | PORT_STATE_STREAMS);

    vector<TrexStream *> streams;
    get_object_list(streams);

    for (auto stream : streams) {
        remove_stream(stream);
        delete stream;
    }
}

/************* Trex Port Owner **************/

TrexPortOwner::TrexPortOwner() {
    m_is_free = true;
    m_session_id = 0;

    /* for handlers random generation */
    m_seed = time(NULL);
}

const std::string TrexPortOwner::g_unowned_name = "<FREE>";
const std::string TrexPortOwner::g_unowned_handler = "";