Flow

utils/gen_babeltrace_emitter.rb

@@ -8,7 +8,7 @@
 #include <babeltrace2/babeltrace.h>
 EOF
 
-
+# We should refractor the class_create to take a parameter
 declare_signed = lambda { |_, name|
   puts <<EOF
     bt_field_class *#{name}_field_class = bt_field_class_integer_signed_create(trace_class);
@@ -39,6 +39,13 @@
 EOF
 }
 
+declare_array_dynamic = lambda { |a, name|
+  #Assume that in the model `_lenth` is provided
+  #And that `_element_field_class` have been set by declare_group
+  puts <<EOF
+    bt_field_class *#{name}_field_class = bt_field_class_array_dynamic_create(trace_class, #{name}_element_field_class, #{name}_length_field_class);
+EOF
+}
 
 def append_member(name, parent_name) 
   puts <<EOF
@@ -56,7 +63,8 @@ def append_member(name, parent_name)
   'unsigned' => declare_unsigned,
   'string' => declare_string,
   'bool' => declare_bool,
-  'structure' => declare_structure
+  'structure' => declare_structure,
+  'array_dynamic' => declare_array_dynamic
 })
 
 def declare_group(type, method, group_name, content)
@@ -65,6 +73,10 @@ def declare_group(type, method, group_name, content)
   content.each { |field|
     name = field[:name]
     klass = field[:class]
+    if klass == "array_dynamic"
+        element_klass = field[:field][:class]
+        $print_declarators[element_klass].call(element_klass, "#{name}_element")
+    end
     $print_declarators[klass].call(klass, name)
     append_member(name, group_name)
   }
@@ -73,6 +85,11 @@ def declare_group(type, method, group_name, content)
     bt_field_class_put_ref(#{group_name}_field_class);
 EOF
   content.each { |field|
+    if field[:class] == "array_dynamic"
+    puts <<EOF
+    bt_field_class_put_ref(#{field[:name]}_element_field_class);
+EOF
+    end
     puts <<EOF
     bt_field_class_put_ref(#{field[:name]}_field_class);
 EOF

utils/gen_babeltrace_event_getter_and_setter.rb

@@ -0,0 +1,138 @@
+require 'yaml'
+model = ARGV[0]
+babeltrace_model = YAML::load_file(model)
+
+puts <<EOF
+#pragma once
+#include <babeltrace2/babeltrace.h>
+EOF
+
+def set_scalar(field_name, index, bt2_class, name)
+    puts <<EOF
+     bt_field *#{name}_field = bt_field_structure_borrow_member_field_by_index(#{field_name}_field,#{index});
+     bt_field_#{get_cetter(bt2_class)}_set_value(#{name}_field, #{name});
+EOF
+end
+
+def set_array_dynamic(field_name, index, bt2_class, name, field)
+    puts <<EOF
+     bt_field *#{name}_field = bt_field_structure_borrow_member_field_by_index(#{field_name}_field,#{index});
+     bt_field_array_dynamic_set_length(#{name}_field, #{name}_length);
+     for (unsigned i=0; i < #{name}_length; i++) {
+        bt_field *#{name}_element_field = bt_field_structure_borrow_member_field_by_index(#{name}_field, i);
+        bt_field_#{get_cetter(field[:class])}_set_value(#{name}_element_field, #{name}[i]);
+     }
+EOF
+end
+
+def set_option(field_name, index, bt2_class, name, field)
+    puts <<EOF
+     bt_field *#{name}_field = bt_field_structure_borrow_member_field_by_index(#{field_name}_field,#{index});
+     if (#{name} != NULL) {
+       bt_field_option_set_has_field(#{name}_field, BT_TRUE);
+       bt_field *#{name}_option_field = bt_field_option_borrow_field(#{name}_field);
+       bt_field_#{get_cetter(bt2_class)}_set_value(#{name}_option_field, #{name});
+     } else {
+        bt_field_option_set_has_field(#{name}_field, FALSE);
+     }
+EOF
+end
+
+def set_field(name, field, index)
+    case field[:class]
+    when "array_dynamic"
+        set_array_dynamic(name, index, field[:class], field[:name], field[:field])
+    when "option"
+        set_option(name, index, field[:class], field[:name], field[:field])
+    else
+        set_scalar(name, index, field[:class], field[:name])
+    end
+end
+
+def get_cetter(klass)
+    case klass
+    when "signed"
+        "integer_signed"
+    when "unsigned"
+        "integer_unsigned"
+    else
+        klass
+    end
+end
+
+def get_c_type(field)
+    return field[:cast_type] if field.key?(:cast_type)
+    bits = field.fetch(:class_properties, {}).fetch(:field_value_range, 32)
+    case field[:class]
+    when "signed"
+        "int#{bits}_t"
+    when "unsigned"
+        "uint#{bits}_t"
+    when "string"
+        "const char*"
+    else
+        field[:class]
+    end
+end
+
+def create_signature_tuple(event)
+    event.flat_map { |field|
+          case field[:class]
+          when "array_dynamic"
+                [ ["#{get_c_type(field[:field])}*", field[:name] ] ]
+          #        ["size_t",  "#{field[:name]}_length"] ]
+          else
+                [ [ get_c_type(field), field[:name] ] ]
+          end
+    }
+end
+
+common_context_field_setter = babeltrace_model[:stream_classes].map { |stream_class|
+    name = stream_class[:name].gsub(':','_')
+
+    l = create_signature_tuple(stream_class[:common_context])
+    signature_str = l.map { |e| e.join(' ') }.join(', ')
+    puts <<EOF
+void bt_event_set_#{name}_common_context_fields(bt_event* event, #{signature_str}) {
+EOF
+    puts <<EOF
+     bt_field *common_context_field = bt_event_borrow_common_context_field(event);
+EOF
+    stream_class[:common_context].each.with_index { |common_context, i|
+        set_field("common_context", common_context, i)
+    }
+    puts <<EOF
+}
+EOF
+    [name, l]
+}.to_h
+
+babeltrace_model[:event_classes].each { |event_class|
+    name_stream_class = event_class.fetch("stream_class", common_context_field_setter.keys[0])
+    l_common = common_context_field_setter[name_stream_class]
+    l_payload = create_signature_tuple(event_class[:payload])
+
+    l_signature_str = (l_common + l_payload).map { |e| e.join(' ') }.join(', ')
+
+    l_common_call_str = l_common.map { |e, n| n}.join(', ')
+    l_call_str = (l_common + l_payload).map { |e, n| n}.join(', ')
+
+    puts <<EOF
+void bt_event_set_#{event_class[:name].gsub(':','_')}(bt_event* event, #{l_signature_str}) {
+     bt_event_set_#{name_stream_class}_common_context_fields(event, #{l_common_call_str});
+     bt_field *payload_field = bt_event_borrow_payload_field(event);
+EOF
+    event_class[:payload].each.with_index { |payload, i|
+        set_field("payload", payload, i)
+    }
+    puts <<EOF
+}
+
+bt_message* bt_message_create_#{event_class[:name].gsub(':','_')}(bt_event_class *event_class, bt_self_message_iterator *message_iterator, bt_stream *stream, #{l_signature_str}) {
+     bt_message *message = bt_message_event_create(message_iterator, event_class, stream);
+     bt_event *downstream_event = bt_message_event_borrow_event(message);
+     bt_event_set_#{event_class[:name].gsub(':','_')}(downstream_event, #{l_call_str});
+     return message;
+}
+EOF
+}

utils/xprof_utils.cpp

@@ -1,4 +1,5 @@
 #include "xprof_utils.hpp"
+#include <set>
 
 const char* borrow_hostname(const bt_event *event){
     const bt_stream *stream = bt_event_borrow_stream_const(event);
@@ -22,7 +23,7 @@ thread_id_t borrow_thread_id(const bt_event *event){
 
 bt_message* create_host_message(const char* hostname, const process_id_t process_id, const thread_id_t thread_id, const char* name,
         const uint64_t ts, const uint64_t duration, const bool err,
-        bt_event_class *event_class, bt_self_message_iterator *message_iterator, bt_stream *stream, backend_t backend) {
+        bt_event_class *event_class, bt_self_message_iterator *message_iterator, bt_stream *stream, backend_t backend, std::set<flow_id_t> flow_ids) {
 
      /* Message creation */
      bt_message *message = bt_message_event_create(
@@ -65,6 +66,21 @@ bt_message* create_host_message(const char* hostname, const process_id_t process
      bt_field *err_field = bt_field_structure_borrow_member_field_by_index(payload_field, 2);
      bt_field_integer_unsigned_set_value(err_field, err);
 
+     // flow id
+     if (!flow_ids.empty() ) {
+        const auto size = flow_ids.size();
+        //bt_field *flow_id_field_length = bt_field_structure_borrow_member_field_by_index(payload_field, 3);
+        //bt_field_integer_unsigned_set_value(flow_id_field_length, size);
+
+        bt_field *flow_id_field = bt_field_structure_borrow_member_field_by_index(payload_field, 4);
+        bt_field_array_dynamic_set_length(flow_id_field, size);
+
+        int i=0;
+        for (auto element: flow_ids) {
+            bt_field *element_field = bt_field_structure_borrow_member_field_by_index(flow_id_field, i++);
+            bt_field_integer_unsigned_set_value(element_field, element);
+        }
+     }
      return message;
 }
 
@@ -125,8 +141,79 @@ bt_message* create_device_message(const char* hostname, const process_id_t proce
      //Metadata
      bt_field *metadata_field = bt_field_structure_borrow_member_field_by_index(payload_field, 5);
      bt_field_string_set_value(metadata_field, metadata);
+     return message;
+}
 
-    return message;
+bt_message* create_device_flow_message(const char* hostname, const process_id_t process_id, const uint64_t uuid,
+                                  const thapi_device_id device_id, const thapi_device_id subdevice_id,
+                                  const char* name, const uint64_t ts, const uint64_t duration, const bool err,
+                                  const char* metadata, const char* queue_name, const flow_id_t flow_id,
+                                  bt_event_class *event_class, bt_self_message_iterator *message_iterator, bt_stream *stream) {
+
+     /* Message creation */
+     bt_message *message = bt_message_event_create(
+                             message_iterator, event_class, stream);
+
+
+     /* event */
+     bt_event *downstream_event = bt_message_event_borrow_event(message);
+
+     /* Common context */
+     bt_field *context_field = bt_event_borrow_common_context_field(downstream_event);
+
+     // Hostname
+     bt_field *hostname_msg_field = bt_field_structure_borrow_member_field_by_index(context_field,0);
+     bt_field_string_set_value(hostname_msg_field, hostname);
+     // pid
+     bt_field *vpid_field = bt_field_structure_borrow_member_field_by_index(context_field,1);
+     bt_field_integer_signed_set_value(vpid_field, process_id);
+     // vid
+     bt_field *vtid_field = bt_field_structure_borrow_member_field_by_index(context_field,2);
+     bt_field_integer_signed_set_value(vtid_field, 0);
+     // ts
+     bt_field *ts_field = bt_field_structure_borrow_member_field_by_index(context_field,3);
+     bt_field_integer_signed_set_value(ts_field, ts);
+
+     /* Payload */
+     bt_field *payload_field = bt_event_borrow_payload_field(downstream_event);
+
+     // name
+     bt_field *name_field = bt_field_structure_borrow_member_field_by_index(payload_field, 0);
+     bt_field_string_set_value(name_field, name);
+
+     // dur
+     bt_field *dur_field = bt_field_structure_borrow_member_field_by_index(payload_field, 1);
+     bt_field_integer_unsigned_set_value(dur_field, duration);
+
+     // did
+     bt_field *device_id_field = bt_field_structure_borrow_member_field_by_index(payload_field,2);
+     bt_field_integer_unsigned_set_value(device_id_field, device_id);
+
+     // sdid
+     bt_field *subdevice_id_field = bt_field_structure_borrow_member_field_by_index(payload_field,3);
+     bt_field_integer_unsigned_set_value(subdevice_id_field, subdevice_id);
+
+     // err
+     bt_field *err_field = bt_field_structure_borrow_member_field_by_index(payload_field, 4);
+     bt_field_integer_unsigned_set_value(err_field, err);
+
+     //Metadata
+     bt_field *metadata_field = bt_field_structure_borrow_member_field_by_index(payload_field, 5);
+     bt_field_string_set_value(metadata_field, metadata);
+
+     // uuid
+     bt_field *uuid_field = bt_field_structure_borrow_member_field_by_index(payload_field, 6);
+     bt_field_integer_unsigned_set_value(uuid_field, uuid);
+
+     //Queue_name
+     bt_field *queue_name_field = bt_field_structure_borrow_member_field_by_index(payload_field, 7);
+     bt_field_string_set_value(queue_name_field, queue_name);
+
+     // flow_id
+     bt_field *flow_field = bt_field_structure_borrow_member_field_by_index(payload_field, 8);
+     bt_field_integer_unsigned_set_value(flow_field, flow_id);
+
+     return message;
 }
 
 bt_message* create_device_name_message(const char* hostname, const process_id_t process_id,

utils/xprof_utils.hpp

@@ -1,6 +1,6 @@
 #pragma once
 
-#include <map>
+#include <set>
 #include <tuple>
 #include <string>
 #include "babeltrace2/babeltrace.h"
@@ -28,10 +28,15 @@ typedef uintptr_t                      thread_id_t;
 typedef std::string                    hostname_t;
 typedef std::string                    thapi_function_name;
 typedef uintptr_t                      thapi_device_id;
+typedef uint64_t flow_id_t;
 
 // Represent a device and a sub device
 typedef std::tuple<thapi_device_id, thapi_device_id> dsd_t;
 typedef std::tuple<hostname_t, thapi_device_id> h_device_t;
+
+typedef std::tuple<hostname_t, thapi_device_id> h_d_t;
+typedef std::tuple<hostname_t, thapi_device_id,thapi_device_id> h_dsd_t;
+
 typedef std::tuple<hostname_t, process_id_t> hp_t;
 typedef std::tuple<hostname_t, process_id_t, thread_id_t> hpt_t;
 typedef std::tuple<hostname_t, process_id_t, thread_id_t, thapi_function_name> hpt_function_name_t;
@@ -120,12 +125,18 @@ thread_id_t borrow_thread_id(const bt_event*);
 
 bt_message* create_host_message(const char *hostname, const process_id_t, const thread_id_t,
                                 const char *name, const uint64_t ts, const uint64_t duration, const bool err,
-                                bt_event_class*, bt_self_message_iterator*, bt_stream*, backend_t = BACKEND_UNKNOWN);
+                                bt_event_class*, bt_self_message_iterator*, bt_stream*, backend_t = BACKEND_UNKNOWN,  std::set<flow_id_t> flow_ids = {});
 
 bt_message* create_device_message(const char *hostname, const process_id_t, const thread_id_t, const thapi_device_id, const thapi_device_id,
                                   const char *name, const uint64_t ts, const uint64_t duration, const bool err, const char* metadata,
                                   bt_event_class*, bt_self_message_iterator*, bt_stream*);
 
+bt_message* create_device_flow_message(const char* hostname, const process_id_t process_id, const uint64_t uuid,
+                                  const thapi_device_id device_id, const thapi_device_id subdevice_id,
+                                  const char* name, const uint64_t ts, const uint64_t duration, const bool err,
+                                  const char* metadata, const char* queue_name, const uint64_t flow_id,
+                                  bt_event_class *event_class, bt_self_message_iterator *message_iterator, bt_stream *stream);
+
 bt_message* create_device_name_message(const char* hostname, const process_id_t process_id,
                                        const thapi_device_id device_id, const char* name,
                                        bt_event_class *event_class, bt_self_message_iterator *message_iterator, bt_stream *stream);

xprof/Makefile.am

@@ -48,10 +48,6 @@ libXProf_la_SOURCES = \
 	my_demangle.h \
 	$(top_srcdir)/utils/include/json.hpp
 
-# Right now we harcode it. We should use pkgtools at some point https://github.com/nlohmann/json/blob/develop/single_include/nlohmann/json.hpp
-
-# -I/home/tapplencourt/tmp/proto/protobuf-3.20.1/ici/include/ -L/home/tapplencourt/tmp/proto/protobuf-3.20.1/ici/lib/ -lprotobuf
-
 # Compiler flags
 libXProf_la_CPPFLAGS = -I$(top_srcdir)/utils/include -I$(srcdir)/include -I./ -I/home/tapplencourt/tmp/proto/protobuf-3.20.1/ici/include/
 libXProf_la_CFLAGS = -Wall -Wextra -Wno-unused-parameter $(WERROR) $(BABELTRACE2_CFLAGS)

xprof/interval.c.erb

@@ -92,8 +92,12 @@ bt_component_class_initialize_method_status <%= namespace %>_dispatch_initialize
     // We don't create a clock, because we ensure monotonic order for the downstream messages.
     populate_ze_stream_class_common_context(trace_class, stream_class);
 
+    // This should be generated!
     dispatch->host_event_class = create_lttng_host_event_class_message(trace_class,stream_class);
+    dispatch->host_flow_event_class = create_lttng_host_flow_event_class_message(trace_class,stream_class);
+
     dispatch->device_event_class = create_lttng_device_event_class_message(trace_class, stream_class);
+    dispatch->device_flow_event_class = create_lttng_device_flow_event_class_message(trace_class, stream_class);
     dispatch->traffic_event_class = create_lttng_traffic_event_class_message(trace_class, stream_class);
     dispatch->device_name_event_class = create_lttng_device_name_event_class_message(trace_class, stream_class);
 

xprof/interval.h.erb

@@ -53,10 +53,11 @@ struct <%= namespace %>_dispatch {
     /* Downstream message */
     bt_stream *stream;
     bt_event_class *host_event_class;
+    bt_event_class *host_flow_event_class;
     bt_event_class *device_event_class;
+    bt_event_class *device_flow_event_class;
     bt_event_class *traffic_event_class;
     bt_event_class *device_name_event_class;
-
     /* Component's input port (weak) */
     bt_self_component_port_input *in_port;
 };