MSG_DSDB_PWD_LOG = 0x0802,
MSG_GROUP_LOG = 0x0803,
+ /*
+ * VFS aio_ratelimit cluster coordination messages
+ * Node-level aggregation via ratelimitd daemon
+ */
+ MSG_VFS_AIO_RATELIMIT_READ_NODE_SUMMARY = 0x0900,
+ MSG_VFS_AIO_RATELIMIT_WRITE_NODE_SUMMARY = 0x0901,
+
/* dbwrap messages 4001-4999 (0x0FA0 - 0x1387) */
/* MSG_DBWRAP_TDB2_CHANGES = 4001, */
/* MSG_DBWRAP_G_LOCK_RETRY = 4002, */
--- /dev/null
+/* Rate limiting protocol definitions for smbd ↔ ratelimitd communication */
+
+#ifndef __RATELIMIT_PROTOCOL_H__
+#define __RATELIMIT_PROTOCOL_H__
+
+#include "lib/util/time.h"
+#include "messages.h"
+
+#define RATELIMITD_SOCKET_NAME "ratelimitd.sock"
+
+/* Protocol version for compatibility checking */
+#define RATELIMIT_PROTOCOL_VERSION 2
+#define RATELIMIT_SHARE_NAME_LEN 256
+
+static inline uint64_t time_now_usec(void)
+{
+ struct timespec ts;
+
+ clock_gettime_mono(&ts);
+ return (uint64_t)ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
+}
+
+/* Operation types for rate limiting */
+enum ratelimit_operation {
+ RATELIMIT_OP_INVALID = 0,
+ RATELIMIT_OP_READ = 1,
+ RATELIMIT_OP_WRITE = 2,
+};
+
+/* Activity report from VFS process to local daemon via Unix socket */
+struct ratelimit_activity_report {
+ uint64_t timestamp_usec;
+ int64_t recent_iops;
+ uint32_t protocol_version;
+ uint32_t operation;
+ int32_t pid;
+ uint32_t inflight_ios;
+ char share_name[RATELIMIT_SHARE_NAME_LEN];
+} PACKED_STRUCT;
+
+/* Node summary broadcast from daemon to cluster */
+struct ratelimit_node_summary {
+ uint32_t vnn;
+ int32_t process_count;
+ uint64_t timestamp_usec;
+ char share_name[RATELIMIT_SHARE_NAME_LEN];
+} PACKED_STRUCT;
+
+static inline uint32_t ratelimit_msg_type_summary(uint32_t operation)
+{
+ switch (operation) {
+ case RATELIMIT_OP_READ:
+ return MSG_VFS_AIO_RATELIMIT_READ_NODE_SUMMARY;
+ case RATELIMIT_OP_WRITE:
+ return MSG_VFS_AIO_RATELIMIT_WRITE_NODE_SUMMARY;
+ default:
+ return 0;
+ }
+}
+
+static inline uint32_t ratelimit_op_from_string(const char *op_str)
+{
+ if (strcmp(op_str, "read") == 0) {
+ return RATELIMIT_OP_READ;
+ } else if (strcmp(op_str, "write") == 0) {
+ return RATELIMIT_OP_WRITE;
+ }
+ return RATELIMIT_OP_INVALID;
+}
+
+static inline const char *ratelimit_op_to_string(uint32_t operation)
+{
+ switch (operation) {
+ case RATELIMIT_OP_READ:
+ return "read";
+ case RATELIMIT_OP_WRITE:
+ return "write";
+ default:
+ return "invalid";
+ }
+}
+
+#endif /* __RATELIMIT_PROTOCOL_H__ */
/*
- * Asynchronous I/O rate-limiting VFS module.
+ * Asynchronous I/O rate-limiting VFS module with cluster-wide coordination.
*
* Copyright (c) 2025 Shachar Sharon <ssharon@redhat.com>
* Copyright (c) 2025 Avan Thakkar <athakkar@redhat.com>
*/
/*
- Token-base rate-limiter using Samba's VFS stack-able module. For each samba
+ Token-based rate-limiter using Samba's VFS stack-able module. For each samba
share a user may define READ/WRITE thresholds in terms of IOPS or BYTES
per-second. If one of those thresholds is exceeded along the asynchronous
I/O path, a delay is injected before sending back a reply to the caller,
persisted to a local TDB, allowing limits to be enforced consistently
across client reconnects and smbd restarts.
+ CLUSTER-WIDE COORDINATION VIA DAEMON:
+ In cluster mode, each smbd process reports its activity to a local
+ ratelimitd daemon via Unix socket. The daemon aggregates activity from
+ all smbd processes on the node and broadcasts node-level summaries to
+ other nodes via Samba's messaging system. This reduces network message
+ volume from O(N²) to O(M²) where N=processes and M=nodes.
+
+ Processes receive node summaries and dynamically recalculate their local
+ rate limits to ensure the global limit is enforced cluster-wide. The
+ global limit is distributed equally among all active smbd processes
+ performing I/O (per-process distribution model).
+
An example to smb.conf segment (zero value implies ignore-this-option):
[share]
#include "lib/util/time.h"
#include "lib/util/tevent_unix.h"
#include "lib/util/util_tdb.h"
+#include "lib/util/server_id.h"
#include "tdb.h"
+#include "messages.h"
#include "system/filesys.h"
+#include "lib/global_contexts.h"
+#include "ratelimit_protocol.h"
#undef DBGC_CLASS
#define DBGC_CLASS DBGC_VFS
/* TDB schema version */
#define RATELIMIT_TDB_VERSION 1
+/* Activity tracking intervals */
+#define ACTIVITY_BROADCAST_INTERVAL_US (1000000L)
+#define ACTIVITY_TIMEOUT_US (5000000L)
+
+/* Initial capacity for node tracking array (grows dynamically) */
+#define INITIAL_TRACKED_CAPACITY 16
+
static unsigned int ref_count = 0;
static TDB_CONTEXT *ratelimit_tdb;
uint8_t reserved[64 - (8 + 4 + 4)];
} PACKED_STRUCT;
+/* Node-level tracking */
+struct node_count {
+ uint32_t vnn;
+ int32_t process_count;
+ uint64_t last_seen_us;
+ bool is_active;
+};
+
/* Token-based rate-limiter control state using a token-bucket. */
struct ratelimiter {
+ struct ratelimiter *prev, *next;
+
const char *op;
uint64_t last_usec;
uint64_t last_save_usec;
float bytes_tokens;
int64_t iops_total;
int64_t bytes_total;
- int64_t iops_limit;
- int64_t bw_limit;
+ int64_t global_iops_limit;
+ int64_t global_bw_limit;
+ int64_t local_iops_limit;
+ int64_t local_bw_limit;
float iops_capacity;
float bytes_capacity;
* are reconfigured in the future (e.g. reload, per-client limits).
*/
float burst_mult;
+
int snum;
+ char share_name[RATELIMIT_SHARE_NAME_LEN];
+
+ /* Cluster coordination via daemon */
+ bool cluster_mode;
+ struct messaging_context *msg_ctx;
+ struct server_id my_server_id;
+ int daemon_sock;
+ uint64_t last_report_to_daemon_us;
+
+ /* Activity tracking */
+ uint32_t inflight_ios;
+ int64_t recent_iops;
+
+ struct node_count *node_counts;
+ int num_tracked_nodes;
+ int max_tracked_nodes;
+
+ int num_active_processes;
+
+ /* Statistics */
+ uint64_t total_reports_sent;
+ uint64_t total_summaries_received;
+ uint64_t total_limit_recalcs;
};
/* In-memory rate-limiting entry per connection */
struct ratelimiter wr_ratelimiter;
};
-static uint64_t time_now_usec(void)
+/*
+ * Process-global dispatch lists, one per op.
+ * Samba's messaging layer delivers a given msg_type to only one registered
+ * callback per process, so we register once per op per process and
+ * fan out to every ratelimiter on the matching list ourselves.
+ */
+static struct ratelimiter *read_ratelimiters_list = NULL;
+static struct ratelimiter *write_ratelimiters_list = NULL;
+
+static struct ratelimiter **ratelimiter_dispatch_list(const char *op)
+{
+ if (strcmp(op, "read") == 0) {
+ return &read_ratelimiters_list;
+ }
+ return &write_ratelimiters_list;
+}
+
+static bool ensure_node_count_capacity(struct ratelimiter *rl)
{
- struct timespec ts;
+ struct node_count *new_array;
+ int new_max;
+
+ if (rl->num_tracked_nodes < rl->max_tracked_nodes) {
+ return true;
+ }
+
+ new_max = rl->max_tracked_nodes * 2;
- clock_gettime_mono(&ts);
- return (uint64_t)ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
+ new_array = talloc_realloc_zero(talloc_parent(rl->node_counts),
+ rl->node_counts,
+ struct node_count,
+ new_max);
+ if (new_array == NULL) {
+ DBG_ERR("[%s snum:%d %s] Failed to grow node_counts: %d -> "
+ "%d\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->max_tracked_nodes,
+ new_max);
+ return false;
+ }
+
+ rl->node_counts = new_array;
+ rl->max_tracked_nodes = new_max;
+
+ return true;
+}
+
+static void update_node_count(struct ratelimiter *rl,
+ uint32_t vnn,
+ int32_t process_count)
+{
+ struct node_count *nc = NULL;
+ uint64_t now = time_now_usec();
+ int i;
+
+ /* Find existing node entry */
+ for (i = 0; i < rl->num_tracked_nodes; i++) {
+ if (rl->node_counts[i].vnn == vnn) {
+ nc = &rl->node_counts[i];
+ break;
+ }
+ }
+
+ if (nc == NULL) {
+ if (!ensure_node_count_capacity(rl)) {
+ return;
+ }
+
+ nc = &rl->node_counts[rl->num_tracked_nodes++];
+ nc->vnn = vnn;
+ nc->is_active = false;
+ }
+
+ nc->process_count = process_count;
+ nc->last_seen_us = now;
+ nc->is_active = (process_count > 0);
+
+ DBG_DEBUG("[%s snum:%d %s] Updated node vnn=%u: count=%d active=%d\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ vnn,
+ process_count,
+ nc->is_active);
+}
+
+static int count_active_processes(struct ratelimiter *rl)
+{
+ uint64_t now = time_now_usec();
+ int total_count = 0;
+ int timed_out = 0;
+ bool i_am_active;
+ int i;
+
+ i_am_active = (rl->inflight_ios > 0) || (rl->recent_iops > 0);
+
+ if (i_am_active) {
+ total_count = 1;
+ DBG_DEBUG("[%s snum:%d %s] I am ACTIVE "
+ "(inflight=%u recent=%" PRId64 ")\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->inflight_ios,
+ rl->recent_iops);
+ } else {
+ DBG_DEBUG("[%s snum:%d %s] I am IDLE "
+ "(inflight=%u recent=%" PRId64 ")\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->inflight_ios,
+ rl->recent_iops);
+ }
+
+ for (i = 0; i < rl->num_tracked_nodes; i++) {
+ struct node_count *nc = &rl->node_counts[i];
+ uint64_t age_us = now - nc->last_seen_us;
+
+ /* Check timeout */
+ if (age_us > ACTIVITY_TIMEOUT_US) {
+ if (nc->is_active) {
+ DBG_NOTICE("[%s snum:%d %s] Node vnn=%u TIMED "
+ "OUT "
+ "(age=%" PRIu64 " ms)\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ nc->vnn,
+ age_us / 1000);
+ nc->is_active = false;
+ timed_out++;
+ }
+ continue;
+ }
+
+ if (!nc->is_active || nc->process_count <= 0) {
+ continue;
+ }
+
+ if (nc->vnn == rl->my_server_id.vnn) {
+ total_count += (nc->process_count - 1);
+ } else {
+ total_count += nc->process_count;
+ }
+ }
+
+ DBG_DEBUG("[%s snum:%d %s] Total=%d active processes "
+ "(timed_out=%d)\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ MAX(total_count, 1),
+ timed_out);
+
+ return MAX(total_count, 1);
+}
+
+/* Recalculate local limits based on active processes */
+static void recalculate_local_limits(struct ratelimiter *rl)
+{
+ int new_active = count_active_processes(rl);
+ int64_t old_local_iops;
+ int64_t old_local_bw;
+ float old_iops_cap;
+ float old_bytes_cap;
+ float old_iops_tokens;
+ float old_bytes_tokens;
+
+ if (new_active == rl->num_active_processes) {
+ return;
+ }
+
+ old_local_iops = rl->local_iops_limit;
+ old_local_bw = rl->local_bw_limit;
+ old_iops_cap = rl->iops_capacity;
+ old_bytes_cap = rl->bytes_capacity;
+
+ DBG_DEBUG("[%s snum:%d %s] *** RECALCULATING LIMITS *** "
+ "Active processes changed: %d -> %d\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->num_active_processes,
+ new_active);
+
+ rl->num_active_processes = new_active;
+ rl->total_limit_recalcs++;
+
+ rl->local_iops_limit = (rl->global_iops_limit > 0)
+ ? MAX(rl->global_iops_limit /
+ new_active,
+ 1)
+ : 0;
+ rl->local_bw_limit = (rl->global_bw_limit > 0)
+ ? MAX(rl->global_bw_limit / new_active, 1)
+ : 0;
+
+ rl->iops_capacity = (float)rl->local_iops_limit * rl->burst_mult;
+ rl->bytes_capacity = (float)rl->local_bw_limit * rl->burst_mult;
+
+ old_iops_tokens = rl->iops_tokens;
+ old_bytes_tokens = rl->bytes_tokens;
+
+ rl->iops_tokens = MIN(rl->iops_tokens, rl->iops_capacity);
+ rl->bytes_tokens = MIN(rl->bytes_tokens, rl->bytes_capacity);
+
+ DBG_DEBUG("[%s snum:%d %s] *** LIMITS UPDATED ***\n"
+ " IOPS: local_limit %" PRId64 " -> %" PRId64
+ " (global=%" PRId64 ")\n"
+ " capacity %.2f -> %.2f, tokens %.2f -> %.2f\n"
+ " BW: local_limit %" PRId64 " -> %" PRId64
+ " (global=%" PRId64 ")\n"
+ " capacity %.2f -> %.2f, tokens %.2f -> %.2f\n"
+ " Active processes: %d, Total recalcs: %" PRIu64 "\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ old_local_iops,
+ rl->local_iops_limit,
+ rl->global_iops_limit,
+ old_iops_cap,
+ rl->iops_capacity,
+ old_iops_tokens,
+ rl->iops_tokens,
+ old_local_bw,
+ rl->local_bw_limit,
+ rl->global_bw_limit,
+ old_bytes_cap,
+ rl->bytes_capacity,
+ old_bytes_tokens,
+ rl->bytes_tokens,
+ rl->num_active_processes,
+ rl->total_limit_recalcs);
+}
+
+static int connect_to_ratelimitd(void)
+{
+ int sock;
+ struct sockaddr_un addr;
+ char *socket_path = NULL;
+ int ret;
+
+ socket_path = state_path(talloc_tos(), RATELIMITD_SOCKET_NAME);
+ if (socket_path == NULL) {
+ DBG_ERR("[%s] Failed to allocate socket path\n", MODULE_NAME);
+ return -1;
+ }
+
+ sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+ if (sock < 0) {
+ DBG_ERR("[%s] socket() failed: %s\n",
+ MODULE_NAME,
+ strerror(errno));
+ TALLOC_FREE(socket_path);
+ return -1;
+ }
+
+ memset(&addr, 0, sizeof(addr));
+ addr.sun_family = AF_UNIX;
+ strlcpy(addr.sun_path, socket_path, sizeof(addr.sun_path));
+
+ ret = connect(sock, (struct sockaddr *)&addr, sizeof(addr));
+ if (ret < 0) {
+ DBG_WARNING(
+ "[%s] connect() to ratelimitd failed: %s (path=%s)\n",
+ MODULE_NAME,
+ strerror(errno),
+ socket_path);
+ close(sock);
+ TALLOC_FREE(socket_path);
+ return -1;
+ }
+
+ DBG_DEBUG("[%s] Connected to ratelimitd: socket=%s\n",
+ MODULE_NAME,
+ socket_path);
+
+ TALLOC_FREE(socket_path);
+ return sock;
+}
+
+static void report_to_daemon(struct ratelimiter *rl)
+{
+ struct ratelimit_activity_report report = {0};
+ ssize_t ret;
+
+ if (rl->daemon_sock < 0) {
+ DBG_DEBUG("[%s snum:%d %s] report_to_daemon: daemon_sock "
+ "invalid, aborting\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op);
+ return;
+ }
+
+ /* Build report */
+ report.protocol_version = RATELIMIT_PROTOCOL_VERSION;
+ report.pid = (int32_t)getpid();
+ strlcpy(report.share_name, rl->share_name, sizeof(report.share_name));
+ report.operation = ratelimit_op_from_string(rl->op);
+ report.recent_iops = rl->recent_iops;
+ report.inflight_ios = rl->inflight_ios;
+ report.timestamp_usec = time_now_usec();
+
+ /* Send to daemon */
+ ret = send(rl->daemon_sock, &report, sizeof(report), MSG_DONTWAIT);
+
+ if (ret != sizeof(report)) {
+ DBG_DEBUG("[%s snum:%d %s] report_to_daemon: send failed "
+ "ret=%zd expected=%zu errno=%d\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ ret,
+ sizeof(report),
+ errno);
+
+ if (errno != EAGAIN && errno != EWOULDBLOCK) {
+ DBG_WARNING("[%s snum:%d %s] Failed to report to "
+ "daemon: %s\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ strerror(errno));
+
+ DBG_DEBUG("[%s snum:%d %s] report_to_daemon: "
+ "attempting reconnect\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op);
+
+ close(rl->daemon_sock);
+ rl->daemon_sock = connect_to_ratelimitd();
+
+ if (rl->daemon_sock < 0) {
+ DBG_ERR("[%s snum:%d %s] Reconnect failed, "
+ "cluster coordination lost\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op);
+ }
+ } else {
+ DBG_DEBUG("[%s snum:%d %s] report_to_daemon: "
+ "EAGAIN/EWOULDBLOCK - "
+ "buffer full, dropping report\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op);
+ }
+ return;
+ }
+
+ DBG_DEBUG("[%s snum:%d %s] Reported to daemon: "
+ "recent_iops=%" PRId64 " inflight=%u total_sent=%" PRIu64
+ "\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->recent_iops,
+ rl->inflight_ios,
+ rl->total_reports_sent + 1);
+
+ rl->total_reports_sent++;
+
+ /* Reset recent counter after report */
+ rl->recent_iops = 0;
+ rl->last_report_to_daemon_us = time_now_usec();
+}
+
+/*
+ * Apply one received node summary to a single ratelimiter that is known
+ * to match (share_name already checked by the caller).
+ */
+static void handle_node_summary_apply(
+ struct ratelimiter *rl,
+ const struct ratelimit_node_summary *summary)
+{
+ int old_active;
+
+ DBG_DEBUG("[%s snum:%d %s] Received node summary from vnn=%u: "
+ "process_count=%d (summaries_received=%" PRIu64 ")\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ summary->vnn,
+ summary->process_count,
+ rl->total_summaries_received + 1);
+
+ rl->total_summaries_received++;
+
+ old_active = rl->num_active_processes;
+
+ update_node_count(rl, summary->vnn, summary->process_count);
+
+ recalculate_local_limits(rl);
+
+ DBG_DEBUG("[%s snum:%d %s] handle_node_summary_apply: "
+ "after update num_active=%d (was %d)\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->num_active_processes,
+ old_active);
+}
+
+/*
+ * Registered once per op per process.
+ */
+static void handle_node_summary_dispatch(struct messaging_context *msg_ctx,
+ void *private_data,
+ uint32_t msg_type,
+ struct server_id server_id,
+ DATA_BLOB *data)
+{
+ struct ratelimiter *list;
+ struct ratelimiter *rl;
+ struct ratelimit_node_summary summary;
+
+ if (data->length != sizeof(struct ratelimit_node_summary)) {
+ DBG_ERR("[%s] Invalid node summary size %zu (expected %zu)\n",
+ MODULE_NAME,
+ data->length,
+ sizeof(struct ratelimit_node_summary));
+ return;
+ }
+
+ memcpy(&summary, data->data, sizeof(summary));
+ summary.share_name[sizeof(summary.share_name) - 1] = '\0';
+
+ list = (msg_type == MSG_VFS_AIO_RATELIMIT_READ_NODE_SUMMARY)
+ ? read_ratelimiters_list
+ : write_ratelimiters_list;
+
+ for (rl = list; rl != NULL; rl = rl->next) {
+ if (strcmp(summary.share_name, rl->share_name) != 0) {
+ DBG_DEBUG("[%s snum:%d %s] Ignoring summary for "
+ "share=%s from vnn=%u\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ summary.share_name,
+ summary.vnn);
+ continue;
+ }
+ handle_node_summary_apply(rl, &summary);
+ }
}
static bool ratelimit_tdb_check_version(void)
TALLOC_FREE(key.dptr);
}
-static void ratelimiter_init(struct ratelimiter *rl,
+static void ratelimiter_cleanup(struct ratelimiter *rl)
+{
+ if (rl == NULL) {
+ return;
+ }
+
+ if (rl->cluster_mode) {
+ struct ratelimiter **list = ratelimiter_dispatch_list(rl->op);
+
+ DLIST_REMOVE(*list, rl);
+
+ if (*list == NULL && rl->msg_ctx != NULL) {
+ /*
+ * Last share for this op in this process -
+ * deregister the shared dispatcher.
+ */
+ messaging_deregister(rl->msg_ctx,
+ ratelimit_msg_type_summary(
+ ratelimit_op_from_string(
+ rl->op)),
+ NULL);
+ }
+ }
+
+ /* Close daemon connection */
+ if (rl->daemon_sock >= 0) {
+ close(rl->daemon_sock);
+ rl->daemon_sock = -1;
+ }
+
+ /* Free node tracking array */
+ TALLOC_FREE(rl->node_counts);
+ rl->num_tracked_nodes = 0;
+ rl->max_tracked_nodes = 0;
+}
+
+static void ratelimiter_init_local_only(struct ratelimiter *rl)
+{
+ rl->num_active_processes = 1;
+ rl->local_iops_limit = rl->global_iops_limit;
+ rl->local_bw_limit = rl->global_bw_limit;
+ rl->daemon_sock = -1;
+ rl->node_counts = NULL;
+ rl->num_tracked_nodes = 0;
+ rl->max_tracked_nodes = 0;
+
+ DBG_NOTICE("[%s snum:%d %s] Cluster mode DISABLED - using "
+ "per-node limits "
+ "samba_clustering=%s\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->cluster_mode ? "enabled" : "disabled");
+}
+
+static void ratelimiter_init(TALLOC_CTX *mem_ctx,
+ struct ratelimiter *rl,
int snum,
const char *op,
int64_t iops_limit,
int64_t bw_limit,
float burst_mult)
{
+ const struct loadparm_substitution
+ *lp_sub = loadparm_s3_global_substitution();
+ const char *servicename = NULL;
+
ZERO_STRUCTP(rl);
rl->op = op;
rl->snum = snum;
- rl->iops_limit = iops_limit;
- rl->bw_limit = bw_limit;
+ servicename = lp_servicename(talloc_tos(), lp_sub, snum);
+ if (servicename != NULL) {
+ strlcpy(rl->share_name, servicename, sizeof(rl->share_name));
+ }
+
+ /* Store both global and local limits */
+ rl->global_iops_limit = iops_limit;
+ rl->global_bw_limit = bw_limit;
rl->burst_mult = burst_mult;
rl->iops_total = 0;
rl->bytes_total = 0;
- rl->iops_capacity = (float)(iops_limit)*burst_mult;
- rl->bytes_capacity = (float)(bw_limit)*burst_mult;
+ rl->cluster_mode = lp_clustering();
+
+ if (rl->cluster_mode) {
+ rl->msg_ctx = global_messaging_context();
+ rl->my_server_id = messaging_server_id(rl->msg_ctx);
+
+ /* Start with single-process assumption */
+ rl->num_active_processes = 1;
+ rl->local_iops_limit = iops_limit;
+ rl->local_bw_limit = bw_limit;
+
+ /* Connect to daemon */
+ rl->daemon_sock = connect_to_ratelimitd();
+ if (rl->daemon_sock < 0) {
+ DBG_WARNING("[%s snum:%d %s] Failed to connect to "
+ "ratelimitd, "
+ "disabling cluster mode\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op);
+ rl->msg_ctx = NULL;
+ rl->cluster_mode = false;
+ ratelimiter_init_local_only(rl);
+ } else {
+ rl->max_tracked_nodes = INITIAL_TRACKED_CAPACITY;
+ rl->num_tracked_nodes = 0;
+ rl->node_counts = talloc_zero_array(
+ mem_ctx,
+ struct node_count,
+ rl->max_tracked_nodes);
+ if (rl->node_counts == NULL) {
+ DBG_ERR("[%s snum:%d %s] Failed to allocate "
+ "node "
+ "tracking\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op);
+ close(rl->daemon_sock);
+ rl->msg_ctx = NULL;
+ rl->cluster_mode = false;
+ ratelimiter_init_local_only(rl);
+ } else {
+ struct ratelimiter **list =
+ ratelimiter_dispatch_list(rl->op);
+ bool need_register = (*list == NULL);
+
+ if (need_register) {
+ /*
+ * First share for this op in this
+ * process, so register the shared
+ * dispatcher once. Samba's messaging
+ * layer only delivers to one callback
+ * per msg_type per process.
+ */
+ messaging_register(
+ rl->msg_ctx,
+ NULL,
+ ratelimit_msg_type_summary(
+ ratelimit_op_from_string(
+ rl->op)),
+ handle_node_summary_dispatch);
+ }
+
+ DLIST_ADD(*list, rl);
+
+ DBG_NOTICE(
+ "[%s snum:%d %s] Cluster mode enabled "
+ "via daemon (capacity=%d, "
+ "registered_dispatcher=%s)\n",
+ MODULE_NAME,
+ rl->snum,
+ rl->op,
+ rl->max_tracked_nodes,
+ need_register ? "yes" : "no");
+ }
+ }
+ } else {
+ ratelimiter_init_local_only(rl);
+ }
+
+ rl->iops_capacity = (float)rl->local_iops_limit * burst_mult;
+ rl->bytes_capacity = (float)rl->local_bw_limit * burst_mult;
rl->last_usec = 0;
rl->last_save_usec = rl->last_usec;
/* Load from global TDB if available */
ratelimit_load_tdb(rl);
- DBG_DEBUG("[%s snum:%d %s] init ratelimiter:"
- " iops_limit=%" PRId64 " bw_limit=%" PRId64
- " burst_mult=%.2f\n",
+ DBG_DEBUG("[%s snum:%d %s] Initialized ratelimiter: "
+ "global_iops_limit=%" PRId64 " global_bw_limit=%" PRId64
+ " burst_mult=%.2f cluster_mode=%s\n",
MODULE_NAME,
rl->snum,
rl->op,
- rl->iops_limit,
- rl->bw_limit,
- rl->burst_mult);
+ rl->global_iops_limit,
+ rl->global_bw_limit,
+ rl->burst_mult,
+ rl->cluster_mode ? "yes" : "no");
}
static bool ratelimiter_enabled(const struct ratelimiter *rl)
{
- return (rl->iops_limit > 0) || (rl->bw_limit > 0);
+ return (rl->local_iops_limit > 0) || (rl->local_bw_limit > 0);
}
static float ratelimiter_calc_refill(uint64_t elapsed,
elapsed = now - rl->last_usec;
- if (rl->iops_limit > 0) {
+ if (rl->local_iops_limit > 0) {
float refill;
refill = ratelimiter_calc_refill(elapsed,
rl->iops_capacity,
- rl->iops_limit);
+ rl->local_iops_limit);
rl->iops_tokens = MIN(rl->iops_tokens + refill,
rl->iops_capacity);
}
- if (rl->bw_limit > 0) {
+ if (rl->local_bw_limit > 0) {
float refill;
refill = ratelimiter_calc_refill(elapsed,
rl->bytes_capacity,
- rl->bw_limit);
+ rl->local_bw_limit);
rl->bytes_tokens = MIN(rl->bytes_tokens + refill,
rl->bytes_capacity);
/* Refill tokens based on elapsed time */
ratelimiter_refill(rl);
+ /* Track in-flight I/O for cluster coordination */
+ if (rl->cluster_mode) {
+ rl->inflight_ios++;
+ }
+
/* Consume tokens for this operation */
- if (rl->iops_limit > 0) {
+ if (rl->local_iops_limit > 0) {
rl->iops_tokens -= 1.0f;
if (rl->iops_tokens < 0.0f) {
iops_deficit = -rl->iops_tokens;
}
}
- if (rl->bw_limit > 0) {
+ if (rl->local_bw_limit > 0) {
rl->bytes_tokens -= (float)nbytes;
if (rl->bytes_tokens < 0.0f) {
bw_deficit = -rl->bytes_tokens;
}
}
- delay_usec = ratelimiter_deficit_to_delay(iops_deficit, rl->iops_limit);
- bw_delay = ratelimiter_deficit_to_delay(bw_deficit, rl->bw_limit);
+ delay_usec = ratelimiter_deficit_to_delay(iops_deficit,
+ rl->local_iops_limit);
+ bw_delay = ratelimiter_deficit_to_delay(bw_deficit,
+ rl->local_bw_limit);
if (bw_delay > delay_usec) {
delay_usec = bw_delay;
rl->iops_total += 1;
rl->bytes_total += nbytes;
+
+ /* Track recent I/O for cluster coordination */
+ if (rl->cluster_mode) {
+ rl->recent_iops++;
+ }
+
now = time_now_usec();
if ((now - rl->last_save_usec) > SAVE_INTERVAL_USEC) {
rl->last_save_usec = now;
}
+ /* Report to daemon for cluster coordination */
+ if (rl->cluster_mode && now - rl->last_report_to_daemon_us >
+ ACTIVITY_BROADCAST_INTERVAL_US)
+ {
+ report_to_daemon(rl);
+ }
+
DBG_DEBUG("[%s snum:%d %s] delay_usec=%" PRIu32
- " iops_tokens=%.2f bytes_tokens=%.2f\n",
+ " iops_tokens=%.2f bytes_tokens=%.2f "
+ "(local limits: iops=%" PRId64 " bw=%" PRId64 ")\n",
MODULE_NAME,
rl->snum,
rl->op,
delay_usec,
rl->iops_tokens,
- rl->bytes_tokens);
+ rl->bytes_tokens,
+ rl->local_iops_limit,
+ rl->local_bw_limit);
return delay_usec;
}
int64_t nbytes_want,
int64_t nbytes_done)
{
- if (rl->bw_limit > 0 && nbytes_done < nbytes_want) {
+ /* Update in-flight counter for cluster coordination */
+ if (rl->cluster_mode && rl->inflight_ios > 0) {
+ rl->inflight_ios--;
+ }
+
+ if (rl->local_bw_limit > 0 && nbytes_done < nbytes_want) {
int64_t unused = nbytes_want - MAX(nbytes_done, (int64_t)0);
rl->bytes_tokens = MIN(rl->bytes_tokens + (float)unused,
}
if (!conv_str_size_error(str, &val)) {
- DBG_ERR("[%s] invalid value for %s: '%s'\n",
+ DBG_ERR("[%s] Invalid value for %s: '%s'\n",
MODULE_NAME,
option,
str);
BURST_MULT_DEF,
100) / 10.0f;
- ratelimiter_init(&config->rd_ratelimiter,
+ ratelimiter_init(config,
+ &config->rd_ratelimiter,
snum,
"read",
iops_limit,
BURST_MULT_DEF,
100) / 10.0f;
- ratelimiter_init(&config->wr_ratelimiter,
+ ratelimiter_init(config,
+ &config->wr_ratelimiter,
snum,
"write",
iops_limit,
ret = vfs_aio_ratelimit_new_config(handle);
if (ret < 0) {
- DBG_ERR("[%s] failed to create new config: "
- "service=%s snum=%d\n",
+ DBG_ERR("[%s] Failed to create config: service=%s snum=%d\n",
MODULE_NAME,
service,
SNUM(handle->conn));
ratelimit_save_tdb(&config->rd_ratelimiter);
ratelimit_save_tdb(&config->wr_ratelimiter);
+ ratelimiter_cleanup(&config->rd_ratelimiter);
+ ratelimiter_cleanup(&config->wr_ratelimiter);
+
ref_count--;
if (ref_count == 0 && ratelimit_tdb != NULL) {
#include "locking/leases_db.h"
#include "smbd/notifyd/notifyd.h"
#include "smbd/smbd_cleanupd.h"
+#ifdef WITH_RATELIMITD
+#include "smbd/smbd_ratelimitd.h"
+#endif
#include "lib/util/sys_rw.h"
#include "cleanupdb.h"
#include "g_lock.h"
struct server_id cleanupd;
struct server_id notifyd;
+#ifdef WITH_RATELIMITD
+ struct server_id ratelimitd;
+#endif
struct tevent_timer *cleanup_te;
}
}
+/**************************************************************************
+ * ratelimitd - cluster-wide rate limit coordination daemon
+ **************************************************************************/
+#ifdef WITH_RATELIMITD
+static void ratelimitd_stopped(struct tevent_req *req)
+{
+ int ret = smbd_ratelimitd_recv(req);
+ if (ret != 0) {
+ DBG_ERR("ratelimitd stopped with error: %s\n", strerror(ret));
+ } else {
+ DBG_NOTICE("ratelimitd stopped cleanly\n");
+ }
+}
+
+static bool smbd_ratelimitd_init(struct messaging_context *msg,
+ bool interactive,
+ struct server_id *ppid)
+{
+ struct tevent_context *ev = messaging_tevent_context(msg);
+ struct tevent_req *req;
+ pid_t pid;
+ NTSTATUS status;
+ bool ok;
+
+ /*
+ * Start daemon unconditionally when clustering is enabled.
+ * VFS modules load dynamically, so we can't detect aio_ratelimit
+ * usage at startup. When idle, daemon only waits in event loop
+ * and broadcasts nothing.
+ */
+ if (!lp_clustering()) {
+ DBG_DEBUG("Clustering disabled, not starting ratelimitd\n");
+ return true;
+ }
+
+ if (interactive) {
+ req = smbd_ratelimitd_send(msg, ev, msg);
+ *ppid = messaging_server_id(msg);
+ return (req != NULL);
+ }
+
+ pid = fork();
+ if (pid == -1) {
+ DBG_ERR("ratelimitd fork failed: %s\n", strerror(errno));
+ return false;
+ }
+
+ if (pid != 0) {
+ DBG_DEBUG("Started ratelimitd pid=%d\n", (int)pid);
+
+ if (am_parent != NULL) {
+ add_child_pid(am_parent, pid);
+ }
+
+ *ppid = pid_to_procid(pid);
+ return true;
+ }
+
+ status = smbd_reinit_after_fork(msg, ev, true);
+ if (!NT_STATUS_IS_OK(status)) {
+ DBG_ERR("reinit_after_fork failed: %s\n", nt_errstr(status));
+ exit(1);
+ }
+
+ process_set_title("smbd-ratelimitd", "ratelimitd");
+ set_remote_machine_name("ratelimitd", false);
+
+ reopen_logs();
+
+ req = smbd_ratelimitd_send(msg, ev, msg);
+ if (req == NULL) {
+ DBG_ERR("smbd_ratelimitd_send failed\n");
+ exit(1);
+ }
+
+ tevent_req_set_callback(req, ratelimitd_stopped, msg);
+
+ ok = tevent_req_poll(req, ev);
+ if (!ok) {
+ DBG_ERR("tevent_req_poll failed: %s\n", strerror(errno));
+ }
+ exit(0);
+}
+
+static void ratelimitd_init_trigger(struct tevent_req *subreq);
+
+struct ratelimitd_init_state {
+ struct tevent_context *ev;
+ struct messaging_context *msg;
+ struct server_id *ppid;
+ bool ok;
+};
+
+static struct tevent_req *ratelimitd_init_send(struct tevent_context *ev,
+ TALLOC_CTX *mem_ctx,
+ struct messaging_context *msg,
+ struct server_id *ppid)
+{
+ struct tevent_req *req = NULL;
+ struct tevent_req *subreq = NULL;
+ struct ratelimitd_init_state *state = NULL;
+
+ req = tevent_req_create(mem_ctx, &state, struct ratelimitd_init_state);
+ if (req == NULL) {
+ return NULL;
+ }
+
+ *state = (struct ratelimitd_init_state){
+ .msg = msg,
+ .ev = ev,
+ .ppid = ppid,
+ };
+
+ subreq = tevent_wakeup_send(state,
+ ev,
+ tevent_timeval_current_ofs(1, 0));
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+
+ tevent_req_set_callback(subreq, ratelimitd_init_trigger, req);
+ return req;
+}
+
+static void ratelimitd_init_trigger(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(subreq,
+ struct tevent_req);
+ struct ratelimitd_init_state *state = tevent_req_data(
+ req, struct ratelimitd_init_state);
+ bool ok;
+
+ DBG_NOTICE("Triggering ratelimitd startup\n");
+
+ ok = tevent_wakeup_recv(subreq);
+ TALLOC_FREE(subreq);
+ if (!ok) {
+ tevent_req_error(req, ENOMEM);
+ return;
+ }
+
+ state->ok = smbd_ratelimitd_init(state->msg, false, state->ppid);
+ if (state->ok) {
+ DBG_NOTICE("ratelimitd restarted\n");
+ tevent_req_done(req);
+ return;
+ }
+
+ DBG_NOTICE("ratelimitd startup failed, rescheduling\n");
+
+ subreq = tevent_wakeup_send(state,
+ state->ev,
+ tevent_timeval_current_ofs(1, 0));
+ if (tevent_req_nomem(subreq, req)) {
+ DBG_ERR("scheduling ratelimitd restart failed, giving up\n");
+ return;
+ }
+
+ tevent_req_set_callback(subreq, ratelimitd_init_trigger, req);
+ return;
+}
+
+static bool ratelimitd_init_recv(struct tevent_req *req)
+{
+ struct ratelimitd_init_state *state = tevent_req_data(
+ req, struct ratelimitd_init_state);
+
+ return state->ok;
+}
+
+static void ratelimitd_started(struct tevent_req *req)
+{
+ bool ok;
+
+ ok = ratelimitd_init_recv(req);
+ TALLOC_FREE(req);
+ if (!ok) {
+ DBG_ERR("Failed to restart ratelimitd, giving up\n");
+ return;
+ }
+}
+#endif /* WITH_RATELIMITD */
+
static void remove_child_pid(struct smbd_parent_context *parent,
pid_t pid,
bool unclean_shutdown,
return;
}
+#ifdef WITH_RATELIMITD
+ if (pid == procid_to_pid(&parent->ratelimitd)) {
+ struct tevent_req *req;
+ struct tevent_context *ev = messaging_tevent_context(
+ parent->msg_ctx);
+
+ server_id_set_disconnected(&parent->ratelimitd);
+
+ DBG_WARNING("Restarting ratelimitd\n");
+ req = ratelimitd_init_send(ev,
+ parent,
+ parent->msg_ctx,
+ &parent->ratelimitd);
+ if (req == NULL) {
+ DBG_ERR("Failed to restart ratelimitd\n");
+ return;
+ }
+ tevent_req_set_callback(req, ratelimitd_started, parent);
+ return;
+ }
+#endif /* WITH_RATELIMITD */
+
ok = cleanupdb_store_child(pid, unclean_shutdown);
if (!ok) {
DBG_ERR("cleanupdb_store_child failed\n");
exit_daemon("Samba cannot init the cleanupd", EACCES);
}
+#ifdef WITH_RATELIMITD
+ if (!smbd_ratelimitd_init(msg_ctx,
+ cmdline_daemon_cfg->interactive,
+ &parent->ratelimitd))
+ {
+ DBG_WARNING("ratelimitd init failed, "
+ "cluster rate limiting will not be available\n");
+ }
+#endif /* WITH_RATELIMITD */
+
if (!W_ERROR_IS_OK(registry_init_full()))
exit_daemon("Samba cannot init registry", EACCES);
--- /dev/null
+/*
+ * Samba rate limiting coordination daemon
+ *
+ * Copyright (c) 2026 Avan Thakkar <athakkar@redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "includes.h"
+#include "smbd_ratelimitd.h"
+#include "ratelimit_protocol.h"
+#include "lib/util/time.h"
+#include "lib/util/tevent_ntstatus.h"
+#include "lib/util/tevent_unix.h"
+#include "messages.h"
+#include "librpc/gen_ndr/messaging.h"
+#include "system/filesys.h"
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#undef DBGC_CLASS
+#define DBGC_CLASS DBGC_VFS
+
+#define MODULE_NAME "ratelimitd"
+
+/* Activity timeout - 5 seconds */
+#define RATELIMITD_ACTIVITY_TIMEOUT_US (5000000L)
+
+/* Broadcast interval - 1 second */
+#define RATELIMITD_BROADCAST_INTERVAL_US (1000000L)
+
+/* Remove processes inactive for more than 1 hour */
+#define INACTIVE_CLEANUP_THRESHOLD_US (3600000000L)
+
+/* Per-process activity tracking */
+struct process_activity {
+ struct process_activity *prev, *next;
+ pid_t pid;
+ char share_name[RATELIMIT_SHARE_NAME_LEN];
+ char op[8];
+ int64_t recent_iops;
+ uint32_t inflight_ios;
+ uint64_t last_seen_usec;
+ bool is_active;
+};
+
+struct smbd_ratelimitd_state {
+ struct tevent_context *ev;
+ struct messaging_context *msg_ctx;
+ int unix_sock;
+ char *socket_path;
+ uint32_t my_vnn;
+
+ struct process_activity *processes;
+
+ struct tevent_timer *broadcast_timer;
+ uint64_t last_broadcast_usec;
+
+ uint64_t total_reports_received;
+ uint64_t total_broadcasts_sent;
+};
+
+static int smbd_ratelimitd_state_destructor(
+ struct smbd_ratelimitd_state *state)
+{
+ int ret;
+
+ if (state->unix_sock != -1) {
+ close(state->unix_sock);
+ state->unix_sock = -1;
+ }
+
+ if (state->socket_path != NULL) {
+ ret = unlink(state->socket_path);
+ if (ret == 0) {
+ DBG_DEBUG("[%s] Removed socket: %s\n",
+ MODULE_NAME,
+ state->socket_path);
+ } else if (errno != ENOENT) {
+ DBG_WARNING("[%s] Failed to remove socket %s: %s\n",
+ MODULE_NAME,
+ state->socket_path,
+ strerror(errno));
+ }
+ }
+
+ return 0;
+}
+
+static struct process_activity *find_process_activity(
+ struct smbd_ratelimitd_state *state,
+ pid_t pid,
+ const char *share_name,
+ const char *op)
+{
+ struct process_activity *proc;
+
+ for (proc = state->processes; proc != NULL; proc = proc->next) {
+ if (proc->pid == pid &&
+ strcmp(proc->share_name, share_name) == 0 &&
+ strcmp(proc->op, op) == 0)
+ {
+ DBG_DEBUG("[%s] find_process_activity: FOUND\n",
+ MODULE_NAME);
+ return proc;
+ }
+ }
+
+ DBG_DEBUG("[%s] find_process_activity: NOT FOUND\n", MODULE_NAME);
+ return NULL;
+}
+
+static void update_process_activity(
+ struct smbd_ratelimitd_state *state,
+ const struct ratelimit_activity_report *report)
+{
+ struct process_activity *proc;
+ const char *op_str;
+ bool is_new = false;
+
+ op_str = ratelimit_op_to_string(report->operation);
+
+ proc = find_process_activity(state,
+ (pid_t)report->pid,
+ report->share_name,
+ op_str);
+
+ if (proc == NULL) {
+ DBG_DEBUG("[%s] update_process_activity: process not found, "
+ "creating new\n",
+ MODULE_NAME);
+
+ proc = talloc_zero(state, struct process_activity);
+ if (proc == NULL) {
+ DBG_ERR("[%s] update_process_activity: failed to "
+ "allocate process entry\n",
+ MODULE_NAME);
+ return;
+ }
+
+ proc->pid = report->pid;
+ strlcpy(proc->share_name,
+ report->share_name,
+ sizeof(proc->share_name));
+ strlcpy(proc->op, op_str, sizeof(proc->op));
+ is_new = true;
+
+ DLIST_ADD(state->processes, proc);
+
+ DBG_DEBUG("[%s] NEW process tracked: pid=%d share=%s op=%s\n",
+ MODULE_NAME,
+ proc->pid,
+ proc->share_name,
+ proc->op);
+ } else {
+ DBG_DEBUG("[%s] update_process_activity: updating existing "
+ "process\n",
+ MODULE_NAME);
+ }
+
+ proc->recent_iops = report->recent_iops;
+ proc->inflight_ios = report->inflight_ios;
+ proc->last_seen_usec = report->timestamp_usec;
+ proc->is_active = (report->recent_iops > 0 ||
+ report->inflight_ios > 0);
+
+ DBG_DEBUG("[%s] Updated: pid=%d share=%s op=%s iops=%" PRId64 " "
+ "inflight=%u active=%d %s\n",
+ MODULE_NAME,
+ proc->pid,
+ proc->share_name,
+ proc->op,
+ proc->recent_iops,
+ proc->inflight_ios,
+ proc->is_active,
+ is_new ? "[NEW]" : "[EXISTING]");
+}
+
+static void handle_unix_socket_read(struct tevent_context *ev,
+ struct tevent_fd *fde,
+ uint16_t flags,
+ void *private_data)
+{
+ struct smbd_ratelimitd_state *state = private_data;
+ struct ratelimit_activity_report report;
+ struct sockaddr_un from;
+ socklen_t fromlen = sizeof(from);
+ ssize_t ret;
+
+ ret = recvfrom(state->unix_sock,
+ &report,
+ sizeof(report),
+ 0,
+ (struct sockaddr *)&from,
+ &fromlen);
+
+ if (ret < 0) {
+ DBG_ERR("[%s] recvfrom() failed: %s\n",
+ MODULE_NAME,
+ strerror(errno));
+ return;
+ }
+
+ if (ret != sizeof(report)) {
+ DBG_ERR("[%s] Short read: got %zd, expected %zu\n",
+ MODULE_NAME,
+ ret,
+ sizeof(report));
+ return;
+ }
+
+ report.share_name[sizeof(report.share_name) - 1] = '\0';
+
+ DBG_DEBUG("[%s] handle_unix_socket_read: received report "
+ "pid=%d share=%s op=%s recent_iops=%" PRId64 "\n",
+ MODULE_NAME,
+ report.pid,
+ report.share_name,
+ ratelimit_op_to_string(report.operation),
+ report.recent_iops);
+
+ if (report.protocol_version != RATELIMIT_PROTOCOL_VERSION) {
+ DBG_ERR("[%s] Protocol mismatch: got %u, expected %u\n",
+ MODULE_NAME,
+ report.protocol_version,
+ RATELIMIT_PROTOCOL_VERSION);
+ return;
+ }
+
+ update_process_activity(state, &report);
+ state->total_reports_received++;
+
+ DBG_DEBUG("[%s] handle_unix_socket_read: DONE total_reports=%" PRIu64
+ "\n",
+ MODULE_NAME,
+ state->total_reports_received);
+}
+
+static void cleanup_inactive_processes(struct smbd_ratelimitd_state *state)
+{
+ uint64_t now = time_now_usec();
+ struct process_activity *proc, *next;
+ unsigned int removed = 0;
+
+ for (proc = state->processes; proc != NULL; proc = next) {
+ uint64_t age_us = now - proc->last_seen_usec;
+ next = proc->next;
+
+ if (!proc->is_active && age_us > INACTIVE_CLEANUP_THRESHOLD_US)
+ {
+ DLIST_REMOVE(state->processes, proc);
+ TALLOC_FREE(proc);
+ removed++;
+ }
+ }
+
+ if (removed > 0) {
+ DBG_NOTICE("[%s] Cleaned up %d inactive processes\n",
+ MODULE_NAME,
+ removed);
+ }
+}
+
+static void broadcast_summary(struct smbd_ratelimitd_state *state,
+ const char *op,
+ const char *share_name,
+ int32_t count,
+ uint64_t timestamp_usec)
+{
+ struct ratelimit_node_summary summary = {0};
+ DATA_BLOB blob;
+
+ if (count <= 0) {
+ return;
+ }
+
+ summary.vnn = state->my_vnn;
+ summary.process_count = count;
+ summary.timestamp_usec = timestamp_usec;
+ strlcpy(summary.share_name, share_name, sizeof(summary.share_name));
+
+ blob = data_blob_const(&summary, sizeof(summary));
+
+ messaging_send_all(state->msg_ctx,
+ ratelimit_msg_type_summary(
+ ratelimit_op_from_string(op)),
+ blob.data,
+ blob.length);
+
+ state->total_broadcasts_sent++;
+
+ DBG_DEBUG("[%s] Broadcast: op=%s share=%s count=%d\n",
+ MODULE_NAME,
+ op,
+ share_name,
+ count);
+}
+
+/*
+ * Broadcast one ratelimit_node_summary per active (snum, op) pair.
+ * Each VFS ratelimiter gets only the count of processes that are
+ * actually doing I/O on a particular share, so it divides the correct
+ * per-share limit.
+ */
+static void broadcast_per_share_summaries(struct smbd_ratelimitd_state *state,
+ uint64_t now)
+{
+ struct process_activity *proc, *inner;
+ int broadcasts = 0;
+ int timed_out = 0;
+
+ for (proc = state->processes; proc != NULL; proc = proc->next) {
+ uint64_t age_us = now - proc->last_seen_usec;
+ bool already_done = false;
+ int32_t count = 0;
+
+ if (age_us > RATELIMITD_ACTIVITY_TIMEOUT_US) {
+ if (proc->is_active) {
+ DBG_NOTICE("[%s] Process TIMED OUT "
+ "(pid=%d share=%s op=%s "
+ "age=%" PRIu64 " ms)\n",
+ MODULE_NAME,
+ proc->pid,
+ proc->share_name,
+ proc->op,
+ age_us / 1000);
+ proc->is_active = false;
+ timed_out++;
+ }
+ continue;
+ }
+
+ if (!proc->is_active) {
+ continue;
+ }
+
+ /*
+ * Skip if an earlier active process with the same
+ * (share_name, op) already triggered a broadcast for this
+ * combination.
+ */
+ for (inner = state->processes; inner != proc;
+ inner = inner->next)
+ {
+ uint64_t inner_age = now - inner->last_seen_usec;
+ if (strcmp(inner->share_name, proc->share_name) == 0 &&
+ strcmp(inner->op, proc->op) == 0 &&
+ inner->is_active &&
+ inner_age <= RATELIMITD_ACTIVITY_TIMEOUT_US)
+ {
+ already_done = true;
+ break;
+ }
+ }
+ if (already_done) {
+ continue;
+ }
+
+ /*
+ * Count all active non-timed-out procs for this
+ * (share_name, op)
+ */
+ for (inner = state->processes; inner != NULL;
+ inner = inner->next)
+ {
+ uint64_t inner_age = now - inner->last_seen_usec;
+ if (strcmp(inner->share_name, proc->share_name) == 0 &&
+ strcmp(inner->op, proc->op) == 0 &&
+ inner->is_active &&
+ inner_age <= RATELIMITD_ACTIVITY_TIMEOUT_US)
+ {
+ count++;
+ }
+ }
+
+ broadcast_summary(
+ state, proc->op, proc->share_name, count, now);
+ broadcasts++;
+ }
+
+ if (broadcasts == 0) {
+ DBG_DEBUG("[%s] No active processes, skipped broadcasts "
+ "(timed_out=%d)\n",
+ MODULE_NAME,
+ timed_out);
+ } else {
+ DBG_DEBUG("[%s] Sent %d per-share broadcasts "
+ "(timed_out=%d)\n",
+ MODULE_NAME,
+ broadcasts,
+ timed_out);
+ }
+}
+
+static void broadcast_timer_handler(struct tevent_context *ev,
+ struct tevent_timer *te,
+ struct timeval current_time,
+ void *private_data)
+{
+ struct smbd_ratelimitd_state *state = private_data;
+ uint64_t now = time_now_usec();
+
+ broadcast_per_share_summaries(state, now);
+
+ state->last_broadcast_usec = now;
+
+ /* Cleanup old inactive entries every 100 timer ticks (~100 seconds) */
+ if ((state->total_broadcasts_sent > 0) &&
+ (state->total_broadcasts_sent % 100 == 0))
+ {
+ cleanup_inactive_processes(state);
+ }
+
+ DBG_DEBUG("[%s] broadcast_timer_handler: rescheduling timer\n",
+ MODULE_NAME);
+
+ state->broadcast_timer = tevent_add_timer(
+ state->ev,
+ state,
+ timeval_current_ofs(RATELIMITD_BROADCAST_INTERVAL_US / 1000000,
+ RATELIMITD_BROADCAST_INTERVAL_US %
+ 1000000),
+ broadcast_timer_handler,
+ state);
+ if (state->broadcast_timer == NULL) {
+ DBG_ERR("[%s] Failed to reschedule broadcast timer: "
+ "cluster coordination suspended\n",
+ MODULE_NAME);
+ }
+}
+
+struct tevent_req *smbd_ratelimitd_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct messaging_context *msg)
+{
+ struct tevent_req *req;
+ struct smbd_ratelimitd_state *state;
+ struct sockaddr_un addr;
+ struct tevent_fd *fde;
+ size_t len;
+ int ret;
+
+ req = tevent_req_create(mem_ctx, &state, struct smbd_ratelimitd_state);
+ if (req == NULL) {
+ return NULL;
+ }
+
+ state->ev = ev;
+ state->msg_ctx = msg;
+ state->my_vnn = get_my_vnn();
+ state->unix_sock = -1;
+ state->socket_path = NULL;
+
+ talloc_set_destructor(state, smbd_ratelimitd_state_destructor);
+
+ state->processes = NULL;
+
+ state->socket_path = state_path(state, RATELIMITD_SOCKET_NAME);
+ if (state->socket_path == NULL) {
+ DBG_ERR("[%s] Failed to allocate socket path\n", MODULE_NAME);
+ tevent_req_error(req, ENOMEM);
+ return tevent_req_post(req, ev);
+ }
+
+ DBG_NOTICE("[%s] Creating socket at: %s\n",
+ MODULE_NAME,
+ state->socket_path);
+
+ state->unix_sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+ if (state->unix_sock < 0) {
+ DBG_ERR("[%s] socket() failed: %s\n",
+ MODULE_NAME,
+ strerror(errno));
+ tevent_req_error(req, errno);
+ return tevent_req_post(req, ev);
+ }
+
+ /* Remove stale socket from previous daemon instance */
+ ret = unlink(state->socket_path);
+ if (ret == 0) {
+ DBG_DEBUG("[%s] Removed stale socket: %s\n",
+ MODULE_NAME,
+ state->socket_path);
+ } else if (errno != ENOENT) {
+ DBG_DEBUG("[%s] unlink(%s) failed: %s (continuing anyway)\n",
+ MODULE_NAME,
+ state->socket_path,
+ strerror(errno));
+ }
+
+ memset(&addr, 0, sizeof(addr));
+ addr.sun_family = AF_UNIX;
+
+ len = strlcpy(addr.sun_path,
+ state->socket_path,
+ sizeof(addr.sun_path));
+ if (len >= sizeof(addr.sun_path)) {
+ DBG_ERR("[%s] Socket path too long: %s\n",
+ MODULE_NAME,
+ state->socket_path);
+ close(state->unix_sock);
+ state->unix_sock = -1;
+ tevent_req_error(req, ENAMETOOLONG);
+ return tevent_req_post(req, ev);
+ }
+
+ ret = bind(state->unix_sock, (struct sockaddr *)&addr, sizeof(addr));
+ if (ret < 0) {
+ DBG_ERR("[%s] bind() failed on %s: %s\n",
+ MODULE_NAME,
+ state->socket_path,
+ strerror(errno));
+ close(state->unix_sock);
+ state->unix_sock = -1;
+ tevent_req_error(req, errno);
+ return tevent_req_post(req, ev);
+ }
+
+ ret = chmod(state->socket_path, 0660);
+ if (ret < 0) {
+ DBG_WARNING("[%s] chmod() failed: %s\n",
+ MODULE_NAME,
+ strerror(errno));
+ }
+
+ fde = tevent_add_fd(state->ev,
+ state,
+ state->unix_sock,
+ TEVENT_FD_READ,
+ handle_unix_socket_read,
+ state);
+ if (fde == NULL) {
+ DBG_ERR("[%s] tevent_add_fd failed\n", MODULE_NAME);
+ close(state->unix_sock);
+ state->unix_sock = -1;
+ tevent_req_error(req, ENOMEM);
+ return tevent_req_post(req, ev);
+ }
+
+ state->broadcast_timer = tevent_add_timer(
+ state->ev,
+ state,
+ timeval_current_ofs(RATELIMITD_BROADCAST_INTERVAL_US / 1000000,
+ RATELIMITD_BROADCAST_INTERVAL_US %
+ 1000000),
+ broadcast_timer_handler,
+ state);
+ if (state->broadcast_timer == NULL) {
+ DBG_ERR("[%s] tevent_add_timer failed\n", MODULE_NAME);
+ close(state->unix_sock);
+ state->unix_sock = -1;
+ tevent_req_error(req, ENOMEM);
+ return tevent_req_post(req, ev);
+ }
+
+ DBG_NOTICE("[%s] Daemon initialized: vnn=%u socket=%s\n",
+ MODULE_NAME,
+ state->my_vnn,
+ state->socket_path);
+
+ return req;
+}
+
+int smbd_ratelimitd_recv(struct tevent_req *req)
+{
+ return tevent_req_simple_recv_unix(req);
+}
--- /dev/null
+/*
+ * Samba rate limiting coordination daemon
+ *
+ * Copyright (c) 2026 Avan Thakkar <athakkar@redhat.com>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef __SMBD_RATELIMITD_H__
+#define __SMBD_RATELIMITD_H__
+
+#include "replace.h"
+#include <tevent.h>
+#include "messages.h"
+
+struct tevent_req *smbd_ratelimitd_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct messaging_context *msg);
+
+/*
+ * Complete ratelimitd request
+ */
+int smbd_ratelimitd_recv(struct tevent_req *req);
+
+#endif /* __SMBD_RATELIMITD_H__ */
########################## BINARIES #################################
bld.SAMBA3_BINARY('smbd/smbd',
- source='smbd/server.c smbd/smbd_cleanupd.c',
+ source='smbd/server.c smbd/smbd_cleanupd.c smbd/smbd_ratelimitd.c',
deps='''
CMDLINE_S3
smbd_base
projects / thirdparty / samba.git / commitdiff
