reactor: Add read_directory() method

The one wraps getdents64 syscall and is used by existing posix_file_impl
to list fd's directs. Next patch will re-use this call for generator
version of the same method.

Another good point of this patch is that reactor friends posix_file_impl
so that this class could access private reactor._thread_pool field. This
patch wraps thread-pool access thus getting us closer to breaking this
friendship.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>

include/seastar/core/reactor.hh

@@ -553,6 +553,8 @@ public:
     future<> link_file(std::string_view oldpath, std::string_view newpath) noexcept;
     future<> chmod(std::string_view name, file_permissions permissions) noexcept;
 
+    future<size_t> read_directory(int fd, char* buffer, size_t buffer_size);
+
     future<int> inotify_add_watch(int fd, std::string_view path, uint32_t flags);
 
     future<std::tuple<file_desc, file_desc>> make_pipe();

src/core/file.cc

@@ -420,16 +420,12 @@ posix_file_impl::list_directory(std::function<future<> (directory_entry de)> nex
         auto eofcond = [w] { return w->eof; };
         return do_until(eofcond, [w, this] {
             if (w->current == w->total) {
-                return engine()._thread_pool->submit<syscall_result<long>>([w , this] () {
-                    auto ret = ::syscall(__NR_getdents64, _fd, reinterpret_cast<linux_dirent64*>(w->buffer), buffer_size);
-                    return wrap_syscall(ret);
-                }).then([w] (syscall_result<long> ret) {
-                    ret.throw_if_error();
-                    if (ret.result == 0) {
+                return engine().read_directory(_fd, w->buffer, buffer_size).then([w] (auto size) {
+                    if (size == 0) {
                         w->eof = true;
                     } else {
                         w->current = 0;
-                        w->total = ret.result;
+                        w->total = size;
                     }
                 });
             }

src/core/reactor.cc

@@ -2013,6 +2013,16 @@ timespec_to_time_point(const timespec& ts) {
     return std::chrono::system_clock::time_point(d);
 }
 
+future<size_t> reactor::read_directory(int fd, char* buffer, size_t buffer_size) {
+    return _thread_pool->submit<syscall_result<long>>([fd, buffer, buffer_size] () {
+        auto ret = ::syscall(__NR_getdents64, fd, reinterpret_cast<linux_dirent64*>(buffer), buffer_size);
+        return wrap_syscall(ret);
+    }).then([] (syscall_result<long> ret) {
+        ret.throw_if_error();
+        return make_ready_future<size_t>(ret.result);
+    });
+}
+
 future<int>
 reactor::inotify_add_watch(int fd, std::string_view path, uint32_t flags) {
     // Allocating memory for a sstring can throw, hence the futurize_invoke