Started implementing perftool sample

CMakeLists.txt

@@ -96,6 +96,7 @@ list(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/cmake/ecaludp-module)
 # Add samples, if enabled
 if (ECALUDP_BUILD_SAMPLES)
     add_subdirectory(samples/ecaludp_sample)
+    add_subdirectory(samples/ecaludp_perftool)
     if (ECALUDP_ENABLE_NPCAP)
         add_subdirectory(samples/ecaludp_sample_npcap)
     endif()

samples/ecaludp_perftool/CMakeLists.txt

@@ -0,0 +1,47 @@
+################################################################################
+# Copyright (c) 2024 Continental Corporation
+# 
+# This program and the accompanying materials are made available under the
+# terms of the Apache License, Version 2.0 which is available at
+# https://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.
+# 
+# SPDX-License-Identifier: Apache-2.0
+################################################################################
+
+cmake_minimum_required(VERSION 3.13)
+
+project(ecaludp_perftool)
+
+find_package(Threads REQUIRED)
+find_package(ecaludp REQUIRED)
+
+set(sources
+  src/main.cpp
+  src/receiver.cpp
+  src/receiver.h
+  src/receiver_sync.cpp
+  src/receiver_sync.h
+  src/sender.cpp
+  src/sender.h
+  src/sender_sync.cpp
+  src/sender_sync.h
+)
+
+add_executable(${PROJECT_NAME} ${sources})
+
+target_link_libraries(${PROJECT_NAME}
+  PRIVATE
+    ecaludp::ecaludp
+    Threads::Threads)
+
+target_compile_features(${PROJECT_NAME} PRIVATE cxx_std_14)
+
+source_group(TREE "${CMAKE_CURRENT_SOURCE_DIR}" FILES 
+    ${sources}
+)

samples/ecaludp_perftool/src/main.cpp

@@ -0,0 +1,160 @@
+/********************************************************************************
+ * Copyright (c) 2024 Continental Corporation
+ * 
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License, Version 2.0 which is available at
+ * https://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.
+ * 
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+#include <array>
+#include <chrono>
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include <asio.hpp> // IWYU pragma: keep
+
+#include "sender_sync.h"
+#include "receiver_sync.h"
+
+void printUsage(const std::string& arg0)
+{
+  std::cout << "Usage:" << std::endl;
+  std::cout << "  " << arg0 << " send <ip>:<port>" << std::endl;
+  std::cout << "or:" << std::endl;
+  std::cout << "  " << arg0 << " receive <ip>:<port>" << std::endl;
+  std::cout << std::endl;
+  std::cout << "Options:" << std::endl;
+  std::cout << "  -h, --help  Show this help message and exit" << std::endl;
+  std::cout << std::endl;
+  std::cout << "  -s, --size <SIZE> Message size to send" << std::endl;
+  std::cout << "  -m, --max-udp-datagram-size <SIZE> Maximum UDP datagram size" << std::endl;
+  std::cout << "      --buffer-size <SIZE> Buffer size for sending & receiving messages" << std::endl;
+  std::cout << std::endl;
+}
+
+int main(int argc, char* argv[])
+{
+  bool sender_mode   = false;
+  bool receiver_mode = false;
+
+  size_t message_size          = 0;
+
+  // Default to max udp datagram size for IPv4
+  size_t max_udp_datagram_size = 65507;
+
+  int buffer_size = -1;
+
+  // convert argc, argv to vector of strings
+  std::vector<std::string> args;
+  args.reserve(static_cast<size_t>(argc));
+  for (int i = 0; i < argc; ++i)
+  {
+    args.emplace_back(argv[i]);
+  }
+
+  // Check for -h / --help
+  if (args.size() < 2
+      || std::find(args.begin(), args.end(), "-h") != args.end()
+      || std::find(args.begin(), args.end(), "--help") != args.end())
+  {
+    printUsage(args[0]);
+    return 0;
+  }
+
+  // Check for -s / --size
+  {
+    auto it = std::find(args.begin(), args.end(), "-s");
+    if (it == args.end())
+    {
+      it = std::find(args.begin(), args.end(), "--size");
+    }
+
+    if (it != args.end())
+    {
+      if (it + 1 == args.end())
+      {
+        std::cerr << "Error: -s / --size requires an argument" << std::endl;
+        return 1;
+      }
+      message_size = std::stoul(*(it + 1));
+    }
+  }
+
+  // Check for -m / --max-udp-datagram-size
+  {
+    auto it = std::find(args.begin(), args.end(), "-m");
+    if (it == args.end())
+    {
+      it = std::find(args.begin(), args.end(), "--max-udp-datagram-size");
+    }
+
+    if (it != args.end())
+    {
+      if (it + 1 == args.end())
+      {
+        std::cerr << "Error: -m / --max-udp-datagram-size requires an argument" << std::endl;
+        return 1;
+      }
+      max_udp_datagram_size = std::stoul(*(it + 1));
+    }
+  }
+
+  // Check for buffer size
+  {
+    auto it = std::find(args.begin(), args.end(), "--buffer-size");
+    if (it != args.end())
+    {
+      if (it + 1 == args.end())
+      {
+        std::cerr << "Error: --buffer-size requires an argument" << std::endl;
+        return 1;
+      }
+      buffer_size = std::stoul(*(it + 1));
+    }
+  }
+
+  // check for mode
+  if (args[1] == "send")
+  {
+    sender_mode = true;
+  }
+  else if (args[1] == "receive")
+  {
+    receiver_mode = true;
+  }
+  else
+  {
+    printUsage(args[0]);
+    return 1;
+  }
+
+  if (sender_mode)
+  {
+    SenderSync sender(message_size, max_udp_datagram_size, buffer_size);
+    sender.start();
+
+    while (true)
+    {
+      std::this_thread::sleep_for(std::chrono::seconds(1));
+    }
+  }
+  else if (receiver_mode)
+  {
+    ReceiverSync receiver(buffer_size);
+    receiver.start();
+
+    while (true)
+    {
+      std::this_thread::sleep_for(std::chrono::seconds(1));
+    }
+  }
+}

samples/ecaludp_perftool/src/receiver.cpp

@@ -0,0 +1,88 @@
+/********************************************************************************
+ * Copyright (c) 2024 Continental Corporation
+ * 
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License, Version 2.0 which is available at
+ * https://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.
+ * 
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+#include "receiver.h"
+
+#include <sstream>
+#include <iostream>
+#include <iomanip>
+
+Receiver::Receiver(int buffer_size)
+  : buffer_size_(buffer_size)
+  , statistics_thread_{std::make_unique<std::thread>([this]() { this->print_statistics(); })}
+{
+  // Print information for debug purposes
+  std::cout << "Receiving data: " << std::endl;
+  std::cout << "  buffer size: " << (buffer_size_ > 0 ? std::to_string(buffer_size_) : "default") << std::endl;
+}
+
+Receiver::~Receiver()
+{
+  {
+    // Stop statistics thread
+    std::lock_guard<std::mutex> lock(statistics_mutex_);
+    is_stopped_ = true;
+    cv_.notify_all();
+  }
+
+  if (statistics_thread_->joinable())
+    statistics_thread_->join();
+}
+
+void Receiver::print_statistics()
+{
+  std::chrono::steady_clock::time_point last_statistics_run;
+
+  while(true)
+  {
+    long long bytes_payload     {0};
+    long long messages_received {0};
+
+    {
+      std::unique_lock<std::mutex> lock(statistics_mutex_);
+      cv_.wait_for(lock, std::chrono::seconds(1), [this]() -> bool { return is_stopped_; });
+
+      if (is_stopped_)
+        return;
+
+      std::swap(bytes_payload_, bytes_payload);
+      std::swap(messages_received_, messages_received);
+    }
+
+    auto now = std::chrono::steady_clock::now();
+
+    // Calculate message per seconds -> frequency)
+    double frequency = 0.0;
+    auto duration = std::chrono::duration_cast<std::chrono::duration<double>>(now - last_statistics_run).count();
+    if (duration > 0)
+    {
+      frequency = static_cast<double>(messages_received) / duration;
+    }
+
+    {
+      std::stringstream ss;
+      ss << "cnt: "   << messages_received;
+      ss << " | ";
+      ss << "rcv pyld: " << bytes_payload;
+      ss << " | ";
+      ss << "freq: " << std::fixed << std::setprecision(1) << frequency;
+
+      std::cout << ss.str() << std::endl;
+    }
+
+    last_statistics_run = now;
+  }
+}

samples/ecaludp_perftool/src/receiver.h

@@ -0,0 +1,51 @@
+/********************************************************************************
+ * Copyright (c) 2024 Continental Corporation
+ * 
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License, Version 2.0 which is available at
+ * https://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.
+ * 
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+#pragma once
+
+#include <thread>
+#include <memory>
+#include <mutex>
+
+
+class Receiver
+{
+public:
+  Receiver(int buffer_size);
+  virtual ~Receiver();
+
+  virtual void start() = 0;
+
+private:
+  void print_statistics();
+
+///////////////////////////////////////////////////////////
+// Member variables
+///////////////////////////////////////////////////////////
+
+protected:
+  int                           buffer_size_;
+
+  bool                          is_stopped_         {false};
+  mutable std::mutex            statistics_mutex_;
+  std::condition_variable       cv_;
+
+  long long bytes_payload_    {0};
+  long long messages_received_{0};
+
+private:
+  std::unique_ptr<std::thread>  statistics_thread_;
+};