编译期进行的序列化与反序列化

好久没有写过博客了, 趁假期有空, 讲一讲今年上半年做的一个工作吧, 仓库地址是https://github.com/ShunzDai/serial_cpp. 这个项目的功能就是把一堆数据打包成堆上的字节流, 或是从字节流中解包出原始数据.

主体是一个类, public 成员是两个对外的接口, 都是静态成员函数, 其中serial::pack用于序列化, serial::unpack 用于反序列化; private 成员有几个工具模板和几个私有成员函数. 所有函数都用 constexpr 修饰, 这意味着在所有入参都在编译期确定的情况下，序列化、反序列化的结果也在编译期确定.

class serial {
  public:
  using ibuf_t = std::basic_string_view<uint8_t>;
  using obuf_t = std::basic_string<uint8_t>;

  template <typename ... arg_t>
  constexpr static obuf_t pack(const arg_t & ... args);
  template <typename ... arg_t>
  constexpr static std::tuple<arg_t ...> unpack(ibuf_t &&ibuf);

  private:
  template<typename first_t, typename ... rest_t>
  struct is_tuple {
    static const bool value =
    std::is_same<first_t, std::tuple<>>::value;
  };

  template<typename first_t, typename ... rest_t>
  struct is_tuple<std::tuple<first_t, rest_t ...>> {
    static const bool value = true;
  };

  template<typename first_t, typename ... rest_t>
  struct is_pair {
    static const bool value = false;
  };

  template<typename first_t, typename ... rest_t>
  struct is_pair<std::pair<first_t, rest_t ...>> {
    static const bool value = true;
  };

  template<typename first_t, typename ... rest_t>
  struct is_array {
    static const bool value = false;
  };

  template<typename first_t, size_t size>
  struct is_array<std::array<first_t, size>> {
    static const bool value = true;
  };

  template <typename arg_t>
  struct is_base {
    static const bool value =
    std::is_integral<arg_t>::value ||
    std::is_floating_point<arg_t>::value;
  };

  template <typename arg_t>
  struct is_string {
    static const bool value =
    std::is_same<arg_t,      std::string>::value ||
    std::is_same<arg_t, std::string_view>::value;
  };

  template <typename arg_t>
  static constexpr obuf_t pack_one(const arg_t &arg);
  template <typename ... arg_t, size_t ... seq>
  static constexpr obuf_t pack_tuple(const std::tuple<arg_t ...> &arg, std::index_sequence<seq ...>);
  template <typename arg_t>
  static constexpr arg_t unpack_one(ibuf_t &ibuf);
  template <typename arg_t, size_t ... seq>
  static constexpr arg_t unpack_tuple(ibuf_t &ibuf, std::index_sequence<seq ...>);
};

从序列化讲起. serial::pack的实现如下

template <typename ... arg_t>
constexpr serial::obuf_t serial::pack(const arg_t & ... args) {
  if constexpr (sizeof...(arg_t) == 0)
    return {};
  else
    return (pack_one(args) + ...);
}

这是一个可变参数模板, 入参是一组将要打包的变量, 取常引用减少拷贝次数; 然后是一个静态条件分支, 如果识别到入参个数为0, 则返回空的序列化结果; 否则执行一个常量折叠表达式, 其中调用了serial::pack_one方法, 它返回的是std::basic_string<>类型, 因此折叠表达式中出现的加法运算符是std::basic_string<>重载的字串拼接方法. 综上所述, serial::pack其实就相当于将每个入参都转换为同一类型的字节流块, 然后再把这些块拼接到一起返回.

serial::pack_one的实现如下

template <typename arg_t>
constexpr serial::obuf_t serial::pack_one(const arg_t &arg) {
  if constexpr (std::is_same<arg_t, const char *>::value) {
    return obuf_t((const uint8_t *)arg, strlen(arg) + 1);
  }
  else if constexpr (is_string<arg_t>::value) {
    return obuf_t((const uint8_t *)arg.data(), arg.size()) + obuf_t((const uint8_t *)"\0", 1);
  }
  else if constexpr (is_tuple<arg_t>::value) {
    return pack_tuple(arg, std::make_index_sequence<std::tuple_size<arg_t>::value> {});
  }
  else if constexpr (is_pair<arg_t>::value) {
    return pack_one(arg.first) + pack_one(arg.second);
  }
  else if constexpr (is_array<arg_t>::value) {
    return obuf_t((uint8_t *)arg.data(), arg.size() * sizeof(typename arg_t::value_type));
  }
  else if constexpr (is_base<arg_t>::value) {
    return obuf_t((const uint8_t *)&arg, sizeof(arg_t));
  }
  else {
    static_assert(!std::is_same<arg_t, arg_t>::value, "unknown arg type");
  }
}

可见这个接口针对不同的数据类型采取了不同的序列化方式, 其中使用了一些工具模板, 实现方式大同小异. 举例来说, 对于std::array<>类型的识别使用了工具模板is_array<>, 定义如下

template<typename first_t, typename ... rest_t>
struct is_array {
  static const bool value = false;
};

template<typename first_t, size_t size>
struct is_array<std::array<first_t, size>> {
  static const bool value = true;
};

serial::pack_one接收的参数类型如果是std::array<>类型就会在编译期的类型识别中自动匹配到特化模板, 使得获取到的value为真, 否则为假.

此外serial::pack_one还用到了serial::pack_tuple, 它可将 tuple 类型的参数解包, 递归传递给serial::pack_one. tuple 解包时使用了std::index_sequence<>这个技术方案, 使用方法读者可以自行搜索, 也有其他技术方案可以达成同样的技术效果.

再来看反序列化. serial::unpack的实现如下

template <typename ... arg_t>
constexpr std::tuple<arg_t ...> serial::unpack(ibuf_t &&ibuf) {
  if constexpr (sizeof...(arg_t))
    return {unpack_one<arg_t>(ibuf) ...};
  else
    return {};
}

可见这基本与序列化的机制对称, 因此不再赘述. 如有疑问可以留言.

这个项目没有复杂的编码逻辑, 没有数据类型的传递, 可以快速部署各种生产消费场景中. 但这也是它的缺点, 如果在迭代中产生未查明的参数变更, 可能产生内存越界访问等异常, 因此这种过于简陋的机制可能存在潜在的风险. 事实上这个项目目前就只用在我们部门内部负责的事件通知模型上, 嵌入到原生环境中非常简单

template <loop loop_id, evid event_id>
class event {
  public:
  template <typename ... arg_t>
  event(void(*cb)(arg_t ...));

  template <typename ... arg_t>
  static void notify(const arg_t & ... arg);

  private:
  std::function<void(serial::ibuf_t)> _cb;
};

template <loop loop_id, evid event_id>
template <typename ... arg_t>
event<loop_id, event_id>::event(void(*cb)(arg_t ...))
 : _cb([cb](serial::ibuf_t ibuf) { std::apply(cb, serial::unpack<arg_t ...>(std::move(ibuf))); }) {
  void regist_impl(loop, int32_t, const std::function<void(serial::ibuf_t)> &);
  regist_impl(loop_id, (int)event_id, _cb);
}

template <loop loop_id, evid event_id>
template <typename ... arg_t>
void event<loop_id, event_id>::notify(const arg_t & ... arg) {
  void notify_impl(loop, int32_t, const serial::obuf_t &);
  notify_impl(loop_id, (int)event_id, serial::pack(arg ...));
}

这里的原生环境是c语言的, 注册器将回调函数注册到不同的事件循环线程中, 不同的线程通过loop id区分, 不同的回调函数通过event id区分, 静态成员函数notify可以通过指定loop id和event id向不同的回调函数发送事件通知. 这里也可以看到版本迭代可能会带给项目的不可控因素, 通过一些技术方案也可以弥补serial类的缺陷, 这就交给读者去思考了.