Working on T-DB.

Author: cliffg-softwarelibre

test/test_data_blaster/README.md

@@ -1,54 +1,75 @@
 # Test Data Blaster
 
-The Test Data Blaster (T-DB) applications are:
-- Data sender / receiver (DSR)
-- Statistics and logging monitor
+The Test Data Blaster (T-DB) is an application suite that performs distributed testing, both TCP and UDP (currently the UDP portion is not implemented). There are multiple binaries in the suite:
+- Data sender / receiver (DSR), a TCP and (future) UDP version
+- Statistics and logging monitor, a C++ and (future) Python version
 
-There are two varieties of the DSR (data sender / receiver), one for TCP/IP and one for UDP/IP. Having separate code bases (and executables) simplifies the command line parameters as well as the internal logic. There is common code, some already factored out in the `shared_test` directory, with additional common code factored out in this (`test_data_blaster`) directory.
+A single instance of the monitor shows statistics from the DSR instances. The monitor is also responsible for shutting down the DSRs. Multiple DSR instances are typically running at the same time. Each TCP DSR instance can be configured for multiple TCP connectors and multiple TCP acceptors at the same time. Each (future) UDP DSR can be configured for multiple senders and receivers at the same time.
 
-The monitor application is the same for both the TCP/IP DSR and the UDP/IP DSR, and the same monitor instance can be run to monitor TCP and UDP DSR testing at the same time (to handle this use case, a protocol flag is part of the monitor message definition).
+The same monitor instance can be run for both TCP and UDP (the messages sent to the monitor from the DSRs specify whether the test data is being sent over TCP or UDP).
 
-While the data sender and receiver apps are written in C++, the monitor app can be written in any language (and both a C++ and Python version is expected, with the Python version having a simple GUI display). The monitor message definition is text only, so binary endianness is not a factor.
+While the DSRs are written in C++, the monitor app can be written in any language (and both a C++ and Python version is expected, with the Python version having a simple GUI display). The monitor message definition is text only, so binary endianness is not a factor.
 
-## Monitor Process
+## Usage
 
-The T-DB Monitor is run as a server (only). It accepts TCP connections on one port, each connection corresponding to one instance of either a TCP T-DB sender / receiver or a UDP T-DB sender / receiver.
+(Fill in command line usage here - note that typically the monitor is started first.)
 
-There are no "begin" or "end" data messages, a new incoming TCP connection is the indication that a new DSR instance has started, and the end of the TCP connection is the indication that the DSR instance has ended.
+(Fill in details about expected output.)
+
+(Fill in details about how to trigger shutdown through the monitor.)
+
+## Building Test Data Blaster
+
+All of the C++ components can be built using CMake. 
+
+(Fill in details, including single build instructions in Linux, and future Python build instructions.)
+
+## Internals and Design
+
+Having separate code bases (and executables) for the TCP and UDP DSRs simplifies the command line parameters as well as the internal logic. There is no specific technical reason why the DSR couldn't handle both TCP and UDP, but the DSR command line is complicated enough as is
+
+There is (obviously) common code between all of the T-DB C++ components, some from the `shared_test` directory (i.e. shared with the Chops Net IP unit tests), and some factored out in this (`test_data_blaster`) directory.
+
+### Monitor Process
+
+The T-DB Monitor is run as a server (only). It accepts TCP connections on one port, each connection corresponding to one instance of either a TCP DSR or a UDP DSR.
+
+There are no "begin" or "end" data messages that the monitor has to process, a new incoming TCP connection is the indication that a new DSR instance has started, and the end of the TCP connection is the indication that the DSR instance has ended.
 
 The DSR will not send a monitor message for every test message sent between DSR instances. A "modulo" command line parameter specifies how often a monitor message will be sent (for each test data set). (A modulo of 1 would generate one monitor message per test data set message.)
 
 The monitor process will notify all DSR processes to shutdown by sending a shutdown message through each monitor connection. The monitor process will have a user initiated way to send the shutdown messages and end the monitor process.
 
-### Monitor Message Definition
+#### Monitor Message Definition
 
-Each message to the Monitor has the following fields:
+Each message from the DSRs to the monitor has the following fields:
 - DSR name
 - DSR protocol, either "tcp" or "udp"
 - Remote endpoint (host, port)
 - Data direction, either "incoming" or "outgoing"
 - Current message number
-- Total messages expected or to be sent
--- with previous field this makes either "100th message out of 10,000 expected / sent" displays possible
+- Total messages expected to be sent (or 0 if incoming messages -- with the previous field this makes "100th message out of 10,000 to be sent" displays possible)
 - Current message size
-- Current message prefix
-- Current message body character
-- Outgoing queue size (only meaningful for outgoing data)
+- Current message beginning (first 15 characters)
+- Outgoing queue size (will always be 0 for incoming data, and 0 for outgoing data if the receiving end is keeping up)
 
-Each of the fields is a text string, and the full message is delimited by
+Each of the fields is a text string, and the full message is delimited by (fill in details).
 
 TODO - define a format that makes Python deserialization easy. Specifically, are each of the text fields nul character terminated, or is there a length prefix? Do we want name / value pairs instead of value only fields? Is the end of the message delimited by a LF (line feed) character or CR / LF (carriage return / line feed)?
 
+The shutdown message is the only message sent from the monitor to the DSRs and has the following fields:
+
+(Fill in details.)
 
-## TCP Data Sender Receiver
+### TCP Data Sender Receiver
 
 Each DSR instance can start multiple TCP connectors and multiple TCP acceptors as provided via the command line arguments.
 
 If the DSR instance is configured to send messages (send count > 0), messages are sent when a connection is made. For TCP connectors this is when the connect succeeds. For acceptors it is when each incoming connection is made. Each connection will send out the full set of test messages.
 
-Degenerate configurations are possible, with infinite message loops or where connection and shutdown timing is not clear. The T-DB is meant to be run by knowledgeable users, not general developers, so no special coding is in the T-DB to protect for bad configurations.
+Degenerate configurations are possible, with infinite message loops or where connection and shutdown timing is not clear. No special coding is in the T-DB to protect for bad configurations.
 
-The internal data messages uses the binary marshalled message format from the `shared_test` facilities (see `msg_handling.hpp` for specific code). The monitor messages are text based messages with all fields in a string format easy to unmarshall for Python applications.
+The DSR internal data messages uses the binary marshalled message format from the `shared_test` facilities (see `msg_handling.hpp` for specific code). The monitor messages are text based messages with all fields in a string format easy to unmarshall for Python applications.
 
 The general logic flow:
 - Process command line arguments

test/test_data_blaster/monitor_msg_handling.hpp

@@ -21,11 +21,11 @@
 #include <cstddef> // std::size_t
 #include <future>
 #include <utility> // std::move
-#include <optional>
 #include <type_traits> // std::is_same
 
 #include "net_ip/net_ip.hpp"
 #include "net_ip/net_entity.hpp"
+#include "net_ip_component/error_delivery.hpp"
 
 #include "asio/ip/basic_endpoint.hpp"
 
@@ -35,16 +35,18 @@ namespace test {
 enum msg_direction { incoming, outgoing };
 
 struct monitor_msg_data {
+
+  static constexpr max_msg_data_to_capture = 15;
+
   std::string    m_dsr_name;
   std::string    m_protocol; // "tcp" or "udp"
   std::string    m_remote_host;
   std::string    m_remote_port;
   msg_direction  m_direction;
   std::size_t    m_curr_msg_num;
   std::size_t    m_curr_msg_size;
-  std::string    m_curr_prefix;
-  char           m_curr_body_char;
-  std::size_t    m_total_msgs_to_send;  // 0 if incoming direction, since total not known
+  std::string    m_curr_msg_beginning; // up to max_msg_data_to_capture chars
+  std::size_t    m_total_msgs_to_send;  // 0 if direction is incoming, since total not known in advance
   std::size_t    m_outgoing_queue_size;
 
   template <typename AsioIpProtocol>
@@ -53,8 +55,7 @@ struct monitor_msg_data {
                     msg_direction direction,
                     std::size_t curr_msg_num,
                     std::size_t curr_msg_size,
-                    const std::string& curr_prefix,
-                    char curr_body_char,
+                    const std::string& curr_msg_beginning,
                     std::size_t total_msgs_to_send,
                     std::size_t outgoing_queue_size) :
       m_dsr_name(dsr_name),
@@ -64,8 +65,7 @@ struct monitor_msg_data {
       m_direction(direction),
       m_curr_msg_num(curr_msg_num),
       m_curr_msg_size(curr_msg_size),
-      m_curr_prefix(curr_prefix),
-      m_curr_body_char(curr_body_char),
+      m_curr_msg_beginning(curr_msg_beginning),
       m_total_msgs_to_send(total_msgs_to_send),
       m_outgoing_queue_size(outgoing_queue_size)
   {
@@ -75,24 +75,27 @@ struct monitor_msg_data {
       m_protocol = "udp";
     }
   }
-                    
-  // more constructors to be added
 };
 
 struct shutdown_msg {
 // to be filled in
 };
 
 
-// shutdown message from monitor process will cause promise to be set, which is signal to DSR to
-// shutdown
+// shutdown message from monitor process will set promise, which causes future to pop, which
+// tells DSR to shutdown
 class monitor_connector {
 public:
 
-  monitor_connector(chops::net::net_ip& net_ip, std::string_view monitor_port, std::string_view monitor_host,
-                    std::promise<void> prom) : // fill in initializer list, promise must use std::move
+  monitor_connector(chops::net::net_ip& net_ip,
+                    std::string_view monitor_port, std::string_view monitor_host,
+                    std::promise<void> prom,
+                    chops::net::err_wait_q& err_wq) : m_monitor(), m_io_output()
+          // fill in initializer list, promise must use std::move
 
   {
+     // make_tcp_connector, start, provide state chg func obj that does start_io,
+     // providing msg handler for shutdown, make_err_func_with_wait_queue for error display
   }
   void send_monitor_msg (const monitor_msg_data& msg_data) const {
 

test/test_data_blaster/tcp_dsr.cpp

@@ -31,30 +31,57 @@
 #include "net_ip/net_entity.hpp"
 
 #include "net_ip_component/worker.hpp"
-#include "net_ip_component/io_output_delivery.hpp"
 #include "net_ip_component/output_queue_stats.hpp"
 #include "net_ip_component/error_delivery.hpp"
 
 #include "shared_test/msg_handling.hpp"
 #include "marshall/shared_buffer.hpp"
 
 #include "test_data_blaster/tcp_dsr_args.hpp"
+#include "test_data_blaster/monitor_msg_handling.hpp"
+
+const std::string_view msg_prefix { "Tasty testing!" };
+
+std::size_t send_msgs_func (chops::net::io_output io_out, char body_char,
+                            int send_count, std::chrono::milliseconds delay) {
+
+  auto msgs = chops::test::make_msg_vec (chops::test::make_variable_len_msg, msg_prefix, 
+                                         body_char, send_count);
+  for (const auto& m : msgs) {
+    if (!io_out.is_valid()) {
+      break;
+    }
+    io_out.send(m);
+    --send_count;
+    std::this_thread::sleep_for(delay);
+  }
+  return send_count; // 0 unless stopped early
+}
+
 
 struct state_chg {
+  int                       m_send_count;
+  char                      m_body_char;
+  std::chrono::milliseconds m_delay;
+  bool                      m_reply;
+
+  state_chg(int send_count, char body_char, std::chrono::milliseconds delay) :
+    m_send_count(send_count), m_body_char(body_char), m_delay(delay) { }
+
+  void operator() (chops::net::tcp_io_interface io_intf, std::size_t, bool starting) {
+    if (starting) {
+      
+    }
+  }
 
 };
 
 
 int main (int argc, char* argv[]) {
 
   auto parms = chops::test::tcp_dsr_args_parse_command_line(argc, argv);
-  chops::test::vec_buf msgs{};
-  int delay{0};
-  if (parms.m_sending_parms) {
-    const auto& t = *parms.m_sending_parms;
-    delay = t.m_delay;
-    msgs = chops::test::make_msg_vec (chops::test::make_variable_len_msg, t.m_prefix, t.m_body_char, t.m_send_count);
-  }
+
+  char body_char { 'a' };
 
   chops::net::worker wk;
   wk.start();
@@ -67,11 +94,16 @@ int main (int argc, char* argv[]) {
   std::promise<void> shutdown_prom;
   auto shutdown_fut = shutdown_prom.get_future();
 
-  chops::test::monitor_connector mon(nip, parms.m_monitor.m_port, parms.m_monitor.m_host, std::move(shutdown_prom));
+  chops::test::monitor_connector mon(nip, parms.m_monitor.m_port, parms.m_monitor.m_host, 
+                                     std::move(shutdown_prom), err_wq);
 
   for (const auto& s : parms.m_acceptors) {
 
   }
+  for (const auto& s : parms.m_connectors) {
+
+  }
+
 
   shutdown_fut.get();
   err_wq.close();

test/test_data_blaster/tcp_dsr_args.hpp

@@ -19,20 +19,11 @@
 #include <string>
 #include <cstddef> // std::size_t
 #include <vector>
-#include <optional>
+#include <chrono>
 
 namespace chops {
 namespace test {
 
-constexpr int prefix_max_size {20}; // arbitrary max size - we can adjust as needed
-
-struct sending_parms {
-  int           m_send_count;    // num msgs to send
-  std::string   m_prefix;        // prefix string, same for each message
-  char          m_body_char;     // each message will be filled with this char, after the prefix
-  int           m_delay;         // delay in milliseconds between messages
-};
-
 struct connector_info {
   std::string     m_port;
   std::string     m_host;
@@ -41,10 +32,11 @@ struct connector_info {
 // command line argument data for use by TCP DSR
 struct tcp_dsr_args {
   std::string     m_dsr_name;      // name of this DSR instance
-  bool            m_reply;         // if true, incoming messages are reflected back to sender
+  bool            m_reply;         // if true, incoming messages reflected back to sender
   int             m_modulus;       // how often to send log msg to monitor, both outgoing and incoming messages
-  std::optional<sending_parms>
-                  m_sending_parms; // if present - count, prefix, body char
+  int             m_send_count;    // if 0, don't send any messages
+  std::chrono::milliseconds
+                  m_delay;
   std::vector<std::string>
                   m_acceptors;     // 0 - N TCP acceptors, port for each entry
   std::vector<connector_info>
@@ -54,7 +46,8 @@ struct tcp_dsr_args {
 
 
 // besides filling in the command line parsing, decide on error handling - possibilities include
-// return std::optiona<tcp_dsr_args>, or throwing an exception
+// return std::optional<tcp_dsr_args>, or return std::expected<tcp_dsr_args>, or throwing an 
+// exception
 inline tcp_dsr_args parse_command_line (int argc, char* argvp[]) {
 
 }