Tutorial 3: Simple Pipeline with RTP

How to run

$ cd FlexPipe/examples
$ bash 3_rtp_pipeline.sh

In this tutorial, by using the given SourceKernel and SinkKernelB (with blocking input), we create a pipeline connected via UDP-based RTP in a single machine. The pipeline topology looks below.

|Source|[o1]----RTP--->[i1, B]|Sink|

where B: blocking

Port Registration by Kernel Developer

Using the same kernels, it is the same as the previous tutorial.

Port Activation by Kernel User

The kernel user uses the given kernels to create a pipeline. The below codes show how the pipeline is created.

/* examples/3_rtp_pipeline.cc */
int main()
{
  /* 1. Create instances of the given kernels */
  SourceKernel *sourceKernel = new SourceKernel("sourceKernel");
  SinkKernelB *sinkKernelB = new SinkKernelB("sinkKernel");

  /* 2. Activate ports that are registered by the kernel developers */
  sourceKernel->portManager.activateOutPortAsRemote(flexpipe::RemoteProtocol::UVG_RTP, "o1", "127.0.0.1", 5555);
  sinkKernelB->portManager.activateInPortAsRemote(flexpipe::RemoteProtocol::UVG_RTP, "i1", 5555);

  /* 3. Run kernels in separate threads as they are not locally linked. */
  std::vector<std::thread> singleKernelThreads;
  vector<flexpipe::Kernel*> separateKernels;
  separateKernels.push_back(sourceKernel);
  separateKernels.push_back(sinkKernelB);

  for(int i = 0; i < separateKernels.size(); i++)
  {
    std::thread singleKernelThread(flexpipe::runSingleKernel, separateKernels[i]);
    singleKernelThreads.push_back(std::move(singleKernelThread));
  }

  for(int i = 0; i < singleKernelThreads.size(); i++) singleKernelThreads[i].join();

  return 0;
}

After creating the kernels, the kernel ports are activated with the protocol::UVG_RTP. FlexPipe's port interface works with different communication channels, and uvgRTP is supported for UDP-based RTP. As the kernels are not linked locally, they run in separate threads.

Compared to the FlexPipe local connection, RTP connection involves data serialization and copy, and it increases the end-to-end latency with the large size of data.