Restore the serialization benchmark program

tests/benchmarks/CMakeLists.txt

@@ -3,4 +3,4 @@ find_package(benchmark QUIET)
 # add_subdirectory(cluster)
 add_subdirectory(fan-out)
 add_subdirectory(routing-table)
-# add_subdirectory(serialization)
+add_subdirectory(serialization)

tests/benchmarks/serialization/CMakeLists.txt

@@ -4,6 +4,7 @@ endif ()
 
 add_executable(broker-serialization-benchmark
   serialization.cc
+  generator.cc
 )
 
 target_include_directories(broker-serialization-benchmark PRIVATE

tests/benchmarks/serialization/generator.cc

@@ -0,0 +1,75 @@
+#include "generator.hh"
+
+using namespace std::literals;
+
+broker::endpoint_id generator::next_endpoint_id() {
+  broker::endpoint_id::array_type bytes;
+  for (auto& x : bytes)
+    x = static_cast<std::byte>(byte_dis_(rng_));
+  return broker::endpoint_id{bytes};
+}
+
+std::string generator::next_string(size_t length) {
+  std::string result;
+  result.resize(length);
+  for (auto& c : result)
+    c = charset[char_dis_(rng_)];
+  return result;
+}
+
+broker::data generator::next_data(int event_type) {
+  using std::chrono::duration_cast;
+  broker::vector result;
+  switch (event_type) {
+    case 1: {
+      result.reserve(2);
+      result.emplace_back(42);
+      result.emplace_back("test"s);
+      break;
+    }
+    case 2: {
+      auto tcp = broker::port::protocol::tcp;
+      broker::address a1;
+      broker::address a2;
+      broker::convert("1.2.3.4", a1);
+      broker::convert("3.4.5.6", a2);
+      result.emplace_back(next_timestamp());
+      result.emplace_back(next_string(10));
+      result.emplace_back(
+        broker::vector{a1, broker::port(4567, tcp), a2, broker::port(80, tcp)});
+      result.emplace_back(broker::enum_value("tcp"));
+      result.emplace_back(next_string(10));
+      result.emplace_back(duration_cast<broker::timespan>(3140ms));
+      result.emplace_back(next_count());
+      result.emplace_back(next_count());
+      result.emplace_back(next_string(5));
+      result.emplace_back(true);
+      result.emplace_back(false);
+      result.emplace_back(next_count());
+      result.emplace_back(next_string(10));
+      result.emplace_back(next_count());
+      result.emplace_back(next_count());
+      result.emplace_back(next_count());
+      result.emplace_back(next_count());
+      result.emplace_back(broker::set({next_string(10), next_string(10)}));
+      break;
+    }
+    case 3: {
+      broker::table m;
+      for (int i = 0; i < 100; ++i) {
+        broker::set s;
+        for (int j = 0; j < 10; ++j)
+          s.insert(next_string(5));
+        m[next_string(15)] = std::move(s);
+      }
+      result.emplace_back(next_timestamp());
+      result.emplace_back(std::move(m));
+      break;
+    }
+    default: {
+      fprintf(stderr, "event type must be 1, 2, or 3; got %d\n", event_type);
+      throw std::logic_error("invalid event type");
+    }
+  }
+  return broker::data{std::move(result)};
+}

tests/benchmarks/serialization/generator.hh

@@ -9,35 +9,39 @@
 
 class generator {
 public:
-  generator();
+  static constexpr std::string_view charset = "0123456789"
+                                              "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+                                              "abcdefghijklmnopqrstuvwxyz";
 
-  broker::endpoint_id next_endpoint_id();
-
-  static auto make_endpoint_id() {
-    generator g;
-    return g.next_endpoint_id();
+  generator()
+    : rng_(0xB7E57), byte_dis_(0, 255), char_dis_(0, charset.size() - 1) {
+    // nop
   }
 
-  broker::count next_count();
-
-  caf::uuid next_uuid();
+  broker::endpoint_id next_endpoint_id();
 
   std::string next_string(size_t length);
 
-  broker::timestamp next_timestamp();
+  broker::count next_count() {
+    return count_dis_(rng_);
+  }
 
-  // Generates events for one of three possible types:
-  // 1. Trivial data consisting of a number and a string.
-  // 2. More complex data that resembles a line in a conn.log.
-  // 3. Large tables of size 100 by 10, filled with random strings.
-  broker::data next_data(size_t event_type);
+  broker::integer next_integer() {
+    return integer_dis_(rng_);
+  }
 
-  static auto make_data(size_t event_type) {
-    generator g;
-    return g.next_data(event_type);
+  broker::timestamp next_timestamp() {
+    ts_ += std::chrono::seconds(byte_dis_(rng_));
+    return ts_;
   }
 
+  broker::data next_data(int event_type);
+
 private:
   std::minstd_rand rng_;
+  std::uniform_int_distribution<> byte_dis_;
+  std::uniform_int_distribution<size_t> char_dis_;
+  std::uniform_int_distribution<broker::count> count_dis_;
+  std::uniform_int_distribution<broker::integer> integer_dis_;
   broker::timestamp ts_;
 };

tests/benchmarks/serialization/serialization.cc

@@ -1,36 +1,25 @@
-#include "main.hh"
+#include "generator.hh"
 
 #include "broker/alm/multipath.hh"
 #include "broker/endpoint.hh"
+#include "broker/format/json.hh"
 #include "broker/fwd.hh"
+#include "broker/internal/wire_format.hh"
 #include "broker/message.hh"
 
 #include <benchmark/benchmark.h>
 
-#include <caf/binary_deserializer.hpp>
-#include <caf/binary_serializer.hpp>
-
 #include <atomic>
 #include <limits>
 #include <random>
 
 using namespace broker;
+using namespace std::literals;
 
 namespace {
 
-using buffer_type = caf::binary_serializer::container_type;
-
-size_t max_size(size_t init) {
-  return init;
-}
-
-template <class T, class... Ts>
-size_t max_size(size_t init, const T& x, const Ts&... xs) {
-  auto combinator = [](size_t init, const buffer_type& buf) {
-    return std::max(init, buf.size());
-  };
-  return max_size(std::accumulate(x.begin(), x.end(), init, combinator), xs...);
-}
+using caf::byte_buffer;
+using char_buffer = std::vector<char>;
 
 class serialization : public benchmark::Fixture {
 public:
@@ -45,25 +34,36 @@ class serialization : public benchmark::Fixture {
     for (size_t index = 0; index < num_message_types; ++index) {
       dmsg[index] = make_data_message("/micro/benchmark",
                                       g.next_data(index + 1));
-      to_bytes(dmsg[index], dmsg_buf[index]);
-      nmsg[index] = make_node_message(dmsg[index], alm::multipath{dst});
-      to_bytes(nmsg[index], nmsg_buf[index]);
-      legacy_nmsg[index] = legacy_node_message{dmsg[index], 20};
-      to_bytes(legacy_nmsg[index], legacy_nmsg_buf[index]);
+      bin_buf.clear();
+      to_bytes(dmsg[index], bin_buf);
+      dmsg_bin[index] = bin_buf;
+      json_buf.clear();
+      to_json(dmsg[index], json_buf);
+      dmsg_json[index] = json_buf;
     }
-    sink_buf.reserve(max_size(0, dmsg_buf, nmsg_buf, legacy_nmsg_buf));
   }
 
-  template <class T>
-  void to_bytes(T&& what, buffer_type& storage) {
-    caf::binary_serializer sink{nullptr, storage};
-    std::ignore = sink.apply(what);
+  void to_bytes(const data_envelope_ptr& msg, byte_buffer& buf) {
+    broker::internal::wire_format::v1::trait trait;
+    if (!trait.convert(msg, buf))
+      throw std::logic_error("serialization failed");
   }
 
-  template <class T>
-  void from_bytes(const buffer_type& storage, T& what) {
-    caf::binary_deserializer source{nullptr, storage};
-    std::ignore = source.apply(what);
+  void from_bytes(const byte_buffer& buf, envelope_ptr& msg) {
+    broker::internal::wire_format::v1::trait trait;
+    if (!trait.convert(buf, msg))
+      throw std::logic_error("deserialization failed");
+  }
+
+  void to_json(const data_envelope_ptr& msg, char_buffer& buf) {
+    broker::format::json::v1::encode(msg, std::back_inserter(buf));
+  }
+
+  void from_json(const char_buffer& buf, envelope_ptr& msg) {
+    auto res = envelope::deserialize_json(buf.data(), buf.size());
+    if (!res)
+      throw std::logic_error("deserialization failed");
+    msg = std::move(*res);
   }
 
   // Dummy node ID for a receiver.
@@ -73,116 +73,62 @@ class serialization : public benchmark::Fixture {
   array_t<data_message> dmsg;
 
   // Serialized versions of dmsg;
-  array_t<buffer_type> dmsg_buf;
-
-  // One node message per type.
-  array_t<node_message> nmsg;
+  array_t<byte_buffer> dmsg_bin;
 
   // Serialized versions of dmsg;
-  array_t<buffer_type> nmsg_buf;
-
-  // One legacy node message per type.
-  array_t<legacy_node_message> legacy_nmsg;
-
-  // Serialized versions of legacy_dmsg;
-  array_t<buffer_type> legacy_nmsg_buf;
+  array_t<char_buffer> dmsg_json;
 
   // A pre-allocated buffer for the benchmarks to serialize into.
-  buffer_type sink_buf;
-
-  template <class T>
-  auto& get_msg(int signed_index) {
-    auto index = static_cast<size_t>(signed_index);
-    if constexpr (std::is_same_v<T, data_message>) {
-      return dmsg[index];
-    } else if constexpr (std::is_same_v<T, node_message>) {
-      return nmsg[index];
-    } else {
-      static_assert(std::is_same_v<T, legacy_node_message>);
-      return legacy_nmsg[index];
-    }
-  }
-
-  template <class T>
-  const buffer_type& get_buf(int signed_index) const {
-    auto index = static_cast<size_t>(signed_index);
-    if constexpr (std::is_same_v<T, data_message>) {
-      return dmsg_buf[index];
-    } else if constexpr (std::is_same_v<T, node_message>) {
-      return nmsg_buf[index];
-    } else {
-      static_assert(std::is_same_v<T, legacy_node_message>);
-      return legacy_nmsg_buf[index];
-    }
-  }
-
-  template <class T>
-  void run_serialization_bench(benchmark::State& state) {
-    const auto& msg = get_msg<T>(state.range(0));
-    caf::binary_serializer sink{nullptr, sink_buf};
-    for (auto _ : state) {
-      sink.seek(0);
-      std::ignore = sink.apply(msg);
-      benchmark::DoNotOptimize(sink_buf);
-    }
-  }
+  byte_buffer bin_buf;
 
-  template <class T>
-  void run_deserialization_bench(benchmark::State& state) {
-    const auto& buf = get_buf<T>(state.range(0));
-    for (auto _ : state) {
-      T msg;
-      caf::binary_deserializer source{nullptr, buf};
-      std::ignore = source.apply(msg);
-      benchmark::DoNotOptimize(msg);
-    }
-  }
+  // A pre-allocated buffer for the benchmarks to serialize into.
+  char_buffer json_buf;
 };
 
 } // namespace
 
 // -- saving and loading data messages -----------------------------------------
 
-BENCHMARK_DEFINE_F(serialization, save_data_message)(benchmark::State& state) {
-  run_serialization_bench<data_message>(state);
-}
-
-BENCHMARK_REGISTER_F(serialization, save_data_message)->DenseRange(0, 2, 1);
-
-BENCHMARK_DEFINE_F(serialization, load_data_message)(benchmark::State& state) {
-  run_deserialization_bench<data_message>(state);
-}
-
-BENCHMARK_REGISTER_F(serialization, load_data_message)->DenseRange(0, 2, 1);
-
-// -- saving and loading node messages -----------------------------------------
-
-BENCHMARK_DEFINE_F(serialization, save_node_message)(benchmark::State& state) {
-  run_serialization_bench<node_message>(state);
+BENCHMARK_DEFINE_F(serialization, save_binary)(benchmark::State& state) {
+  const auto& msg = dmsg[static_cast<size_t>(state.range(0))];
+  for (auto _ : state) {
+    bin_buf.clear();
+    to_bytes(msg, bin_buf);
+    benchmark::DoNotOptimize(bin_buf);
+  }
 }
 
-BENCHMARK_REGISTER_F(serialization, save_node_message)->DenseRange(0, 2, 1);
+BENCHMARK_REGISTER_F(serialization, save_binary)->DenseRange(0, 2, 1);
 
-BENCHMARK_DEFINE_F(serialization, load_node_message)(benchmark::State& state) {
-  run_deserialization_bench<node_message>(state);
+BENCHMARK_DEFINE_F(serialization, load_binary)(benchmark::State& state) {
+  const auto& buf = dmsg_bin[static_cast<size_t>(state.range(0))];
+  for (auto _ : state) {
+    broker::envelope_ptr msg;
+    from_bytes(buf, msg);
+    benchmark::DoNotOptimize(msg);
+  }
 }
 
-BENCHMARK_REGISTER_F(serialization, load_node_message)->DenseRange(0, 2, 1);
-
-// -- saving and loading legacy node messages ----------------------------------
+BENCHMARK_REGISTER_F(serialization, load_binary)->DenseRange(0, 2, 1);
 
-BENCHMARK_DEFINE_F(serialization, save_legacy_node_message)
-(benchmark::State& state) {
-  run_serialization_bench<legacy_node_message>(state);
+BENCHMARK_DEFINE_F(serialization, save_json)(benchmark::State& state) {
+  const auto& msg = dmsg[static_cast<size_t>(state.range(0))];
+  for (auto _ : state) {
+    json_buf.clear();
+    to_json(msg, json_buf);
+    benchmark::DoNotOptimize(json_buf);
+  }
 }
 
-BENCHMARK_REGISTER_F(serialization, save_legacy_node_message)
-  ->DenseRange(0, 2, 1);
+BENCHMARK_REGISTER_F(serialization, save_json)->DenseRange(0, 2, 1);
 
-BENCHMARK_DEFINE_F(serialization, load_legacy_node_message)
-(benchmark::State& state) {
-  run_deserialization_bench<legacy_node_message>(state);
+BENCHMARK_DEFINE_F(serialization, load_json)(benchmark::State& state) {
+  const auto& buf = dmsg_json[static_cast<size_t>(state.range(0))];
+  for (auto _ : state) {
+    broker::envelope_ptr msg;
+    from_json(buf, msg);
+    benchmark::DoNotOptimize(msg);
+  }
 }
 
-BENCHMARK_REGISTER_F(serialization, load_legacy_node_message)
-  ->DenseRange(0, 2, 1);
+BENCHMARK_REGISTER_F(serialization, load_json)->DenseRange(0, 2, 1);