flowmatch_example_receiver.cpp Source File

/*
 *
 * Copyright 2023 Napatech A/S. All Rights Reserved.
 *
 * 1. Copying, modification, and distribution of this file, or executable
 * versions of this file, is governed by the terms of the Napatech Software
 * license agreement under which this file was made available. If you do not
 * agree to the terms of the license do not install, copy, access or
 * otherwise use this file.
 *
 * 2. Under the Napatech Software license agreement you are granted a
 * limited, non-exclusive, non-assignable, copyright license to copy, modify
 * and distribute this file in conjunction with Napatech SmartNIC's and
 * similar hardware manufactured or supplied by Napatech A/S.
 *
 * 3. The full Napatech Software license agreement is included in this
 * distribution, please see "NP-0405 Napatech Software license
 * agreement.pdf"
 *
 * 4. Redistributions of source code must retain this copyright notice,
 * list of conditions and the following disclaimer.
 *
 * THIS SOFTWARE IS PROVIDED "AS IS" WITHOUT ANY WARRANTIES, EXPRESS OR
 * IMPLIED, AND NAPATECH DISCLAIMS ALL IMPLIED WARRANTIES INCLUDING ANY
 * IMPLIED WARRANTY OF TITLE, MERCHANTABILITY, NONINFRINGEMENT, OR OF
 * FITNESS FOR A PARTICULAR PURPOSE. TO THE EXTENT NOT PROHIBITED BY
 * APPLICABLE LAW, IN NO EVENT SHALL NAPATECH BE LIABLE FOR PERSONAL INJURY,
 * OR ANY INCIDENTAL, SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES WHATSOEVER,
 * INCLUDING, WITHOUT LIMITATION, DAMAGES FOR LOSS OF PROFITS, CORRUPTION OR
 * LOSS OF DATA, FAILURE TO TRANSMIT OR RECEIVE ANY DATA OR INFORMATION,
 * BUSINESS INTERRUPTION OR ANY OTHER COMMERCIAL DAMAGES OR LOSSES, ARISING
 * OUT OF OR RELATED TO YOUR USE OR INABILITY TO USE NAPATECH SOFTWARE OR
 * SERVICES OR ANY THIRD PARTY SOFTWARE OR APPLICATIONS IN CONJUNCTION WITH
 * THE NAPATECH SOFTWARE OR SERVICES, HOWEVER CAUSED, REGARDLESS OF THE THEORY
 * OF LIABILITY (CONTRACT, TORT OR OTHERWISE) AND EVEN IF NAPATECH HAS BEEN
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. SOME JURISDICTIONS DO NOT ALLOW
 * THE EXCLUSION OR LIMITATION OF LIABILITY FOR PERSONAL INJURY, OR OF
 * INCIDENTAL OR CONSEQUENTIAL DAMAGES, SO THIS LIMITATION MAY NOT APPLY TO YOU.
 *
 *

 */

// Include this in order to access the Napatech API
#include <nt.h>

#include <atomic>
#include <chrono>
#include <cstring>
#include <iostream>
#include <memory>
#include <thread>
#include <vector>

#include "flowmatch_example_helper.h"

#include "flowmatch_example_receiver.h"

// This application requires 64-bit pointers
static_assert(sizeof(NtFlow_t*) == sizeof(uint64_t), "Pointers must be 64bit");

namespace {

/**
 * Print flow info and status records related to flow stream.
 */
void printFlowStreamInfo(NtFlowStream_t& flowStream, std::vector<std::unique_ptr<NtFlow_t>>& learnedFlowList,
  std::atomic<uint64_t>* ipv4counter, std::atomic<uint64_t>* ipv6counter)
{
  const char* ip;
  NtFlowInfo_t flowInfo;
  NtFlowStatus_t flowStatus;

  // For each element in internal flow stream queue print the flow status record.
  while (NT_FlowStatusRead(flowStream, &flowStatus) == NT_SUCCESS) {
    ip = learnedFlowList[flowStatus.id]->keyId == KEY_ID_IPV4 ? " (IPv4): " : " (IPv6): ";
    std::cout << "Flow status for ID " << flowStatus.id << ip;

    // The flags returned from NT_FlowStatusRead depend on ntservice.ini settings.
    switch (flowStatus.flags) {
      case NT_FLOW_STAT_LDS: std::cout << "NT_FLOW_STAT_LDS" << std::endl; break;
      case NT_FLOW_STAT_LFS: std::cout << "NT_FLOW_STAT_LFS" << std::endl; break;
      case NT_FLOW_STAT_LIS: std::cout << "NT_FLOW_STAT_LIS" << std::endl; break;
      case NT_FLOW_STAT_UDS: std::cout << "NT_FLOW_STAT_UDS" << std::endl; break;
      case NT_FLOW_STAT_UIS: std::cout << "NT_FLOW_STAT_UIS" << std::endl; break;
      default: std::cout << "Unknown flag" << std::endl; break;
    }
  }

  std::cout << std::endl;

  // For each element in internal flow stream queue print the flow info record.
  // The flow info record is only available when NtFlow_t.gfi is set to 1.
  // Maintaining the flow info record has a performance overhead, so if the
  // info record is not need, it is recommended to set NtFlow_t.gfi to 0.
  while (NT_FlowRead(flowStream, &flowInfo, 0) == NT_SUCCESS) {
    if (learnedFlowList[flowInfo.id]->keyId == KEY_ID_IPV4) {
      ip = " (IPv4):";
      *ipv4counter += (flowInfo.packetsA + flowInfo.packetsB);
    }
    else {
      ip = " (IPv6):";
      *ipv6counter += (flowInfo.packetsA + flowInfo.packetsB);
    }

    std::cout << "NT_FlowRead of flow ID " << flowInfo.id << ip << std::endl
      << "CSA: Packages: " << flowInfo.packetsA
      << ", Octets: "      << flowInfo.octetsA << std::endl
      << "CSB: Packages: " << flowInfo.packetsB
      << ", Octets: "      << flowInfo.octetsB << std::endl
      << "Time stamp: "    << flowInfo.ts << std::endl
      << "TCP flags A: "   << flowInfo.flagsA
      << ", TCP flags B: " << flowInfo.flagsB << std::endl;

    switch (flowInfo.cause) {
      case 0:  std::cout << "Unlearn cause: Software" << std::endl; break;
      case 1:  std::cout << "Unlearn cause: Timeout" << std::endl; break;
      case 2:  std::cout << "Unlearn cause: TCP flow termination" << std::endl; break;
      default: std::cout << "Unlearn cause: Not supported" << std::endl; break;
    }
    std::cout << std::endl;
  }
}

}  // Unnamed namespace

void taskReceiverMiss(const char* streamName, uint32_t streamId,
  std::atomic<uint64_t>* ipv4counter, std::atomic<uint64_t>* ipv6counter)
{
  int status;
  uint64_t idCounter = 0U;
  std::vector<std::unique_ptr<NtFlow_t>> learnedFlowList;

  NtFlowAttr_t    flowAttr;
  NtFlowStream_t  flowStream;

  NtNetStreamRx_t hNetRx;
  NtNetBuf_t      hNetBuffer;

  // Initialize flow stream attributes and set adapter number attribute.
  NT_FlowOpenAttrInit(&flowAttr);
  NT_FlowOpenAttrSetAdapterNo(&flowAttr, ADAPTER_NO);

  // Opens a flow programming stream and returns a stream handle (flowStream).
  status = NT_FlowOpen_Attr(&flowStream, "flowmatch_example_receiver_attr", &flowAttr);
  handleErrorStatus(status, "NT_FlowOpen_Attr() failed");

  // Retrieve handle to rx stream.
  status = NT_NetRxOpen(&hNetRx, streamName, NT_NET_INTERFACE_PACKET, streamId, -1);
  handleErrorStatus(status, "NT_NetRxOpen() failed");

  // As long as packages are being transmitted, this loop will handle all
  // packages from the MISSED filters. When a flow has been learned, it will no
  // longer be received here.
  while (applicationRunning()) {
    // Get package from rx stream.
    status = NT_NetRxGetNextPacket(hNetRx, &hNetBuffer, 100);
    if (status == NT_STATUS_TIMEOUT || status == NT_STATUS_TRYAGAIN) continue;
    handleErrorStatus(status, "NT_NetRxGetNextPacket() failed");

    // Here a package has successfully been received, and the parameters for the
    // next flow to be learned will be set up.
    auto flow = std::unique_ptr<NtFlow_t>(new NtFlow_t);
    std::memset(flow.get(), 0x0, sizeof(NtFlow_t));

    // In this example, the ID is a simple incremental value that can be used
    // for lookup in the std::vector learnedFlowList. However, any value can be used,
    // including the raw value of pointers.
    flow->id              = idCounter++;  // User defined ID
    flow->color           = 0;            // Flow color
    flow->overwrite       = 0;            // Overwrite filter action (1: enable, 0: disable)
    flow->streamId        = 0;            // Marks the stream id if overwrite filter action is enabled
    flow->ipProtocolField = 6;            // IP protocol number of next header (6: TCP)
    flow->keySetId        = KEY_SET_ID;   // Key Set ID as used in the NTPL filter
    flow->op              = 1;            // Flow programming operation (1: learn, 0: un-learn)
    flow->gfi             = 1;            // Generate flow info record (1: generate, 0: do not generate)
    flow->tau             = 0;            // TCP auto unlearn (1: auto unlearn enable, 0: auto unlearn disable)

    // For this example the descriptor DYN3 is used, which is set up by NTPL.
    NtDyn4Descr_t* dyn4 = _NT_NET_GET_PKT_DESCR_PTR_DYN4(hNetBuffer);
    uint8_t* packet = reinterpret_cast<uint8_t*>(dyn4) + dyn4->descrLength;

    // Because colormask was used in the filters, it is very easy to check for
    // the IP type.
    // The filters also set up an alternative offset0, such that it points
    // directly to the IP source address.
    switch (dyn4->color0 & (COLOR_IPV4 | COLOR_IPV6)) {
      case COLOR_IPV4: {
        *ipv4counter += 1U;
        std::memcpy(flow->keyData,      packet + dyn4->offset0,     4);  // IPv4 src
        std::memcpy(flow->keyData + 4,  packet + dyn4->offset0 + 4, 4);  // IPv4 dst
        std::memcpy(flow->keyData + 8,  packet + dyn4->offset1,     2);  // TCP port src
        std::memcpy(flow->keyData + 10, packet + dyn4->offset1 + 2, 2);  // TCP port dst
        flow->keyId = KEY_ID_IPV4;  // Key ID as used in the NTPL Key Test
        break;
      }
      case COLOR_IPV6: {
        *ipv6counter += 1U;
        std::memcpy(flow->keyData,      packet + dyn4->offset0,      16);  // IPv6 src
        std::memcpy(flow->keyData + 16, packet + dyn4->offset0 + 16, 16);  // IPv6 dst
        std::memcpy(flow->keyData + 32, packet + dyn4->offset1,      2);   // TCP port src
        std::memcpy(flow->keyData + 34, packet + dyn4->offset1 + 2,  2);   // TCP port dst
        flow->keyId = KEY_ID_IPV6;  // Key ID as used in the NTPL Key Test
        break;
      }
    }

    // Program the flow into the adapter.
    status = NT_FlowWrite(flowStream, flow.get(), -1);
    handleErrorStatus(status, "NT_FlowWrite() failed");

    learnedFlowList.push_back(std::move(flow));
  }

  // Unlearn all stored flows
  for (auto&& flow : learnedFlowList) {
    flow->op = 0;
    status = NT_FlowWrite(flowStream, flow.get(), -1);
    handleErrorStatus(status, "NT_FlowWrite() failed");
  }

  // While it doesn't take a full 1000 ms for the internal flow status queue to
  // be updated, it is still a slow operation, thus waiting to a bit after
  // NT_FlowWrite is a good idea.
  std::this_thread::sleep_for(std::chrono::milliseconds(1000));
  printFlowStreamInfo(flowStream, learnedFlowList, ipv4counter, ipv6counter);

  // Closes rx stream.
  status = NT_NetRxClose(hNetRx);
  handleErrorStatus(status, "NT_NetRxClose() failed");

  // Closes flow programming stream
  status = NT_FlowClose(flowStream);
  handleErrorStatus(status, "NT_FlowClose() failed");
}

void taskReceiverCounter(const char* streamName, uint32_t streamId,
  std::atomic<uint64_t>* counter)
{
  int status;

  NtNetStreamRx_t hNetRx;
  NtNetBuf_t      hNetBuffer;

  // Retrieve handle to rx stream.
  status = NT_NetRxOpen(&hNetRx, streamName, NT_NET_INTERFACE_PACKET, streamId, -1);
  handleErrorStatus(status, "NT_NetRxOpen() failed");

  while (applicationRunning()) {
    // Get package from rx stream.
    status = NT_NetRxGetNextPacket(hNetRx, &hNetBuffer, 100);
    if (status == NT_STATUS_TIMEOUT || status == NT_STATUS_TRYAGAIN) continue;
    handleErrorStatus(status, "NT_NetRxGetNextPacket() failed");

    // When a package has successfully been received increment the counter.
    *counter += 1U;
  }

  // Closes rx stream.
  status = NT_NetRxClose(hNetRx);
  handleErrorStatus(status, "NT_NetRxClose() failed");
}