Cvb/CppRdma

// ---------------------------------------------------------------------------
// ---------------------------------------------------------------------------
 
#include <iostream>
#include <string>
#include <vector>
 
#include <cvb/device_factory.hpp>
#include <cvb/global.hpp>
#include <cvb/driver/composite_stream.hpp>
#include <cvb/genapi/node_map_enumerator.hpp>
 
static const constexpr auto TIMEOUT = std::chrono::milliseconds(3000);
static const constexpr int NUM_ELEMENTS_TO_ACQUIRE = 5000; // number of elements to be acquired
static const constexpr int FEEDBACK_INTERVAL = NUM_ELEMENTS_TO_ACQUIRE / 10; // print progress
static const constexpr int NUM_FLOW_SETS = 10; // number flow sets to announce (usually buffers)
 
int main()
{
  try
  {
    // RDMA transfer is part of the socket driver (GevSD) so the filter driver (GevFD) must be ignored.
    auto itemList = Cvb::DeviceFactory::Discover(Cvb::DiscoverFlags::IgnoreVins | Cvb::DiscoverFlags::IgnoreGevFD);
 
    // The following will only work if there is a RDMA capable device connected to an RDMA capable 
    // network interface. Usually also a driver that allows user space access is required. See also 
    // * Windows: Network Direct Service Provider Interface (NDSPI)
    // * Linux: librdmacm.so and libibverbs.so 
    auto& item = itemList.at(0);
    // RDMA must be switched on explicitly otherwise the classic GVSP protocol will be used.
    // This is just a request and may fail silently if RDMA is not supported.
    // To verify that RDMA is supported and switched on you can check
    // the "GevDataStreamingMode" in the TLDevice (Cvb::NodeMapID::TLDevice) node map and validate that it is
    // set to "RDMA". The "RDMA" entry is only available if the device is RDMA capable.
    item.SetParameter(CVB_LIT("UseRdma"), CVB_LIT("1"));
 
 
    auto device = Cvb::DeviceFactory::Open<Cvb::GenICamDevice>(item.AccessToken(), Cvb::AcquisitionStack::GenTL);
    auto payloadSize = static_cast<double>(device->NodeMap(Cvb::NodeMapID::Device)->Node<Cvb::IntegerNode>(CVB_LIT("PayloadSize"))->Value());
    // Note that RDMA requires GenDC containers as payload.
    // The container will automatically be mapped to a composite for
    // further processing. 
    auto dataStream = device->Stream<Cvb::CompositeStream>();
 
    // The default and minimal number of flow sets is three but it is usually 
    // advisable to increase the number so that about one second can be buffered.
    dataStream->RegisterManagedFlowSetPool(NUM_FLOW_SETS);
 
 
    dataStream->Start();
    auto startTime = std::chrono::high_resolution_clock::now();
    for (auto i = 0; i < NUM_ELEMENTS_TO_ACQUIRE; i++)
    {
      Cvb::CompositePtr composite;
      Cvb::WaitStatus waitStatus;
      Cvb::NodeMapEnumerator enumerator;
      std::tie(composite, waitStatus, enumerator) = dataStream->WaitFor(TIMEOUT);
      switch (waitStatus)
      {
      default:
        // abort cannot happen in single threaded scenarios
      case Cvb::WaitStatus::Abort:
        throw std::runtime_error("wait status unknown.");
 
      case Cvb::WaitStatus::Timeout:
        throw std::runtime_error("time out during wait");
 
      case Cvb::WaitStatus::Ok:
        break;
      }
 
      if (i && i % FEEDBACK_INTERVAL == 0)
      {
        // try to get the composite's first element as an image
        auto firstElement = composite->ItemAt(0);
        if (!Cvb::holds_alternative<Cvb::ImagePtr>(firstElement))
        {
          std::cout << "acquired composites: " << i << " which is not an image" << std::endl;
        }
        else
        {
          auto image = Cvb::get<Cvb::ImagePtr>(firstElement);
          std::cout << "acquired images: " << i << " | size: " << image->Width() << " x " << image->Height() << std::endl;
        }
      }
    }
    auto endTime = std::chrono::high_resolution_clock::now();
    auto durationMs = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime).count();
    dataStream->TryAbort();
 
    // For RDMA streaming the data rate is usually interesting...
    auto dataRate = static_cast<int>((static_cast<double>(NUM_ELEMENTS_TO_ACQUIRE) * payloadSize) / (1000.0 * static_cast<double>(durationMs)));
    std::cout << std::endl << "acquired " << NUM_ELEMENTS_TO_ACQUIRE << " buffers in " << durationMs << " ms ( " << dataRate << " MB/s )" << std::endl;
  }
  catch (const std::exception& e)
  {
    std::cout << e.what() << std::endl;
  }
  return 0;
}