[Project] Implement srv commands broadcast

2 files changed, 32 insertions, 5 deletions

diff --git a/src/libserver/rspamd_control.c b/src/libserver/rspamd_control.c
index 2e6ad21f0..699317ff5 100644
--- a/src/libserver/rspamd_control.c
+++ b/src/libserver/rspamd_control.c
@@ -103,6 +103,8 @@ static const struct rspamd_control_cmd_match {
 		},
 };
 
+static void rspamd_control_ignore_io_handler (int fd, short what, void *ud);
+
 void
 rspamd_control_send_error (struct rspamd_control_session *session,
 		gint code, const gchar *error_msg, ...)
@@ -388,7 +390,8 @@ rspamd_control_broadcast_cmd (struct rspamd_main *rspamd_main,
 							  struct rspamd_control_command *cmd,
 							  gint attached_fd,
 							  rspamd_ev_cb handler,
-							  gpointer ud)
+							  gpointer ud,
+							  pid_t except_pid)
 {
 	GHashTableIter it;
 	struct rspamd_worker *wrk;
@@ -409,6 +412,10 @@ rspamd_control_broadcast_cmd (struct rspamd_main *rspamd_main,
 			continue;
 		}
 
+		if (except_pid != 0 && wrk->pid == except_pid) {
+			continue;
+		}
+
 		memset (&msg, 0, sizeof (msg));
 
 		/* Attach fd to the message */
@@ -458,6 +465,15 @@ rspamd_control_broadcast_cmd (struct rspamd_main *rspamd_main,
 	return res;
 }
 
+void
+rspamd_control_broadcast_srv_cmd (struct rspamd_main *rspamd_main,
+								  struct rspamd_control_command *cmd,
+								  pid_t except_pid)
+{
+	rspamd_control_broadcast_cmd (rspamd_main, cmd, -1,
+			rspamd_control_ignore_io_handler, NULL, except_pid);
+}
+
 static gint
 rspamd_control_finish_handler (struct rspamd_http_connection *conn,
 		struct rspamd_http_message *msg)
@@ -496,7 +512,7 @@ rspamd_control_finish_handler (struct rspamd_http_connection *conn,
 			/* Send command to all workers */
 			session->replies = rspamd_control_broadcast_cmd (
 					session->rspamd_main, &session->cmd, -1,
-					rspamd_control_wrk_io, session);
+					rspamd_control_wrk_io, session, 0);
 
 			DL_FOREACH (session->replies, cur) {
 				session->replies_remain ++;
@@ -890,7 +906,7 @@ rspamd_srv_handler (EV_P_ ev_io *w, int revents)
 						sizeof (wcmd.cmd.hs_loaded.cache_dir));
 				wcmd.cmd.hs_loaded.forced = cmd.cmd.hs_loaded.forced;
 				rspamd_control_broadcast_cmd (srv, &wcmd, rfd,
-						rspamd_control_ignore_io_handler, NULL);
+						rspamd_control_ignore_io_handler, NULL, worker->pid);
 				break;
 			case RSPAMD_SRV_MONITORED_CHANGE:
 				/* Broadcast command to all workers */
@@ -902,14 +918,14 @@ rspamd_srv_handler (EV_P_ ev_io *w, int revents)
 				wcmd.cmd.monitored_change.alive = cmd.cmd.monitored_change.alive;
 				wcmd.cmd.monitored_change.sender = cmd.cmd.monitored_change.sender;
 				rspamd_control_broadcast_cmd (srv, &wcmd, rfd,
-						rspamd_control_ignore_io_handler, NULL);
+						rspamd_control_ignore_io_handler, NULL, 0);
 				break;
 			case RSPAMD_SRV_LOG_PIPE:
 				memset (&wcmd, 0, sizeof (wcmd));
 				wcmd.type = RSPAMD_CONTROL_LOG_PIPE;
 				wcmd.cmd.log_pipe.type = cmd.cmd.log_pipe.type;
 				rspamd_control_broadcast_cmd (srv, &wcmd, rfd,
-						rspamd_control_log_pipe_io_handler, NULL);
+						rspamd_control_log_pipe_io_handler, NULL, 0);
 				break;
 			case RSPAMD_SRV_ON_FORK:
 				rdata->rep.reply.on_fork.status = 0;
diff --git a/src/libserver/rspamd_control.h b/src/libserver/rspamd_control.h
index ebb564d7e..b04fca09d 100644
--- a/src/libserver/rspamd_control.h
+++ b/src/libserver/rspamd_control.h
@@ -241,6 +241,17 @@ void rspamd_srv_send_command (struct rspamd_worker *worker,
 							  gpointer ud);
 
 /**
+ * Broadcast srv cmd from rspamd_main to workers
+ * @param rspamd_main
+ * @param cmd
+ * @param except_pid
+ */
+void
+rspamd_control_broadcast_srv_cmd (struct rspamd_main *rspamd_main,
+								  struct rspamd_control_command *cmd,
+								  pid_t except_pid);
+
+/**
  * Returns command from a specified string (case insensitive)
  * @param str
  * @return