async: fix fd leaking and invalid memory overwrite

Also simplify the async interface to prevent dereferencing the fd.
This commit combines the following master commits:

9b45482
2b264d0
ac549a0

Kudos to Liviu Chircu for brainstorming and optimizing the initial fix.

(cherry picked from commit 989b7b9)

async.c

@@ -93,15 +93,15 @@ int register_async_fd(int fd, async_resume_fd *f, void *resume_param)
 }
 
 
-int async_fd_resume(int *fd, void *param)
+int async_fd_resume(int fd, void *param)
 {
 	async_ctx *ctx = (async_ctx *)param;
 	int ret;
 
 	async_status = ASYNC_DONE; /* assume default status as done */
 
 	/* call the resume function in order to read and handle data */
-	ret = ((async_resume_fd*)ctx->resume_f)( *fd, ctx->resume_param );
+	ret = ((async_resume_fd*)ctx->resume_f)( fd, ctx->resume_param );
 	if (async_status==ASYNC_CONTINUE) {
 		/* leave the fd into the reactor*/
 		return 0;
@@ -110,25 +110,25 @@ int async_fd_resume(int *fd, void *param)
 			LM_ERR("ASYNC_CHANGE_FD: given file descriptor shall be "
 				"positive!\n");
 			return 0;
-		} else if (ret>0 && ret==*fd) {
+		} else if (ret>0 && ret==fd) {
 			/*trying to add the same fd; shall continue*/
 			LM_CRIT("You are trying to replace the old fd with the same fd!"
 					"Will act as in ASYNC_CONTINUE!\n");
 			return 0;
 		}
 
 		/* remove the old fd from the reactor */
-		reactor_del_reader( *fd, -1, IO_FD_CLOSING);
-		*fd=ret;
+		reactor_del_reader(fd, -1, IO_FD_CLOSING);
+		fd=ret;
 
 		/* insert the new fd inside the reactor */
-		if (reactor_add_reader(*fd,F_FD_ASYNC,RCT_PRIO_ASYNC,(void*)ctx)<0 ) {
+		if (reactor_add_reader(fd,F_FD_ASYNC,RCT_PRIO_ASYNC,(void*)ctx)<0 ) {
 			LM_ERR("failed to add async FD to reactor -> act in sync mode\n");
 			do {
 				async_status = ASYNC_DONE;
-				ret = ((async_resume_fd*)ctx->resume_f)(*fd,ctx->resume_param);
+				ret = ((async_resume_fd*)ctx->resume_f)(fd,ctx->resume_param);
 				if (async_status == ASYNC_CHANGE_FD)
-					*fd=ret;
+					fd=ret;
 			} while(async_status==ASYNC_CONTINUE||async_status==ASYNC_CHANGE_FD);
 			goto done;
 		} else {
@@ -139,11 +139,11 @@ int async_fd_resume(int *fd, void *param)
 	}
 
 	/* remove from reactor, we are done */
-	reactor_del_reader( *fd, -1, IO_FD_CLOSING);
+	reactor_del_reader(fd, -1, IO_FD_CLOSING);
 
 done:
 	if (async_status == ASYNC_DONE_CLOSE_FD)
-		close(*fd);
+		close(fd);
 
 	return 0;
 }
@@ -163,7 +163,7 @@ int async_fd_resume(int *fd, void *param)
 		(_req).rcv.dst_ip.af = AF_INET;\
 	} while(0)
 
-int async_launch_resume(int *fd, void *param)
+int async_launch_resume(int fd, void *param)
 {
 	struct sip_msg req;
 	async_launch_ctx *ctx = (async_launch_ctx *)param;
@@ -175,7 +175,7 @@ int async_launch_resume(int *fd, void *param)
 
 	/* call the resume function in order to read and handle data */
 	return_code = ((async_resume_module*)(ctx->async.resume_f))
-		( *fd, &req, ctx->async.resume_param );
+		( fd, &req, ctx->async.resume_param );
 
 	if (async_status==ASYNC_CONTINUE) {
 		/* do not run the report route, leave the fd into the reactor*/
@@ -190,27 +190,27 @@ int async_launch_resume(int *fd, void *param)
 			LM_ERR("ASYNC_CHANGE_FD: given file descriptor must be "
 				"positive!\n");
 			goto restore;
-		} else if (return_code>0 && return_code==*fd) {
+		} else if (return_code>0 && return_code==fd) {
 			/*trying to add the same fd; shall continue*/
 			LM_CRIT("You are trying to replace the old fd with the same fd!"
 					"Will act as in ASYNC_CONTINUE!\n");
 			goto restore;
 		}
 
 		/* remove the old fd from the reactor */
-		reactor_del_reader( *fd, -1, IO_FD_CLOSING);
-		*fd=return_code;
+		reactor_del_reader(fd, -1, IO_FD_CLOSING);
+		fd=return_code;
 
 		/* insert the new fd inside the reactor */
-		if (reactor_add_reader( *fd, F_LAUNCH_ASYNC, RCT_PRIO_ASYNC,
+		if (reactor_add_reader(fd, F_LAUNCH_ASYNC, RCT_PRIO_ASYNC,
 		(void*)ctx)<0 ) {
 			LM_ERR("failed to add async FD to reactor -> act in sync mode\n");
 			do {
 				async_status = ASYNC_DONE;
 				return_code = ((async_resume_module*)(ctx->async.resume_f))
-					( *fd, &req, ctx->async.resume_param );
+					(fd, &req, ctx->async.resume_param );
 				if (async_status == ASYNC_CHANGE_FD)
-					*fd=return_code;
+					fd=return_code;
 			} while(async_status==ASYNC_CONTINUE||async_status==ASYNC_CHANGE_FD);
 			goto run_route;
 		} else {
@@ -221,11 +221,11 @@ int async_launch_resume(int *fd, void *param)
 	}
 
 	/* remove from reactor, we are done */
-	reactor_del_reader( *fd, -1, IO_FD_CLOSING);
+	reactor_del_reader(fd, -1, IO_FD_CLOSING);
 
 run_route:
 	if (async_status == ASYNC_DONE_CLOSE_FD)
-		close(*fd);
+		close(fd);
 
 	if (ctx->report_route!=-1) {
 		LM_DBG("runinng report route for a launch job\n");

async.h

@@ -73,8 +73,18 @@ extern int async_status;
 typedef int (async_script_start_function)
 	(struct sip_msg *msg, struct action* a , int resume_route);
 
+/* Handles periodic progress (data arrival) on behalf of the contained,
+ * module-specific resume function, which it must also call
+ *
+ * Parameters:
+ *   @fd: file to resume on.  If no descriptor is available,
+ *        provide ASYNC_FD_NONE
+ *   @param: private data, stored by (async_script_start_function)
+ */
 typedef int (async_script_resume_function)
-	(int *fd, void *param);
+	(int fd, void *param);
+#define ASYNC_FD_NONE -1
+#define valid_async_fd(fd) (fd >= 0)
 
 /* internal used functions to start (from script) and
  * to continue (from reactor) async I/O ops */
@@ -86,8 +96,12 @@ int register_async_script_handlers(async_script_start_function *f1,
 											async_script_resume_function *f2);
 
 /* async related functions to be used by the
- * functions exported by modules */
-typedef int (async_resume_module)
+ * functions exported by modules
+ *
+ * NOTE: This function may be triggered even without any pending data!
+ *       If this is the case, @fd == ASYNC_FD_NONE
+ */
+typedef enum async_ret_code (async_resume_module)
 	(int fd, struct sip_msg *msg, void *param);
 
 
@@ -111,11 +125,11 @@ typedef int (async_resume_fd)
  */
 int register_async_fd(int fd, async_resume_fd *f, void *param);
 
-/* Reseum function for the registered async fd. This is internally called
- * by the reactor via the handle_io() routine
-   Function only for internal usage.
+/* Resume function for the registered async fd. This is internally called
+ * by the reactor via the handle_io() routine.  Function only for internal
+ * usage.  @fd is always valid.
  */
-int async_fd_resume(int *fd, void *param);
+int async_fd_resume(int fd, void *param);
 
 
 
@@ -124,7 +138,8 @@ int async_fd_resume(int *fd, void *param);
 int async_script_launch(struct sip_msg *msg, struct action* a,
 		int report_route);
 
-int async_launch_resume(int *fd, void *param);
+/* @fd is always valid */
+int async_launch_resume(int fd, void *param);
 
 
 #endif

modules/event_routing/ebr_data.c

@@ -563,7 +563,7 @@ void handle_ebr_ipc(int sender, void *payload)
 		((async_ctx*)job->data)->resume_param = job->avps;
 
 		/* invoke the global resume ASYNC function */
-		async_script_resume_f( NULL, job->data /*the async ctx*/ );
+		async_script_resume_f(ASYNC_FD_NONE, job->data /*the async ctx*/ );
 
 		shm_free(job);
 

modules/tm/async.c

@@ -70,7 +70,7 @@ static inline void run_resume_route( int resume_route, struct sip_msg *msg,
 
 /* function triggered from reactor in order to continue the processing
  */
-int t_resume_async(int *fd, void *param)
+int t_resume_async(int fd, void *param)
 {
 	static struct sip_msg faked_req;
 	static struct ua_client uac;
@@ -83,8 +83,8 @@ int t_resume_async(int *fd, void *param)
 	struct cell *t= ctx->t;
 	int route;
 
-	if (fd)
-		LM_DBG("resuming on fd %d, transaction %p \n",*fd, t);
+	if (valid_async_fd(fd))
+		LM_DBG("resuming on fd %d, transaction %p \n", fd, t);
 	else
 		LM_DBG("resuming without a fd, transaction %p \n", t);
 
@@ -126,7 +126,7 @@ int t_resume_async(int *fd, void *param)
 	async_status = ASYNC_DONE; /* assume default status as done */
 	/* call the resume function in order to read and handle data */
 	return_code = ((async_resume_module*)(ctx->async.resume_f))
-		( (fd ? *fd: -1), &faked_req, ctx->async.resume_param );
+		( (valid_async_fd(fd) ? fd: ASYNC_FD_NONE), &faked_req, ctx->async.resume_param );
 	if (async_status==ASYNC_CONTINUE) {
 		/* do not run the resume route */
 		goto restore;
@@ -137,29 +137,26 @@ int t_resume_async(int *fd, void *param)
 		if (return_code<0) {
 			LM_ERR("ASYNC_CHANGE_FD: given file descriptor shall be positive!\n");
 			goto restore;
-		} else if (return_code > 0 && fd && return_code == *fd) {
+		} else if (return_code > 0 && valid_async_fd(fd) && return_code == fd) {
 			/*trying to add the same fd; shall continue*/
 			LM_CRIT("You are trying to replace the old fd with the same fd!"
 					"Will act as in ASYNC_CONTINUE!\n");
 			goto restore;
 		}
 
 		/* if there was a file descriptor, remove it from the reactor */
-		if (fd) {
-			/* remove the old fd from the reactor */
-			reactor_del_reader( *fd, -1, IO_FD_CLOSING);
-			*fd=return_code;
-		}
+		reactor_del_reader(fd, -1, IO_FD_CLOSING);
+		fd=return_code;
 
 		/* insert the new fd inside the reactor */
-		if(reactor_add_reader(return_code,F_SCRIPT_ASYNC,RCT_PRIO_ASYNC,(void*)ctx)<0){
+		if(reactor_add_reader(fd,F_SCRIPT_ASYNC,RCT_PRIO_ASYNC,(void*)ctx)<0){
 			LM_ERR("failed to add async FD to reactor -> act in sync mode\n");
 			do {
 				async_status = ASYNC_DONE;
 				return_code = ((async_resume_module*)(ctx->async.resume_f))
-					( return_code, &faked_req, ctx->async.resume_param );
-				if (async_status == ASYNC_CHANGE_FD && fd)
-					*fd=return_code;
+					(fd, &faked_req, ctx->async.resume_param );
+				if (async_status == ASYNC_CHANGE_FD)
+					fd=return_code;
 			} while(async_status==ASYNC_CONTINUE||async_status==ASYNC_CHANGE_FD);
 			goto route;
 		}
@@ -168,14 +165,14 @@ int t_resume_async(int *fd, void *param)
 		goto restore;
 	}
 
-	if (fd) {
+	if (valid_async_fd(fd)) {
 		/* remove from reactor, we are done */
-		reactor_del_reader( *fd, -1, IO_FD_CLOSING);
+		reactor_del_reader(fd, -1, IO_FD_CLOSING);
 	}
 
 route:
-	if (async_status == ASYNC_DONE_CLOSE_FD && fd)
-		close(*fd);
+	if (async_status == ASYNC_DONE_CLOSE_FD && valid_async_fd(fd))
+		close(fd);
 
 	/* run the resume_route (some type as the original one) */
 	swap_route_type(route, ctx->route_type);

modules/tm/async.h

@@ -34,7 +34,7 @@
  */
 int t_handle_async(struct sip_msg *msg, struct action* a , int resume_route);
 
-int t_resume_async(int *fd, void *param);
+int t_resume_async(int fd, void *param);
 
 
 #endif

net/net_tcp_proc.c

@@ -130,13 +130,13 @@ inline static int handle_io(struct fd_map* fm, int idx,int event_type)
 			handle_timer_job();
 			break;
 		case F_SCRIPT_ASYNC:
-			async_script_resume_f( &fm->fd, fm->data);
+			async_script_resume_f( fm->fd, fm->data);
 			return 0;
 		case F_FD_ASYNC:
-			async_fd_resume( &fm->fd, fm->data);
+			async_fd_resume( fm->fd, fm->data);
 			return 0;
 		case F_LAUNCH_ASYNC:
-			async_launch_resume( &fm->fd, fm->data);
+			async_launch_resume( fm->fd, fm->data);
 			return 0;
 		case F_IPC:
 			ipc_handle_job();

net/net_udp.c

@@ -264,13 +264,13 @@ inline static int handle_io(struct fd_map* fm, int idx,int event_type)
 			handle_timer_job();
 			return 0;
 		case F_SCRIPT_ASYNC:
-			async_script_resume_f( &fm->fd, fm->data);
+			async_script_resume_f( fm->fd, fm->data);
 			return 0;
 		case F_FD_ASYNC:
-			async_fd_resume( &fm->fd, fm->data);
+			async_fd_resume( fm->fd, fm->data);
 			return 0;
 		case F_LAUNCH_ASYNC:
-			async_launch_resume( &fm->fd, fm->data);
+			async_launch_resume( fm->fd, fm->data);
 			return 0;
 		case F_IPC:
 			ipc_handle_job();

timer.c

@@ -612,13 +612,13 @@ inline static int handle_io(struct fd_map* fm, int idx,int event_type)
 			handle_timer_job();
 			return 0;
 		case F_SCRIPT_ASYNC:
-			async_script_resume_f( &fm->fd, fm->data);
+			async_script_resume_f( fm->fd, fm->data);
 			return 0;
 		case F_FD_ASYNC:
-			async_fd_resume( &fm->fd, fm->data);
+			async_fd_resume( fm->fd, fm->data);
 			return 0;
 		case F_LAUNCH_ASYNC:
-			async_launch_resume( &fm->fd, fm->data);
+			async_launch_resume( fm->fd, fm->data);
 			return 0;
 		case F_IPC:
 			ipc_handle_job();