tango-tfe/cache/tango_cache_client.cpp

#include <sys/types.h>
#include <sys/stat.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 <MESA/MESA_prof_load.h>

#include "tango_cache_client_in.h"
#include "tango_cache_transfer.h"
#include "tango_cache_tools.h"

int TANGO_CACHE_VERSION_20180910=0;

void caculate_sha256(const char *data, unsigned long len, char *result, u_int32_t size)
{
	SHA256_CTX c;
	unsigned char sha256[128];

	SHA256_Init(&c);
	SHA256_Update(&c, data, len);
	SHA256_Final(sha256, &c);
	
	for(u_int32_t i=0; i<32 && i<size/2; i++)
	{
		sprintf(result + i*2, "%02x", sha256[i]);
	}
}

enum CACHE_ERR_CODE tango_cache_get_last_error(const struct tango_cache_ctx *ctx)
{
	return ctx->error_code;
}
enum CACHE_ERR_CODE tango_cache_ctx_error(const struct tango_cache_instance *instance)
{
	return instance->error_code;
}

void tango_cache_get_statistics(const struct tango_cache_instance *instance, struct cache_statistics *out)
{
	out->get_recv_num = instance->statistic.get_recv_num;
	out->get_succ_num = instance->statistic.get_succ_num;
	out->get_fail_num = instance->statistic.get_fail_num;
	out->put_recv_num = instance->statistic.put_recv_num;
	out->put_succ_num = instance->statistic.put_succ_num;
	out->put_fail_num = instance->statistic.put_fail_num;
	out->session_num  = instance->statistic.session_num;
	out->memory_used  = instance->statistic.memory_used;
}

const char *tango_cache_get_object_key(struct tango_cache_ctx *ctx)
{
	return ctx->file_key;
}

static inline void update_statistics(enum CACHE_REQUEST_METHOD method, bool fail_state, struct cache_statistics *statistic)
{
	if(method == CACHE_REQUEST_PUT)
	{
		if(fail_state)
		{
			statistic->put_fail_num += 1;
		}
		else
		{
			statistic->put_succ_num += 1;
		}
	}
	else
	{
		if(fail_state)
		{
			statistic->get_fail_num += 1;
		}
		else
		{
			statistic->get_succ_num += 1;
		}
	}
}

void response_buffer_destroy(struct easy_string *estr)
{
	if(estr->buff != NULL)
	{
		free(estr->buff);
		estr->buff = NULL;
 		estr->len = estr->size = 0;
	}
}

void buffer_cache_destroy(struct cache_buffer *cache, struct tango_cache_instance *instance)
{
	if(cache->buf)
	{
		free(cache->buf);
	}
	instance->statistic.memory_used -= instance->block_len;
	free(cache);
}

static inline struct cache_buffer *buffer_cache_new(struct tango_cache_instance *instance)
{
	struct cache_buffer *cache = (struct cache_buffer *)malloc(sizeof(struct cache_buffer));
	
	cache->buf = (char *)malloc(instance->block_len);
	cache->len = 0;
	cache->off = 0;
	instance->statistic.memory_used += instance->block_len;
	return cache;
}

void buffer_cache_list_destroy(struct buffer_cache_list *list, struct tango_cache_ctx *ctx)
{
	struct cache_buffer *pnode;
	
	TAILQ_FOREACH(pnode, &list->cache_list, node)
	{
		TAILQ_REMOVE(&list->cache_list, pnode, node);
		buffer_cache_destroy(pnode, ctx->instance);
	}
	if(list->cache_cur)
	{
		buffer_cache_destroy(list->cache_cur, ctx->instance);
	}
	if(list->curl)
	{
		curl_multi_remove_handle(ctx->instance->multi_hd, list->curl);
		curl_easy_cleanup(list->curl);
	}
	if(list->etag)
	{
		free(list->etag);
	}
	free(list);
}

static struct buffer_cache_list *buffer_cache_list_new(struct tango_cache_ctx *ctx)
{
	struct buffer_cache_list *list = (struct buffer_cache_list *)calloc(1, sizeof(struct buffer_cache_list));

	list->part_number = ++ctx->part_index;
	list->cache_cur = buffer_cache_new(ctx->instance);
	list->ctx = ctx;
	TAILQ_INIT(&list->cache_list);
	return list;
}

void tango_cache_ctx_destroy(struct tango_cache_ctx *ctx)
{
	if(ctx->curl != NULL)
	{
		curl_multi_remove_handle(ctx->instance->multi_hd, ctx->curl);
		curl_easy_cleanup(ctx->curl);
	}	
	response_buffer_destroy(&ctx->response);
	
	if(ctx->method == CACHE_REQUEST_PUT)
	{
		struct buffer_cache_list *list;

		if(ctx->uploadID != NULL) free(ctx->uploadID);
		if(ctx->combine_xml != NULL) free(ctx->combine_xml);
		if(ctx->headers != NULL) curl_slist_free_all(ctx->headers);
		if(ctx->headers_puts != NULL) curl_slist_free_all(ctx->headers_puts);		
		if(ctx->list_cur != NULL) buffer_cache_list_destroy(ctx->list_cur, ctx);

		TAILQ_FOREACH(list, &ctx->cache_head, node)
		{
			TAILQ_REMOVE(&ctx->cache_head, list, node);
			buffer_cache_list_destroy(list, ctx);
		}
		if(ctx->future != NULL)
		{
			if(ctx->fail_state)
			{
				promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, ctx->error);
			}			
			else
			{
				promise_success(future_to_promise(ctx->future), NULL);
			}
		}
	}
	update_statistics(ctx->method, ctx->fail_state, &ctx->instance->statistic);
	free(ctx);
}

void tango_cache_update_end(struct tango_cache_ctx *ctx)
{
	cache_kick_upload_minio_end(ctx);
}

//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>buffer<65><72><EFBFBD><EFBFBD><EFBFBD>ڴ<EFBFBD><DAB4><EFBFBD><EFBFBD><EFBFBD>
void tango_cache_kick_new_buffer(struct tango_cache_ctx *ctx, const char *data, size_t size)
{
	struct buffer_cache_list *list = ctx->list_cur;
	
	if(list->length >= ctx->instance->upload_block_size)
	{
		TAILQ_INSERT_TAIL(&ctx->cache_head, list, node);
		cache_kick_upload_minio_multipart(ctx, list);
		list = buffer_cache_list_new(ctx);
		ctx->list_cur = list;
	}
	else
	{
		list->cache_cur = buffer_cache_new(ctx->instance);
	}
	memcpy(list->cache_cur->buf, data, size);
	list->cache_cur->len = size;
}

int tango_cache_update_frag(struct tango_cache_ctx *ctx, const char *data, size_t size)
{
	struct buffer_cache_list *list_cur = ctx->list_cur;
	struct cache_buffer *cache_cur = list_cur->cache_cur;
	u_int32_t copy_len;

	if(ctx->fail_state)
	{
		return -1;
	}

	//ADD data to Cache Buffer
	if(cache_cur->len + size > ctx->instance->block_len)
	{
		copy_len = ctx->instance->block_len - cache_cur->len;
		memcpy(cache_cur->buf+cache_cur->len, data, copy_len);
		cache_cur->len = ctx->instance->block_len;
		TAILQ_INSERT_TAIL(&list_cur->cache_list, cache_cur, node);
		list_cur->length += ctx->instance->block_len;
		list_cur->cache_cur = NULL;
		size -= copy_len;
		data += copy_len;

		while(size >= ctx->instance->block_len)
		{
			tango_cache_kick_new_buffer(ctx, data, ctx->instance->block_len); //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>buffer<65><72>
			list_cur = ctx->list_cur;
			TAILQ_INSERT_TAIL(&list_cur->cache_list, list_cur->cache_cur, node);
			list_cur->length += ctx->instance->block_len;
			list_cur->cache_cur = NULL;

			size -= ctx->instance->block_len;
			data += ctx->instance->block_len;
		}
		tango_cache_kick_new_buffer(ctx, data, size);
	}
	else
	{
		memcpy(cache_cur->buf+cache_cur->len, data, size);
		cache_cur->len += size;
	}
	return 0;
}

struct tango_cache_ctx *tango_cache_update_prepare(struct tango_cache_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
	struct tango_cache_ctx *ctx;	
	char buffer[256]={0};
	int other_len;

	if((u_int64_t)instance->statistic.memory_used >= instance->cache_limit_size)
	{
		instance->error_code = CACHE_OUTOF_MEMORY;
		return NULL;
	}
	instance->statistic.put_recv_num += 1;
	instance->error_code = CACHE_OK;

	ctx = (struct tango_cache_ctx *)calloc(1, sizeof(struct tango_cache_ctx));
	ctx->instance = instance;
	ctx->future = future;
	ctx->method = CACHE_REQUEST_PUT;

	if(instance->hash_object_key)
	{
		caculate_sha256(meta->url, strlen(meta->url), buffer, 72);
		snprintf(ctx->file_key, 72, "%c%c/%c%c/%s", buffer[0], buffer[1], buffer[2], buffer[3], buffer+4);
	}
	else
	{
		snprintf(ctx->file_key, 72, "%s", meta->url);
	}

	//Expires<65>ֶΣ<D6B6><CEA3><EFBFBD><EFBFBD>ڻ<EFBFBD><DABB><EFBFBD><EFBFBD>ڲ<EFBFBD><DAB2>ж<EFBFBD><D0B6><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƿ<EFBFBD><C7B7><EFBFBD>ʱ
	ctx->relative_age = (meta->relative_ttl==0||meta->relative_ttl>instance->relative_ttl)?instance->relative_ttl:meta->relative_ttl;
	if(expires_timestamp2hdr_str(time(NULL)+ctx->relative_age, buffer, 256))
	{
		ctx->headers_puts = curl_slist_append(ctx->headers_puts, buffer);
	}
	//<2F>б<EFBFBD><D0B1><EFBFBD>֧<EFBFBD>ֵı<D6B5>׼ͷ<D7BC><CDB7>
	for(int i=0; i<HDR_CONTENT_NUM; i++)
	{
		if(meta->std_hdr[i] != NULL)
		{
			ctx->headers_puts = curl_slist_append(ctx->headers_puts, meta->std_hdr[i]);
		}
	}
	//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͷ<EFBFBD><CDB7><EFBFBD><EFBFBD>GETʱ<54><CAB1>ԭ<EFBFBD><D4AD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
	if(meta->other_hdr!=NULL && (other_len = strlen(meta->other_hdr))<1535)
	{
		char *p = (char *)malloc((other_len/3 + 1)*4 + 18); //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ռ䣻18=17+1: ͷ<><CDB7>+<2B>ַ<EFBFBD><D6B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
		memcpy(p, "x-amz-meta-user: ", 17);
		Base64_EncodeBlock((unsigned char*)meta->other_hdr, other_len, (unsigned char*)p+17);
		ctx->headers_puts = curl_slist_append(ctx->headers_puts, p);
		free(p);
	}

	return ctx;
}

struct tango_cache_ctx *tango_cache_update_start(struct tango_cache_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
	struct tango_cache_ctx *ctx;	

	ctx = tango_cache_update_prepare(instance, future, meta);
	if(ctx == NULL)
	{
		return NULL;
	}

	ctx->list_cur = buffer_cache_list_new(ctx);
	TAILQ_INIT(&ctx->cache_head);
	return ctx;
}

int tango_cache_upload_once(struct tango_cache_instance *instance, struct future* future, enum PUT_MEMORY_COPY_WAY way, const char *data, size_t size, struct tango_cache_meta *meta, char *filename, size_t namelen)
{
	struct tango_cache_ctx *ctx;	

	ctx = tango_cache_update_prepare(instance, future, meta);
	if(ctx == NULL)
	{
		return -1;
	}
	ctx->way = way;
	if(filename != NULL)
	{
		snprintf(filename, namelen, "%s", ctx->file_key);
	}

	return tango_cache_upload_once_start(ctx, data, size);
}

struct tango_cache_ctx *tango_cache_fetch_prepare(struct tango_cache_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
	struct tango_cache_ctx *ctx;	
	char sha256[72]={0};

	ctx = (struct tango_cache_ctx *)calloc(1, sizeof(struct tango_cache_ctx));
	ctx->instance = instance;
	ctx->future = future;
	ctx->method = CACHE_REQUEST_GET;
	ctx->get_state = GET_STATE_START;
	ctx->expire_comes = false;
	instance->statistic.get_recv_num += 1;

	if(instance->hash_object_key)
	{
		caculate_sha256(meta->url, strlen(meta->url), sha256, 72);
		snprintf(ctx->file_key, 72, "%c%c/%c%c/%s", sha256[0], sha256[1], sha256[2], sha256[3], sha256+4);
	}
	else
	{
		snprintf(ctx->file_key, 72, "%s", meta->url);
	}

	return ctx;
}

int tango_cache_fetch(struct tango_cache_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
	return tango_cache_fetch_start(tango_cache_fetch_prepare(instance, future, meta));
}

static void check_multi_info(CURLM *multi)
{
	CURLMsg *msg;
	int msgs_left;
	struct tango_cache_ctx *ctx;
	CURL *easy;
	CURLcode res;
	long res_code;

	while((msg = curl_multi_info_read(multi, &msgs_left))) 
	{
		if(msg->msg != CURLMSG_DONE)
		{
			continue;
		}
		
		easy = msg->easy_handle;
		res = msg->data.result;
		curl_easy_getinfo(easy, CURLINFO_PRIVATE, &ctx);			
		curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &res_code);

		if(ctx->method == CACHE_REQUEST_GET)
		{
			tango_cache_curl_get_done(easy, ctx, res, res_code);
		}
		else
		{
			tango_cache_curl_put_done(easy, ctx, res, res_code);
		}
	}
}

/* Called by libevent when we get action on a multi socket */
static void libevent_socket_event_cb(int fd, short action, void *userp)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)userp; //from event_assign
	CURLMcode rc;
	int what, still_running;

	what = ((action&EV_READ)?CURL_CSELECT_IN:0) | ((action & EV_WRITE)?CURL_CSELECT_OUT:0);

	rc = curl_multi_socket_action(instance->multi_hd, fd, what, &still_running);
	instance->statistic.session_num = still_running;
	assert(rc==CURLM_OK);

	check_multi_info(instance->multi_hd);
	if(still_running<=0 && evtimer_pending(&instance->timer_event, NULL))
	{
		evtimer_del(&instance->timer_event);
	}
}

/* Called by libevent when our timeout expires */
static void libevent_timer_event_cb(int fd, short kind, void *userp)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)userp;
	CURLMcode rc;
	int still_running;

	rc = curl_multi_socket_action(instance->multi_hd, CURL_SOCKET_TIMEOUT, 0, &still_running);	
	instance->statistic.session_num = still_running;
	assert(rc==CURLM_OK);
	check_multi_info(instance->multi_hd);
}

static int curl_socket_function_cb(CURL *curl, curl_socket_t sockfd, int what, void *userp, void *sockp)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)userp; //from multi handle
	struct curl_socket_data *sockinfo = (struct curl_socket_data *)sockp; //curl_multi_assign, for socket
	int action;

	if(what == CURL_POLL_REMOVE)
	{
		if(sockinfo != NULL)
		{
			event_del(&sockinfo->sock_event);
			free(sockinfo);
		}
	}
	else
	{
		if(sockinfo == NULL)
		{
			sockinfo = (struct curl_socket_data *)calloc(1, sizeof(struct curl_socket_data));
			curl_multi_assign(instance->multi_hd, sockfd, sockinfo);
		}
		else
		{
			event_del(&sockinfo->sock_event);
		}
		
		action = (what&CURL_POLL_IN?EV_READ:0)|(what&CURL_POLL_OUT?EV_WRITE:0)|EV_PERSIST;
		event_assign(&sockinfo->sock_event, instance->evbase, sockfd, action, libevent_socket_event_cb, instance);
		event_add(&sockinfo->sock_event, NULL);
	}

	return 0;
}

static int curl_timer_function_cb(CURLM *multi, long timeout_ms, void *userp)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)userp;
	struct timeval timeout;
	CURLMcode rc;
	int still_running;

	timeout.tv_sec = timeout_ms/1000;
	timeout.tv_usec = (timeout_ms%1000)*1000;

	if(timeout_ms == 0)
	{
		//timeout_ms is 0 means we should call curl_multi_socket_action/curl_multi_perform at once.
		//To initiate the whole process(inform CURLMOPT_SOCKETFUNCTION callback) or when timeout occurs.
		rc = curl_multi_socket_action(multi, CURL_SOCKET_TIMEOUT, 0, &still_running);
		instance->statistic.session_num = still_running;
		assert(rc==CURLM_OK);
	}
	else if(timeout_ms == -1) //timeout_ms is -1 means we should delete the timer.
	{
		evtimer_del(&instance->timer_event);
	}
	else //update the timer to the new value.
	{
		evtimer_add(&instance->timer_event, &timeout);
	}
	
	return 0; //0-success; -1-error
}

static int load_local_configure(struct tango_cache_instance *instance, const char* profile_path, const char* section)
{
	u_int32_t intval;
	u_int64_t longval;

	MESA_load_profile_uint_def(profile_path, section, "MAX_CONNECTION_PER_HOST", &intval, 0);
	instance->max_cnn_host = intval;
	MESA_load_profile_uint_def(profile_path, section, "MAX_USED_MEMORY_SIZE_MB", &intval, 5120);
	longval = intval;
	instance->cache_limit_size = longval * 1024 * 1024;
	MESA_load_profile_string_def(profile_path, section, "CACHE_BUCKET_NAME", instance->bucketname, 256, "openbucket");
	MESA_load_profile_uint_def(profile_path, section, "CACHE_OBJECT_KEY_HASH_SWITCH", &instance->hash_object_key, 1);
	if(MESA_load_profile_string_nodef(profile_path, section, "MINIO_BROKERS_LIST", instance->minio_hostlist, 64) < 0)
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_FATAL, "Load config %s [%s] MINIO_BROKERS_LIST not found.\n", profile_path, section);
		return -1;
	}
	MESA_load_profile_uint_def(profile_path, section, "CACHE_BLOCK_MAX_SIZE", &instance->block_len, 8192);
	if(instance->block_len > 16777216)
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_FATAL, "Load config %s [%s] CACHE_BLOCK_MAX_SIZE too large, must smaller than 16777216(16MB).\n", profile_path, section);
		return -1;
	}
	MESA_load_profile_uint_def(profile_path, section, "CACHE_UPLOAD_BLOCK_SIZE", &instance->upload_block_size, 5242880);
	if(instance->upload_block_size < 5242880)
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_FATAL, "Load config %s [%s] CACHE_UPLOAD_BLOCK_SIZE too small, must bigger than 5242880(5MB).\n", profile_path, section);
		return -1;
	}
	MESA_load_profile_uint_def(profile_path, section, "CACHE_DEFAULT_TTL_SECOND", &intval, 3600);
	if(intval < 60)
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_FATAL, "Load config %s [%s] CACHE_DEFAULT_TTL_SECOND too small, must bigger than 60s.\n", profile_path, section);
		return -1;
	}
	instance->relative_ttl = intval;

	return 0;
}

struct tango_cache_instance *tango_cache_instance_new(struct event_base* evbase,const char* profile_path, const char* section, void *runtimelog)
{
	struct tango_cache_instance *instance;

	instance = (struct tango_cache_instance *)malloc(sizeof(struct tango_cache_instance));
	memset(instance, 0, sizeof(struct tango_cache_instance));
	
	if(load_local_configure(instance, profile_path, section))
	{
		free(instance);
		return NULL;
	}

	instance->evbase = evbase;
	instance->multi_hd = curl_multi_init();
	instance->runtime_log = runtimelog;

	curl_multi_setopt(instance->multi_hd, CURLMOPT_PIPELINING, CURLPIPE_HTTP1);
	curl_multi_setopt(instance->multi_hd, CURLMOPT_MAX_HOST_CONNECTIONS, instance->max_cnn_host); 
	curl_multi_setopt(instance->multi_hd, CURLMOPT_SOCKETFUNCTION, curl_socket_function_cb);
	curl_multi_setopt(instance->multi_hd, CURLMOPT_SOCKETDATA, instance); //curl_socket_function_cb *userp
	curl_multi_setopt(instance->multi_hd, CURLMOPT_TIMERFUNCTION, curl_timer_function_cb);
	curl_multi_setopt(instance->multi_hd, CURLMOPT_TIMERDATA, instance);

	evtimer_assign(&instance->timer_event, evbase, libevent_timer_event_cb, instance);

	return instance;
}