gfwleak/tango-tfe
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构cache目录，将libxml2、libcurl放到vendor目录。

Browse Source
This commit is contained in:
zhengchao
2018-10-14 17:11:45 +08:00
parent cb27db40f2
commit 12d4370b3f
120 changed files with 508 additions and 29642 deletions
Download Patch File Download Diff File Expand all files Collapse all files

552
cache/src/cache_evbase_client.cpp vendored Normal file
View File

@@ -0,0 +1,552 @@
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/types.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/prctl.h>
#include <string.h>
#include <pthread.h>
#include "cache_evbase_client.h"
#include "tango_cache_transfer.h"
#include "tango_cache_tools.h"
enum CACHE_ASYN_CMD
{
CACHE_ASYN_FETCH=0,
CACHE_ASYN_UPLOAD_ONCE_DATA,
CACHE_ASYN_UPLOAD_ONCE_EVBUF,
CACHE_ASYN_UPLOAD_START,
CACHE_ASYN_UPLOAD_FRAG_DATA,
CACHE_ASYN_UPLOAD_FRAG_EVBUF,
CACHE_ASYN_UPLOAD_END,
CACHE_ASYN_DELETE,
};
struct databuffer
{
char *data;
size_t size;
struct evbuffer *evbuf;
enum CACHE_ASYN_CMD cmd_type;
struct cache_evbase_ctx *ctx_asyn;
};
enum CACHE_ERR_CODE cache_evbase_get_last_error(const struct cache_evbase_ctx *ctx_asyn)
{
return tango_cache_get_last_error(ctx_asyn->ctx);
}
enum CACHE_ERR_CODE cache_evbase_ctx_error(const struct cache_evbase_instance *instance)
{
return tango_cache_ctx_error(instance->instance);
}
void cache_evbase_get_statistics(const struct cache_evbase_instance *instance, struct cache_statistics *out)
{
tango_cache_get_statistics(instance->instance, out);
}
struct tango_cache_result *cache_evbase_read_result(void *promise_result)
{
return tango_cache_read_result(promise_result);
}
void cache_evbase_get_object_path(const struct cache_evbase_ctx *ctx_asyn, char *path, size_t pathsize)
{
tango_cache_get_object_path(ctx_asyn->ctx, path, pathsize);
}
static int create_notification_pipe(evutil_socket_t sv[2])
{
if(evutil_socketpair(AF_LOCAL, SOCK_STREAM, 0, sv) == -1)
{
return -1;
}
if(evutil_make_socket_nonblocking(sv[0])<0 || evutil_make_socket_nonblocking(sv[1])<0)
{
return -1;
}
if(evutil_make_socket_closeonexec(sv[0])<0 || evutil_make_socket_closeonexec(sv[1])<0)
{
return -1;
}
return 0;
}
static int32_t mesa_tcp_sock_write (int32_t write_fd, void *buf, int32_t bufsize)
{
int32_t res;
do{
res = send(write_fd, buf, bufsize, MSG_NOSIGNAL);
}while (res==-1 &&(errno == EINTR));
return res;
}
static int32_t iothread_notify_event(int32_t socket_fd, void *content, int32_t len, int32_t s_time_out)
{
fd_set w_set, e_set;
struct timeval tv;
int32_t res=0, sndlen=0, sendsize=0;
while(len > sndlen)
{
FD_ZERO (&w_set);
FD_ZERO (&e_set);
FD_SET (socket_fd, &w_set);
FD_SET (socket_fd, &e_set);
if(s_time_out == 0)
{
res = select (socket_fd + 1, NULL, &w_set, &e_set, NULL);
}
else
{
tv.tv_sec = s_time_out;
tv.tv_usec = 0;
res = select (socket_fd + 1, NULL, &w_set, &e_set, &tv);
}
if(res <= 0)
{
printf("log_error: select io res=%d, error: %s\n", res, strerror(errno));
return -1;
}
if(FD_ISSET(socket_fd, &e_set))
{
printf("log_error: select io is in efds, error: %s\n", strerror(errno));
return -2;
}
if(FD_ISSET(socket_fd, &w_set))
{
sendsize = mesa_tcp_sock_write(socket_fd, (char*)content + sndlen, len - sndlen);
if (sendsize < 0)
{
if(errno == EAGAIN)
{
continue;
}
return -1;
}
sndlen += sendsize;
}
}
return sndlen;
}
static void cache_asyn_ctx_destroy(struct cache_evbase_ctx *ctx_asyn)
{
free(ctx_asyn);
}
static void cache_asyn_ioevent_dispatch(struct databuffer *buffer)
{
struct cache_evbase_ctx *ctx_asyn=buffer->ctx_asyn;
switch(buffer->cmd_type)
{
case CACHE_ASYN_FETCH:
tango_cache_fetch_start(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
break;
case CACHE_ASYN_DELETE:
cache_delete_minio_object(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
break;
case CACHE_ASYN_UPLOAD_ONCE_DATA:
tango_cache_upload_once_start_data(ctx_asyn->ctx, PUT_MEM_FREE, buffer->data, buffer->size);
cache_asyn_ctx_destroy(ctx_asyn);
break;
case CACHE_ASYN_UPLOAD_ONCE_EVBUF:
tango_cache_upload_once_start_evbuf(ctx_asyn->ctx, EVBUFFER_MOVE, buffer->evbuf);
evbuffer_free(buffer->evbuf);
cache_asyn_ctx_destroy(ctx_asyn);
break;
case CACHE_ASYN_UPLOAD_START:
ctx_asyn->ctx->instance->statistic.put_recv_num += 1;
ctx_asyn->ctx->instance->error_code = CACHE_OK;
break;
case CACHE_ASYN_UPLOAD_FRAG_DATA:
tango_cache_update_frag_data(ctx_asyn->ctx, buffer->data, buffer->size);
free(buffer->data);
break;
case CACHE_ASYN_UPLOAD_FRAG_EVBUF:
tango_cache_update_frag_evbuf(ctx_asyn->ctx, EVBUFFER_MOVE, buffer->evbuf);
evbuffer_free(buffer->evbuf);
break;
case CACHE_ASYN_UPLOAD_END:
tango_cache_update_end(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
break;
default: assert(0);break;
}
}
static void sockpair_notification_handler(evutil_socket_t fd, short events, void *arg)
{
ssize_t readlen, needlen;
struct cache_evbase_instance *instance_asyn = (struct cache_evbase_instance *)arg;
struct databuffer *buffer;
while(1)
{
needlen=sizeof(struct cache_evbase_ctx *);
readlen = recv(fd, &buffer, needlen, 0);
if(readlen == needlen)
{
cache_asyn_ioevent_dispatch(buffer);
free(buffer);
}
else
{
if(errno!=EWOULDBLOCK && errno!=EAGAIN)
{
MESA_HANDLE_RUNTIME_LOGV2(instance_asyn->instance->runtime_log, RLOG_LV_FATAL, "read pipe error: %s.", strerror(errno));
assert(0);
return;
}
break;
}
}
}
static void* thread_listen_sockpair(void *arg)
{
struct cache_evbase_instance *instance_asyn = (struct cache_evbase_instance *)arg;
struct event listen_event;
prctl(PR_SET_NAME, "tango_cache");
event_assign(&listen_event, instance_asyn->evbase, instance_asyn->notify_readfd, EV_READ|EV_PERSIST, sockpair_notification_handler, instance_asyn);
event_add(&listen_event, NULL);
event_base_dispatch(instance_asyn->evbase);
printf("Libevent dispath error, should not run here.\n");
MESA_HANDLE_RUNTIME_LOGV2(instance_asyn->instance->runtime_log, RLOG_LV_FATAL, "Libevent dispath error, should not run here.");
assert(0);
return NULL;
}
void cache_evbase_update_end(struct cache_evbase_ctx *ctx_asyn)
{
struct databuffer *buffer;
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_UPLOAD_END;
if(iothread_notify_event(ctx_asyn->instance_asyn->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
if(!ctx_asyn->ctx->fail_state)
{
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
}
tango_cache_ctx_destroy(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
free(buffer);
}
}
int cache_evbase_update_frag_data(struct cache_evbase_ctx *ctx_asyn, enum PUT_MEMORY_COPY_WAY way, const char *data, size_t size)
{
struct databuffer *buffer;
if(ctx_asyn->ctx->fail_state)
{
return -1;
}
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
if(way == PUT_MEM_COPY)
{
buffer->data = (char *)malloc(size);
memcpy(buffer->data, data, size);
}
else
{
buffer->data = (char*)data;
}
buffer->size = size;
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_UPLOAD_FRAG_DATA;
if(iothread_notify_event(ctx_asyn->instance_asyn->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
if(!ctx_asyn->ctx->fail_state)
{
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
}
free(buffer->data);
free(buffer);
return -2;
}
return 0;
}
int cache_evbase_update_frag_evbuf(struct cache_evbase_ctx *ctx_asyn, struct evbuffer *evbuf)
{
struct databuffer *buffer;
if(ctx_asyn->ctx->fail_state)
{
return -1;
}
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_UPLOAD_FRAG_EVBUF;
buffer->evbuf = evbuffer_new();
evbuffer_add_buffer(buffer->evbuf, evbuf);
if(iothread_notify_event(ctx_asyn->instance_asyn->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
if(!ctx_asyn->ctx->fail_state)
{
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
}
evbuffer_free(buffer->evbuf);
free(buffer);
return -2;
}
return 0;
}
struct cache_evbase_ctx *cache_evbase_update_start(struct cache_evbase_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
struct cache_evbase_ctx *ctx_asyn;
struct tango_cache_ctx *ctx;
struct databuffer *buffer;
ctx = tango_cache_update_prepare(instance->instance, future, meta);
if(ctx == NULL)
{
return NULL;
}
ctx_asyn = (struct cache_evbase_ctx *)calloc(1, sizeof(struct cache_evbase_ctx));
ctx_asyn->instance_asyn = instance;
ctx_asyn->ctx = ctx;
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_UPLOAD_START;
//<2F>¼<EFBFBD>֪ͨ<CDA8><D6AA>Ϊ<EFBFBD><CEAA><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͳ<EFBFBD><CDB3><EFBFBD><EFBFBD>Ϣ
if(iothread_notify_event(instance->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
tango_cache_ctx_destroy(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
free(buffer);
return NULL;
}
return ctx_asyn;
}
int cache_evbase_upload_once_data(struct cache_evbase_instance *instance, struct future* future,
enum PUT_MEMORY_COPY_WAY way, const char *data, size_t size, struct tango_cache_meta *meta, char *path, size_t pathsize)
{
struct cache_evbase_ctx *ctx_asyn;
struct tango_cache_ctx *ctx;
struct databuffer *buffer;
ctx = tango_cache_update_prepare(instance->instance, future, meta);
if(ctx == NULL)
{
return -1;
}
if(path != NULL)
{
snprintf(path, pathsize, "http://%s/%s/%s", ctx->hostaddr, instance->instance->bucketname, ctx->object_key);
}
ctx_asyn = (struct cache_evbase_ctx *)calloc(1, sizeof(struct cache_evbase_ctx));
ctx_asyn->instance_asyn = instance;
ctx_asyn->ctx = ctx;
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
if(way == PUT_MEM_COPY)
{
buffer->data = (char *)malloc(size);
memcpy(buffer->data, data, size);
}
else
{
buffer->data = (char*)data;
}
buffer->size = size;
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_UPLOAD_ONCE_DATA;
if(iothread_notify_event(instance->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
free(buffer->data);
free(buffer);
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
tango_cache_ctx_destroy(ctx);
cache_asyn_ctx_destroy(ctx_asyn);
return -2;
}
return 0;
}
int cache_evbase_upload_once_evbuf(struct cache_evbase_instance *instance, struct future* future,
struct evbuffer *evbuf, struct tango_cache_meta *meta, char *path, size_t pathsize)
{
struct cache_evbase_ctx *ctx_asyn;
struct tango_cache_ctx *ctx;
struct databuffer *buffer;
ctx = tango_cache_update_prepare(instance->instance, future, meta);
if(ctx == NULL)
{
return -1;
}
if(path != NULL)
{
snprintf(path, pathsize, "http://%s/%s/%s", ctx->hostaddr, instance->instance->bucketname, ctx->object_key);
}
ctx_asyn = (struct cache_evbase_ctx *)calloc(1, sizeof(struct cache_evbase_ctx));
ctx_asyn->instance_asyn = instance;
ctx_asyn->ctx = ctx;
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_UPLOAD_ONCE_EVBUF;
buffer->evbuf = evbuffer_new();
evbuffer_add_buffer(buffer->evbuf, evbuf);
if(iothread_notify_event(instance->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
evbuffer_free(buffer->evbuf);
free(buffer);
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
tango_cache_ctx_destroy(ctx);
cache_asyn_ctx_destroy(ctx_asyn);
return -2;
}
return 0;
}
int cache_evbase_fetch_object(struct cache_evbase_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
struct cache_evbase_ctx *ctx_asyn;
struct databuffer *buffer;
ctx_asyn = (struct cache_evbase_ctx *)calloc(1, sizeof(struct cache_evbase_ctx));
ctx_asyn->instance_asyn = instance;
ctx_asyn->ctx = tango_cache_fetch_prepare(instance->instance, future, meta);
if(ctx_asyn->ctx == NULL)
{
free(ctx_asyn);
return -1;
}
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_FETCH;
if(iothread_notify_event(instance->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
promise_failed(future_to_promise(future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx_asyn->ctx));
tango_cache_ctx_destroy(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
free(buffer);
return -2;
}
return 0;
}
int cache_evbase_delete_object(struct cache_evbase_instance *instance, struct future* future, const char *objkey)
{
struct cache_evbase_ctx *ctx_asyn;
struct databuffer *buffer;
ctx_asyn = (struct cache_evbase_ctx *)calloc(1, sizeof(struct cache_evbase_ctx));
ctx_asyn->instance_asyn = instance;
ctx_asyn->ctx = tango_cache_delete_prepare(instance->instance, future, objkey);
if(ctx_asyn->ctx == NULL)
{
free(ctx_asyn);
return -1;
}
buffer = (struct databuffer *)malloc(sizeof(struct databuffer));
buffer->ctx_asyn = ctx_asyn;
buffer->cmd_type = CACHE_ASYN_DELETE;
//<2F>ο<EFBFBD>Unix<69>߼<EFBFBD><DFBC><EFBFBD><EFBFBD><EFBFBD>432ҳ<32><D2B3><EFBFBD>ڶ<EFBFBD><DAB6>߳<EFBFBD>д<EFBFBD>İ<EFBFBD>ȫ<EFBFBD><C8AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
if(iothread_notify_event(instance->notify_sendfd, &buffer, sizeof(void *), 2) != sizeof(void *))
{
tango_cache_set_fail_state(ctx_asyn->ctx, CACHE_ERR_SOCKPAIR);
if(future != NULL)
{
promise_failed(future_to_promise(future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx_asyn->ctx));
}
tango_cache_ctx_destroy(ctx_asyn->ctx);
cache_asyn_ctx_destroy(ctx_asyn);
free(buffer);
return -2;
}
return 0;
}
struct cache_evbase_instance *cache_evbase_instance_new(const char* profile_path, const char* section, void *runtimelog)
{
evutil_socket_t notification_fd[2];
struct cache_evbase_instance *instance_asyn;
struct event_base *evbase;
pthread_t thread_tid;
pthread_attr_t attr;
if(create_notification_pipe(notification_fd))
{
return NULL;
}
if((evbase = event_base_new()) == NULL)
{
return NULL;
}
instance_asyn = (struct cache_evbase_instance *)calloc(1, sizeof(struct cache_evbase_instance));
instance_asyn->evbase = evbase;
instance_asyn->notify_readfd = notification_fd[0];
instance_asyn->notify_sendfd = notification_fd[1];
instance_asyn->instance = tango_cache_instance_new(evbase, profile_path, section, runtimelog);
if(instance_asyn->instance == NULL)
{
free(instance_asyn);
return NULL;
}
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
if(pthread_create(&thread_tid, &attr, thread_listen_sockpair, instance_asyn))
{
free(instance_asyn);
return NULL;
}
return instance_asyn;
}
void cache_evbase_global_init(void)
{
tango_cache_global_init();
}

795
cache/src/tango_cache_client.cpp vendored Normal file
View File

@@ -0,0 +1,795 @@
#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 <openssl/md5.h>
#include <MESA/MESA_prof_load.h>
#include "tango_cache_client_in.h"
#include "tango_cache_transfer.h"
#include "tango_cache_tools.h"
#include "tango_cache_xml.h"
int TANGO_CACHE_VERSION_20181009=0;
static int caculate_base64_md5(const char *data, unsigned long len, unsigned char *result, unsigned int size)
{
MD5_CTX c;
unsigned char md5[17]={0};
if(size < 33)
return -1;
MD5_Init(&c);
MD5_Update(&c, data, len);
MD5_Final(md5, &c);
Base64_EncodeBlock(md5, 16, result);
return 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]);
}
}
static int wired_load_balancer_lookup(WLB_handle_t wiredlb, const char *key, int keylen, char *host, size_t hostsize)
{
struct WLB_consumer_t chosen;
if(wiredLB_lookup(wiredlb, key, keylen, &chosen))
{
return -1;
}
snprintf(host, hostsize, "%s:%u", chosen.ip_addr, chosen.data_port);
return 0;
}
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_set_fail_state(struct tango_cache_ctx *ctx, enum CACHE_ERR_CODE error_code)
{
ctx->fail_state = true;
ctx->error_code = error_code;
}
const char *tango_cache_get_errstring(const struct tango_cache_ctx *ctx)
{
switch(ctx->error_code)
{
case CACHE_CACHE_MISS: return "cache not hit";
case CACHE_TIMEOUT: return "cache not fresh";
case CACHE_OUTOF_MEMORY:return "outof memory";
case CACHE_ERR_WIREDLB: return "wiredlb error";
case CACHE_ERR_SOCKPAIR:return "socketpair error";
case CACHE_ERR_INTERNAL:return "cache Expires or x-amz-meta-lm not found";
default: return ctx->error;
}
}
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_error_num= instance->statistic.get_error_num;
out->get_miss_num = instance->statistic.get_miss_num;
out->put_recv_num = instance->statistic.put_recv_num;
out->put_succ_num = instance->statistic.put_succ_num;
out->put_error_num= instance->statistic.put_error_num;
out->del_recv_num = instance->statistic.del_recv_num;
out->del_succ_num = instance->statistic.del_succ_num;
out->del_error_num= instance->statistic.del_error_num;
out->totaldrop_num= instance->statistic.totaldrop_num;
out->session_num = instance->statistic.session_num;
out->memory_used = instance->statistic.memory_used;
}
struct tango_cache_result *tango_cache_read_result(void *promise_result)
{
return (struct tango_cache_result *)promise_result;
}
void tango_cache_get_object_path(const struct tango_cache_ctx *ctx, char *path, size_t pathsize)
{
snprintf(path, pathsize, "http://%s/%s/%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key);
}
static void update_statistics(struct tango_cache_ctx *ctx, struct cache_statistics *statistic)
{
switch(ctx->method)
{
case CACHE_REQUEST_PUT:
if(ctx->fail_state)
{
statistic->put_error_num += 1;
}
else
{
statistic->put_succ_num += 1;
}
break;
case CACHE_REQUEST_GET:
if(ctx->fail_state)
{
if(ctx->error_code == CACHE_ERR_CURL)
statistic->get_error_num += 1;
else
statistic->get_miss_num += 1;
}
else
{
statistic->get_succ_num += 1;
}
break;
case CACHE_REQUEST_DELETE:
if(ctx->fail_state)
{
statistic->del_error_num += 1;
}
else
{
statistic->del_succ_num += 1;
}
break;
case CACHE_REQUEST_DELETE_MUL:
statistic->del_succ_num += ctx->del.succ_num;
statistic->del_error_num += ctx->del.fail_num;
break;
default:break;
}
}
void easy_string_destroy(struct easy_string *estr)
{
if(estr->buff != NULL)
{
free(estr->buff);
estr->buff = NULL;
estr->len = estr->size = 0;
}
}
void easy_string_savedata(struct easy_string *estr, const char *data, size_t len)
{
if(estr->size-estr->len < len+1)
{
estr->size += len*4+1;
estr->buff = (char*)realloc(estr->buff, estr->size);
}
memcpy(estr->buff+estr->len, data, len);
estr->len += len;
estr->buff[estr->len]='\0';
}
void tango_cache_ctx_destroy(struct tango_cache_ctx *ctx)
{
struct multipart_etag_list *etag;
if(ctx->curl != NULL)
{
curl_multi_remove_handle(ctx->instance->multi_hd, ctx->curl);
curl_easy_cleanup(ctx->curl);
}
easy_string_destroy(&ctx->response);
switch(ctx->method)
{
case CACHE_REQUEST_GET:
easy_string_destroy(&ctx->get.response_tag);
break;
case CACHE_REQUEST_PUT:
if(ctx->put.uploadID != NULL) free(ctx->put.uploadID);
if(ctx->put.combine_xml != NULL) free(ctx->put.combine_xml);
if(ctx->put.evbuf!=NULL)
{
ctx->instance->statistic.memory_used -= evbuffer_get_length(ctx->put.evbuf);
evbuffer_free(ctx->put.evbuf);
}
TAILQ_FOREACH(etag, &ctx->put.etag_head, node)
{
TAILQ_REMOVE(&ctx->put.etag_head, etag, node);
free(etag->etag);
free(etag);
}//no break here
case CACHE_REQUEST_DELETE_MUL:
if(ctx->headers != NULL)
{
curl_slist_free_all(ctx->headers);
}//no break here
case CACHE_REQUEST_DELETE:
if(ctx->future != NULL)
{
if(ctx->fail_state)
{
promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
}
else
{
promise_success(future_to_promise(ctx->future), NULL);
}
}
break;
default: break;
}
update_statistics(ctx, &ctx->instance->statistic);
free(ctx);
}
void tango_cache_update_end(struct tango_cache_ctx *ctx)
{
cache_kick_upload_minio_end(ctx);
}
int tango_cache_update_frag_data(struct tango_cache_ctx *ctx, const char *data, size_t size)
{
if(ctx->fail_state)
{
return -1;
}
if(evbuffer_add(ctx->put.evbuf, data, size))
{
return -1;
}
ctx->instance->statistic.memory_used += size;
if(evbuffer_get_length(ctx->put.evbuf) >= ctx->instance->upload_block_size)
{
cache_kick_upload_minio_multipart(ctx, ctx->instance->upload_block_size);
}
return 0;
}
int tango_cache_update_frag_evbuf(struct tango_cache_ctx *ctx, enum EVBUFFER_COPY_WAY way, struct evbuffer *evbuf)
{
size_t size;
if(ctx->fail_state)
{
return -1;
}
size = evbuffer_get_length(evbuf);
if(way == EVBUFFER_MOVE)
{
if(evbuffer_add_buffer(ctx->put.evbuf, evbuf))
{
return -1;
}
}
else
{
if(evbuffer_add_buffer_reference(ctx->put.evbuf, evbuf))
{
return -1;
}
}
ctx->instance->statistic.memory_used += size;
if(evbuffer_get_length(ctx->put.evbuf) >= ctx->instance->upload_block_size)
{
cache_kick_upload_minio_multipart(ctx, ctx->instance->upload_block_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};
time_t expires, now, last_modify;
if((u_int64_t)instance->statistic.memory_used >= instance->cache_limit_size)
{
instance->error_code = CACHE_OUTOF_MEMORY;
instance->statistic.totaldrop_num += 1;
return NULL;
}
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->object_key, 256, "%c%c/%c%c/%s", buffer[0], buffer[1], buffer[2], buffer[3], buffer+4);
}
else
{
snprintf(ctx->object_key, 256, "%s", meta->url);
}
if(wired_load_balancer_lookup(instance->wiredlb, meta->url, strlen(meta->url), ctx->hostaddr, 48))
{
instance->error_code = CACHE_ERR_WIREDLB;
instance->statistic.totaldrop_num += 1;
free(ctx);
return NULL;
}
//Expires<65>ֶΣ<D6B6><CEA3><EFBFBD><EFBFBD>ڻ<EFBFBD><DABB><EFBFBD><EFBFBD>ڲ<EFBFBD><DAB2>ж<EFBFBD><D0B6><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƿ<EFBFBD><C7B7><EFBFBD>ʱ
now = time(NULL);
expires = (meta->put.timeout==0||meta->put.timeout>instance->relative_ttl)?instance->relative_ttl:meta->put.timeout;
if(expires_timestamp2hdr_str(now + expires, buffer, 256))
{
ctx->headers = curl_slist_append(ctx->headers, buffer);
}
//Last-Modify<66>ֶΣ<D6B6><CEA3><EFBFBD><EFBFBD><EFBFBD>GETʱ<54>ж<EFBFBD><D0B6>Ƿ<EFBFBD><C7B7><EFBFBD><EFBFBD><EFBFBD>
last_modify = (meta->put.date > meta->put.last_modified)?meta->put.date:meta->put.last_modified;
if(last_modify == 0)
{
last_modify = get_gmtime_timestamp(now);
}
sprintf(buffer, "x-amz-meta-lm: %lu", last_modify);
ctx->headers = curl_slist_append(ctx->headers, 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 = curl_slist_append(ctx->headers, meta->std_hdr[i]);
}
}
if(meta->std_hdr[HDR_CONTENT_TYPE] == NULL)
{
ctx->headers = curl_slist_append(ctx->headers, "Content-Type:");
}
ctx->headers = curl_slist_append(ctx->headers, "Expect:");//ע<><D7A2>POST<53><54><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Expect<63><74>ϵ<EFBFBD><CFB5>Ҫ<EFBFBD><D2AA>ȷ<EFBFBD><C8B7><EFBFBD><EFBFBD>CURLOPT_POSTFIELDSIZE
//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͷ<EFBFBD><CDB7><EFBFBD><EFBFBD>GETʱ<54><CAB1>ԭ<EFBFBD><D4AD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
if(meta->usertag_len>0 && meta->usertag_len<=USER_TAG_MAX_LEN)
{
char *p = (char *)malloc((meta->usertag_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((const unsigned char*)meta->usertag, meta->usertag_len, (unsigned char*)p+17);
ctx->headers = curl_slist_append(ctx->headers, p);
free(p);
}
ctx->put.evbuf = evbuffer_new();
TAILQ_INIT(&ctx->put.etag_head);
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->instance->statistic.put_recv_num += 1;
ctx->instance->error_code = CACHE_OK;
return ctx;
}
int tango_cache_upload_once_data(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 *path, size_t pathsize)
{
struct tango_cache_ctx *ctx;
ctx = tango_cache_update_prepare(instance, future, meta);
if(ctx == NULL)
{
return -1;
}
if(path != NULL)
{
snprintf(path, pathsize, "http://%s/%s/%s", ctx->hostaddr, instance->bucketname, ctx->object_key);
}
return tango_cache_upload_once_start_data(ctx, way, data, size);
}
int tango_cache_upload_once_evbuf(struct tango_cache_instance *instance, struct future* future,
enum EVBUFFER_COPY_WAY way, struct evbuffer *evbuf, struct tango_cache_meta *meta, char *path, size_t pathsize)
{
struct tango_cache_ctx *ctx;
ctx = tango_cache_update_prepare(instance, future, meta);
if(ctx == NULL)
{
return -1;
}
if(path != NULL)
{
snprintf(path, pathsize, "http://%s/%s/%s", ctx->hostaddr, instance->bucketname, ctx->object_key);
}
return tango_cache_upload_once_start_evbuf(ctx, way, evbuf);
}
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->get.max_age = meta->get.max_age;
ctx->get.min_fresh = meta->get.min_fresh;
if(instance->hash_object_key)
{
caculate_sha256(meta->url, strlen(meta->url), sha256, 72);
snprintf(ctx->object_key, 256, "%c%c/%c%c/%s", sha256[0], sha256[1], sha256[2], sha256[3], sha256+4);
}
else
{
snprintf(ctx->object_key, 256, "%s", meta->url);
}
if(wired_load_balancer_lookup(instance->wiredlb, meta->url, strlen(meta->url), ctx->hostaddr, 48))
{
instance->error_code = CACHE_ERR_WIREDLB;
instance->statistic.totaldrop_num += 1;
free(ctx);
return NULL;
}
return ctx;
}
int tango_cache_fetch_object(struct tango_cache_instance *instance, struct future* future, struct tango_cache_meta *meta)
{
struct tango_cache_ctx *ctx;
ctx = tango_cache_fetch_prepare(instance, future, meta);
if(ctx == NULL)
{
return -1;
}
return tango_cache_fetch_start(ctx);
}
struct tango_cache_ctx *tango_cache_delete_prepare(struct tango_cache_instance *instance, struct future* future, const char *objkey)
{
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_DELETE;
if(instance->hash_object_key)
{
caculate_sha256(objkey, strlen(objkey), sha256, 72);
snprintf(ctx->object_key, 256, "%c%c/%c%c/%s", sha256[0], sha256[1], sha256[2], sha256[3], sha256+4);
}
else
{
snprintf(ctx->object_key, 256, "%s", objkey);
}
if(wired_load_balancer_lookup(instance->wiredlb, objkey, strlen(objkey), ctx->hostaddr, 48))
{
instance->error_code = CACHE_ERR_WIREDLB;
instance->statistic.totaldrop_num += 1;
free(ctx);
return NULL;
}
return ctx;
}
int tango_cache_delete_object(struct tango_cache_instance *instance, struct future* future, const char *objkey)
{
struct tango_cache_ctx *ctx;
ctx = tango_cache_delete_prepare(instance, future, objkey);
if(ctx == NULL)
{
return -1;
}
return (cache_delete_minio_object(ctx)==1)?0:-1;
}
struct tango_cache_ctx *tango_cache_multi_delete_prepare(struct tango_cache_instance *instance, struct future* future, char *objlist[], u_int32_t num)
{
struct tango_cache_ctx *ctx;
char md5[48]={0}, content_md5[48];
ctx = (struct tango_cache_ctx *)calloc(1, sizeof(struct tango_cache_ctx));
ctx->instance = instance;
ctx->future = future;
ctx->method = CACHE_REQUEST_DELETE_MUL;
ctx->del.succ_num = num;
if(wired_load_balancer_lookup(instance->wiredlb, objlist[0], strlen(objlist[0]), ctx->hostaddr, 48))
{
instance->error_code = CACHE_ERR_WIREDLB;
instance->statistic.totaldrop_num += num;
free(ctx);
return NULL;
}
construct_multiple_delete_xml(ctx->instance->bucketname, objlist, num, instance->hash_object_key, &ctx->response.buff, &ctx->response.size);
caculate_base64_md5(ctx->response.buff, ctx->response.size, (unsigned char *)md5, 48);
sprintf(content_md5, "Content-MD5: %s", md5);
ctx->headers = curl_slist_append(ctx->headers, content_md5);
ctx->headers = curl_slist_append(ctx->headers, "Content-Type: application/xml");
return ctx;
}
//TODO: AccessDenied
int tango_cache_multi_delete(struct tango_cache_instance *instance, struct future* future, char *objlist[], u_int32_t num)
{
struct tango_cache_ctx *ctx;
ctx = tango_cache_multi_delete_prepare(instance, future, objlist, num);
if(ctx == NULL)
{
return -1;
}
return tango_cache_multi_delete_start(ctx);
}
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);
curl_multi_remove_handle(multi, easy);
curl_easy_cleanup(easy);
ctx->curl = NULL;
ctx->res_code = 0;
switch(ctx->method)
{
case CACHE_REQUEST_GET:
tango_cache_curl_get_done(ctx, res, res_code);
break;
case CACHE_REQUEST_PUT:
tango_cache_curl_put_done(ctx, res, res_code);
break;
case CACHE_REQUEST_DELETE:
tango_cache_curl_del_done(ctx, res, res_code);
break;
case CACHE_REQUEST_DELETE_MUL:
tango_cache_curl_muldel_done(ctx, res, res_code);
break;
default: break;
}
}
}
/* 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 wired_load_balancer_init(struct tango_cache_instance *instance)
{
instance->wiredlb = wiredLB_create(instance->wiredlb_topic, instance->wiredlb_group, WLB_PRODUCER);
if(instance->wiredlb == NULL)
{
MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_FATAL, "wiredLB_create failed.\n");
return -1;
}
wiredLB_set_opt(instance->wiredlb, WLB_OPT_ENABLE_OVERRIDE, &instance->wiredlb_override, sizeof(instance->wiredlb_override));
wiredLB_set_opt(instance->wiredlb, WLB_PROD_OPT_DATACENTER, instance->wiredlb_datacenter, strlen(instance->wiredlb_datacenter)+1);
if(instance->wiredlb_override)
{
wiredLB_set_opt(instance->wiredlb, WLB_PROD_OPT_OVERRIDE_PRIMARY_IP, instance->minio_iplist, strlen(instance->minio_iplist)+1);
wiredLB_set_opt(instance->wiredlb, WLB_PROD_OPT_OVERRIDE_DATAPORT, &instance->minio_port, sizeof(instance->minio_port));
}
if(wiredLB_init(instance->wiredlb) < 0)
{
MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_FATAL, "wiredLB_init failed.\n");
return -1;
}
return 0;
}
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);
MESA_load_profile_uint_def(profile_path, section, "MINIO_LISTEN_PORT", &instance->minio_port, 9000);
if(MESA_load_profile_string_nodef(profile_path, section, "MINIO_IP_LIST", instance->minio_iplist, 4096) < 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_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, 999999999);
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;
//Wired_LB<4C><42><EFBFBD><EFBFBD>
MESA_load_profile_string_def(profile_path, section, "WIREDLB_TOPIC", instance->wiredlb_topic, 64, "TANGO_CACHE_PRODUCER");
MESA_load_profile_string_def(profile_path, section, "WIREDLB_GROUP", instance->wiredlb_group, 64, "KAZAKHSTAN");
MESA_load_profile_string_def(profile_path, section, "WIREDLB_DATACENTER", instance->wiredlb_datacenter, 64, "ASTANA");
MESA_load_profile_uint_def(profile_path, section, "WIREDLB_OVERRIDE", &instance->wiredlb_override, 1);
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;
}
if(wired_load_balancer_init(instance))
{
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 | CURLPIPE_MULTIPLEX);
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;
}
void tango_cache_global_init(void)
{
curl_global_init(CURL_GLOBAL_NOTHING);
}

151
cache/src/tango_cache_client_in.h vendored Normal file
View File

@@ -0,0 +1,151 @@
#ifndef __TANGO_CACHE_CLIENT_IN_H__
#define __TANGO_CACHE_CLIENT_IN_H__
#include <curl/curl.h>
#include <sys/queue.h>
#include <event2/event.h>
#include <event.h>
#include <MESA/wiredLB.h>
#include "tango_cache_client.h"
#define RESPONSE_HDR_EXPIRES 1
#define RESPONSE_HDR_LAST_MOD 2
#define RESPONSE_HDR_ALL 3
enum CACHE_REQUEST_METHOD
{
CACHE_REQUEST_GET=0,
CACHE_REQUEST_PUT,
CACHE_REQUEST_DELETE,
CACHE_REQUEST_DELETE_MUL,
};
enum GET_OBJECT_STATE
{
GET_STATE_START=0,
GET_STATE_DELETE,
GET_STATE_END,
};
enum PUT_OBJECT_STATE
{
PUT_STATE_START=0,
PUT_STATE_WAIT_START,
PUT_STATE_PART,
PUT_STATE_END,
PUT_STATE_CANCEL,
};
struct easy_string
{
char* buff;
size_t len;
size_t size;
};
struct tango_cache_instance
{
char minio_iplist[4096];
char bucketname[256];
char wiredlb_topic[64];
char wiredlb_group[64];
char wiredlb_datacenter[64];
u_int32_t minio_port;
u_int32_t wiredlb_override;
struct event_base* evbase;
struct event timer_event;
struct cache_statistics statistic;
CURLM *multi_hd;
void *runtime_log;
WLB_handle_t wiredlb;
time_t relative_ttl; //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><D0A7>
u_int64_t cache_limit_size;
long max_cnn_host;
u_int32_t upload_block_size; //minio<69>ֶ<EFBFBD><D6B6>ϴ<EFBFBD><CFB4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>С<EFBFBD><D0A1><EFBFBD><EFBFBD>
enum CACHE_ERR_CODE error_code;
u_int32_t hash_object_key;
};
struct multipart_etag_list
{
char *etag;
u_int32_t part_number;
TAILQ_ENTRY(multipart_etag_list) node;
};
struct cache_ctx_data_get
{
time_t max_age;//Get
time_t min_fresh;//Get
time_t expires;
time_t last_modify;
u_int32_t need_hdrs;
enum GET_OBJECT_STATE state;
struct easy_string response_tag;
};
struct cache_ctx_data_put
{
struct evbuffer *evbuf;
size_t upload_length;
size_t upload_offset;
char *uploadID;
char *combine_xml;
TAILQ_HEAD(__etag_list_head, multipart_etag_list) etag_head;
enum PUT_OBJECT_STATE state;
u_int32_t part_index; //<2F><>RESPONSE_HDR_
bool close_state; //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ùر<C3B9>
};
struct cache_ctx_multi_delete
{
u_int32_t succ_num;
u_int32_t fail_num;
};
struct tango_cache_ctx
{
CURL *curl;
struct curl_slist *headers;
struct future* future;
char error[CURL_ERROR_SIZE];
char object_key[256];
char hostaddr[48];
enum CACHE_REQUEST_METHOD method;
enum CACHE_ERR_CODE error_code;
struct easy_string response;
bool fail_state;
long res_code;
union{
struct cache_ctx_data_put put;
struct cache_ctx_data_get get;
struct cache_ctx_multi_delete del;
};
struct tango_cache_instance *instance;
};
struct curl_socket_data
{
struct event sock_event;
};
void caculate_sha256(const char *data, unsigned long len, char *result, u_int32_t size);
void easy_string_savedata(struct easy_string *estr, const char *data, size_t len);
void easy_string_destroy(struct easy_string *estr);
void tango_cache_ctx_destroy(struct tango_cache_ctx *ctx);
void tango_cache_set_fail_state(struct tango_cache_ctx *ctx, enum CACHE_ERR_CODE error_code);
const char *tango_cache_get_errstring(const struct tango_cache_ctx *ctx);
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 *tango_cache_fetch_prepare(struct tango_cache_instance *instance, struct future* future, struct tango_cache_meta *meta);
struct tango_cache_ctx *tango_cache_delete_prepare(struct tango_cache_instance *instance, struct future* future, const char *objkey);
#endif

50
cache/src/tango_cache_pending.cpp vendored
View File

@@ -1,4 +1,4 @@
#include"proxy_cache.h"
#include"tango_cache_pending.h"
#include<assert.h>
#include<string.h>
#include<stdlib.h>
@@ -63,19 +63,17 @@ enum cache_pending_action request_cache_control(const char* value, struct reques
}
bool cache_vertify(const struct tfe_http_field *http_fields, size_t n_fields)
bool cache_vertify(const struct tfe_http_half *request)
{
int i = 0;
for (; i < n_fields; i++)
{
if ( http_fields[i].http_field == TLF_HTTP_IF_MATCH ||
http_fields[i].http_field == TLF_HTTP_IF_NONE_MATCH ||
http_fields[i].http_field == TLF_HTTP_IF_MODIFIED_SINCE ||
http_fields[i].http_field == TLF_HTTP_IF_UNMODIFIED_SINCE )
{
return true;
}
}
if( !tfe_http_std_field_read(request,TFE_HTTP_IF_MATCH) ||
!tfe_http_std_field_read(request,TFE_HTTP_IF_NONE_MATCH) ||
!tfe_http_std_field_read(request,TFE_HTTP_IF_MODIFIED_SINCE) ||
!tfe_http_std_field_read(request,TFE_HTTP_IF_UNMODIFIED_SINCE)
)
{
return true;
}
return false;
}
@@ -105,28 +103,28 @@ enum cache_pending_action get_pragma_action(const char * value)
}
enum cache_pending_action tfe_cache_get_pending(const struct tfe_http_field *request, size_t n_fields, struct request_freshness* restrict)
enum cache_pending_action tfe_cache_get_pending(const struct tfe_http_half *request, struct request_freshness* restrict)
{
enum cache_pending_action res = UNDEFINED;
int i = 0;
int index = 0;
const char *value = NULL;
memset(restrict,0,sizeof(struct request_freshness));
value = get_head_value(request, n_fields, TFE_HTTP_PRAGMA);
value = tfe_http_std_field_read(request, TFE_HTTP_PRAGMA);
if (value != NULL)
{
res = get_pragma_action(value);
}
else
{
value = get_head_value(request, n_fields, TFE_HTTP_CACHE_CONTROL);
value = tfe_http_std_field_read(request, TFE_HTTP_CACHE_CONTROL);
if (value != NULL)
{
res = request_cache_control(value, restrict);
}
else
{
if (cache_vertify(request, n_fields))
if (cache_vertify(request))
{
res = VERIFY;
}
@@ -199,13 +197,13 @@ time_t get_response_maxage(const char* cache_ctl)
}
void get_response_freshness(const struct tfe_http_field *response, size_t n_fields, struct response_freshness* freshness)
void get_response_freshness(const struct tfe_http_half *response, struct response_freshness* freshness)
{
time_t expire_rel_time = 0;
time_t cur_rel_time = 0;
struct tm cur_gmt_time;
const char* field_value = NULL;
field_value = get_head_value(response, n_fields, TFE_HTTP_CACHE_CONTROL);
field_value = tfe_http_std_field_read(response, TFE_HTTP_CACHE_CONTROL);
if (field_value != NULL)
{
freshness->timeout = get_response_s_maxage(field_value);
@@ -216,7 +214,7 @@ void get_response_freshness(const struct tfe_http_field *response, size_t n_fiel
}
else
{
field_value = get_head_value(response, n_fields, TFE_HTTP_EXPIRES);
field_value = tfe_http_std_field_read(response, TFE_HTTP_EXPIRES);
if (field_value != NULL)
{
assert(is_standard_gmt_format(field_value));
@@ -230,13 +228,13 @@ void get_response_freshness(const struct tfe_http_field *response, size_t n_fiel
freshness->timeout = expire_rel_time - cur_rel_time;
}
}
field_value = get_head_value(response, n_fields, TFE_HTTP_DATE);
field_value = tfe_http_std_field_read(response, TFE_HTTP_DATE);
if (field_value != NULL)
{
assert(is_standard_gmt_format(field_value));
freshness->date = absolute_to_relative_time(field_value);;
}
field_value = get_head_value(response, n_fields, TLF_HTTP_LAST_MODIFIED);
field_value = tfe_http_std_field_read(response, TFE_HTTP_LAST_MODIFIED);
if (field_value != NULL)
{
assert(is_standard_gmt_format(field_value));
@@ -268,28 +266,28 @@ enum cache_pending_action response_cache_control(const char* value)
}
enum cache_pending_action tfe_cache_put_pending(const struct tfe_http_field *response, size_t n_fields, struct response_freshness* freshness)
enum cache_pending_action tfe_cache_put_pending(const struct tfe_http_half *response, struct response_freshness* freshness)
{
enum cache_pending_action res = UNDEFINED;
int i = 0;
int index = 0;
const char *value = NULL;
memset(freshness,0,sizeof(struct response_freshness));
value = get_head_value(response, n_fields, TFE_HTTP_PRAGMA);
value = tfe_http_std_field_read(response, TFE_HTTP_PRAGMA);
if (value != NULL)
{
res = get_pragma_action(value);
}
else
{
value = get_head_value(response, n_fields, TFE_HTTP_CACHE_CONTROL);
value = tfe_http_std_field_read(response, TFE_HTTP_CACHE_CONTROL);
if (value != NULL)
{
res = response_cache_control(value);
}
else
{
value = get_head_value(response, n_fields, TFE_HTTP_EXPIRES);
value = tfe_http_std_field_read(response, TFE_HTTP_EXPIRES);
if (value != NULL)
{
res = ALLOWED;
@@ -298,7 +296,7 @@ enum cache_pending_action tfe_cache_put_pending(const struct tfe_http_field *res
}
if (res == ALLOWED)
{
get_response_freshness(response, n_fields, freshness);
get_response_freshness(response, freshness);
}
return res;
}

259
cache/src/tango_cache_tools.cpp vendored Normal file
View File

@@ -0,0 +1,259 @@
#include <sys/time.h>
#include <time.h>
#include <string.h>
#include <ctype.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include "tango_cache_tools.h"
static const char data_bin2ascii[65] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
#define conv_bin2ascii(a) (data_bin2ascii[(a)&0x3f])
#define B64_EOLN 0xF0
#define B64_CR 0xF1
#define B64_EOF 0xF2
#define B64_WS 0xE0
#define B64_ERROR 0xFF
#define B64_NOT_BASE64(a) (((a)|0x13) == 0xF3)
#define B64_BASE64(a) !B64_NOT_BASE64(a)
static const unsigned char data_ascii2bin[128] = {
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xE0, 0xF0, 0xFF, 0xFF, 0xF1, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xE0, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0x3E, 0xFF, 0xF2, 0xFF, 0x3F,
0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3A, 0x3B,
0x3C, 0x3D, 0xFF, 0xFF, 0xFF, 0x00, 0xFF, 0xFF,
0xFF, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06,
0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E,
0x0F, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16,
0x17, 0x18, 0x19, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F, 0x20,
0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28,
0x29, 0x2A, 0x2B, 0x2C, 0x2D, 0x2E, 0x2F, 0x30,
0x31, 0x32, 0x33, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
};
//<2F>мDz<D0BC><C7B2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>˵<EFBFBD><CBB5><EFBFBD>: conv_ascii2bin(x++); <20><>Ϊ<EFBFBD><CEAA>++<2B><><EFBFBD><EFBFBD>
#define conv_ascii2bin(aa) (((aa) & 0x80)?(0xFF):data_ascii2bin[(aa)])
/*********************************************************************
<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƣ<EFBFBD>Base64_EncodeBlock
<EFBFBD><EFBFBD><EFBFBD>ܼ<EFBFBD><EFBFBD>飺<EFBFBD><EFBFBD>һ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>BASE64<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>б<EFBFBD><EFBFBD><EFBFBD>
<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>in<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ַ<EFBFBD><EFBFBD><EFBFBD>
inl<6E><6C>in<69>ij<EFBFBD><C4B3><EFBFBD>
<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>out<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>洢<EFBFBD>Ļ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֵ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ij<EFBFBD><EFBFBD><EFBFBD>
*********************************************************************/
int Base64_EncodeBlock(const unsigned char *in, int inl, unsigned char *out)
{
int i, ret = 0;
unsigned long l;
for (i = inl; i > 0; i -= 3) {
if (i >= 3) {
l = (((unsigned long)in[0]) << 16L) |
(((unsigned long)in[1]) << 8L) | in[2];
*(out++) = conv_bin2ascii(l >> 18L);
*(out++) = conv_bin2ascii(l >> 12L);
*(out++) = conv_bin2ascii(l >> 6L);
*(out++) = conv_bin2ascii(l);
} else {
l = ((unsigned long)in[0]) << 16L;
if (i == 2)
l |= ((unsigned long)in[1] << 8L);
*(out++) = conv_bin2ascii(l >> 18L);
*(out++) = conv_bin2ascii(l >> 12L);
*(out++) = (i == 1) ? '=' : conv_bin2ascii(l >> 6L);
*(out++) = '=';
}
ret += 4;
in += 3;
}
*out = '\0';
return (ret);
}
/*********************************************************************
<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƣ<EFBFBD>Base64_DecodeBlock
<EFBFBD><EFBFBD><EFBFBD>ܼ<EFBFBD><EFBFBD>飺<EFBFBD><EFBFBD>һ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>BASE64<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>н<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Զ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>β<EFBFBD><EFBFBD>BASE64<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ַ<EFBFBD>
<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>in<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ַ<EFBFBD><EFBFBD><EFBFBD>
inl<6E><6C>in<69>ij<EFBFBD><C4B3><EFBFBD>
<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>out<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>洢<EFBFBD>Ļ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
<09><><EFBFBD>뱣֤<EBB1A3><D6A4><EFBFBD>㹻<EFBFBD>Ŀռ䣬һ<E4A3AC><D2BB><EFBFBD>ﵽ@inl<6E><6C>С<EFBFBD><D0A1><EFBFBD>ɣ<EFBFBD>
<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֵ<EFBFBD><EFBFBD><0<><30>ʧ<EFBFBD>ܣ<EFBFBD>>=0<><30><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ij<EFBFBD><C4B3><EFBFBD>
*********************************************************************/
int Base64_DecodeBlock(const unsigned char *in, int inl, unsigned char *out, int outsize)
{
int i, ret = 0;
unsigned char a, b, c, d;
unsigned long l;
/* ignore not-base64-encoded charactor. */
while ((conv_ascii2bin(*in) == B64_WS) && (inl > 0))
{
in++;
inl--;
}
while ((inl > 3) && (B64_NOT_BASE64(conv_ascii2bin(in[inl - 1]))))
inl--;
if (inl % 4 != 0)
return -1;
if(outsize < (inl*3)/4)
return -2;
for (i = 0; i < inl; i += 4)
{
a = conv_ascii2bin(*(in));
b = conv_ascii2bin(*(in+1));
c = conv_ascii2bin(*(in+2));
d = conv_ascii2bin(*(in+3));
if ((a & 0x80) || (b & 0x80) || (c & 0x80) || (d & 0x80))
return (-1);
l = ((((unsigned long)a) << 18L) | (((unsigned long)b) << 12L) |
(((unsigned long)c) << 6L) | (((unsigned long)d)));
*(out++) = (unsigned char)(l >> 16L) & 0xff;
*(out++) = (unsigned char)(l >> 8L) & 0xff;
*(out++) = (unsigned char)(l) & 0xff;
ret += 3;
in+=4;
}
for(i = inl; i > 0; i -= 4)
{
if(*(in-3) == '=')
{
in -= 4;
ret -= 3;
continue;
}
while(*(--in) == '=')
ret -= 1;
break;
}
return ret;
}
//<2F><>֪<EFBFBD><D6AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͬ<EFBFBD><CDAC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD><C2A3>Ƚ<EFBFBD><C8BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƿ<EFBFBD><C7B7><EFBFBD><EFBFBD><EFBFBD>
int strcmp_one_word_mesa_equal_len(const char *s1_lowercase, const char *s1_uppercase, const char *s2, size_t len)
{
unsigned char *s1,*s12;
if (s2[len-1]-'a'>=0)
{
s1 = (unsigned char *)s1_lowercase;
s12= (unsigned char *)s1_uppercase;
}
else
{
s1 = (unsigned char *)s1_uppercase;
s12= (unsigned char *)s1_lowercase;
}
do {
if (*s1 == *s2 || *s12 == *s2)
{
++s1;
++s12;
++s2;
continue;
}
return 0;
} while (--len);
return 1;
}
int mkdir_according_path(const char * path)
{
char buffer[256];
const char *ps=path, *pc;
if(*ps == '/')
ps += 1;
while((pc = strchr(ps, '/')) != NULL)
{
while(*(pc+1) == '/')
pc++;
memcpy(buffer, path, pc - path);
buffer[pc-path] = '\0';
if(access(buffer, F_OK))
{
if(mkdir(buffer, 0777) && errno!=EEXIST)
{
return -1;
}
}
ps = pc + 1;
}
if(access(path, F_OK))
{
if(mkdir(path, 0777))
{
return -1;
}
}
return 0;
}
//<2F><>ʱ<EFBFBD><CAB1><EFBFBD>ַ<EFBFBD><D6B7><EFBFBD>ת<EFBFBD><D7AA>Ϊʱ<CEAA><CAB1><EFBFBD><EFBFBD>
time_t expires_hdr2timestamp(const char *expires_val, int len)
{
struct tm tm;
time_t expire;
while(len > 0 && (*expires_val==' '||*expires_val=='\t'||*expires_val=='\r'||*expires_val=='\n'))
{
expires_val++;
len--;
}
if(len == 0)
{
return 0;
}
if(strptime(expires_val, "%a, %d %b %Y %T", &tm) == NULL)
{
return 0;
}
expire = mktime(&tm);
return expire;
}
//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><CAB1><EFBFBD><EFBFBD>ת<EFBFBD><D7AA>ΪGMTʱ<54><CAB1><EFBFBD>ַ<EFBFBD><D6B7><EFBFBD>
size_t expires_timestamp2hdr_str(time_t seconds, char *buffer, size_t size)
{
struct tm save;
return strftime(buffer, size, "Expires: %a, %d %b %Y %T GMT", gmtime_r(&seconds, &save));
}
//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><CAB1><EFBFBD><EFBFBD>ת<EFBFBD><D7AA>ΪGMTʱ<54><CAB1><EFBFBD><EFBFBD>
time_t get_gmtime_timestamp(time_t seconds)
{
struct tm *tm, save;
tm = gmtime_r(&seconds, &save);
return mktime(tm);
}

20
cache/src/tango_cache_tools.h vendored Normal file
View File

@@ -0,0 +1,20 @@
#ifndef __TANGO_CACHE_TOOLS_H__
#define __TANGO_CACHE_TOOLS_H__
#include <MESA/MESA_handle_logger.h>
#define MESA_HANDLE_RUNTIME_LOGV2(handle, lv, fmt, args...) \
MESA_handle_runtime_log((handle), (lv), "TANGO_CACHE", "%s:%d, " fmt, __FILE__, __LINE__, ##args)
int Base64_EncodeBlock(const unsigned char *in, int inl, unsigned char *out);
int Base64_DecodeBlock(const unsigned char *in, int inl, unsigned char *out, int outsize);
int strcmp_one_word_mesa_equal_len(const char *s1_lowercase, const char *s1_uppercase, const char *s2, size_t len);
int mkdir_according_path(const char * path);
time_t get_gmtime_timestamp(time_t seconds);
time_t expires_hdr2timestamp(const char *expires_val, int len);
size_t expires_timestamp2hdr_str(time_t seconds, char *buffer, size_t size);
#endif

876
cache/src/tango_cache_transfer.cpp vendored Normal file
View File

@@ -0,0 +1,876 @@
#include <sys/types.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 <curl/curl.h>
#include "tango_cache_transfer.h"
#include "tango_cache_xml.h"
#include "tango_cache_tools.h"
//response body<64>̻ܶ򲻹<CCBB><F2B2BBB9><EFBFBD>ʱ
size_t curl_response_any_cb(void *ptr, size_t size, size_t count, void *userp)
{
return size*count;
}
static size_t curl_put_multipart_header_cb(void *ptr, size_t size, size_t count, void *userp)
{
struct tango_cache_ctx *ctx = (struct tango_cache_ctx *)userp;
size_t totallen = size*count;
char *start = (char *)ptr, *end = start + totallen;
struct multipart_etag_list *etag;
if(!strncmp(start, "Etag:", totallen>5?5:totallen))
{
start += 5; end -= 1; totallen -= 5;
while(totallen>0 && (*start==' ')) {start++; totallen--;}
while(totallen>0 && (*end=='\r'||*end=='\n')) {end--; totallen--;}
if(totallen > 0)
{
etag = (struct multipart_etag_list *)malloc(sizeof(struct multipart_etag_list));
totallen = end - start + 1;
etag->etag = (char *)malloc(totallen + 1);
etag->part_number = ctx->put.part_index;
memcpy(etag->etag, start, totallen);
*(etag->etag + totallen) = '\0';
TAILQ_INSERT_TAIL(&ctx->put.etag_head, etag, node);
}
}
return size*count;
}
static size_t curl_put_once_send_cb(void *ptr, size_t size, size_t count, void *userp)
{
size_t len;
struct tango_cache_ctx *ctx = (struct tango_cache_ctx *)userp;
if(size==0 || count==0 || ctx->response.len>=ctx->response.size)
{
return 0; //<2F><>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
}
len = ctx->response.size - ctx->response.len; //ʣ<><CAA3><EFBFBD><EFBFBD><EFBFBD>ϴ<EFBFBD><CFB4>ij<EFBFBD><C4B3><EFBFBD>
if(len > size * count)
{
len = size * count;
}
memcpy(ptr, ctx->response.buff + ctx->response.len, len);
ctx->response.len += len;
if(ctx->response.len >= ctx->response.size)
{
ctx->instance->statistic.memory_used -= ctx->response.size; //δʹ<CEB4><CAB9>cache buffer<65><72><EFBFBD>Լ<EFBFBD><D4BC><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڴ<EFBFBD><DAB4><EFBFBD><EFBFBD><EFBFBD>
easy_string_destroy(&ctx->response);
}
return len;
}
static size_t curl_put_multipart_send_cb(void *ptr, size_t size, size_t count, void *userp)
{
size_t len, space=size*count, send_len;
struct tango_cache_ctx *ctx = (struct tango_cache_ctx *)userp;
if(size==0 || count==0 || ctx->put.upload_offset>=ctx->put.upload_length)
{
return 0;
}
len = ctx->put.upload_length - ctx->put.upload_offset;
if(len > space)
{
len = space;
}
send_len = evbuffer_remove(ctx->put.evbuf, ptr, len);
assert(send_len>0);
ctx->put.upload_offset += send_len;
ctx->instance->statistic.memory_used -= send_len;
return send_len;
}
//return value: <0:fail; =0: not exec; >0: OK
static int http_put_bodypart_request_evbuf(struct tango_cache_ctx *ctx, bool full)
{
CURLMcode rc;
char minio_url[256];
if(NULL == (ctx->curl=curl_easy_init()))
{
return -1;
}
ctx->put.upload_offset = 0;
if(full)
{
snprintf(minio_url, 256, "http://%s/%s/%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key);
}
else
{
snprintf(minio_url, 256, "http://%s/%s/%s?partNumber=%d&uploadId=%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key, ++ctx->put.part_index, ctx->put.uploadID);
curl_easy_setopt(ctx->curl, CURLOPT_HEADERFUNCTION, curl_put_multipart_header_cb);
curl_easy_setopt(ctx->curl, CURLOPT_HEADERDATA, ctx);
}
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_any_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
curl_easy_setopt(ctx->curl, CURLOPT_HTTPHEADER, ctx->headers);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_TIME, 2L);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_LIMIT, 100L);
curl_easy_setopt(ctx->curl, CURLOPT_UPLOAD, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_INFILESIZE, ctx->put.upload_length);
curl_easy_setopt(ctx->curl, CURLOPT_READFUNCTION, curl_put_multipart_send_cb);
curl_easy_setopt(ctx->curl, CURLOPT_READDATA, ctx);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return 1;
}
static size_t curl_response_body_save_cb(void *ptr, size_t size, size_t count, void *userp)
{
struct tango_cache_ctx *ctx = (struct tango_cache_ctx *)userp;
struct easy_string *estr = &ctx->response;
CURLcode code;
if(ctx->fail_state)
{
return size*count;
}
if(ctx->res_code == 0)
{
code = curl_easy_getinfo(ctx->curl, CURLINFO_RESPONSE_CODE, &ctx->res_code);
if(code != CURLE_OK || ctx->res_code!=200L)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
return size*count;
}
}
easy_string_savedata(estr, (const char*)ptr, size*count);
return size*count;
}
int curl_get_minio_uploadID(struct tango_cache_ctx *ctx)
{
CURLMcode rc;
char minio_url[256];
if(NULL == (ctx->curl=curl_easy_init()))
{
free(ctx);
return -1;
}
snprintf(minio_url, 256, "http://%s/%s/%s?uploads", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key);
curl_easy_setopt(ctx->curl, CURLOPT_POST, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_POSTFIELDSIZE, 0); //Ĭ<><C4AC>ʹ<EFBFBD>ûص<C3BB><D8B5><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>fread<61><64><EFBFBD><EFBFBD><EFBFBD>Է<EFBFBD><D4B7>ֹر<D6B9>Expectʱ<74>ᵼ<EFBFBD>¿<EFBFBD><C2BF><EFBFBD>curl_multi_socket_action
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL,1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_body_save_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
curl_easy_setopt(ctx->curl, CURLOPT_HTTPHEADER, ctx->headers);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_TIME, 2L);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_LIMIT, 100L);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return 1;
}
int cache_delete_minio_object(struct tango_cache_ctx *ctx)
{
CURLMcode rc;
char minio_url[256];
ctx->instance->statistic.del_recv_num += 1;
if(NULL == (ctx->curl=curl_easy_init()))
{
return 0;
}
snprintf(minio_url, 256, "http://%s/%s/%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key);
curl_easy_setopt(ctx->curl, CURLOPT_CUSTOMREQUEST, "DELETE");
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_any_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return 1;
}
//return value: true-<2D>ɹ<EFBFBD><C9B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>¼<EFBFBD><C2BC><EFBFBD>false-δ<><CEB4><EFBFBD><EFBFBD><EFBFBD>¼<EFBFBD>
bool cache_cancel_upload_minio(struct tango_cache_ctx *ctx)
{
CURLMcode rc;
char minio_url[256];
if(NULL == (ctx->curl=curl_easy_init()))
{
return false;
}
snprintf(minio_url, 256, "http://%s/%s/%s?uploadId=%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key, ctx->put.uploadID);
curl_easy_setopt(ctx->curl, CURLOPT_CUSTOMREQUEST, "DELETE");
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_any_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return true;
}
//return value: true-<2D>ɹ<EFBFBD><C9B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>¼<EFBFBD><C2BC><EFBFBD>false-δ<><CEB4><EFBFBD><EFBFBD><EFBFBD>¼<EFBFBD>
bool cache_kick_combine_minio(struct tango_cache_ctx *ctx)
{
int len=0;
CURLMcode rc;
char minio_url[256];
if(NULL == (ctx->curl=curl_easy_init()))
{
return false;
}
construct_complete_xml(ctx, &ctx->put.combine_xml, &len);
snprintf(minio_url, 256, "http://%s/%s/%s?uploadId=%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key, ctx->put.uploadID);
curl_easy_setopt(ctx->curl, CURLOPT_POST, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL,1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_any_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
curl_easy_setopt(ctx->curl, CURLOPT_POSTFIELDSIZE, len); //<2F><><EFBFBD><EFBFBD>Content-Length
curl_easy_setopt(ctx->curl, CURLOPT_POSTFIELDS, ctx->put.combine_xml);
if(ctx->headers != NULL)
{
curl_slist_free_all(ctx->headers);
ctx->headers = NULL;
}
ctx->headers = curl_slist_append(ctx->headers, "Content-Type: application/xml");
curl_easy_setopt(ctx->curl, CURLOPT_HTTPHEADER, ctx->headers);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return true;
}
//return value: true-<2D>ɹ<EFBFBD><C9B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>¼<EFBFBD><C2BC><EFBFBD>false-δ<><CEB4><EFBFBD><EFBFBD><EFBFBD>¼<EFBFBD>
bool cache_kick_upload_minio_multipart(struct tango_cache_ctx *ctx, size_t block_len)
{
int ret = 1;
switch(ctx->put.state)
{
case PUT_STATE_START:
ctx->put.state = PUT_STATE_WAIT_START;
ret = curl_get_minio_uploadID(ctx);
break;
case PUT_STATE_PART:
if(ctx->curl == NULL)
{
ctx->put.upload_length = block_len;
ret = http_put_bodypart_request_evbuf(ctx, false);
}
break;
default: break;//nothing to do
}
if(ret <= 0)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
return false;
}
return true;
}
int http_put_complete_part_evbuf(struct tango_cache_ctx *ctx)
{
int ret=0;
ctx->put.state = PUT_STATE_END;
ctx->put.upload_length = evbuffer_get_length(ctx->put.evbuf);
if(ctx->put.upload_length > 0)
{
ret = http_put_bodypart_request_evbuf(ctx, true);
if(ret <= 0)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
tango_cache_ctx_destroy(ctx);
}
}
else
{
tango_cache_ctx_destroy(ctx);
}
return ret;
}
int cache_kick_upload_minio_end(struct tango_cache_ctx *ctx)
{
int ret = 0;
ctx->put.close_state = true;//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>״̬<D7B4><CCAC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>رգ<D8B1><D5A3>ڲ<EFBFBD>״̬<D7B4><CCAC><EFBFBD><EFBFBD>ת<EFBFBD><D7AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ٹر<D9B9>
if(ctx->fail_state)
{
tango_cache_ctx_destroy(ctx);
return 0;
}
switch(ctx->put.state)
{
case PUT_STATE_START:
http_put_complete_part_evbuf(ctx);
break;
case PUT_STATE_PART:
if(ctx->curl == NULL)
{
ctx->put.upload_length = evbuffer_get_length(ctx->put.evbuf);
if(ctx->put.upload_length == 0)
{
if(cache_kick_combine_minio(ctx))
{
ctx->put.state = PUT_STATE_END;
}
else
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
tango_cache_ctx_destroy(ctx);
}
}
else
{
ret = http_put_bodypart_request_evbuf(ctx, false);
if(ret <= 0)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
if(cache_cancel_upload_minio(ctx))
{
ctx->put.state = PUT_STATE_CANCEL;
}
else
{
tango_cache_ctx_destroy(ctx);
}
}
}
}
break;
case PUT_STATE_END: assert(0); //<2F>û<EFBFBD><C3BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>endʱ<64><CAB1><EFBFBD><EFBFBD><EFBFBD>ܴ<EFBFBD><DCB4>ڴ<EFBFBD>״̬
case PUT_STATE_WAIT_START: //<2F><>ʱδ<CAB1><CEB4>ȡ<EFBFBD><C8A1>uploadId<49><64><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>޷<EFBFBD><DEB7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ϴ<EFBFBD>
default: break;
}
return ret;
}
void tango_cache_curl_put_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code)
{
switch(ctx->put.state)
{
case PUT_STATE_WAIT_START:
if(res!=CURLE_OK||res_code!=200L|| ctx->fail_state || !parse_uploadID_xml(ctx->response.buff, ctx->response.len, &ctx->put.uploadID))
{
easy_string_destroy(&ctx->response);
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
if(ctx->put.close_state)
{
tango_cache_ctx_destroy(ctx);
}
}
else
{
easy_string_destroy(&ctx->response);
ctx->put.state = PUT_STATE_PART;
if(ctx->put.close_state)
{
cache_kick_upload_minio_end(ctx);
}
else
{
size_t upload_length = evbuffer_get_length(ctx->put.evbuf);
if(upload_length >= ctx->instance->upload_block_size)
{
cache_kick_upload_minio_multipart(ctx, upload_length);
}
}
}
break;
case PUT_STATE_PART:
if(res != CURLE_OK || res_code!=200L)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
}
if(ctx->fail_state)
{
if(cache_cancel_upload_minio(ctx))
{
ctx->put.state = PUT_STATE_CANCEL;
}
else if(ctx->put.close_state)
{
tango_cache_ctx_destroy(ctx);
}
}
else if(ctx->put.close_state)
{
cache_kick_upload_minio_end(ctx);
}
else
{
size_t upload_length = evbuffer_get_length(ctx->put.evbuf);
if(upload_length >= ctx->instance->upload_block_size)
{
cache_kick_upload_minio_multipart(ctx, upload_length);
}
}
break;
case PUT_STATE_CANCEL: //<2F>ȴ<EFBFBD><C8B4>ر<EFBFBD>
if(ctx->put.close_state)
{
tango_cache_ctx_destroy(ctx);
}
break;
case PUT_STATE_END:
if(res != CURLE_OK || res_code!=200L)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
}
tango_cache_ctx_destroy(ctx);
break;
default: break;
}
}
int tango_cache_upload_once_start_data(struct tango_cache_ctx *ctx, enum PUT_MEMORY_COPY_WAY way, const char *data, size_t size)
{
CURLMcode rc;
char minio_url[256];
ctx->instance->statistic.put_recv_num += 1;
ctx->instance->error_code = CACHE_OK;
if(NULL == (ctx->curl=curl_easy_init()))
{
tango_cache_ctx_destroy(ctx);
if(way == PUT_MEM_FREE)
{
free((void *)data);
}
return -1;
}
ctx->put.state = PUT_STATE_END;
snprintf(minio_url, 256, "http://%s/%s/%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key);
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_any_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
curl_easy_setopt(ctx->curl, CURLOPT_HTTPHEADER, ctx->headers);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_TIME, 2L);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_LIMIT, 100L);
if(way == PUT_MEM_COPY)
{
ctx->response.buff = (char *)malloc(size);
memcpy(ctx->response.buff, data, size);
}
else
{
ctx->response.buff = (char *)data;
}
ctx->response.size = size;
ctx->response.len = 0;
ctx->instance->statistic.memory_used += size;
curl_easy_setopt(ctx->curl, CURLOPT_UPLOAD, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_INFILESIZE, ctx->response.size);
curl_easy_setopt(ctx->curl, CURLOPT_READFUNCTION, curl_put_once_send_cb);
curl_easy_setopt(ctx->curl, CURLOPT_READDATA, ctx);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return 0;
}
int tango_cache_upload_once_start_evbuf(struct tango_cache_ctx *ctx, enum EVBUFFER_COPY_WAY way, struct evbuffer *evbuf)
{
size_t size;
ctx->instance->statistic.put_recv_num += 1;
ctx->instance->error_code = CACHE_OK;
size = evbuffer_get_length(evbuf);
if(way == EVBUFFER_MOVE)
{
if(evbuffer_add_buffer(ctx->put.evbuf, evbuf))
{
return -1;
}
}
else
{
if(evbuffer_add_buffer_reference(ctx->put.evbuf, evbuf))
{
return -1;
}
}
ctx->instance->statistic.memory_used += size;
return http_put_complete_part_evbuf(ctx);
}
void tango_cache_curl_del_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code)
{
if(res!=CURLE_OK || (res_code!=204L && res_code!=200L ))
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
}
tango_cache_ctx_destroy(ctx);
}
void tango_cache_curl_muldel_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code)
{
u_int32_t errnum=0;
if(res!=CURLE_OK || (res_code!=204L && res_code!=200L ))
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
ctx->del.fail_num = ctx->del.succ_num;
ctx->del.succ_num = 0;
}
else
{
if(!parse_multidelete_xml(ctx->response.buff, ctx->response.len, &errnum, ctx->error, CURL_ERROR_SIZE))
{
ctx->del.fail_num = ctx->del.succ_num;
ctx->del.succ_num = 0;
}
else
{
ctx->del.fail_num = errnum;
ctx->del.succ_num -= errnum;
}
if(ctx->del.fail_num > 0)
{
tango_cache_set_fail_state(ctx, CACHE_ERR_CURL);
}
}
tango_cache_ctx_destroy(ctx);
}
int tango_cache_multi_delete_start(struct tango_cache_ctx *ctx)
{
CURLMcode rc;
char minio_url[256];
ctx->instance->statistic.del_recv_num += ctx->del.succ_num;
ctx->instance->error_code = CACHE_OK;
if(NULL == (ctx->curl=curl_easy_init()))
{
tango_cache_ctx_destroy(ctx);
return -1;
}
snprintf(minio_url, 256, "http://%s/%s/?delete", ctx->hostaddr, ctx->instance->bucketname);
curl_easy_setopt(ctx->curl, CURLOPT_POST, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_POSTFIELDSIZE, ctx->response.size); //<2F><><EFBFBD><EFBFBD>Content-Length<74><68><EFBFBD><EFBFBD>CURLOPT_COPYPOSTFIELDS֮ǰ<D6AE><C7B0><EFBFBD><EFBFBD>
curl_easy_setopt(ctx->curl, CURLOPT_COPYPOSTFIELDS, ctx->response.buff);
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_HTTPHEADER, ctx->headers);
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_response_body_save_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_TIME, 2L);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_LIMIT, 100L);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
easy_string_destroy(&ctx->response);
return 0;
}
void tango_cache_curl_get_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code)
{
switch(ctx->get.state)
{
case GET_STATE_START:
if(!ctx->fail_state)
{
if(res!=CURLE_OK || res_code!=200L)
{
tango_cache_set_fail_state(ctx, (res!=CURLE_OK)?CACHE_ERR_CURL:CACHE_CACHE_MISS);
promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
}
else
{
promise_success(future_to_promise(ctx->future), NULL);
}
}
tango_cache_ctx_destroy(ctx);
break;
case GET_STATE_DELETE:
if(cache_delete_minio_object(ctx))
{
ctx->get.state = GET_STATE_END;
}
else
{
tango_cache_ctx_destroy(ctx);
}
break;
case GET_STATE_END:
tango_cache_ctx_destroy(ctx);
break;
default: assert(0);break;
}
}
static size_t curl_get_response_body_cb(void *ptr, size_t size, size_t count, void *userp)
{
struct tango_cache_ctx *ctx = (struct tango_cache_ctx *)userp;
struct tango_cache_result result;
if(ctx->fail_state || ctx->get.state==GET_STATE_DELETE)
{
return size*count;
}
if(ctx->get.need_hdrs!=RESPONSE_HDR_ALL) //<2F><>Expiresʱ
{
tango_cache_set_fail_state(ctx, CACHE_ERR_INTERNAL);
ctx->get.state = GET_STATE_DELETE;
promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
return size*count;
}
if(ctx->get.response_tag.len > 0)
{
result.data_frag = ctx->get.response_tag.buff;
result.size = ctx->get.response_tag.len;
result.type = RESULT_TYPE_USERTAG;
promise_success(future_to_promise(ctx->future), &result);
easy_string_destroy(&ctx->get.response_tag);
}
if(ctx->response.len > 0)
{
result.data_frag = ctx->response.buff;
result.size = ctx->response.len;
result.type = RESULT_TYPE_HEADER;
promise_success(future_to_promise(ctx->future), &result);
easy_string_destroy(&ctx->response);
}
result.data_frag = ptr;
result.size = size * count;
result.type = RESULT_TYPE_BODY;
promise_success(future_to_promise(ctx->future), &result);
return size*count;
}
static bool check_expires_header(struct tango_cache_ctx *ctx, const char *expires_val, size_t len)
{
time_t time_gmt;
ctx->get.expires = expires_hdr2timestamp(expires_val, len);
time_gmt = get_gmtime_timestamp(time(NULL));
if(time_gmt > ctx->get.expires) //<2F><><EFBFBD><EFBFBD>ʧЧ<CAA7><D0A7>TODO relative_age<67>ĺ<EFBFBD><C4BA><EFBFBD><EFBFBD><EFBFBD>ɶ
{
tango_cache_set_fail_state(ctx, CACHE_TIMEOUT);
ctx->get.state = GET_STATE_DELETE; //<2F><><EFBFBD><EFBFBD>ʧЧʱ<D0A7><CAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><CAB1><EFBFBD><EFBFBD>ɾ<EFBFBD><C9BE><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
easy_string_destroy(&ctx->response);
promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
return false;
}
return true;
}
static bool check_fresh_header(struct tango_cache_ctx *ctx)
{
time_t now_gmt;
if(ctx->get.need_hdrs != RESPONSE_HDR_ALL)
return true;
now_gmt = get_gmtime_timestamp(time(NULL));
if(ctx->get.last_modify+ctx->get.max_age > now_gmt || now_gmt+ctx->get.min_fresh>ctx->get.expires)
{
tango_cache_set_fail_state(ctx, CACHE_TIMEOUT);
easy_string_destroy(&ctx->response);
promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
return false;
}
return true;
}
static bool check_get_result_code(struct tango_cache_ctx *ctx)
{
CURLcode code;
code = curl_easy_getinfo(ctx->curl, CURLINFO_RESPONSE_CODE, &ctx->res_code);
if(code != CURLE_OK || ctx->res_code!=200L)
{
tango_cache_set_fail_state(ctx, (code!=CURLE_OK)?CACHE_ERR_CURL:CACHE_CACHE_MISS);
promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
return false;
}
return true;
}
static size_t curl_get_response_header_cb(void *ptr, size_t size, size_t count, void *userp)
{
struct tango_cache_ctx *ctx = (struct tango_cache_ctx *)userp;
char *start=(char *)ptr, *pos_colon;
size_t raw_len = size*count, hdrlen=size*count;
char usertag[2048];
size_t datalen;
if(ctx->fail_state || ctx->get.state==GET_STATE_DELETE)
{
return raw_len;
}
if(ctx->res_code==0 && !check_get_result_code(ctx)) //<2F>״<EFBFBD>Ӧ<EFBFBD><D3A6>ʱ<EFBFBD>ȿ<EFBFBD>Ӧ<EFBFBD><D3A6><EFBFBD><EFBFBD><EFBFBD>Ƿ<EFBFBD><C7B7><EFBFBD>200
{
return raw_len;
}
pos_colon = (char*)memchr(start, ':', raw_len);
if(pos_colon == NULL)
{
return raw_len;
}
datalen = pos_colon - start;
switch(datalen)
{
case 7:
if(strcmp_one_word_mesa_equal_len("expires", "EXPIRES", start, 7))
{
ctx->get.need_hdrs |= RESPONSE_HDR_EXPIRES;
if(!check_expires_header(ctx, pos_colon + 1, raw_len - datalen - 1) || !check_fresh_header(ctx))
{
return raw_len;
}
}
break;
case 13:
if(strcmp_one_word_mesa_equal_len("x-amz-meta-lm", "X-AMZ-META-LM", start, 13))
{
ctx->get.need_hdrs |= RESPONSE_HDR_LAST_MOD;
sscanf(pos_colon+1, "%lu", &ctx->get.last_modify);
if(!check_fresh_header(ctx))
{
return raw_len;
}
}
break;
case 15:
if(strcmp_one_word_mesa_equal_len("x-amz-meta-user", "X-AMZ-META-USER", start, 15))
{
if((hdrlen = Base64_DecodeBlock((unsigned char*)pos_colon+1, raw_len-datalen-1, (unsigned char*)usertag, 2048))>0)
{
easy_string_savedata(&ctx->get.response_tag, usertag, hdrlen);
}
}
break;
case 11: if(strcmp_one_word_mesa_equal_len("content-md5", "CONTENT-MD5", start, 11)) easy_string_savedata(&ctx->response, (const char*)ptr, raw_len); break;
case 12: if(strcmp_one_word_mesa_equal_len("content-type", "CONTENT-TYPE", start, 12)) easy_string_savedata(&ctx->response, (const char*)ptr, raw_len); break;
case 14: if(strcmp_one_word_mesa_equal_len("content-length", "CONTENT-LENGTH", start, 14)) easy_string_savedata(&ctx->response, (const char*)ptr, raw_len); break;
case 16: if(strcmp_one_word_mesa_equal_len("content-encoding", "CONTENT-ENCODING", start, 16)) easy_string_savedata(&ctx->response, (const char*)ptr, raw_len); break;
case 19: if(strcmp_one_word_mesa_equal_len("content-disposition", "CONTENT-DISPOSITION", start, 19)) easy_string_savedata(&ctx->response, (const char*)ptr, raw_len); break;
default: break;
}
return raw_len;
}
int tango_cache_fetch_start(struct tango_cache_ctx *ctx)
{
CURLMcode rc;
char minio_url[256];
ctx->instance->statistic.get_recv_num += 1;
if(NULL == (ctx->curl=curl_easy_init()))
{
tango_cache_ctx_destroy(ctx);
return -1;
}
snprintf(minio_url, 256, "http://%s/%s/%s", ctx->hostaddr, ctx->instance->bucketname, ctx->object_key);
curl_easy_setopt(ctx->curl, CURLOPT_URL, minio_url);
curl_easy_setopt(ctx->curl, CURLOPT_USERAGENT, "aws-sdk-cpp/1.5.24 Linux/3.10.0-327.el7.x86_64 x86_64 pangu_cache");
curl_easy_setopt(ctx->curl, CURLOPT_NOSIGNAL,1L);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEFUNCTION, curl_get_response_body_cb);
curl_easy_setopt(ctx->curl, CURLOPT_WRITEDATA, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_ERRORBUFFER, ctx->error);
curl_easy_setopt(ctx->curl, CURLOPT_PRIVATE, ctx);
curl_easy_setopt(ctx->curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(ctx->curl, CURLOPT_CONNECTTIMEOUT_MS, 500L);
curl_easy_setopt(ctx->curl, CURLOPT_HEADERFUNCTION, curl_get_response_header_cb);
curl_easy_setopt(ctx->curl, CURLOPT_HEADERDATA, ctx);
//ctx->error="Operation too slow. Less than 1024 bytes/sec transferred the last 3 seconds"
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_TIME, 2L);
curl_easy_setopt(ctx->curl, CURLOPT_LOW_SPEED_LIMIT, 100L);
rc = curl_multi_add_handle(ctx->instance->multi_hd, ctx->curl);
assert(rc==CURLM_OK);
return 0;
}

26
cache/src/tango_cache_transfer.h vendored Normal file
View File

@@ -0,0 +1,26 @@
#ifndef __TANGO_CACHE_UPLOAD_H__
#define __TANGO_CACHE_UPLOAD_H__
#include <sys/queue.h>
#include <curl/curl.h>
#include "tango_cache_client_in.h"
void tango_cache_curl_put_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code);
void tango_cache_curl_get_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code);
void tango_cache_curl_del_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code);
void tango_cache_curl_muldel_done(struct tango_cache_ctx *ctx, CURLcode res, long res_code);
int cache_delete_minio_object(struct tango_cache_ctx *ctx);
int tango_cache_multi_delete_start(struct tango_cache_ctx *ctx);
int cache_kick_upload_minio_end(struct tango_cache_ctx *ctx);
bool cache_kick_upload_minio_multipart(struct tango_cache_ctx *ctx, size_t block_len);
int tango_cache_upload_once_start_data(struct tango_cache_ctx *ctx, enum PUT_MEMORY_COPY_WAY way, const char *data, size_t size);
int tango_cache_upload_once_start_evbuf(struct tango_cache_ctx *ctx, enum EVBUFFER_COPY_WAY way, struct evbuffer *evbuf);
int tango_cache_fetch_start(struct tango_cache_ctx *ctx);
#endif

171
cache/src/tango_cache_xml.cpp vendored Normal file
View File

@@ -0,0 +1,171 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <libxml/parser.h>
#include "tango_cache_xml.h"
bool parse_uploadID_xml(const char *content, int len, char **uploadID)
{
xmlDoc *pdoc;
xmlNode *pcur;
if((pdoc = xmlParseMemory(content, len)) == NULL)
{
return false;
}
if((pcur = xmlDocGetRootElement(pdoc)) == NULL)
{
xmlFreeDoc(pdoc);
return false;
}
while(pcur->type != XML_ELEMENT_NODE)
pcur = pcur->next;
if(xmlStrcmp(pcur->name, (const xmlChar *)"InitiateMultipartUploadResult"))
{
xmlFreeDoc(pdoc);
return false;
}
pcur = pcur->children;
while(pcur != NULL)
{
if(pcur->type != XML_ELEMENT_NODE || xmlStrcmp(pcur->name, (const xmlChar *)"UploadId"))
{
pcur = pcur->next;
continue;
}
*uploadID = (char *)xmlNodeGetContent(pcur);
xmlFreeDoc(pdoc);
return true;
}
xmlFreeDoc(pdoc);
return false;
}
void construct_complete_xml(struct tango_cache_ctx *ctx, char **xml, int *len)
{
struct multipart_etag_list *etag;
xmlDoc *pdoc;
xmlNode *root, *child;
char number[20];
pdoc = xmlNewDoc((const xmlChar *)"1.0");
root = xmlNewNode(NULL, (const xmlChar *)"CompleteMultipartUpload");
xmlNewProp(root, (const xmlChar *)"xmlns",(const xmlChar *)"http://s3.amazonaws.com/doc/2006-03-01/");
xmlDocSetRootElement(pdoc, root);
TAILQ_FOREACH(etag, &ctx->put.etag_head, node)
{
sprintf(number, "%u", etag->part_number);
child = xmlNewChild(root, NULL, (const xmlChar*)"Part", NULL);
xmlNewChild(child, NULL, (const xmlChar*)"ETag", (const xmlChar*)etag->etag);
xmlNewChild(child, NULL, (const xmlChar*)"PartNumber", (const xmlChar*)number);
}
xmlDocDumpFormatMemory(pdoc, (xmlChar **)xml, len, 1);
xmlFreeDoc(pdoc);
}
static void fill_multidelete_xml_errcode(xmlNode *error, char *out, int size)
{
xmlChar *errcode;
xmlNode *child = error->children;
while(child != NULL)
{
if(child->type != XML_ELEMENT_NODE || xmlStrcmp(child->name, (const xmlChar *)"Message"))
{
child = child->next;
continue;
}
errcode = xmlNodeGetContent(child);
snprintf(out, size, "%s", (char *)errcode);
xmlFree(errcode);
break;
}
}
bool parse_multidelete_xml(const char *xml, int len, u_int32_t *errnum, char *errstr, int size)
{
xmlDoc *pdoc;
xmlNode *pcur;
int errornum=0;
if((pdoc = xmlParseMemory(xml, len)) == NULL)
{
return false;
}
if((pcur = xmlDocGetRootElement(pdoc)) == NULL)
{
xmlFreeDoc(pdoc);
return false;
}
while(pcur->type != XML_ELEMENT_NODE)
pcur = pcur->next;
if(xmlStrcmp(pcur->name, (const xmlChar *)"DeleteResult"))
{
xmlFreeDoc(pdoc);
return false;
}
pcur = pcur->children;
while(pcur != NULL)
{
if(pcur->type != XML_ELEMENT_NODE || xmlStrcmp(pcur->name, (const xmlChar *)"Error"))
{
pcur = pcur->next;
continue;
}
if(errornum == 0)
{
fill_multidelete_xml_errcode(pcur, errstr, size);
}
errornum++;
pcur = pcur->next;
}
*errnum = errornum;
xmlFreeDoc(pdoc);
return true;
}
void construct_multiple_delete_xml(const char *bucket, char *key[], u_int32_t num, int is_hash, char **xml, size_t *len)
{
xmlDoc *pdoc;
xmlNode *root, *child;
int xmllen;
pdoc = xmlNewDoc((const xmlChar *)"1.0");
root = xmlNewNode(NULL, (const xmlChar *)"Delete");
xmlDocSetRootElement(pdoc, root);
xmlNewChild(root, NULL, (const xmlChar*)"Quiet", (const xmlChar*)"true");
if(is_hash)
{
char hashkey[72], sha256[72];
for(u_int32_t i=0; i<num; i++)
{
child = xmlNewChild(root, NULL, (const xmlChar*)"Object", NULL);
caculate_sha256(key[i], strlen(key[i]), sha256, 72);
snprintf(hashkey, 256, "%c%c/%c%c/%s", sha256[0], sha256[1], sha256[2], sha256[3], sha256+4);
xmlNewChild(child, NULL, (const xmlChar*)"Key", (const xmlChar*)hashkey);
}
}
else
{
for(u_int32_t i=0; i<num; i++)
{
child = xmlNewChild(root, NULL, (const xmlChar*)"Object", NULL);
xmlNewChild(child, NULL, (const xmlChar*)"Key", (const xmlChar*)key[i]);
}
}
xmlDocDumpFormatMemoryEnc(pdoc, (xmlChar **)xml, &xmllen, "UTF-8", 1);
xmlFreeDoc(pdoc);
*len = xmllen;
}

12
cache/src/tango_cache_xml.h vendored Normal file
View File

@@ -0,0 +1,12 @@
#ifndef __TANGO_CACHE_XML_H__
#define __TANGO_CACHE_XML_H__
#include "tango_cache_client_in.h"
bool parse_uploadID_xml(const char *content, int len, char **uploadID);
void construct_complete_xml(struct tango_cache_ctx *ctx, char **xml, int *len);
void construct_multiple_delete_xml(const char *bucket, char *key[], u_int32_t num, int is_hash, char **xml, size_t *len);
bool parse_multidelete_xml(const char *xml, int len, u_int32_t *errnum, char *errstr, int size);
#endif