core: add label to io_threaded_fallbacks to categorize operations

Author: ballard26

src/core/file.cc

@@ -230,7 +230,7 @@ future<int>
 posix_file_impl::ioctl(uint64_t cmd, void* argp) noexcept {
     return engine()._thread_pool->submit<syscall_result<int>>([this, cmd, argp] () mutable {
         return wrap_syscall<int>(::ioctl(_fd, cmd, argp));
-    }).then([] (syscall_result<int> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result<int> sr) {
         sr.throw_if_error();
         // Some ioctls require to return a positive integer back.
         return make_ready_future<int>(sr.result);
@@ -251,7 +251,7 @@ future<int>
 posix_file_impl::fcntl(int op, uintptr_t arg) noexcept {
     return engine()._thread_pool->submit<syscall_result<int>>([this, op, arg] () mutable {
         return wrap_syscall<int>(::fcntl(_fd, op, arg));
-    }).then([] (syscall_result<int> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result<int> sr) {
         sr.throw_if_error();
         // Some fcntls require to return a positive integer back.
         return make_ready_future<int>(sr.result);
@@ -273,7 +273,7 @@ posix_file_impl::discard(uint64_t offset, uint64_t length) noexcept {
     return engine()._thread_pool->submit<syscall_result<int>>([this, offset, length] () mutable {
         return wrap_syscall<int>(::fallocate(_fd, FALLOC_FL_PUNCH_HOLE|FALLOC_FL_KEEP_SIZE,
             offset, length));
-    }).then([] (syscall_result<int> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result<int> sr) {
         sr.throw_if_error();
         return make_ready_future<>();
     });
@@ -294,7 +294,7 @@ posix_file_impl::allocate(uint64_t position, uint64_t length) noexcept {
             supported = false; // Racy, but harmless.  At most we issue an extra call or two.
         }
         return wrap_syscall<int>(ret);
-    }).then([] (syscall_result<int> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result<int> sr) {
         sr.throw_if_error();
         return make_ready_future<>();
     });
@@ -336,7 +336,7 @@ posix_file_impl::close() noexcept {
             try {
                 return engine()._thread_pool->submit<syscall_result<int>>([fd] {
                     return wrap_syscall<int>(::close(fd));
-                });
+                }, submit_reason::file_operation);
             } catch (...) {
                 report_exception("Running ::close() in reactor thread, submission failed with exception", std::current_exception());
                 return make_ready_future<syscall_result<int>>(wrap_syscall<int>(::close(fd)));
@@ -358,7 +358,7 @@ blockdev_file_impl::size() noexcept {
         uint64_t size;
         int ret = ::ioctl(_fd, BLKGETSIZE64, &size);
         return wrap_syscall(ret, size);
-    }).then([] (syscall_result_extra<uint64_t> ret) {
+    }, submit_reason::file_operation).then([] (syscall_result_extra<uint64_t> ret) {
         ret.throw_if_error();
         return make_ready_future<uint64_t>(ret.extra);
     });
@@ -401,7 +401,7 @@ posix_file_impl::list_directory(std::function<future<> (directory_entry de)> nex
                 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) {
+                }, submit_reason::file_operation).then([w] (syscall_result<long> ret) {
                     ret.throw_if_error();
                     if (ret.result == 0) {
                         w->eof = true;
@@ -631,7 +631,7 @@ blockdev_file_impl::discard(uint64_t offset, uint64_t length) noexcept {
     return engine()._thread_pool->submit<syscall_result<int>>([this, offset, length] () mutable {
         uint64_t range[2] { offset, length };
         return wrap_syscall<int>(::ioctl(_fd, BLKDISCARD, &range));
-    }).then([] (syscall_result<int> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result<int> sr) {
         sr.throw_if_error();
         return make_ready_future<>();
     });

src/core/reactor.cc

@@ -1984,7 +1984,7 @@ reactor::open_file_dma(std::string_view nameref, open_flags flags, file_open_opt
             }
             close_fd.cancel();
             return wrap_syscall(fd, st);
-        }).then([&options, name = std::move(name), &open_flags] (syscall_result_extra<struct stat> sr) {
+        }, submit_reason::file_operation).then([&options, name = std::move(name), &open_flags] (syscall_result_extra<struct stat> sr) {
             sr.throw_fs_exception_if_error("open failed", name);
             return make_file_impl(sr.result, options, open_flags, sr.extra);
         }).then([] (shared_ptr<file_impl> impl) {
@@ -1999,7 +1999,7 @@ reactor::remove_file(std::string_view pathname) noexcept {
     return futurize_invoke([this, pathname] {
         return _thread_pool->submit<syscall_result<int>>([pathname = sstring(pathname)] {
             return wrap_syscall<int>(::remove(pathname.c_str()));
-        }).then([pathname = sstring(pathname)] (syscall_result<int> sr) {
+        }, submit_reason::file_operation).then([pathname = sstring(pathname)] (syscall_result<int> sr) {
             sr.throw_fs_exception_if_error("remove failed", pathname);
             return make_ready_future<>();
         });
@@ -2012,7 +2012,8 @@ reactor::rename_file(std::string_view old_pathname, std::string_view new_pathnam
     return futurize_invoke([this, old_pathname, new_pathname] {
         return _thread_pool->submit<syscall_result<int>>([old_pathname = sstring(old_pathname), new_pathname = sstring(new_pathname)] {
             return wrap_syscall<int>(::rename(old_pathname.c_str(), new_pathname.c_str()));
-        }).then([old_pathname = sstring(old_pathname), new_pathname = sstring(new_pathname)] (syscall_result<int> sr) {
+        }, submit_reason::file_operation
+        ).then([old_pathname = sstring(old_pathname), new_pathname = sstring(new_pathname)] (syscall_result<int> sr) {
             sr.throw_fs_exception_if_error("rename failed",  old_pathname, new_pathname);
             return make_ready_future<>();
         });
@@ -2025,7 +2026,7 @@ reactor::link_file(std::string_view oldpath, std::string_view newpath) noexcept
     return futurize_invoke([this, oldpath, newpath] {
         return _thread_pool->submit<syscall_result<int>>([oldpath = sstring(oldpath), newpath = sstring(newpath)] {
             return wrap_syscall<int>(::link(oldpath.c_str(), newpath.c_str()));
-        }).then([oldpath = sstring(oldpath), newpath = sstring(newpath)] (syscall_result<int> sr) {
+        }, submit_reason::file_operation).then([oldpath = sstring(oldpath), newpath = sstring(newpath)] (syscall_result<int> sr) {
             sr.throw_fs_exception_if_error("link failed", oldpath, newpath);
             return make_ready_future<>();
         });
@@ -2039,7 +2040,7 @@ reactor::chmod(std::string_view name, file_permissions permissions) noexcept {
     return futurize_invoke([name, mode, this] {
         return _thread_pool->submit<syscall_result<int>>([name = sstring(name), mode] {
             return wrap_syscall<int>(::chmod(name.c_str(), mode));
-        }).then([name = sstring(name), mode] (syscall_result<int> sr) {
+        }, submit_reason::file_operation).then([name = sstring(name), mode] (syscall_result<int> sr) {
             if (sr.result == -1) {
                 auto reason = format("chmod(0{:o}) failed", mode);
                 sr.throw_fs_exception(reason, fs::path(name));
@@ -2083,7 +2084,7 @@ reactor::file_type(std::string_view name, follow_symlink follow) noexcept {
             auto stat_syscall = follow ? stat : lstat;
             auto ret = stat_syscall(name.c_str(), &st);
             return wrap_syscall(ret, st);
-        }).then([name = sstring(name)] (syscall_result_extra<struct stat> sr) {
+        }, submit_reason::file_operation).then([name = sstring(name)] (syscall_result_extra<struct stat> sr) {
             if (long(sr.result) == -1) {
                 if (sr.error != ENOENT && sr.error != ENOTDIR) {
                     sr.throw_fs_exception_if_error("stat failed", name);
@@ -2218,7 +2219,7 @@ reactor::spawn(std::string_view pathname,
                     return wrap_syscall<int>(::posix_spawn(&child_pid, pathname.c_str(), &actions, &attr,
                                                            const_cast<char* const *>(argv.data()),
                                                            const_cast<char* const *>(env.data())));
-            });
+            }, submit_reason::process_operation);
         }).finally([&actions, &attr] {
             posix_spawn_file_actions_destroy(&actions);
             posix_spawnattr_destroy(&attr);
@@ -2269,7 +2270,7 @@ future<int> reactor::waitpid(pid_t pid) {
                                            &wait_timeout] {
                     return _thread_pool->submit<syscall_result<pid_t>>([pid, &wstatus] {
                         return wrap_syscall<pid_t>(::waitpid(pid, &wstatus, WNOHANG));
-                    }).then([&wstatus, &wait_timeout] (syscall_result<pid_t> ret) mutable {
+                    }, submit_reason::process_operation).then([&wstatus, &wait_timeout] (syscall_result<pid_t> ret) mutable {
                         if (ret.result == 0) {
                             wait_timeout = next_waitpid_timeout(wait_timeout);
                             return ::seastar::sleep(wait_timeout).then([] {
@@ -2289,7 +2290,7 @@ future<int> reactor::waitpid(pid_t pid) {
                 return pidfd.readable().then([pid, &wstatus, this] {
                     return _thread_pool->submit<syscall_result<pid_t>>([pid, &wstatus] {
                         return wrap_syscall<pid_t>(::waitpid(pid, &wstatus, WNOHANG));
-                    });
+                    }, submit_reason::process_operation);
                 }).then([&wstatus] (syscall_result<pid_t> ret) {
                     ret.throw_if_error();
                     assert(ret.result > 0);
@@ -2314,7 +2315,7 @@ reactor::file_stat(std::string_view pathname, follow_symlink follow) noexcept {
             auto stat_syscall = follow ? stat : lstat;
             auto ret = stat_syscall(pathname.c_str(), &st);
             return wrap_syscall(ret, st);
-        }).then([pathname = sstring(pathname)] (syscall_result_extra<struct stat> sr) {
+        }, submit_reason::file_operation).then([pathname = sstring(pathname)] (syscall_result_extra<struct stat> sr) {
             sr.throw_fs_exception_if_error("stat failed", pathname);
             struct stat& st = sr.extra;
             stat_data sd;
@@ -2352,7 +2353,7 @@ reactor::file_accessible(std::string_view pathname, access_flags flags) noexcept
             auto aflags = std::underlying_type_t<access_flags>(flags);
             auto ret = ::access(pathname.c_str(), aflags);
             return wrap_syscall(ret);
-        }).then([pathname = sstring(pathname), flags] (syscall_result<int> sr) {
+        }, submit_reason::file_operation).then([pathname = sstring(pathname), flags] (syscall_result<int> sr) {
             if (sr.result < 0) {
                 if ((sr.error == ENOENT && flags == access_flags::exists) ||
                     (sr.error == EACCES && flags != access_flags::exists)) {
@@ -2374,7 +2375,7 @@ reactor::file_system_at(std::string_view pathname) noexcept {
             struct statfs st;
             auto ret = statfs(pathname.c_str(), &st);
             return wrap_syscall(ret, st);
-        }).then([pathname = sstring(pathname)] (syscall_result_extra<struct statfs> sr) {
+        }, submit_reason::file_operation).then([pathname = sstring(pathname)] (syscall_result_extra<struct statfs> sr) {
             static std::unordered_map<long int, fs_type> type_mapper = {
                 { internal::fs_magic::xfs, fs_type::xfs },
                 { internal::fs_magic::ext2, fs_type::ext2 },
@@ -2401,7 +2402,7 @@ reactor::fstatfs(int fd) noexcept {
         struct statfs st;
         auto ret = ::fstatfs(fd, &st);
         return wrap_syscall(ret, st);
-    }).then([] (syscall_result_extra<struct statfs> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result_extra<struct statfs> sr) {
         sr.throw_if_error();
         struct statfs st = sr.extra;
         return make_ready_future<struct statfs>(std::move(st));
@@ -2416,7 +2417,7 @@ reactor::statvfs(std::string_view pathname) noexcept {
             struct statvfs st;
             auto ret = ::statvfs(pathname.c_str(), &st);
             return wrap_syscall(ret, st);
-        }).then([pathname = sstring(pathname)] (syscall_result_extra<struct statvfs> sr) {
+        }, submit_reason::file_operation).then([pathname = sstring(pathname)] (syscall_result_extra<struct statvfs> sr) {
             sr.throw_fs_exception_if_error("statvfs failed", pathname);
             struct statvfs st = sr.extra;
             return make_ready_future<struct statvfs>(std::move(st));
@@ -2440,7 +2441,7 @@ reactor::open_directory(std::string_view name) noexcept {
                 }
             }
             return wrap_syscall(fd, st);
-        }).then([name = sstring(name), oflags] (syscall_result_extra<struct stat> sr) {
+        }, submit_reason::file_operation).then([name = sstring(name), oflags] (syscall_result_extra<struct stat> sr) {
             sr.throw_fs_exception_if_error("open failed", name);
             return make_file_impl(sr.result, file_open_options(), oflags, sr.extra);
         }).then([] (shared_ptr<file_impl> file_impl) {
@@ -2456,7 +2457,7 @@ reactor::make_directory(std::string_view name, file_permissions permissions) noe
         return _thread_pool->submit<syscall_result<int>>([name = sstring(name), permissions] {
             auto mode = static_cast<mode_t>(permissions);
             return wrap_syscall<int>(::mkdir(name.c_str(), mode));
-        }).then([name = sstring(name)] (syscall_result<int> sr) {
+        }, submit_reason::file_operation).then([name = sstring(name)] (syscall_result<int> sr) {
             sr.throw_fs_exception_if_error("mkdir failed", name);
         });
     });
@@ -2469,7 +2470,7 @@ reactor::touch_directory(std::string_view name, file_permissions permissions) no
         return _thread_pool->submit<syscall_result<int>>([name = sstring(name), permissions] {
             auto mode = static_cast<mode_t>(permissions);
             return wrap_syscall<int>(::mkdir(name.c_str(), mode));
-        }).then([name = sstring(name)] (syscall_result<int> sr) {
+        }, submit_reason::file_operation).then([name = sstring(name)] (syscall_result<int> sr) {
             if (sr.result == -1 && sr.error != EEXIST) {
                 sr.throw_fs_exception("mkdir failed", fs::path(name));
             }
@@ -2514,7 +2515,7 @@ reactor::fdatasync(int fd) noexcept {
     }
     return _thread_pool->submit<syscall_result<int>>([fd] {
         return wrap_syscall<int>(::fdatasync(fd));
-    }).then([] (syscall_result<int> sr) {
+    }, submit_reason::file_operation).then([] (syscall_result<int> sr) {
         sr.throw_if_error();
         return make_ready_future<>();
     });
@@ -2658,6 +2659,12 @@ void reactor::register_metrics() {
 
     namespace sm = seastar::metrics;
 
+    auto io_fallback_counter = [this](const sstring& reason_str, submit_reason r) {
+        static auto reason_label = sm::label("reason");
+        return sm::make_counter("io_threaded_fallbacks", std::bind(&thread_pool::count, _thread_pool.get(), r),
+                sm::description("Total number of io-threaded-fallbacks operations"), { reason_label(reason_str), });
+    };
+
     _metric_groups.add_group("reactor", {
             sm::make_gauge("tasks_pending", std::bind(&reactor::pending_task_count, this), sm::description("Number of pending tasks in the queue")),
             // total_operations value:DERIVE:0:U
@@ -2683,9 +2690,13 @@ void reactor::register_metrics() {
             // total_operations value:DERIVE:0:U
             sm::make_counter("fsyncs", _fsyncs, sm::description("Total number of fsync operations")),
             // total_operations value:DERIVE:0:U
-            sm::make_counter("io_threaded_fallbacks", std::bind(&thread_pool::operation_count, _thread_pool.get()),
-                    sm::description("Total number of io-threaded-fallbacks operations")),
-
+            io_fallback_counter("aio_fallback", submit_reason::aio_fallback),
+            // total_operations value:DERIVE:0:U
+            io_fallback_counter("file_operation", submit_reason::file_operation),
+            // total_operations value:DERIVE:0:U
+            io_fallback_counter("process_operation", submit_reason::process_operation),
+            // total_operations value:DERIVE:0:U
+            io_fallback_counter("unknown", submit_reason::unknown),
     });
 
     _metric_groups.add_group("memory", {

src/core/reactor_backend.cc

@@ -260,7 +260,7 @@ void aio_storage_context::schedule_retry() {
         return _r._thread_pool->submit<syscall_result<int>>([this] () mutable {
             auto r = io_submit(_io_context, _aio_retries.size(), _aio_retries.data());
             return wrap_syscall<int>(r);
-        }).then_wrapped([this] (future<syscall_result<int>> f) {
+        }, submit_reason::aio_fallback).then_wrapped([this] (future<syscall_result<int>> f) {
             // If submit failed, just log the error and exit the loop.
             // The next call to submit_work will call schedule_retry again.
             if (f.failed()) {

src/core/thread_pool.hh

@@ -27,9 +27,36 @@ namespace seastar {
 
 class reactor;
 
+// Reasons for why a function had to be submitted to the thread_pool 
+enum class submit_reason : size_t {
+    // Used for aio operations what would block in `io_submit`.
+    aio_fallback,
+    // Used for file operations that don't have non-blocking alternatives.
+    file_operation,
+    // Used for process operations that don't have non-blocking alternatives.
+    process_operation,
+    // Used by default when a caller doesn't specify a submission reason.
+    unknown,
+};
+
+class submit_metrics {
+    uint64_t _counters[static_cast<size_t>(submit_reason::unknown) + 1]{};
+
+public:
+    void record_reason(submit_reason reason) {
+        reason = std::min(reason, submit_reason::unknown);
+        ++_counters[static_cast<size_t>(reason)];
+    }
+
+    uint64_t count_for(submit_reason reason) const {
+        reason = std::min(reason, submit_reason::unknown);
+        return _counters[static_cast<size_t>(reason)];
+    }
+};
+
 class thread_pool {
     reactor* _reactor;
-    uint64_t _aio_threaded_fallbacks = 0;
+    submit_metrics metrics;
 #ifndef HAVE_OSV
     syscall_work_queue inter_thread_wq;
     posix_thread _worker_thread;
@@ -39,11 +66,11 @@ public:
     explicit thread_pool(reactor* r, sstring thread_name);
     ~thread_pool();
     template <typename T, typename Func>
-    future<T> submit(Func func) noexcept {
-        ++_aio_threaded_fallbacks;
+    future<T> submit(Func func, submit_reason reason = submit_reason::unknown) noexcept {
+        metrics.record_reason(reason);
         return inter_thread_wq.submit<T>(std::move(func));
     }
-    uint64_t operation_count() const { return _aio_threaded_fallbacks; }
+    uint64_t count(submit_reason r) const { return metrics.count_for(r); }
 
     unsigned complete() { return inter_thread_wq.complete(); }
     // Before we enter interrupt mode, we must make sure that the syscall thread will properly
@@ -61,7 +88,7 @@ public:
 #else
 public:
     template <typename T, typename Func>
-    future<T> submit(Func func) { std::cerr << "thread_pool not yet implemented on osv\n"; abort(); }
+    future<T> submit(Func func, submit_reason reason = submit_reason::unknown) { std::cerr << "thread_pool not yet implemented on osv\n"; abort(); }
 #endif
 private:
     void work(sstring thread_name);