创建

cache/test_demo/Makefile

@@ -0,0 +1,33 @@
+CCC=g++
+
+INC_PATH=-I../src/include -I../src/include/libevent2
+CFLAGS=-Wall -g $(INC_PATH) 
+
+LIBS = -lMESA_handle_logger -lMESA_htable -lMESA_prof_load -lwiredcfg
+LIBS += -lssl -lcrypto -lcurl -levent -lxml2
+LIBS += ../src/pangu_tango_cache.a
+
+OBJS = tango_cache_test.o
+OBJS_EVBASE=cache_evbase_test.o
+
+TARGET_EXE=tango_cache_test
+TARGET_EXE_EVBASE=cache_evbase_test
+
+ALL:$(TARGET_EXE) $(TARGET_EXE_EVBASE)
+
+$(TARGET_EXE):$(OBJS)
+	$(CCC) $(LDFLAGS) $^ -o $@ $(LIBS)
+
+$(TARGET_EXE_EVBASE):$(OBJS_EVBASE)
+	$(CCC) $(LDFLAGS) $^ -o $@ $(LIBS) -lpthread
+
+.c.o:
+	$(CCC) $(CFLAGS) -c $<
+.cpp.o:
+	$(CCC) $(CFLAGS) -c $<
+
+-include $(DEPS)
+
+clean:
+	rm -rf $(OBJS) $(TARGET_EXE) $(OBJS_EVBASE) $(TARGET_EXE_EVBASE)
+

cache/test_demo/README.txt

@@ -0,0 +1,16 @@
+1<EFBFBD><EFBFBD>tango_cache_test.c<><63><EFBFBD>칷ʹ<ECB9B7>õ<EFBFBD>demo<6D><6F><EFBFBD><EFBFBD>Ҫ<EFBFBD>ⲿ<EFBFBD><E2B2BF><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>event_base<73><65>ִ<EFBFBD>з<EFBFBD><D0B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD>
+	<09><><EFBFBD><EFBFBD>һ<EFBFBD><D2BB><EFBFBD>նˣ<D5B6>ִ<EFBFBD><D6B4>./tango_cache_test<73><74>
+	<09><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>һ<EFBFBD><D2BB><EFBFBD>նˣ<D5B6>ִ<EFBFBD>У<EFBFBD>
+	<09><>ȡ<EFBFBD><C8A1>echo GET:tcpdump_all_1.pcap > cache.fifo
+	<09><>ʽ<EFBFBD>ϴ<EFBFBD><CFB4><EFBFBD>
+echo PUT:tcpdump_all_1.pcap > cache.fifo
+	<09><><EFBFBD><EFBFBD><EFBFBD>ϴ<EFBFBD><CFB4><EFBFBD>
+echo PUTONCE:tcpdump_all_1.pcap > cache.fifo
+
+
+2<EFBFBD><EFBFBD>cache_evbase_test.c<>Ƕ<EFBFBD><C7B6>칷cache<68>ķ<EFBFBD>װ<EFBFBD><D7B0><EFBFBD>û<EFBFBD><C3BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Լ<EFBFBD><D4BC><EFBFBD><EFBFBD><EFBFBD>event_base<73><65><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڲ<EFBFBD><DAB2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ִ<EFBFBD>з<EFBFBD><D0B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD>
+	<09><>ȡ<EFBFBD><C8A1>./cache_evbase_test GET:tcpdump_all_1.pcap
+	<09><>ʽ<EFBFBD>ϴ<EFBFBD><CFB4><EFBFBD>./cache_evbase_test PUT:tcpdump_all_1.pcap
+	<09><><EFBFBD><EFBFBD><EFBFBD>ϴ<EFBFBD><CFB4><EFBFBD>./cache_evbase_test PUTONCE:tcpdump_all_1.pcap
+
+<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ж<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>tcpdump_all_1.pcap<61><70><EFBFBD>Եģ<D4B5>ʵ<EFBFBD><CAB5>Ӧ<EFBFBD><D3A6>ʱ<EFBFBD>滻<EFBFBD>ɾ<EFBFBD><C9BE><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD><C4BC><EFBFBD><EFBFBD><EFBFBD>

cache/test_demo/cache_evbase_test.cpp

@@ -0,0 +1,224 @@
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <netinet/in.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/prctl.h>
+#include <sys/un.h>
+#include <unistd.h>
+#include <sys/poll.h>
+#include <curl/curl.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <errno.h>
+#include <sys/cdefs.h>
+#include <MESA/MESA_handle_logger.h>
+
+#include "cache_evbase_client.h"
+
+struct cache_evbase_instance *instance_asyn;
+int still_runing=1;
+
+struct future_pdata
+{
+	struct future * future;
+	FILE *fp;
+	char filename[256];
+};
+
+void get_future_success(future_result_t* result, void * user)
+{
+	struct tango_cache_result *res = (struct tango_cache_result *)result;
+	struct future_pdata *pdata = (struct future_pdata *)user;
+	char buffer[1024];
+
+	if(res != NULL)
+	{
+		if(res->type==RESULT_TYPE_BODY)
+		{
+			fwrite(res->data_frag, res->size, 1, pdata->fp);
+		}
+		else
+		{
+			memcpy(buffer, res->data_frag, res->size>=1024?1023:res->size);
+			buffer[res->size] = '\0';
+			printf("%s", buffer);
+		}
+	}
+	else //<2F><><EFBFBD><EFBFBD>
+	{
+		future_destroy(pdata->future);
+		fclose(pdata->fp);
+		free(pdata);
+		still_runing = 0;
+	}
+}
+
+void get_future_failed(enum e_future_error err, const char * what, void * user)
+{
+	printf("GET fail: %s\n", what);
+	sleep(5);
+	still_runing = 0;
+}
+
+void put_future_success(future_result_t* result, void * user)
+{
+	struct future_pdata *pdata = (struct future_pdata *)user;
+
+	printf("PUT %s succ\n", pdata->filename);
+	future_destroy(pdata->future);
+	free(pdata);
+	still_runing = 0;
+}
+void put_future_failed(enum e_future_error err, const char * what, void * user)
+{
+	struct future_pdata *pdata = (struct future_pdata *)user;
+
+	printf("PUT %s fail: %s\n", what, pdata->filename);
+	future_destroy(pdata->future);
+	free(pdata);
+	still_runing = 0;
+}
+
+char * get_file_content(const char *filename, size_t *filelen_out)
+{
+	char *buffer;
+	FILE *fp;
+	size_t filelen = 0;
+	struct stat filestat;
+	int readlen;
+
+	fp = fopen(filename, "rb");
+	if(fp == NULL)
+	{
+		printf("fopen file %s failed.\n", filename);
+		return NULL;
+	}
+	if(fstat(fileno(fp), &filestat))
+	{
+		printf("fstat %s failed.\n", filename);
+		return NULL;
+	}
+
+	buffer = (char *)malloc(filestat.st_size);
+
+	while(filelen < (size_t)filestat.st_size)
+	{
+		readlen = fread(buffer + filelen, 1, filestat.st_size - filelen, fp);
+		if(readlen < 0)
+		{
+			printf("read error: %s\n", strerror(errno));
+			return NULL;
+		}
+
+		filelen += readlen;
+	}
+	fclose(fp);
+	*filelen_out = filestat.st_size;
+
+	return buffer;
+}
+
+int main(int argc, char **argv)
+{
+	int n, index=0;
+	char method[16], filename_in[256], filename_out[256], *p;
+	struct tango_cache_meta meta;
+	struct future_pdata *pdata;
+	struct cache_evbase_ctx *ctx;
+	void *runtime_log;
+
+	if(argc != 2 && argc!=3)
+	{
+		printf("USGAE: %s <PUT/PUTONCE/GET:filename> [get_out_file_index]\n", argv[0]);
+		return -1;
+	}
+	if(argc==3)
+	{
+		index = atoi(argv[2]);
+	}
+	
+	runtime_log = MESA_create_runtime_log_handle("./runtime.log", 10);
+	if(NULL==runtime_log)
+	{
+		return -1;
+	}
+		
+	instance_asyn = cache_evbase_instance_new("./pangu_tg_cahce.conf", "TANGO_CACHE", runtime_log);
+	assert(instance_asyn!=NULL);
+
+	pdata = (struct future_pdata *)malloc(sizeof(struct future_pdata));
+
+	if(sscanf(argv[1], "%[^:]:%1023s%n", method, filename_in, &n) != 2)
+	{
+		assert(0);
+	}
+	if(strlen(filename_in) > 0) 
+	{
+		memset(&meta, 0, sizeof(struct tango_cache_meta));
+		meta.url = filename_in;
+		meta.std_hdr[HDR_CONTENT_TYPE] = "Content-Type: maintype/subtype";
+		meta.std_hdr[HDR_CONTENT_ENCODING] = "Content-Encoding: gzip";
+		meta.other_hdr = "Etag: hgdkqkwdwqekdfjwjfjwelkjfkwfejwhf\r\n";
+		
+		p = method;
+		while(*p=='\r'||*p=='\n') p++;
+		assert(*p!='\0');
+
+		if(!strcasecmp(p, "GET"))
+		{
+			sprintf(filename_out, "file_index_%u.bin", index);
+			pdata->future = future_create(get_future_success, get_future_failed, pdata);
+			promise_set_ctx(future_to_promise(pdata->future), NULL, NULL);
+			pdata->fp = fopen(filename_out, "w");
+			
+			cache_evbase_fetch(instance_asyn, pdata->future, &meta);
+		}
+		else if(!strcasecmp(p, "PUTONCE"))
+		{
+			size_t filelen;
+			p = get_file_content(filename_in, &filelen);
+			pdata->future = future_create(put_future_success, put_future_failed, pdata);
+			promise_set_ctx(future_to_promise(pdata->future), NULL, NULL);
+			
+			cache_evbase_upload_once(instance_asyn, pdata->future, PUT_ONCE_FREE, p, filelen, &meta, pdata->filename, 256);
+		}
+		else
+		{
+			pdata->future = future_create(put_future_success, put_future_failed, pdata);
+			promise_set_ctx(future_to_promise(pdata->future), NULL, NULL);
+
+			ctx = cache_evbase_update_start(instance_asyn, pdata->future, &meta);
+			sprintf(pdata->filename, "%s", cache_evbase_get_object_key(ctx));
+
+			char buffer[1024];
+			FILE *fp = fopen(filename_in, "r");
+			while(!feof(fp))
+			{
+				n = fread(buffer, 1, 1024, fp);
+				assert(n>=0);
+				cache_evbase_update_frag(ctx, PUT_ONCE_COPY, buffer, n);
+			}
+			
+			cache_evbase_update_end(ctx);
+		}
+	}
+
+	printf("Waiting to finish.......\n");
+	while(still_runing)
+	{
+		sleep(1);
+	}
+
+	struct cache_statistics out;
+	cache_evbase_get_statistics(instance_asyn, &out);
+	printf("get_recv: %llu, get_succ: %llu, get_fail: %llu, put_recv: %llu, put_succ: %llu, put_fail: %llu, session: %llu, memory: %llu\n",
+		out.get_recv_num, out.get_succ_num, out.get_fail_num, out.put_recv_num, out.put_succ_num, out.put_fail_num, out.session_num, out.memory_used);
+	return 0;
+}
+

cache/test_demo/pangu_tg_cahce.conf

@@ -0,0 +1,22 @@
+[TANGO_CACHE]
+#MINIO IP地址，目前只支持一个
+MINIO_BROKERS_LIST=192.168.10.64:9000
+
+#每个域名最多开启的链接数
+MAX_CONNECTION_PER_HOST=10
+
+#bucket的名称
+CACHE_BUCKET_NAME=openbucket
+
+#内部分配缓存块的大小，根据上传的文件大小分布而定，最大16777216（16MB）
+CACHE_BLOCK_MAX_SIZE=8192
+
+#缓存最大占用的内存空间大小，超出空间时上传失败
+MAX_USED_MEMORY_SIZE_MB=5120
+
+#上传时Expires头部的过期时间，单位秒，最小60（1分钟）
+CACHE_DEFAULT_TTL_SECOND=3600
+
+#是否对对象的名称进行哈希，开启哈希有助于提高上传下载的速率
+CACHE_OBJECT_KEY_HASH_SWITCH=1
+

cache/test_demo/tango_cache_test.c

@@ -0,0 +1,298 @@
+#include <event2/event.h>  
+#include <event2/bufferevent.h> 
+#include <event.h>
+#include <event2/event.h>
+#include <event2/buffer.h>
+#include <event2/http.h>
+#include <event2/http_struct.h>
+#include <event2/keyvalq_struct.h>
+
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <netinet/in.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/prctl.h>
+#include <sys/un.h>
+#include <unistd.h>
+#include <sys/poll.h>
+#include <curl/curl.h>
+#include <event2/event.h>
+#include <event2/event_struct.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <errno.h>
+#include <sys/cdefs.h>
+#include <MESA/MESA_handle_logger.h>
+
+#include "tango_cache_client.h"
+
+struct event_base *ev_base;
+struct tango_cache_instance *tango_instance;
+
+#define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */
+
+struct future_pdata
+{
+	struct future * future;
+	FILE *fp;
+	char filename[256];
+};
+
+void get_future_success(future_result_t* result, void * user)
+{
+	struct tango_cache_result *res = (struct tango_cache_result *)result;
+	struct future_pdata *pdata = (struct future_pdata *)user;
+	char buffer[1024];
+
+	if(res != NULL)
+	{
+		if(res->type==RESULT_TYPE_BODY)
+		{
+			fwrite(res->data_frag, res->size, 1, pdata->fp);
+		}
+		else
+		{
+			memcpy(buffer, res->data_frag, res->size>=1024?1023:res->size);
+			buffer[res->size] = '\0';
+			printf("%s", buffer);
+		}
+	}
+	else
+	{
+		future_destroy(pdata->future);
+		fclose(pdata->fp);
+		free(pdata);
+	}
+}
+
+void get_future_failed(enum e_future_error err, const char * what, void * user)
+{
+	printf("GET fail: %s\n", what);
+}
+
+void put_future_success(future_result_t* result, void * user)
+{
+	struct future_pdata *pdata = (struct future_pdata *)user;
+
+	printf("PUT %s succ\n", pdata->filename);
+	future_destroy(pdata->future);
+	free(pdata);
+}
+void put_future_failed(enum e_future_error err, const char * what, void * user)
+{
+	struct future_pdata *pdata = (struct future_pdata *)user;
+
+	printf("PUT %s fail: %s\n", what, pdata->filename);
+	future_destroy(pdata->future);
+	free(pdata);
+}
+
+char * get_file_content(const char *filename, size_t *filelen_out)
+{
+	char *buffer;
+	FILE *fp;
+	size_t filelen = 0;
+	struct stat filestat;
+	int readlen;
+
+	fp = fopen(filename, "rb");
+	if(fp == NULL)
+	{
+		printf("fopen file %s failed.\n", filename);
+		return NULL;
+	}
+	if(fstat(fileno(fp), &filestat))
+	{
+		printf("fstat %s failed.\n", filename);
+		return NULL;
+	}
+
+	buffer = (char *)malloc(filestat.st_size);
+
+	while(filelen < (size_t)filestat.st_size)
+	{
+		readlen = fread(buffer + filelen, 1, filestat.st_size - filelen, fp);
+		if(readlen < 0)
+		{
+			printf("read error: %s\n", strerror(errno));
+			return NULL;
+		}
+
+		filelen += readlen;
+	}
+	fclose(fp);
+	*filelen_out = filestat.st_size;
+
+	return buffer;
+}
+
+static void dummy_accept_callback(evutil_socket_t fd, short events, void *arg)
+{
+	char s[1024];
+	long int rv = 0;
+	int n = 0;
+	FILE *input = (FILE *)arg;
+	static int index=0;
+	char filename[128], method[16], buffer[1024], *p;
+	struct tango_cache_meta meta;
+	struct future_pdata *pdata;
+
+	struct tango_cache_ctx *ctx;
+	
+	pdata = (struct future_pdata *)malloc(sizeof(struct future_pdata));
+
+	do 
+	{
+		s[0]='\0';
+		rv = fscanf(input, "%[^:]:%1023s%n", method, s, &n);
+		if(strlen(s) > 0) 
+		{
+			p = method;
+
+			memset(&meta, 0, sizeof(struct tango_cache_meta));
+			meta.url = s;
+			meta.std_hdr[HDR_CONTENT_TYPE] = "Content-Type: maintype/subtype";
+			meta.std_hdr[HDR_CONTENT_ENCODING] = "Content-Encoding: gzip";
+			meta.other_hdr = "Etag: hgdkqkwdwqekdfjwjfjwelkjfkwfejwhf\r\n";
+			
+			while(*p=='\r'||*p=='\n')p++;
+			if(*p=='\0') continue;
+			if(!strcasecmp(p, "GET"))
+			{				
+				sprintf(filename, "file_index_%u.bin", index++);
+				pdata->fp = fopen(filename, "w");
+				pdata->future = future_create(get_future_success, get_future_failed, pdata);
+				promise_set_ctx(future_to_promise(pdata->future), NULL, NULL);
+
+				tango_cache_fetch(tango_instance, pdata->future, &meta);
+			}
+			else if(!strcasecmp(p, "PUTONCE"))
+			{
+				size_t filelen;
+				p = get_file_content(s, &filelen);
+				pdata->future = future_create(put_future_success, put_future_failed, pdata);
+				promise_set_ctx(future_to_promise(pdata->future), NULL, NULL);
+				
+				tango_cache_upload_once(tango_instance, pdata->future, PUT_ONCE_FREE, p, filelen, &meta, pdata->filename, 256);
+			}
+			else
+			{
+				pdata->future = future_create(put_future_success, put_future_failed, pdata);
+				promise_set_ctx(future_to_promise(pdata->future), NULL, NULL);
+			
+				ctx = tango_cache_update_start(tango_instance, pdata->future, &meta);
+				sprintf(pdata->filename, "%s", tango_cache_get_object_key(ctx));
+
+				FILE *fp = fopen(s, "r");
+				while(!feof(fp))
+				{
+					n = fread(buffer, 1, 1024, fp);
+					assert(n>=0);
+					tango_cache_update_frag(ctx, buffer, n);
+				}
+				tango_cache_update_end(ctx);
+			}
+		}
+		else
+		{
+			break;
+		}
+	} while(rv != EOF);
+}
+
+struct event fifo_event;
+static int init_fifo(void)
+{
+  struct stat st;
+  curl_socket_t sockfd;
+  static const char *fifo = "cache.fifo";
+
+  fprintf(MSG_OUT, "Creating named pipe \"%s\"\n", fifo);
+  if(lstat (fifo, &st) == 0) {
+    if((st.st_mode & S_IFMT) == S_IFREG) {
+      errno = EEXIST;
+      perror("lstat");
+      exit(1);
+    }
+  }
+  unlink(fifo);
+  if(mkfifo (fifo, 0600) == -1) {
+    perror("mkfifo");
+    exit(1);
+  }
+  sockfd = open(fifo, O_RDWR | O_NONBLOCK, 0);
+  if(sockfd == -1) {
+    perror("open");
+    exit(1);
+  }
+  FILE *input = fdopen(sockfd, "r");
+
+	memset(&fifo_event, 0, sizeof(struct event));
+  fprintf(MSG_OUT, "Now, pipe some URL's into > %s\n", fifo);
+  event_assign(&fifo_event, ev_base, sockfd, EV_READ|EV_PERSIST,
+               dummy_accept_callback, input);
+  event_add(&fifo_event, NULL);
+  return (0);
+}
+
+void timer_cb(evutil_socket_t fd, short what, void *arg)
+{
+	struct timeval tv;
+	struct cache_statistics out;
+
+	tv.tv_sec = 10;
+	tv.tv_usec = 0;
+
+	/*static int num=0;
+	num++;
+	if(ctx_global!=NULL && num>5)
+	{
+		tango_cache_update_end(ctx_global);
+		ctx_global = NULL;
+	}*/
+
+	tango_cache_get_statistics(tango_instance, &out);
+	printf("get_recv: %llu, get_succ: %llu, get_fail: %llu, put_recv: %llu, put_succ: %llu, put_fail: %llu, session: %llu, memory: %llu\n",
+		out.get_recv_num, out.get_succ_num, out.get_fail_num, out.put_recv_num, out.put_succ_num, out.put_fail_num, out.session_num, out.memory_used);
+
+	event_add((struct event *)arg, &tv);
+}
+
+int main(int crgc, char  **arg)
+{	
+	struct event ev_timer;
+	struct timeval tv;
+	void *runtime_log;
+	
+	runtime_log = MESA_create_runtime_log_handle("./runtime.log", 10);
+	if(NULL==runtime_log)
+	{
+		return -1;
+	}
+
+	ev_base = event_base_new();
+	if (!ev_base) 
+	{  
+		printf("create socket error!\n");  
+		return 0;
+	}  
+	init_fifo();
+	
+	tango_instance = tango_cache_instance_new(ev_base, "./pangu_tg_cahce.conf", "TANGO_CACHE", runtime_log);
+
+	tv.tv_sec = 10;
+	tv.tv_usec = 0;
+	evtimer_assign(&ev_timer, ev_base, timer_cb, &ev_timer);
+	evtimer_add(&ev_timer, &tv);
+
+	event_base_dispatch(ev_base);
+	return 0;  
+}
+
+

cache/test_demo/urllist_get.txt

@@ -0,0 +1,3 @@
+GET:openbucket/curl_put_chunk_1.pcap
+GET:openbucket/tcpdump_all_1.pcap
+GET:openbucket/tcpdump_big_1.pcap

cache/test_demo/urllist_put.txt

@@ -0,0 +1,3 @@
+PUT:openbucket/curl_put_chunk_1.pcap
+PUT:openbucket/tcpdump_all_1.pcap
+PUT:openbucket/tcpdump_big_1.pcap