argonne-lcf · Kerilk · Jul 11, 2023 · Jun 16, 2023 · Jun 16, 2023 · Jun 16, 2023
diff --git a/utils/babeltrace_thapi.in b/utils/babeltrace_thapi.in
@@ -100,11 +100,11 @@ text_pretty = BT2::BTPlugin.find('text').get_sink_component_class_by_name('prett
 
 thapi_plugins = BT2::BTPlugin.find_all_from_dir(File.join("@prefix@", "lib"))
 
+# Used for tally and timeline
 interval_plugins = thapi_plugins.filter_map do |plugin|
   plugin.get_filter_component_class_by_name('interval') if $options[:backends].include?(plugin.name.chomp('interval'))
 end
 
-
 # Fix segfault
 ARGV.uniq!
 if !$options[:live]
@@ -171,6 +171,11 @@ serial_comps =
     }
     graph.add_simple_sink('babeltrace_thapi', consume)
   when 'tally'
+
+    thapi_plugins.each do |plugin|
+      interval_plugins << plugin.get_filter_component_class_by_name('aggreg') if plugin.name == 'btx_aggreg'
+    end
+
     comp_intervals = interval_plugins.map.with_index do |comp, i|
       graph.add_component(comp, "interval_#{i}")
     end

diff --git a/xprof/Makefile.am b/xprof/Makefile.am
@@ -20,40 +20,51 @@ xprof_utils.hpp: $(top_srcdir)/utils/xprof_utils.hpp
 %.pb.h %.pb.cc: %.proto
 	$(PROTOC) -I=$(srcdir) --cpp_out=$(builddir) $<
 
-btx_timeline_upstream_model.yaml: btx_interval_model.yaml
-	cp $< $(builddir)/$@
-
-btx_tally_upstream_model.yaml: btx_interval_model.yaml
-	cp $< $(builddir)/$@
-
-btx_%/metababel/metababel.h btx_%/metababel/btx_component.h btx_%/metababel/btx_component.c btx_%/metababel/btx_upstream.h btx_%/metababel/btx_upstream.c btx_%/btx_main.c: btx_%_upstream_model.yaml btx_%_params.yaml
-	$(METABABEL) -u $(builddir)/btx_$*_upstream_model.yaml -t SINK -o btx_$* -p btx_$* -c $* --params $(srcdir)/btx_$*_params.yaml
-
 BTX_TALLY_GENERATED = \
-	btx_tally/metababel/metababel.h \
-	btx_tally/metababel/btx_component.h \
-	btx_tally/metababel/btx_component.c \
-	btx_tally/metababel/btx_upstream.h \
-	btx_tally/metababel/btx_upstream.c \
-	btx_tally/btx_main.c
+	btx_sink_tally/metababel/metababel.h \
+	btx_sink_tally/metababel/btx_component.h \
+	btx_sink_tally/metababel/btx_component.c \
+	btx_sink_tally/metababel/btx_upstream.h \
+	btx_sink_tally/metababel/btx_upstream.c \
+	btx_sink_tally/btx_main.c
 
 BTX_TIMELINE_GENERATED = \
-	btx_timeline/metababel/metababel.h \
-	btx_timeline/metababel/btx_component.h \
-	btx_timeline/metababel/btx_component.c \
-	btx_timeline/metababel/btx_upstream.h \
-	btx_timeline/metababel/btx_upstream.c \
-	btx_timeline/btx_main.c
+	btx_sink_timeline/metababel/metababel.h \
+	btx_sink_timeline/metababel/btx_component.h \
+	btx_sink_timeline/metababel/btx_component.c \
+	btx_sink_timeline/metababel/btx_upstream.h \
+	btx_sink_timeline/metababel/btx_upstream.c \
+	btx_sink_timeline/btx_main.c
+
+BTX_AGGREG_GENERATED = \
+        btx_filter_aggreg/metababel/metababel.h \
+        btx_filter_aggreg/metababel/btx_component.h \
+        btx_filter_aggreg/metababel/btx_component.c \
+        btx_filter_aggreg/metababel/btx_upstream.h \
+        btx_filter_aggreg/metababel/btx_upstream.c \
+        btx_filter_aggreg/metababel/btx_downstream.h \
+        btx_filter_aggreg/metababel/btx_downstream.c \
+        btx_filter_aggreg/btx_main.c
+
+$(BTX_TIMELINE_GENERATED) &: btx_interval_model.yaml btx_timeline_params.yaml
+	$(METABABEL) -u $(srcdir)/btx_interval_model.yaml -t SINK -o btx_sink_timeline -p btx_timeline -c timeline --params $(srcdir)/btx_timeline_params.yaml
+
+$(BTX_TALLY_GENERATED) &: btx_aggreg_model.yaml btx_tally_params.yaml
+	$(METABABEL) -u $(srcdir)/btx_aggreg_model.yaml -t SINK -o btx_sink_tally -p btx_tally -c tally --params $(srcdir)/btx_tally_params.yaml
+
+$(BTX_AGGREG_GENERATED) &: btx_interval_model.yaml btx_aggreg_model.yaml btx_aggreg_params.yaml
+	$(METABABEL) -u $(srcdir)/btx_interval_model.yaml -d $(srcdir)/btx_aggreg_model.yaml -t FILTER -o btx_filter_aggreg -p btx_aggreg -c aggreg --params $(srcdir)/btx_aggreg_params.yaml
 
 BUILT_SOURCES = \
 	$(BTX_TALLY_GENERATED) \
 	$(BTX_TIMELINE_GENERATED) \
+        $(BTX_AGGREG_GENERATED) \
 	xprof_utils.hpp \
 	perfetto_prunned.pb.h \
 	perfetto_prunned.pb.cc
 
 # Library to be packaged
-lib_LTLIBRARIES = libXTally.la libXTimeline.la
+lib_LTLIBRARIES = libXTally.la libXTimeline.la libXAggreg.la
 
 # File generated but not distributed
 nodist_libXTally_la_SOURCES = \
@@ -66,30 +77,45 @@ nodist_libXTimeline_la_SOURCES = \
 	perfetto_prunned.pb.h \
 	perfetto_prunned.pb.cc
 
-# File generated
+nodist_libXAggreg_la_SOURCES = \
+        $(BTX_AGGREG_GENERATED) \
+	xprof_utils.hpp
+
+# File not generated
 libXTally_la_SOURCES = \
-	tally.hpp \
-	tally.cpp \
+	tally_core.hpp \
+	btx_tally.hpp \
+	btx_tally.cpp \
 	my_demangle.c \
 	my_demangle.h \
 	$(top_srcdir)/utils/include/json.hpp
 
 libXTimeline_la_SOURCES = \
-	timeline.cpp
+	btx_timeline.cpp
+
+libXAggreg_la_SOURCES = \
+	tally_core.hpp \
+	btx_aggreg.cpp
 
 # JSON: Right now we harcode it, We should use pkgtools at some point https://github.com/nlohmann/json/blob/develop/single_include/nlohmann/json.hpp
 
 # Compiler flags
-libXTally_la_CPPFLAGS = -I$(top_srcdir)/utils/include -I$(srcdir)/include -I./ -I./btx_tally
+libXTally_la_CPPFLAGS = -I$(top_srcdir)/utils/include -I$(srcdir)/include -I./ -I./btx_sink_tally
 libXTally_la_CFLAGS = -Wall -Wextra -Wno-unused-parameter $(WERROR) -fno-fast-math $(BABELTRACE2_CFLAGS)
 libXTally_la_CXXFLAGS = -std=c++17 -Wall -Wextra -Wno-unused-parameter $(WERROR) -fno-fast-math $(BABELTRACE2_CFLAGS)
 libXTally_la_LDFLAGS = $(BABELTRACE2_LIBS) -avoid-version -module
 
-libXTimeline_la_CPPFLAGS = -I$(top_srcdir)/utils/include -I$(srcdir)/include -I./ -I./btx_timeline
+libXTimeline_la_CPPFLAGS = -I$(top_srcdir)/utils/include -I$(srcdir)/include -I./ -I./btx_sink_timeline
 libXTimeline_la_CFLAGS = -Wall -Wextra -Wno-unused-parameter $(WERROR) -fno-fast-math $(BABELTRACE2_CFLAGS)
 libXTimeline_la_CXXFLAGS = -std=c++17 -Wall -Wextra -Wno-unused-parameter $(WERROR) -fno-fast-math $(BABELTRACE2_CFLAGS) $(PROTOBUF_CFLAGS)
 libXTimeline_la_LDFLAGS = $(BABELTRACE2_LIBS) $(PROTOBUF_LIBS) -avoid-version -module
 
+
+libXAggreg_la_CPPFLAGS = -I$(top_srcdir)/utils/include -I$(srcdir)/include -I./ -I./btx_filter_aggreg
+libXAggreg_la_CFLAGS = -Wall -Wextra -Wno-unused-parameter $(WERROR) -fno-fast-math $(BABELTRACE2_CFLAGS)
+libXAggreg_la_CXXFLAGS = -std=c++17 -Wall -Wextra -Wno-unused-parameter $(WERROR) -fno-fast-math $(BABELTRACE2_CFLAGS)
+libXAggreg_la_LDFLAGS = $(BABELTRACE2_LIBS) -avoid-version -module
+
 # Cannot use check_LTLIBRARIES because we need the shared version of those
 # Thanks Vincent Danjean
 #   noinst_LTLIBRARIES would be the correct thing but then libtool
@@ -124,9 +150,11 @@ EXTRA_DIST = \
 	interval.c.erb \
 	interval.h.erb \
 	btx_interval_model.yaml \
+        btx_aggreg_model.yaml \
 	interval_model.yaml \
 	btx_timeline_params.yaml \
 	btx_tally_params.yaml \
+	btx_aggreg_params.yaml \
 	$(TRACE_FILES) \
 	tests/tally.dust.erb \
 	perfetto_prunned.proto
@@ -137,7 +165,6 @@ CLEANFILES = \
 	$(TALLY_DUST_FILES) \
 	perfetto_prunned.pb.h \
 	perfetto_prunned.pb.cc \
-	btx_timeline_upstream_model.yaml \
-	btx_tally_upstream_model.yaml \
 	$(BTX_TIMELINE_GENERATED) \
-	$(BTX_TALLY_GENERATED)
+	$(BTX_TALLY_GENERATED) \
+	$(BTX_AGGREG_GENERATED)
diff --git a/xprof/btx_aggreg.cpp b/xprof/btx_aggreg.cpp
@@ -0,0 +1,78 @@
+#include "tally_core.hpp"
+#include "xprof_utils.hpp"
+#include <metababel/metababel.h>
+#include <unordered_map>
+#include <cstdint>
+
+typedef std::string name_t;
+typedef std::string metadata_t;
+typedef std::tuple<hostname_t, process_id_t, thread_id_t, backend_t, name_t> _hptbn_t;
+typedef std::tuple<hostname_t, process_id_t, thread_id_t, thapi_device_id, thapi_device_id, name_t, metadata_t> _hptn_dsb_nm_t;
+
+struct aggreg_s {
+  std::unordered_map<_hptbn_t, TallyCoreBase> host;
+  std::unordered_map<_hptbn_t, TallyCoreBase> traffic;
+  std::unordered_map<_hptn_dsb_nm_t, TallyCoreBase> device;
+};
+typedef struct aggreg_s aggreg_t;
+
+static void initialize_usr_data_callback(void *btx_handle, void **usr_data) { *usr_data = new aggreg_t; }
+
+static void finalize_usr_data_callback(void *btx_handle, void *usr_data) {
+  auto *data = (aggreg_t *)usr_data;
+  // Host
+  for (const auto &[hptb_function_name, t] : data->host) {
+    const auto [hostname, vpid, vtid, backend_id, name] = hptb_function_name;
+    btx_push_message_aggreg_host(btx_handle, hostname.c_str(), vpid, vtid, name.c_str(), t.min, t.max, t.duration,
+                                t.count, (int64_t)backend_id, t.error);
+  }
+  // Traffic
+  for (const auto &[hptb_function_name, t] : data->traffic) {
+    const auto &[hostname, vpid, vtid, backend_id, name] = hptb_function_name;
+    btx_push_message_aggreg_traffic(btx_handle, hostname.c_str(), vpid, vtid, name.c_str(), t.min, t.max, t.duration,
+                                   t.count, (int64_t)backend_id);
+  }
+
+  // Device
+  for (const auto &[hptb_function_name, t] : data->device) {
+    const auto &[hostname, vpid, vtid, did, sdid, name, metadata] = hptb_function_name;
+    btx_push_message_aggreg_device(btx_handle, hostname.c_str(), vpid, vtid, name.c_str(), t.min, t.max, t.duration,
+                                  t.count, did, sdid, metadata.c_str());
+  }
+
+  delete data;
+}
+
+static void host_usr_callback(void *btx_handle, void *usr_data, const char *hostname, int64_t vpid,
+                              uint64_t vtid, int64_t ts, int64_t backend_id, const char *name,
+                              uint64_t dur, bt_bool err) {
+  auto *data = (aggreg_t *)usr_data;
+  data->host[{hostname, vpid, vtid, (backend_t)backend_id, name}] += {dur, (bool) err};
+}
+
+static void traffic_usr_callback(void *btx_handle, void *usr_data, const char *hostname,
+                                 int64_t vpid, uint64_t vtid, int64_t ts, int64_t backend_id,
+                                 const char *name, uint64_t size) {
+
+  auto *data = (aggreg_t *)usr_data;
+  data->traffic[{hostname, vpid, vtid, (backend_t)backend_id, name}] += {size, 0};
+}
+
+static void device_usr_callback(void *btx_handle, void *usr_data, const char *hostname,
+                                int64_t vpid, uint64_t vtid, int64_t ts, int64_t backend_id,
+                                const char *name, uint64_t dur, uint64_t did, uint64_t sdid,
+                                bt_bool err, const char *metadata) {
+
+  auto *data = (aggreg_t *)usr_data;
+  data->device[{hostname, vpid, vtid, did, sdid, name, metadata}] += {dur, (bool) err};
+}
+
+void btx_register_usr_callbacks(void *btx_handle) {
+
+  btx_register_callbacks_initialize_usr_data(btx_handle, &initialize_usr_data_callback);
+  btx_register_callbacks_finalize_usr_data(btx_handle, &finalize_usr_data_callback);
+
+  btx_register_callbacks_lttng_host(btx_handle, &host_usr_callback);
+  btx_register_callbacks_lttng_traffic(btx_handle, &traffic_usr_callback);
+  btx_register_callbacks_lttng_device(btx_handle, &device_usr_callback);
+}
diff --git a/xprof/btx_aggreg_model.yaml b/xprof/btx_aggreg_model.yaml
@@ -0,0 +1,135 @@
+:stream_classes:
+  - :name: interval_aggreg
+    :event_common_context_field_class:
+      :type: structure
+      :members:
+        - :name: hostname
+          :field_class:
+            :type: string
+            :cast_type: const char*
+        - :name: vpid
+          :field_class:
+            :type: integer_signed
+            :field_value_range: 64
+            :cast_type: int64_t
+        - :name: vtid
+          :field_class:
+            :type: integer_unsigned
+            :field_value_range: 64
+            :cast_type: uint64_t
+        - :name: name
+          :field_class:
+            :type: string
+            :cast_type: const char*
+        - :name: min
+          :field_class:
+            :type: integer_unsigned
+            :field_value_range: 64
+            :cast_type: uint64_t
+        - :name: max
+          :field_class:
+            :type: integer_unsigned
+            :field_value_range: 64
+            :cast_type: uint64_t
+        - :name: total
+          :field_class:
+            :type: integer_unsigned
+            :field_value_range: 64
+            :cast_type: uint64_t
+        - :name: count
+          :field_class:
+            :type: integer_unsigned
+            :field_value_range: 64
+            :cast_type: uint64_t
+    :event_classes:
+      - :name: aggreg:host
+        :payload_field_class:
+          :type: structure
+          :members:
+            - :name: backend
+              :field_class:
+                :type: integer_unsigned
+                :field_value_range: 64
+                :cast_type: uint64_t
+            - :name: err_count
+              :field_class:
+                :type: integer_unsigned
+                :field_value_range: 64
+                :cast_type: uint64_t
+      - :name: aggreg:device
+        :payload_field_class:
+          :type: structure
+          :members:
+            - :name: did
+              :field_class:
+                :type: integer_unsigned
+                :field_value_range: 64
+                :cast_type: uint64_t
+            - :name: sdid
+              :field_class:
+                :type: integer_unsigned
+                :field_value_range: 64
+                :cast_type: uint64_t
+            - :name: metadata
+              :field_class:
+                :type: string
+                :cast_type: const char*
+      - :name: aggreg:traffic
+        :payload_field_class:
+          :type: structure
+          :members:
+            - :name: backend
+              :field_class:
+                :type: integer_unsigned
+                :field_value_range: 64
+                :cast_type: uint64_t
+  - :name: interval
+    :event_common_context_field_class:
+      :type: structure
+      :members:
+        - :name: hostname
+          :field_class:
+            :type: string
+            :cast_type: const char*
+        - :name: vpid
+          :field_class:
+            :type: integer_signed
+            :field_value_range: 64
+            :cast_type: int64_t
+        - :name: vtid
+          :field_class:
+            :type: integer_unsigned
+            :field_value_range: 64
+            :cast_type: uint64_t
+        - :name: ts
+          :field_class:
+            :type: integer_signed
+            :field_value_range: 64
+            :cast_type: int64_t
+        - :name: backend
+          :field_class:
+            :type: integer_signed
+            :field_value_range: 64
+            :cast_type: int64_t
+    :event_classes:
+      - :name: lttng:device_name
+        :payload_field_class:
+          :type: structure
+          :members:
+            - :name: name
+              :field_class:
+                :type: string
+                :cast_type: const char*
+            - :name: did
+              :field_class:
+                :type: integer_unsigned
+                :field_value_range: 64
+                :cast_type: uint64_t
+      - :name: lttng_ust_thapi:metadata
+        :payload_field_class:
+          :type: structure
+          :members:
+            - :name: metadata
+              :field_class:
+                :type: string
+                :cast_type: const char*