posix_data_source_impl: adaptive prefetch size

Signed-off-by: Yingxin <[email protected]>

include/seastar/net/posix-stack.hh

@@ -109,7 +109,7 @@ class posix_data_source_impl final : public data_source_impl {
     size_t _buf_size;
 public:
     explicit posix_data_source_impl(lw_shared_ptr<pollable_fd> fd, compat::polymorphic_allocator<char>* allocator=memory::malloc_allocator,
-        size_t buf_size = 8192) : _buffer_allocator(allocator), _fd(std::move(fd)),
+        size_t buf_size = 1 << 13) : _buffer_allocator(allocator), _fd(std::move(fd)),
         _buf(make_temporary_buffer<char>(_buffer_allocator, buf_size)), _buf_size(buf_size) {}
     future<temporary_buffer<char>> get() override;
     future<size_t, temporary_buffer<char>> get_direct(char* buf, size_t size) override;

src/net/posix-stack.cc

@@ -315,9 +315,17 @@ posix_ap_server_socket_impl<Transport>::move_connected_socket(socket_address sa,
     }
 }
 
+static constexpr size_t min_buf_size = 1 << 9;
+static constexpr size_t max_buf_size = 1 << 15;
+
 future<temporary_buffer<char>>
 posix_data_source_impl::get() {
     return _fd->read_some(_buf.get_write(), _buf_size).then([this] (size_t size) {
+        if (_buf_size == size) {
+            _buf_size = std::min(max_buf_size, _buf_size << 2);
+        } else if (size < (_buf_size >> 2)) {
+            _buf_size = std::max(min_buf_size, _buf_size >> 2);
+        }
         _buf.trim(size);
         auto ret = std::move(_buf);
         _buf = make_temporary_buffer<char>(_buffer_allocator, _buf_size);
@@ -329,7 +337,12 @@ future<size_t, temporary_buffer<char>>
 posix_data_source_impl::get_direct(char* buf, size_t size) {
     if (size > _buf_size / 2) {
         // this was a large read, we don't prefetch
-        return _fd->read_some(buf, size).then([] (auto read_size) {
+        return _fd->read_some(buf, size).then([this] (auto read_size) {
+            if (_buf_size == read_size) {
+                _buf_size = std::min(max_buf_size, _buf_size << 2);
+            } else if (read_size < (_buf_size >> 2)) {
+                _buf_size = std::max(min_buf_size, _buf_size >> 2);
+            }
             return make_ready_future<size_t, temporary_buffer<char>>(
                 read_size, temporary_buffer<char>());
         });