ipc: add new set of synchronous pipes

These new pipe set for each process can be used to synchronously send a
command from one process to the other. When using these pipes, make sure
that there's no other job that can overlap, otherwise the receiver will
not be albe to handle it

Author: razvancrainea

ipc.c

@@ -106,6 +106,12 @@ int create_ipc_pipes( int proc_no )
 				i, errno, strerror(errno));
 			return -1;
 		}
+
+		if (pipe(pt[i].ipc_sync_pipe_holder)<0) {
+			LM_ERR("failed to create IPC sync pipe for process %d, err %d/%s\n",
+				i, errno, strerror(errno));
+			return -1;
+		}
 	}
 	return 0;
 }
@@ -193,6 +199,39 @@ int ipc_dispatch_rpc( ipc_rpc_f *rpc, void *param)
 	return __ipc_send_job(ipc_shared_pipe[1], ipc_rpc_type, rpc, param);
 }
 
+int ipc_send_sync_reply(int dst_proc, void *param)
+{
+	int n;
+
+again:
+	n = write(IPC_FD_SYNC_WRITE(dst_proc), &param, sizeof(param));
+	if (n<0) {
+		if (errno==EINTR)
+			goto again;
+		LM_ERR("sending sync rpc %d[%s]\n", errno, strerror(errno));
+		return -1;
+	}
+	return 0;
+}
+
+int ipc_recv_sync_reply(void **param)
+{
+	void *ret;
+	int n;
+
+again:
+	n = read(IPC_FD_SYNC_READ_SELF, &ret, sizeof(ret));
+	if (n < sizeof(*ret)) {
+		if (errno == EINTR)
+			goto again;
+		/* if we got here, it's definitely an error, because the socket is
+		 * blocking, so we can't read partial messages */
+		LM_ERR("read failed:[%d] %s\n", errno, strerror(errno));
+		return -1;
+	}
+	*param = ret;
+	return 0;
+}
 
 void ipc_handle_job(int fd)
 {

ipc.h

@@ -32,6 +32,9 @@ extern int ipc_shared_fd_read;
 #define IPC_FD_WRITE(_proc_no)  pt[_proc_no].ipc_pipe[1]
 #define IPC_FD_READ_SELF        IPC_FD_READ(process_no)
 #define IPC_FD_READ_SHARED      ipc_shared_fd_read
+#define IPC_FD_SYNC_READ(_proc_no)   pt[_proc_no].ipc_sync_pipe[0]
+#define IPC_FD_SYNC_WRITE(_proc_no)  pt[_proc_no].ipc_sync_pipe[1]
+#define IPC_FD_SYNC_READ_SELF        IPC_FD_SYNC_READ(process_no)
 
 /* prototype of IPC handler - function called by the IPC engine
  * when the a job with the correspoding type was received */
@@ -68,6 +71,25 @@ int ipc_send_job(int dst_proc, ipc_handler_type type, void *payload);
 int ipc_send_rpc(int dst_proc, ipc_rpc_f *rpc, void *param);
 
 
+/*
+ * Send a synchronous message to a specific "dst_proc" process
+ * Use this command when you are sure that the "dst_proc" is waiting only for
+ * this specific "response", and cannot overlap with a different task
+ *
+ * Return: 0 on success, -1 on failure
+ */
+int ipc_send_sync_reply(int dst_proc, void *param);
+
+
+/*
+ * Wait for a message sent by a different process synchronously using the
+ * ipc_send_sync_reply() function.
+ *
+ * Return: 0 on success, -1 on failure
+ */
+int ipc_recv_sync_reply(void **param);
+
+
 /*
  * Push a job for the next available OpenSIPS worker and quickly return
  *

pt.c

@@ -81,6 +81,7 @@ int init_multi_proc_support(void)
 		pt[i].unix_sock = -1;
 		pt[i].pid = -1;
 		pt[i].ipc_pipe[0] = pt[i].ipc_pipe[1] = -1;
+		pt[i].ipc_sync_pipe[0] = pt[i].ipc_sync_pipe[1] = -1;
 	}
 
 	/* create the load-related stats (initially marked as hidden */
@@ -199,6 +200,7 @@ void reset_process_slot( int p_id )
 	pt[p_id].flags = 0;
 
 	pt[p_id].ipc_pipe[0] = pt[p_id].ipc_pipe[1] = -1;
+	pt[p_id].ipc_sync_pipe[0] = pt[p_id].ipc_sync_pipe[1] = -1;
 	pt[p_id].unix_sock = -1;
 
 	pt[p_id].log_level = pt[p_id].default_log_level = 0; /*not really needed*/
@@ -260,12 +262,16 @@ int internal_fork(char *proc_desc, unsigned int flags,
 		/* advertise no IPC to the rest of the procs */
 		pt[new_idx].ipc_pipe[0] = -1;
 		pt[new_idx].ipc_pipe[1] = -1;
+		pt[new_idx].ipc_sync_pipe[0] = -1;
+		pt[new_idx].ipc_sync_pipe[1] = -1;
 		/* NOTE: the IPC fds will remain open in the other processes,
 		 * but they will not be known */
 	} else {
 		/* activate the IPC pipes */
 		pt[new_idx].ipc_pipe[0]=pt[new_idx].ipc_pipe_holder[0];
 		pt[new_idx].ipc_pipe[1]=pt[new_idx].ipc_pipe_holder[1];
+		pt[new_idx].ipc_sync_pipe[0]=pt[new_idx].ipc_sync_pipe_holder[0];
+		pt[new_idx].ipc_sync_pipe[1]=pt[new_idx].ipc_sync_pipe_holder[1];
 	}
 
 	pt[new_idx].pid = 0;
@@ -388,6 +394,8 @@ void dynamic_process_final_exit(void)
 	/* prevent any more IPC */
 	pt[process_no].ipc_pipe[0] = -1;
 	pt[process_no].ipc_pipe[1] = -1;
+	pt[process_no].ipc_sync_pipe[0] = -1;
+	pt[process_no].ipc_sync_pipe[1] = -1;
 
 	/* clear the per-process connection from the DB queues */
 	ql_force_process_disconnect(process_no);

pt.h

@@ -56,6 +56,14 @@ struct process_table {
 	 * does not exist */
 	int ipc_pipe_holder[2];
 
+	/* pipe used by the process to receive a synchronoys job
+	 * this pipe should only be used by a process to synchronously receive a
+	 * message after he knows that some other process will send it for sure,
+	 * and there's no other job that can overlap in the meantime */
+	int ipc_sync_pipe[2];
+	/* same as above, but holder for non-existing processes */
+	int ipc_sync_pipe_holder[2];
+
 	/* holder for the unixsocks used by TCP layer for inter-proc communication;
 	 * used when the corresponding process does not exist */
 	int tcp_socks_holder[2];