simple-stream: avoid memcpy calls in fragmented streams for constant sizes

In 3a24d6d ("simple_stream: add [[gnu::always_inline]]"), we
sprinkled [[gnu::always_inline]] to encourage constant propagation
of the size parameter, in non-fragmented streams. When the size is a
constant (which it is, when reading serialized integrals), the memcpy()
call can be optimized into a single instruction to read memory.

Here, we do the same for fragmented streams. Since the code is prepared
for the integral to span two (or more) fragments, it will issue tiny memcpy
calls which expend a large amount of instructions to figure out the right
path, then copy memory using a single instruction and return.

To fix this, we do the following:

1. split for_each_fragment() into a fast-path and slow-path.
2. select the fast path when the size is a constant and it happens
   to fit into the first fragment (which is very likely, as constant
   sizes are usually for the various integral types).
3. encourage inlining (without which we don't get constant propagation)
   with [[gnu::always_inline]].

The fast-path is guarded with __builtin_constant_p. For non-constant
data, we'll call memcpy() anyway, so we don't get much from splitting the
paths.

A demonstration of the optimization is available in [1].

[1] https://godbolt.org/z/rWdMa7bfK

Author: avikivity

include/seastar/core/simple-stream.hh

@@ -129,11 +129,22 @@ class fragmented_memory_output_stream {
 private:
     template<typename Func>
     //requires requires(Func f, view bv) { { f(bv) } -> void; }
+    [[gnu::always_inline]]
     void for_each_fragment(size_t size, Func&& func) {
         if (size > _size) {
             throw std::out_of_range("serialization buffer overflow");
         }
         _size -= size;
+        // Fast path, avoids memcpy for constant sizes.
+        if constexpr (__builtin_constant_p(size)) {
+            if (size <= _current.size()) [[likely]] {
+                func(_current.write_substream(size));
+                return;
+            }
+        }
+        for_each_fragment_slowpath(size, std::forward<Func>(func));
+    }
+    void for_each_fragment_slowpath(size_t size, std::invocable<simple_memory_output_stream> auto&& func) {
         while (size) {
             if (!_current.size()) {
                 _current = simple(reinterpret_cast<char*>((*_it).get_write()), (*_it).size());
@@ -156,9 +167,11 @@ public:
         : _it(it), _size(size) {
     }
 
+    [[gnu::always_inline]]
     void skip(size_t size) {
         for_each_fragment(size, [] (auto) { });
     }
+    [[gnu::always_inline]]
     memory_output_stream<Iterator> write_substream(size_t size) {
         if (size > _size) {
             throw std::out_of_range("serialization buffer overflow");
@@ -171,12 +184,14 @@ public:
         skip(size);
         return substream;
     }
+    [[gnu::always_inline]]
     void write(const char* p, size_t size) {
         for_each_fragment(size, [&p] (auto bv) {
             std::copy_n(p, bv.size(), bv.begin());
             p += bv.size();
         });
     }
+    [[gnu::always_inline]]
     void fill(char c, size_t size) {
         for_each_fragment(size, [c] (simple fragment) {
             std::fill_n(fragment.begin(), fragment.size(), c);
@@ -391,11 +406,22 @@ class fragmented_memory_input_stream {
 private:
     template<typename Func>
     //requires requires(Func f, view bv) { { f(bv) } -> void; }
+    [[gnu::always_inline]]
     void for_each_fragment(size_t size, Func&& func) {
         if (size > _size) {
             throw std::out_of_range("deserialization buffer underflow");
         }
         _size -= size;
+        // Fast path, avoids memcpy for constant sizes.
+        if constexpr (__builtin_constant_p(size)) {
+            if (size <= _current.size()) [[likely]] {
+                func(_current.read_substream(size));
+                return;
+            }
+        }
+        for_each_fragment_slowpath(size, std::forward<Func>(func));
+    }
+    void for_each_fragment_slowpath(size_t size, std::invocable<simple_memory_input_stream> auto&& func) {
         while (size) {
             if (!_current.size()) {
                 _current = simple(reinterpret_cast<const char*>((*_it).begin()), (*_it).size());
@@ -416,9 +442,11 @@ public:
         : _it(it), _size(size) {
     }
 
+    [[gnu::always_inline]]
     void skip(size_t size) {
         for_each_fragment(size, [] (auto) { });
     }
+    [[gnu::always_inline]]
     fragmented read_substream(size_t size) {
         if (size > _size) {
             throw std::out_of_range("deserialization buffer underflow");
@@ -427,12 +455,14 @@ public:
         skip(size);
         return substream;
     }
+    [[gnu::always_inline]]
     void read(char* p, size_t size) {
         for_each_fragment(size, [&p] (auto bv) {
             p = std::copy_n(bv.begin(), bv.size(), p);
         });
     }
     template<typename Output>
+    [[gnu::always_inline]]
     void copy_to(Output& out) {
         for_each_fragment(_size, [&out] (auto bv) {
             bv.copy_to(out);