memory_limiter: Encapsulate memory limiting facility

The storage service carries sempaphore and a size_t value
to facilitate the memory limiting for client services.

This patch encapsulates both fields on a separate helper
class that will be used by whoever needs it without
messing with the storage service.

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

service/memory_limiter.hh

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2021 ScyllaDB
+ *
+ */
+
+#pragma once
+
+#include <seastar/core/semaphore.hh>
+
+namespace service {
+
+class storage_service;
+
+class memory_limiter final {
+    friend class storage_service; // temporary
+    size_t _mem_total;
+    semaphore _sem;
+
+public:
+    memory_limiter(size_t available_memory) noexcept
+        : _mem_total(available_memory / 10)
+        , _sem(_mem_total) {}
+
+    future<> stop() {
+        return _sem.wait(_mem_total);
+    }
+
+    size_t total_memory() const noexcept { return _mem_total; }
+    semaphore& get_semaphore() noexcept { return _sem; }
+};
+
+} // namespace service

service/storage_service.cc

@@ -106,8 +106,7 @@ storage_service::storage_service(abort_source& abort_source, distributed<databas
         , _gossiper(gossiper)
         , _mnotifier(mn)
         , _messaging(ms)
-        , _service_memory_total(config.available_memory / 10)
-        , _service_memory_limiter(_service_memory_total)
+        , _service_memory_limiter(config.available_memory)
         , _for_testing(for_testing)
         , _node_ops_abort_thread(node_ops_abort_thread())
         , _shared_token_metadata(stm)
@@ -1485,7 +1484,7 @@ future<> storage_service::gossip_sharder() {
 future<> storage_service::stop() {
     // make sure nobody uses the semaphore
     node_ops_singal_abort(std::nullopt);
-    return _service_memory_limiter.wait(_service_memory_total).finally([this] {
+    return _service_memory_limiter.stop().finally([this] {
         _listeners.clear();
         return _schema_version_publisher.join();
     }).finally([this] {

service/storage_service.hh

@@ -66,6 +66,7 @@
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/core/lowres_clock.hh>
 #include "locator/snitch_base.hh"
+#include "service/memory_limiter.hh"
 
 class node_ops_cmd_request;
 class node_ops_cmd_response;
@@ -176,8 +177,7 @@ private:
     bool _ms_stopped = false;
     bool _stream_manager_stopped = false;
     seastar::metrics::metric_groups _metrics;
-    size_t _service_memory_total;
-    semaphore _service_memory_limiter;
+    memory_limiter _service_memory_limiter;
     using client_shutdown_hook = noncopyable_function<void()>;
     std::vector<std::pair<std::string, client_shutdown_hook>> _client_shutdown_hooks;
     std::vector<std::any> _listeners;
@@ -264,11 +264,11 @@ public:
     const gms::feature_service& features() const { return _feature_service; }
 
     size_t service_memory_total() const {
-        return _service_memory_total;
+        return _service_memory_limiter._mem_total;
     }
 
     semaphore& service_memory_limiter() {
-        return _service_memory_limiter;
+        return _service_memory_limiter._sem;
     }
 
     cdc::generation_service& get_cdc_generation_service() {