创建

client/doris_client_http.cpp

@@ -0,0 +1,374 @@
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <arpa/inet.h>
+#include <net/if.h>
+#include <unistd.h> 
+#include <stdio.h> 
+#include <stdlib.h>
+#include <assert.h>
+#include <errno.h>
+#include <sys/time.h>
+#include <time.h>
+#include <string.h>
+#include <openssl/sha.h>
+#include <openssl/md5.h>
+
+#include <MESA/MESA_prof_load.h>
+
+#include "doris_client_http.h"
+
+int32_t param_get_connected_hosts(struct doris_http_parameter *param)
+{
+	return param->connected_hosts;
+}
+int32_t param_get_failed_hosts(struct doris_http_parameter *param)
+{
+	return param->failed_hosts;
+}
+
+static int _unfold_IP_range(char* ip_range, char***ip_list, int size)
+{
+	int i=0,count=0, ret=0;
+	int range_digits[5];
+	memset(range_digits,0,sizeof(range_digits));
+	ret=sscanf(ip_range,"%d.%d.%d.%d-%d",&range_digits[0],&range_digits[1],&range_digits[2],&range_digits[3],&range_digits[4]);
+	if(ret!=4&&ret!=5)
+	{
+		return 0;
+	}
+	if(ret==4&&range_digits[4]==0)
+	{
+		range_digits[4]=range_digits[3];
+	}
+	for(i=0;i<5;i++)
+	{
+		if(range_digits[i]<0||range_digits[i]>255)
+		{
+			return 0;
+		}
+	}
+	count=range_digits[4]-range_digits[3]+1;
+	*ip_list=(char**)realloc(*ip_list, sizeof(char*)*(size+count));
+	for(i=0;i<count;i++)
+	{
+		(*ip_list)[size+i]=(char*)malloc(64);
+		snprintf((*ip_list)[size+i],64,"%d.%d.%d.%d",range_digits[0],range_digits[1],range_digits[2],range_digits[3]+i);
+	}
+	return count;
+}
+
+static int unfold_IP_range(const char* ip_range, char***ip_list)
+{
+	char *token=NULL,*sub_token=NULL,*saveptr;
+	char *buffer=(char*)calloc(sizeof(char),strlen(ip_range)+1);
+	int count=0;
+	strcpy(buffer,ip_range);
+	for (token = buffer; ; token= NULL)
+	{
+		sub_token= strtok_r(token,";", &saveptr);
+		if (sub_token == NULL)
+			break;
+		count+=_unfold_IP_range(sub_token, ip_list,count);	
+	}
+	free(buffer);
+	return count;
+}
+
+static int decode_one_specific_group_ip(const char *iplist, struct dst_ipaddr_group *dstaddr)
+{
+	int i, ipaddr_num;
+	char **balance_iplist=NULL;
+	
+	ipaddr_num = unfold_IP_range(iplist, &balance_iplist);
+	if(ipaddr_num == 0)
+	{
+		free(dstaddr);
+		return -1;
+	}
+	dstaddr->dstaddr_num = ipaddr_num;
+	dstaddr->dstaddrs = (u_int32_t *)calloc(1, sizeof(u_int32_t)*ipaddr_num);
+	for(i=0; i<ipaddr_num; i++)
+	{
+		if(inet_pton(AF_INET, balance_iplist[i], (void *)&dstaddr->dstaddrs[i]) != 1)
+		{
+			free(dstaddr->dstaddrs);
+			return -1;
+		}
+		free(balance_iplist[i]);
+	}
+	free(balance_iplist);
+	return 0;
+}
+
+static int32_t load_and_init_server_group(char *dst_ipaddr, struct dst_ipaddr_group *dstaddrs, void *runtime_log)
+{
+	u_int32_t intval;
+
+	intval = strlen(dst_ipaddr);
+	if(dst_ipaddr[intval-1] != ';')
+	{
+		dst_ipaddr[intval] = ';';
+		dst_ipaddr[intval+1] = '\0';
+	}
+	if(decode_one_specific_group_ip(dst_ipaddr, dstaddrs) < 0)
+	{
+		return -2;
+	}
+	return 0;
+}
+
+static void conhash_delay_destroy_timer_cb(int fd, short kind, void *userp)
+{
+	struct time_event *delay_event=(struct time_event *)userp;
+
+	conhash_instance_free((struct consistent_hash *)delay_event->data);
+	free(delay_event);
+}
+
+static void load_balance_common_timer_start(struct event *time_event)
+{
+	struct timeval tv;
+
+	tv.tv_sec = 2;
+	tv.tv_usec = 0;
+	evtimer_add(time_event, &tv);
+}
+
+static void conhash_handle_delay_destroy(struct event_base *evbase, struct consistent_hash *conhash)
+{
+	struct time_event *delay_event;
+
+	delay_event = (struct time_event *)malloc(sizeof(struct time_event));
+	delay_event->data = conhash;
+	evtimer_assign(&delay_event->timer_event, evbase, conhash_delay_destroy_timer_cb, delay_event);
+	load_balance_common_timer_start(&delay_event->timer_event);
+}
+
+static void conhash_insert_dest_host(struct dst_host_cnn_balance *balance)
+{
+	struct conhash_bucket bucket;
+	struct consistent_hash *tmphash, *newhash=NULL;
+	enum CONHASH_ERRCODE code;
+	
+	bucket.bucket_id = balance->dstip;
+	bucket.point_num = DEFAULT_HOST_CAPACITY * LOAD_BALANC_VIRT_TIMES;;
+	bucket.tag = NULL;
+
+	newhash = conhash_instance_copy(balance->param->conhash);
+	code = conhash_insert_bucket(newhash, &bucket);
+	assert(code == CONHASH_OK);
+	
+	tmphash = balance->param->conhash;
+	balance->param->conhash = newhash;
+	conhash_handle_delay_destroy(balance->param->evbase, tmphash);
+}
+
+static void conhash_remove_dest_host(struct dst_host_cnn_balance *balance)
+{
+	struct consistent_hash *tmphash, *newhash=NULL;
+	enum CONHASH_ERRCODE code;
+	
+	newhash = conhash_instance_copy(balance->param->conhash);
+	code = conhash_remove_bucket(newhash, balance->dstip, NULL);
+	assert(code == CONHASH_OK || code==CONHASH_BUCKET_NOT_FOUND);
+	
+	tmphash = balance->param->conhash;
+	balance->param->conhash = newhash;
+	conhash_handle_delay_destroy(balance->param->evbase, tmphash);
+}
+
+static void client_bufferevent_error_cb(struct bufferevent *bev, short event, void *arg)  
+{
+	struct dst_host_cnn_balance *balance = (struct dst_host_cnn_balance *)arg;
+	const char *errtype;
+	
+	if(event & BEV_EVENT_CONNECTED) 
+	{  
+		errtype = "connected";
+		balance->connection_status = TCP_STATUS_CONNECTED;
+		balance->param->connected_hosts += 1; //<2F><><EFBFBD>й<EFBFBD><D0B9><EFBFBD><EFBFBD>߳<EFBFBD>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>̽<EFBFBD><CCBD>
+		if(balance->connect_failed)
+		{
+			balance->connect_failed = false;
+			balance->param->failed_hosts -= 1;
+		}
+		conhash_insert_dest_host(balance);
+		assert(balance->param->connected_hosts > 0);
+		assert(balance->param->failed_hosts >= 0);
+	}
+	else
+	{
+		if (event & BEV_EVENT_TIMEOUT) {  
+			errtype = "Timed out";  
+		}
+		else if (event & BEV_EVENT_EOF) {  
+			errtype = "disconnected";  
+		}
+		else if (event & BEV_EVENT_ERROR) {
+			errtype = "some other error"; 
+		}
+		else {
+			errtype = "unkonwn error";
+		}
+		bufferevent_free(bev);
+		balance->bev = NULL;
+
+		if(balance->connection_status == TCP_STATUS_CONNECTED)
+		{
+			balance->param->connected_hosts -= 1;
+		}
+		if(!balance->connect_failed)
+		{
+			balance->connect_failed = true;
+			balance->param->failed_hosts += 1;
+		}
+		balance->connection_status = TCP_STATUS_DISCONNECT;
+	
+		load_balance_common_timer_start(&balance->timer_detect);
+		conhash_remove_dest_host(balance);
+
+		assert(balance->param->connected_hosts >= 0);
+		assert(balance->param->failed_hosts > 0);
+	}
+	
+	MESA_HANDLE_RUNTIME_LOGV2(balance->param->runtime_log, RLOG_LV_INFO, "connection event: %s, addr: %s", errtype, balance->srvaddr);
+}
+
+int do_bufferevent_connection(struct doris_http_parameter *param, struct dst_host_cnn_balance *balance, struct sockaddr *server_addr)
+{
+	if(balance->connection_status==TCP_STATUS_CONNECTING)
+	{
+		return 0;
+	}
+
+	if(NULL == (balance->bev = bufferevent_socket_new(param->evbase, -1, BEV_OPT_CLOSE_ON_FREE)))
+	{
+		assert(0);return -1;
+	}
+	
+	if(bufferevent_socket_connect(balance->bev, server_addr, sizeof(struct sockaddr_in)))
+	{
+		bufferevent_free(balance->bev);
+		balance->bev = NULL;
+		MESA_HANDLE_RUNTIME_LOGV2(param->runtime_log, RLOG_LV_FATAL, "bufferevent_socket_connect error: %s", strerror(errno));
+		assert(0);return -2;
+	}
+	balance->connection_status = TCP_STATUS_CONNECTING;
+	bufferevent_setcb(balance->bev, NULL, NULL, client_bufferevent_error_cb, balance);
+	bufferevent_setwatermark(balance->bev, EV_WRITE, 100*1024*1024UL, 0);
+	bufferevent_enable(balance->bev, EV_READ|EV_WRITE|EV_PERSIST);
+	return 0;
+}
+
+void param_connection_detect_timer_cb(int fd, short kind, void *userp)
+{
+	struct dst_host_cnn_balance *balance = (struct dst_host_cnn_balance *)userp;
+
+	if(do_bufferevent_connection(balance->param, balance, (struct sockaddr*)&balance->addr))
+	{
+		assert(0);
+	}
+}
+
+static int32_t doris_launch_group_connection(struct doris_http_parameter *param, struct event_base* evbase)
+{
+	char ipaddr[64];
+
+	param->balance = (struct dst_host_cnn_balance *)calloc(1, sizeof(struct dst_host_cnn_balance)*param->ipgroup.dstaddr_num);
+	
+	for(u_int32_t i=0; i<param->ipgroup.dstaddr_num; i++) //<2F><><EFBFBD><EFBFBD>
+	{
+		param->balance[i].dstip = param->ipgroup.dstaddrs[i];
+		param->balance[i].addr.sin_family = AF_INET;
+		param->balance[i].addr.sin_port = htons(param->manage_port);
+		param->balance[i].addr.sin_addr.s_addr = param->ipgroup.dstaddrs[i];
+		param->balance[i].connection_status = TCP_STATUS_IDLE;
+		param->balance[i].param = param;
+	
+		inet_ntop(AF_INET, &param->ipgroup.dstaddrs[i], ipaddr, 64);
+		snprintf(param->balance[i].srvaddr, 64, "%s:%u", ipaddr, param->server_port);
+
+		evtimer_assign(&param->balance[i].timer_detect, evbase, param_connection_detect_timer_cb, &param->balance[i]);
+	
+		if(do_bufferevent_connection(param, &param->balance[i], (struct sockaddr *)&param->balance[i].addr))
+		{
+			MESA_HANDLE_RUNTIME_LOGV2(param->runtime_log, RLOG_LV_FATAL, "do_bufferevent_connection error: %s.", strerror(errno));
+			return -1;
+		}
+	}
+	return 0;
+}
+
+struct doris_http_parameter *doris_http_parameter_new(const char* profile_path, const char* section, struct event_base* evbase, void *runtime_log)
+{
+	struct doris_http_parameter *param;
+	u_int32_t intval;
+	char dst_ipaddr[8192];
+
+	param = (struct doris_http_parameter *)calloc(1, sizeof(struct doris_http_parameter));
+	param->runtime_log = runtime_log;
+	param->evbase = evbase;
+
+	//multi curl
+	MESA_load_profile_uint_def(profile_path, section, "max_connection_per_host", &intval, 1);
+	param->maximum_host_cnns = intval;
+	MESA_load_profile_uint_def(profile_path, section, "max_cnnt_pipeline_num", &intval, 20);
+	param->maximum_pipelines = intval;
+	MESA_load_profile_uint_def(profile_path, section, "max_curl_transfer_timeout_s", &intval, 0);
+	param->transfer_timeout = intval;
+	MESA_load_profile_int_def(profile_path, section, "https_connection_on", &param->ssl_connection, 0);
+	MESA_load_profile_uint_def(profile_path, section, "max_curl_session_num", &param->max_http_sessions, 30);
+
+	//Server
+	if(MESA_load_profile_uint_def(profile_path, section, "http_server_listen_port", &param->server_port, 9898) < 0)
+	{
+		free(param);
+		MESA_HANDLE_RUNTIME_LOGV2(runtime_log, RLOG_LV_FATAL, "Load config %s [%s] http_server_listen_port not found.", profile_path, section);
+		return NULL;
+	}
+	MESA_load_profile_uint_def(profile_path, section, "http_server_manage_port", &param->manage_port, 2233);
+
+	if(MESA_load_profile_string_nodef(profile_path, section, "http_server_ip_list", dst_ipaddr, 8192) < 0)
+	{
+		free(param);
+		MESA_HANDLE_RUNTIME_LOGV2(runtime_log, RLOG_LV_FATAL, "Load config %s [%s] http_server_ip_list not found.", profile_path, section);
+		return NULL;
+	}
+	
+	if(load_and_init_server_group(dst_ipaddr, &param->ipgroup, runtime_log))
+	{
+		MESA_HANDLE_RUNTIME_LOGV2(runtime_log, RLOG_LV_FATAL, "Decode %s [%s] http_server_ip_list format error: %s", profile_path, section, dst_ipaddr);
+		assert(0);return NULL;
+	}
+	param->conhash = conhash_instance_new(NULL, 0);;
+	if(doris_launch_group_connection(param, evbase))
+	{
+		assert(0);return NULL;
+	}
+	return param;
+}
+
+struct doris_http_instance *doris_http_instance_new(struct doris_http_parameter *param, struct event_base* evbase, void *runtimelog)
+{
+	struct doris_http_instance *instance;
+	struct doris_curl_multihd *multihd;
+
+	instance = (struct doris_http_instance *)calloc(1, sizeof(struct doris_http_instance));
+	instance->runtime_log = runtimelog;	
+	instance->evbase = evbase;
+	instance->param = param;
+	
+	//Ϊÿһ<C3BF><D2BB>minio IP<49><50><EFBFBD><EFBFBD><EFBFBD><EFBFBD>һ<EFBFBD><D2BB>multi_handle
+	instance->server_hosts = new map<u_int32_t, doris_curl_multihd*>;
+	for(u_int32_t i=0; i<param->ipgroup.dstaddr_num; i++)
+	{
+		multihd = doris_initialize_multihd_for_host(instance, &param->balance[i]);
+		instance->server_hosts->insert(make_pair(param->ipgroup.dstaddrs[i], multihd));
+	}
+	return instance;
+}
+