New source for sockets

Author: davidtrihy-genesys

modules/rest_client/rest_cb.c

@@ -109,6 +109,16 @@ size_t header_func(char *ptr, size_t size, size_t nmemb, void *userdata)
 	return len;
 }
 
+int timer_cb(CURLM *multi_handle, long timeout_ms, void *cbp)
+{
+	LM_DBG("multi_handle timer called %d\n", timeout_ms);
+	long *p = (long*) cbp;
+
+	*p = timeout_ms;
+
+  	return 0;
+}
+
 int prereq_callback(void *cbp,
                            char *conn_primary_ip,
                            char *conn_local_ip,

modules/rest_client/rest_cb.h

@@ -47,6 +47,7 @@ enum curl_status {
 
 size_t write_func(char *ptr, size_t size, size_t nmemb, void *userdata);
 size_t header_func(char *ptr, size_t size, size_t nmemb, void *userdata);
+int timer_cb(CURLM *multi_handle, long timeout_ms, void *cbp);
 int prereq_callback(void *cbp,  char *conn_primary_ip,  char *conn_local_ip, int conn_primary_port, int conn_local_port);
 
 #endif /* _REST_CB_H_ */

modules/rest_client/rest_client.c

@@ -28,6 +28,8 @@
 #include <stdlib.h>
 #include <curl/curl.h>
 
+#include <sys/resource.h>
+
 #include "../../async.h"
 #include "../../sr_module.h"
 #include "../../dprint.h"
@@ -397,7 +399,7 @@ static int child_init(int rank)
 		return 0;
 	}
 
-	if (init_process_limits() != 0) {
+	if (init_process_limits(lim.rlim_cur) != 0) {
 		LM_WARN("Could not set file descriptor limits\n");
 		return 0;
 	}

modules/rest_client/rest_client.h

@@ -26,8 +26,6 @@
 #ifndef _REST_CLIENT_
 #define _REST_CLIENT_
 
-#include <sys/resource.h>
-
 enum tr_rest_subtype {
 	TR_REST_ESCAPE, TR_REST_UNESCAPE
 };
@@ -40,6 +38,5 @@ typedef struct _preconnect_urls {
 
 extern int enable_expect_100;
 extern unsigned int max_transfer_size;
-extern struct rlimit lim;
 
 #endif /* _REST_CLIENT_ */

modules/rest_client/rest_methods.c

@@ -41,6 +41,7 @@
 #include "rest_client.h"
 #include "rest_methods.h"
 #include "rest_cb.h"
+#include "rest_sockets.h"
 
 #define REST_CORRELATION_COOKIE "RESTCORR"
 
@@ -70,10 +71,12 @@ extern int _async_resume_retr_itv;
 #define MAX_HOST_LENGTH 128
 
 /* file descriptor limits */
-#define WORD_SIZE_BITS 64
-#define WORD_SIZE_BYTES 16
-#define BYTE_LEN 8
-extern struct rlimit lim;
+// #define WORD_SIZE_BITS 64
+// #define WORD_SIZE_BYTES 16
+// #define BYTE_LEN 8
+// extern struct rlimit lim;
+
+static CURLSH *curl_share;
 
 #define REST_TRACE_API_MODULE "proto_hep"
 extern int rest_proto_id;
@@ -459,168 +462,6 @@ static int init_transfer(CURL *handle, char *url, unsigned long timeout_s)
 	return -1;
 }
 
-typedef struct _file_descriptors {
-	unsigned char *tracked_socks;
-	int max_fd_index;
-} file_descriptors;
-
-static CURLSH *curl_share = NULL;
-static file_descriptors fds;
-size_t aligned_bitset_len;
-int total_chunks;
-
-int init_process_limits(void) {
-	aligned_bitset_len = (((lim.rlim_cur + WORD_SIZE_BITS - 1) / WORD_SIZE_BITS) * WORD_SIZE_BITS) / WORD_SIZE_BITS;
-
-	fds.tracked_socks = (unsigned char*) pkg_malloc(aligned_bitset_len);
-	fds.max_fd_index = 0;
-	total_chunks = aligned_bitset_len / sizeof(uint64_t);
-
-	if (fds.tracked_socks == NULL) {
-		return -1;
-	}
-
-	return 0;
-}
-
-static inline int get_max_fd(file_descriptors *fds) {
-	uint64_t *socket_bitmask;
-	uint64_t sockets;
-	int start_index, bitmask_index;
-
-	bitmask_index = fds->max_fd_index -1;
-
-	if (bitmask_index < 0) {
-		return -2;
-	}
-
-	socket_bitmask = (uint64_t*) fds->tracked_socks;
-	sockets = socket_bitmask[bitmask_index];
-
-	if (!sockets) {
-		fds->max_fd_index--;
-		return -1;
-	}
-
-	start_index = (fds->max_fd_index * (WORD_SIZE_BITS)) - 1;
-	
-    return start_index - __builtin_clzll(sockets);
-}
-
-/*
- * Adds the socket to a bitmask and calculates index of that socket in the bitmask
- * eg. s = 63 then index is 1 and if s = 65 then index is 2, in increments of 64 bits
- * The index is the max index to check when actioning on the sockets
- */
-static inline void add_sock(file_descriptors *fds, int s) {
-	int sock_index = (s >> 6) + 1;
-
-	if (sock_index > fds->max_fd_index) {
-		fds->max_fd_index = sock_index;
-	}
-
-    fds->tracked_socks[s / BYTE_LEN] |= (1 << (s % BYTE_LEN));
-}
-
-/*
- * Removes the socket from the bitmask and then checks the max bitmask for any setbits
- * If no bits are set then the max index is decremented regardless if the socket removed was in that chunk
- */
-static inline void remove_sock(file_descriptors *fds, int s) {
-	uint64_t sockets;
-	uint64_t *socket_bitmask;
-	int bitmask_index;
-
-	fds->tracked_socks[s / BYTE_LEN] &= ~(1 << (s % BYTE_LEN));
-
-	bitmask_index = fds->max_fd_index -1;
-    socket_bitmask = (uint64_t*) fds->tracked_socks;
-
-	if (bitmask_index > 0) {
-		memcpy(&sockets, socket_bitmask + fds->max_fd_index - 1, sizeof(sockets));
-
-		if (!sockets) {
-			fds->max_fd_index--;
-		}
-	}
-}
-
-static int socket_action_cb(CURL *e, curl_socket_t s, int event, void *cbp, void *sockp)
-{
-	LM_DBG("called for socket %d status %d\n", s, event);
-	file_descriptors *fds = (file_descriptors*) cbp;
-
-	if (event != CURL_POLL_REMOVE) {
-		add_sock(fds, s);
-	} else if (event == CURL_POLL_REMOVE) {
-		remove_sock(fds, s);
-	}
-
-  	return 0;
-}
-
-static int timer_cb(CURLM *multi_handle, long timeout_ms, void *cbp)
-{
-	LM_DBG("multi_handle timer called %d\n", timeout_ms);
-	long *p = (long*) cbp;
-
-	*p = timeout_ms;
-
-  	return 0;
-}
-
-static int start_multi_socket(CURLM *multi_handle) {
-	CURLMcode mrc;
-	int running;
-
-	memset(fds.tracked_socks, 0, aligned_bitset_len);
-	fds.max_fd_index = 0;
-	mrc = curl_multi_socket_action(multi_handle, CURL_SOCKET_TIMEOUT, 0, &running);
-
-	if (mrc != CURLM_OK) {
-		LM_ERR("curl_multi_socket_action: %s\n", curl_multi_strerror(mrc));
-		return -1;
-	}
-
-	return running;
-}
-
-static int run_multi_socket(CURLM *multi_handle) {
-	CURLMcode mrc;
-	int running;
-	uint64_t sockets;
-	uint64_t *socket_bitmask;
-	
-	socket_bitmask = (uint64_t*) fds.tracked_socks;
-
-	if (fds.max_fd_index > 0) {
-		memcpy(&sockets, socket_bitmask + fds.max_fd_index - 1, sizeof(sockets));
-    
-    	if (!sockets) {
-    		fds.max_fd_index--;
-    	}   
-	}
-
-	for (int i = 0; i < fds.max_fd_index; i++) {
-		memcpy(&sockets, socket_bitmask + i, sizeof(sockets));
-		
-		while (sockets) {
-			int curl_s = (i * WORD_SIZE_BITS) + __builtin_ctzll(sockets);
-			LM_DBG("Action on socket %d\n", curl_s);
-		
-			mrc = curl_multi_socket_action(multi_handle, curl_s, 0, &running);
-			if (mrc != CURLM_OK) {
-				LM_ERR("curl_multi_socket_action: %s\n", curl_multi_strerror(mrc));
-				return -1;
-			}
-
-			sockets &= sockets - 1;
-		}
-	}
-
-	return running;
-}
-
 int connect_only(preconnect_urls *precon_urls, int total_cons) {
 	CURLcode rc;
 	CURLMcode mrc;
@@ -680,13 +521,14 @@ int connect_only(preconnect_urls *precon_urls, int total_cons) {
 
 	busy_wait = connect_poll_interval;
 
+	if (setsocket_callback(multi_handle) != 0) {
+		goto cleanup;
+	}
+
 	// TODO Need to expose these settings for connect and async req
 	w_curl_multi_setopt(multi_handle, CURLMOPT_MAX_HOST_CONNECTIONS, (long) num_of_connections);
 	w_curl_multi_setopt(multi_handle, CURLMOPT_MAXCONNECTS, (long) num_of_connections);
 
-	w_curl_multi_setopt(multi_handle, CURLMOPT_SOCKETFUNCTION, socket_action_cb);
-	w_curl_multi_setopt(multi_handle, CURLMOPT_SOCKETDATA, &fds);
-
 	w_curl_multi_setopt(multi_handle, CURLMOPT_TIMERFUNCTION, timer_cb);
   	w_curl_multi_setopt(multi_handle, CURLMOPT_TIMERDATA, &timer);
 
@@ -1192,6 +1034,10 @@ int start_async_http_req(struct sip_msg *msg, enum rest_client_method method,
 	multi_handle = multi_list->multi_handle;
 	curl_multi_add_handle(multi_handle, handle);
 
+	if (setsocket_callback(multi_handle) != 0) {
+		goto cleanup;
+	}
+
 	w_curl_multi_setopt(multi_handle, CURLMOPT_MAX_HOST_CONNECTIONS, (long) max_host_connection);
 	w_curl_multi_setopt(multi_handle, CURLMOPT_MAXCONNECTS, (long) max_async_transfers);
 
@@ -1203,9 +1049,6 @@ int start_async_http_req(struct sip_msg *msg, enum rest_client_method method,
 	timeout = connect_timeout;
 	busy_wait = connect_poll_interval;
 
-	w_curl_multi_setopt(multi_handle, CURLMOPT_SOCKETFUNCTION, socket_action_cb);
-	w_curl_multi_setopt(multi_handle, CURLMOPT_SOCKETDATA, &fds);
-
 	w_curl_multi_setopt(multi_handle, CURLMOPT_TIMERFUNCTION, timer_cb);
   	w_curl_multi_setopt(multi_handle, CURLMOPT_TIMERDATA, &timer);
 
@@ -1236,7 +1079,7 @@ int start_async_http_req(struct sip_msg *msg, enum rest_client_method method,
 	async_parm->handle = handle;
 	async_parm->multi_list = multi_list;
 	header_list = NULL;
-	*out_fd = get_max_fd(&fds); // Running only one socket at a time so it's always the max
+	*out_fd = get_max_fd(); // Running only one socket at a time so it's always the max
 	return RCL_OK;
 
 error:

modules/rest_client/rest_methods.h

@@ -145,7 +145,6 @@ enum async_ret_code time_out_async_http_req(int fd, struct sip_msg *msg, void *_
 int rest_append_hf_method(struct sip_msg *msg, str *hfv);
 int rest_init_client_tls(struct sip_msg *msg, str *tls_client_dom);
 int connect_only(preconnect_urls *precon_urls, int total_cons);
-int init_process_limits(void);
 
 #endif /* _REST_METHODS_ */