EPOL-19: Parse latency histogram

VNFs/DPPD-PROX/cmd_parser.c

@@ -2005,11 +2005,58 @@ static void handle_latency_histogram(unsigned lcore_id, unsigned task_id, struct
 	}
 }
 
+static void parse_bucket_size_freq(struct input *input)
+{
+	uint32_t bucket_size = stats_get_latency_bucket_size();
+	uint64_t hz = rte_get_tsc_hz();
+
+	if (input->reply) {
+			char buf[128];
+			snprintf(buf, sizeof(buf),"Bucket size is %d, freq is %ld\n", bucket_size, hz);
+			input->reply(input, buf, strlen(buf));
+	}
+	else {
+			plog_info("Bucket size is %d, freq is %ld\n", bucket_size, hz);
+	}
+}
+
 static void handle_stats_and_packets(unsigned lcore_id, unsigned task_id, struct input *input)
 {
+	parse_bucket_size_freq(input);
 	handle_lat_stats(lcore_id, task_id, input);
 	handle_latency_histogram(lcore_id, task_id, input);
 }
+
+static void handle_total_latency_histogram(unsigned lcore_id, unsigned task_id, struct input *input)
+{
+	parse_bucket_size_freq(input);
+
+    uint64_t *buckets;
+
+    stats_core_lat_total_histogram(lcore_id, task_id, &buckets);
+
+    if (buckets == NULL) {
+        if (input->reply) {
+            char buf[128];
+            snprintf(buf, sizeof(buf), "error: unexpected NULL bucket\n");
+            input->reply(input, buf, strlen(buf));
+        }
+        return;
+    }
+
+    if (input->reply) {
+        char buf[4096] = {0};
+        for (size_t i = 0; i < LAT_BUCKET_COUNT; i++)
+            sprintf(buf+strlen(buf), "Total Bucket [%zu]: %"PRIu64"\n", i, buckets[i]);
+        input->reply(input, buf, strlen(buf));
+
+    }
+    else {
+        for (size_t i = 0; i < LAT_BUCKET_COUNT; i++)
+            if (buckets[i])
+                plog_info("Total Bucket [%zu]: %"PRIu64"\n", i, buckets[i]);
+    }
+}
 #endif
 
 static int parse_cmd_dp_core_stats(const char *str, struct input *input)
@@ -2056,6 +2103,22 @@ static int parse_cmd_lat_stats_and_packets(const char *str, struct input *input)
 	return 0;
 }
 
+static int parse_cmd_lat_tot_stats(const char *str, struct input *input)
+{
+#ifdef LATENCY_HISTOGRAM
+    handle_cores_tasks(str, input, "lat", "latency", handle_total_latency_histogram);
+#else
+    if (input->reply) {
+        char buf[128];
+        snprintf(buf, sizeof(buf), "error: LATENCY_HISTOGRAM disabled\n");
+        input->reply(input, buf, strlen(buf));
+    } else {
+        plog_info("LATENCY_HISTOGRAMS disabled\n");
+    }
+#endif
+    return 0;
+}
+
 static int parse_cmd_show_irq_buckets(const char *str, struct input *input)
 {
 	char buf[4096] = {0};
@@ -2354,6 +2417,7 @@ static struct cmd_str cmd_strings[] = {
 	{"show irq buckets", "<core id> <task id>", "Print irq buckets", parse_cmd_show_irq_buckets},
 	{"lat packets", "<core id> <task id>", "Print the latency for each of the last set of packets", parse_cmd_lat_packets},
 	{"lat all stats", "<core id> <task id>", "Print the latency for each of the last set of packets as well as latency distribution", parse_cmd_lat_stats_and_packets},
+	{"lat tot stats", "<core id> <task id>", "Print total latency histogram since reset", parse_cmd_lat_tot_stats},
 	{"accuracy limit", "<core id> <task id> <nsec>", "Only consider latency of packets that were measured with an error no more than <nsec>", parse_cmd_accuracy},
 	{"core stats", "<core id> <task id>", "Print rx/tx/drop for task <task id> running on core <core id>", parse_cmd_core_stats},
 	{"dp core stats", "<core id> <task id>", "Print rx/tx/non_dp_rx/non_dp_tx/drop for task <task id> running on core <core id>", parse_cmd_dp_core_stats},

VNFs/DPPD-PROX/display.c

@@ -21,6 +21,7 @@
 #include <signal.h>
 #include <math.h>
 #include <signal.h>
+#include <pthread.h>
 
 #include "display_latency.h"
 #include "display_mempools.h"

VNFs/DPPD-PROX/helper-scripts/prox_client.py

@@ -21,7 +21,7 @@ class ProxUtility(object):
 
     ALLOWED_OPERATIONS = [
         'start', 'stop', 'tot_stats', 'reset_stats', 'quit',
-        'lat_stats', 'dp_core_stats']
+        'lat_stats', 'lat_all_stats', 'lat_tot_stats', 'dp_core_stats']
     DEFAULT_PORT = 8474
 
     def __init__(self, sock):
@@ -47,6 +47,142 @@ def _recv(self):
         self._received = self._received[pos + 1:]
         return response
 
+    def _recv_multiline(self):
+        """ Receive multiline response from PROX instance (for histogram data). """
+        try:
+            full_response = b''
+
+            # Set a short timeout for multiline responses
+            original_timeout = self._sock.gettimeout()
+            self._sock.settimeout(2.0)  # 2 second timeout for histogram data
+
+            try:
+                while True:
+                    try:
+                        chunk = self._sock.recv(4096)
+                        if not chunk:
+                            # Empty chunk means no more data or connection closed
+                            break
+                        full_response += chunk
+                    except socket.timeout:
+                        # Timeout means no more data is coming
+                        break
+            finally:
+                # Restore original timeout
+                if original_timeout is not None:
+                    self._sock.settimeout(original_timeout)
+                else:
+                    self._sock.settimeout(None)
+
+            # Parse the response
+            if full_response:
+                response_lines = full_response.decode('utf-8', errors='ignore').strip().split('\n')
+                # Filter out empty lines
+                response_lines = [line.strip() for line in response_lines if line.strip()]
+                return response_lines
+
+            return []
+        except Exception as e:
+            raise RuntimeError(f'PROX failed to read multi-line response: {e}')
+
+    def _parse_histogram_response(self, response_lines, bucket_prefix="Bucket"):
+        """
+        Parse histogram response lines from PROX latency commands.
+
+        This method processes multiline responses from PROX 'lat all stats' and 'lat tot stats'
+        commands, extracting bucket histogram data and configuration information.
+
+        :param response_lines: List of response lines received from PROX socket
+        :param bucket_prefix: Bucket identifier prefix - "Bucket" for lat_all_stats, 
+                             "Total Bucket" for lat_tot_stats
+        :return: tuple of (histogram_data, bucket_info) where:
+                 - histogram_data: dict mapping bucket names to packet counts
+                 - bucket_info: dict with bucket_size and frequency_hz metadata
+        """
+        histogram_data = {}
+        bucket_info = {
+            'bucket_size': None,
+            'frequency_hz': None
+        }
+
+        for line in response_lines:
+            # Parse bucket size and frequency info
+            if 'Bucket size is' in line and 'freq is' in line:
+                try:
+                    parts = line.split(',')
+                    for part in parts:
+                        part = part.strip()
+                        if part.startswith('Bucket size is'):
+                            bucket_size_str = part.replace('Bucket size is', '').strip()
+                            bucket_info['bucket_size'] = int(bucket_size_str)
+                        elif part.startswith('freq is'):
+                            freq_str = part.replace('freq is', '').strip()
+                            bucket_info['frequency_hz'] = int(freq_str)
+                except (ValueError, IndexError):
+                    continue
+
+            # Parse bucket data with flexible prefix
+            elif line.startswith(f'{bucket_prefix} [') and ']:' in line:
+                try:
+                    bucket_part, count_part = line.split(']: ', 1)
+                    bucket_num = bucket_part.split('[')[1]  # Extract X from "Bucket [X" or "Total Bucket [X"
+                    count = int(count_part)
+                    histogram_data[f'{bucket_prefix} [{bucket_num}]'] = count
+                except (ValueError, IndexError):
+                    continue
+
+        return histogram_data, bucket_info
+
+    def _get_latency_histogram_stats(self, command, bucket_prefix, **kwargs):
+        """
+        Common implementation for collecting latency histogram statistics from PROX.
+
+        This method handles the complete workflow of sending histogram commands to PROX,
+        receiving multiline responses, parsing bucket data, and formatting results.
+        Used by both lat_all_stats() and lat_tot_stats() methods.
+
+        :param command: PROX socket command string ('lat all stats' or 'lat tot stats')
+        :param bucket_prefix: Bucket line prefix for parsing ('Bucket' or 'Total Bucket')
+        :param kwargs: Command parameters containing:
+                      - iterations: Number of histogram collections to perform
+                      - core: CPU core ID for latency measurement
+                      - task: Task ID on the specified core
+                      - sleep: Sleep duration between iterations in seconds
+        :return: JSON string containing histogram data for all iterations with metadata
+        """
+        iterations = kwargs['iterations']
+        core = kwargs['core']
+        task = kwargs['task']
+        sleep_in_seconds = kwargs['sleep']
+        data_dict = {}
+
+        for iteration_string in map(str, range(1, iterations + 1)):
+            time.sleep(sleep_in_seconds)
+            self._send('{} {} {}'.format(command, core, task))
+
+            dt = datetime.datetime.now(timezone.utc)
+            utc_time = dt.replace(tzinfo=timezone.utc)
+            utc_timestamp = utc_time.timestamp()
+
+            # Read the multiline response
+            response_lines = self._recv_multiline()
+
+            # Parse histogram data using common method
+            histogram_data, bucket_info = self._parse_histogram_response(response_lines, bucket_prefix)
+
+            iter_data = {
+                'Iteration': iteration_string,
+                'Timestamp': utc_timestamp,
+                'Core': core,
+                'Task': task,
+                'Bucket_Info': bucket_info,
+                'Histogram': histogram_data,
+                'Bucket_Count': len(histogram_data)
+            }
+            data_dict.update({iteration_string: iter_data})
+
+        return json.dumps(data_dict)
+
     def start(self, **kwargs):
         """ Method that starts all tasks """
         if kwargs.get('core') and kwargs.get('task'):
@@ -155,6 +291,44 @@ def lat_stats(self, **kwargs):
 
         return data_dict
 
+    def lat_all_stats(self, **kwargs):
+        """
+        Collect current latency histogram distribution from PROX showing packets per second.
+
+        Executes the PROX 'lat all stats' command to get instantaneous latency 
+        histogram data showing the distribution of packet latencies across buckets
+        as packets per second rates. This provides a snapshot of current latency
+        distribution rates at the time of measurement.
+
+        :param kwargs: Parameters passed to _get_latency_histogram_stats():
+                      - iterations: Number of histogram snapshots to collect
+                      - core: Core ID of the latency measurement core  
+                      - task: Task ID on the specified core (typically 0 for latency tasks)
+                      - sleep: Sleep duration between iterations in seconds
+        :return: JSON string containing histogram data with bucket distribution,
+                 bucket configuration (size, frequency), and timestamps
+        """
+        return self._get_latency_histogram_stats('lat all stats', 'Bucket', **kwargs)
+
+
+    def lat_tot_stats(self, **kwargs):
+        """
+        Collect accumulated latency histogram data from PROX showing total packet counts.
+
+        Executes the PROX 'lat tot stats' command to get cumulative latency 
+        histogram data that has been accumulated since the last statistics reset.
+        This provides the total count of packets in each latency bucket measured
+        during the entire test period, useful for comprehensive latency analysis.
+
+        :param kwargs: Parameters passed to _get_latency_histogram_stats():
+                      - iterations: Number of total histogram snapshots to collect
+                      - core: Core ID of the latency measurement core
+                      - task: Task ID on the specified core (typically 0 for latency tasks)  
+                      - sleep: Sleep duration between iterations in seconds
+        :return: JSON string containing accumulated histogram data with total bucket
+                 packet counts, bucket configuration (size, frequency), and timestamps
+        """
+        return self._get_latency_histogram_stats('lat tot stats', 'Total Bucket', **kwargs)
 
 # **************** Command Parser and main method *************************** #
 
@@ -178,7 +352,7 @@ def get_parser():
         '-m', '--command', type=str, required=True,
         action=ValidateOperation,
         help='PROX operation to be performed ('
-             'start/stop/tot_stats/reset_stats/lat_stats)'
+             'start/stop/tot_stats/reset_stats/lat_stats/lat_all_stats/lat_tot_stats)'
     )
     parser.add_argument(
         '-c', '--core-id', type=str, required=False, default='all',
@@ -227,6 +401,10 @@ def execute_prox_command(args):
         method_cb = prox_wrapper.quit
     elif command == 'lat_stats':
         method_cb = prox_wrapper.lat_stats
+    elif command == 'lat_all_stats':
+        method_cb = prox_wrapper.lat_all_stats
+    elif command == 'lat_tot_stats':
+        method_cb = prox_wrapper.lat_tot_stats
     elif command == 'dp_core_stats':
         method_cb = prox_wrapper.dp_core_stats
     else:

VNFs/DPPD-PROX/lconf.c

@@ -13,6 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 */
+#include <pthread.h>
 
 #include "prox_malloc.h"
 #include "lconf.h"

VNFs/DPPD-PROX/stats_latency.c

@@ -194,6 +194,18 @@ void stats_core_lat_histogram(uint8_t lcore_id, uint8_t task_id, uint64_t **buck
 	else
 		*buckets = NULL;
 }
+
+void stats_core_lat_total_histogram(uint8_t lcore_id, uint8_t task_id, uint64_t **buckets)
+{
+    struct stats_latency_manager_entry *lat_stats;
+
+    lat_stats = stats_latency_entry_find(lcore_id, task_id);
+
+    if (lat_stats)
+		*buckets = lat_stats->tot_lat_test.buckets;
+    else
+		*buckets = NULL;
+}
 #endif
 
 static void stats_latency_fetch_entry(struct stats_latency_manager_entry *entry)

VNFs/DPPD-PROX/stats_latency.h

@@ -55,6 +55,7 @@ int stats_get_latency_bucket_size(void);
 
 #ifdef LATENCY_HISTOGRAM
 void stats_core_lat_histogram(uint8_t lcore_id, uint8_t task_id, uint64_t **buckets);
+void stats_core_lat_total_histogram(uint8_t lcore_id, uint8_t task_id, uint64_t **buckets);
 #endif
 
 #endif /* _STATS_LATENCY_H_ */