net: use adjustable_semaphore in server

Author: WillemKauf

src/v/net/BUILD

@@ -70,6 +70,7 @@ redpanda_cc_library(
         "//src/v/ssx:mutex",
         "//src/v/ssx:semaphore",
         "//src/v/ssx:sformat",
+        "//src/v/utils:adjustable_semaphore",
         "//src/v/utils:inet_address",
         "//src/v/utils:log_hist",
         "//src/v/utils:named_type",

src/v/net/server.cc

@@ -36,7 +36,7 @@ namespace net {
 server::server(server_configuration c, ss::logger& log)
   : cfg(std::move(c))
   , _log(log)
-  , _memory{size_t{static_cast<size_t>(cfg.max_service_memory_per_core)}, "net/server-mem"}
+  , _memory{static_cast<uint64_t>(cfg.max_service_memory_per_core), "net/server-mem"}
   , _probe(std::make_unique<server_probe>()) {
     vlog(
       _log.info, "Creating net::server for {} with config {}", cfg.name, cfg);
@@ -325,18 +325,22 @@ ss::future<> server::stop() {
     return wait_for_shutdown();
 }
 
+void server::set_memory_capacity(uint64_t new_capacity) {
+    _memory.set_capacity(new_capacity);
+}
+
 void server::setup_metrics() {
     namespace sm = ss::metrics;
     _metrics.add_group(
       prometheus_sanitize::metrics_name(cfg.name),
       {sm::make_total_bytes(
          "max_service_mem_bytes",
-         [this] { return cfg.max_service_memory_per_core; },
+         [this] { return _memory.capacity(); },
          sm::description(
            ssx::sformat("{}: Maximum memory allowed for RPC", cfg.name))),
        sm::make_total_bytes(
          "consumed_mem_bytes",
-         [this] { return cfg.max_service_memory_per_core - _memory.current(); },
+         [this] { return _memory.outstanding(); },
          sm::description(
            ssx::sformat(
              "{}: Memory consumed by request processing", cfg.name))),

src/v/net/server.h

@@ -19,6 +19,7 @@
 #include "net/connection_rate.h"
 #include "net/types.h"
 #include "ssx/semaphore.h"
+#include "utils/adjustable_semaphore.h"
 #include "utils/log_hist.h"
 
 #include <seastar/core/abort_source.hh>
@@ -141,11 +142,12 @@ class server {
     virtual ss::future<> apply(ss::lw_shared_ptr<net::connection>) = 0;
 
     server_probe& probe() { return *_probe; }
-    ssx::semaphore& memory() { return _memory; }
+    ssx::semaphore& memory() { return _memory.underlying(); }
     ss::gate& conn_gate() { return _conn_gate; }
     hist_t& hist() { return _hist; }
     ss::abort_source& abort_source() { return _as; }
     bool abort_requested() const { return _as.abort_requested(); }
+    void set_memory_capacity(uint64_t new_capacity);
 
 private:
     struct listener {
@@ -166,11 +168,9 @@ class server {
     print_exceptional_future(ss::future<>, const char*, ss::socket_address);
     ss::future<>
       apply_proto(ss::lw_shared_ptr<net::connection>, conn_quota::units);
-    void setup_metrics();
-    void setup_public_metrics();
 
     ss::logger& _log;
-    ssx::semaphore _memory;
+    adjustable_semaphore _memory;
     std::vector<std::unique_ptr<listener>> _listeners;
     boost::intrusive::list<net::connection> _connections;
     ss::abort_source _as;