非结构化接口线程安全

cache/cache_evbase_client.cpp

@@ -23,6 +23,7 @@ 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,
@@ -172,6 +173,11 @@ static void cache_asyn_ioevent_dispatch(struct databuffer *buffer)
 			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);
@@ -292,7 +298,7 @@ int cache_evbase_update_frag_data(struct cache_evbase_ctx *ctx_asyn, enum PUT_ME
 		}
 		free(buffer->data);
 		free(buffer);
-		return -1;
+		return -2;
 	}
 	return 0;
 }
@@ -320,7 +326,7 @@ int cache_evbase_update_frag_evbuf(struct cache_evbase_ctx *ctx_asyn, struct evb
 		}
 		evbuffer_free(buffer->evbuf);
 		free(buffer);
-		return -1;
+		return -2;
 	}
 	return 0;
 }
@@ -329,8 +335,9 @@ struct cache_evbase_ctx *cache_evbase_update_start(struct cache_evbase_instance
 {
 	struct cache_evbase_ctx *ctx_asyn;
 	struct tango_cache_ctx *ctx;
+	struct databuffer *buffer;
 
-	ctx = tango_cache_update_start(instance->instance, future, meta);
+	ctx = tango_cache_update_prepare(instance->instance, future, meta);
 	if(ctx == NULL)
 	{
 		return NULL;
@@ -340,6 +347,19 @@ struct cache_evbase_ctx *cache_evbase_update_start(struct cache_evbase_instance
 	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 *), 0) != sizeof(void *))
+	{
+		ctx_asyn->ctx->fail_state = true;
+		tango_cache_ctx_destroy(ctx_asyn->ctx);
+		cache_asyn_ctx_destroy(ctx_asyn);
+		free(buffer);
+		return NULL;
+	}
 	return ctx_asyn;
 }
 
@@ -382,9 +402,10 @@ int cache_evbase_upload_once_data(struct cache_evbase_instance *instance, struct
 	{
 		free(buffer->data);
 		free(buffer);
+		ctx_asyn->ctx->fail_state = true;
 		tango_cache_ctx_destroy(ctx);
 		cache_asyn_ctx_destroy(ctx_asyn);
-		return -1;
+		return -2;
 	}
 	return 0;
 }
@@ -420,9 +441,10 @@ int cache_evbase_upload_once_evbuf(struct cache_evbase_instance *instance, struc
 	{
 		evbuffer_free(buffer->evbuf);
 		free(buffer);
+		ctx_asyn->ctx->fail_state = true;
 		tango_cache_ctx_destroy(ctx);
 		cache_asyn_ctx_destroy(ctx_asyn);
-		return -1;
+		return -2;
 	}
 	return 0;
 }
@@ -465,6 +487,7 @@ int cache_evbase_delete_object(struct cache_evbase_instance *instance, struct fu
 
 	if(iothread_notify_event(instance->notify_sendfd, &buffer, sizeof(void *), 0) != sizeof(void *))
 	{
+		ctx_asyn->ctx->fail_state = true;
 		tango_cache_ctx_destroy(ctx_asyn->ctx);
 		cache_asyn_ctx_destroy(ctx_asyn);
 		free(buffer);

cache/include/tango_cache_client.h

@@ -42,6 +42,7 @@ struct cache_statistics
 	long long del_recv_num;	//<2F><><EFBFBD><EFBFBD>DELETE<54>Ĵ<EFBFBD><C4B4><EFBFBD>
 	long long del_succ_num;	//DELETE<54>ɹ<EFBFBD><C9B9>Ĵ<EFBFBD><C4B4><EFBFBD>
 	long long del_error_num;//DELETE<54>ɹ<EFBFBD><C9B9>Ĵ<EFBFBD><C4B4><EFBFBD>
+	long long totaldrop_num;//<2F>ڴ<EFBFBD><DAB4><EFBFBD>DROP<4F>Ĵ<EFBFBD><C4B4><EFBFBD>
 	long long memory_used;	//<2F><>ǰUPLOAD BODY<44><59>ռ<EFBFBD>ڴ<EFBFBD><DAB4><EFBFBD>С
 	long long session_num;	//<2F><>ǰ<EFBFBD><C7B0><EFBFBD>ڽ<EFBFBD><DABD><EFBFBD>GET/PUT<55><54>HTTP<54>Ự<EFBFBD><E1BBB0>
 };

cache/tango_cache_client.cpp

@@ -54,6 +54,7 @@ void tango_cache_get_statistics(const struct tango_cache_instance *instance, str
 	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;
 }
@@ -245,8 +246,6 @@ struct tango_cache_ctx *tango_cache_update_prepare(struct tango_cache_instance *
 		instance->error_code = CACHE_OUTOF_MEMORY;
 		return NULL;
 	}
-	instance->statistic.put_recv_num += 1;
-	instance->error_code = CACHE_OK;
 
 	ctx = (struct tango_cache_ctx *)calloc(1, sizeof(struct tango_cache_ctx));
 	ctx->instance = instance;
@@ -301,6 +300,8 @@ struct tango_cache_ctx *tango_cache_update_prepare(struct tango_cache_instance *
 		free(p);
 	}
 
+	ctx->put.evbuf = evbuffer_new();
+	TAILQ_INIT(&ctx->put.etag_head);
 	return ctx;
 }
 
@@ -313,9 +314,8 @@ struct tango_cache_ctx *tango_cache_update_start(struct tango_cache_instance *in
 	{
 		return NULL;
 	}
-
-	ctx->put.evbuf = evbuffer_new();
-	TAILQ_INIT(&ctx->put.etag_head);
+	ctx->instance->statistic.put_recv_num += 1;
+	ctx->instance->error_code = CACHE_OK;
 	return ctx;
 }
 
@@ -367,7 +367,6 @@ struct tango_cache_ctx *tango_cache_fetch_prepare(struct tango_cache_instance *i
 	ctx->get.state = GET_STATE_START;
 	ctx->get.max_age = meta->get.max_age;
 	ctx->get.min_fresh = meta->get.min_fresh;
-	instance->statistic.get_recv_num += 1;
 
 	if(instance->hash_object_key)
 	{
@@ -396,7 +395,6 @@ struct tango_cache_ctx *tango_cache_delete_prepare(struct tango_cache_instance *
 	ctx->instance = instance;
 	ctx->future = future;
 	ctx->method = CACHE_REQUEST_DELETE;
-	instance->statistic.del_recv_num += 1;
 
 	if(instance->hash_object_key)
 	{

cache/tango_cache_transfer.cpp

@@ -204,6 +204,7 @@ 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;
@@ -505,6 +506,8 @@ int tango_cache_upload_once_start_data(struct tango_cache_ctx *ctx, enum PUT_MEM
 	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);
@@ -556,7 +559,9 @@ int tango_cache_upload_once_start_evbuf(struct tango_cache_ctx *ctx, enum EVBUFF
 {
 	size_t size;
 
-	ctx->put.evbuf = evbuffer_new();
+	ctx->instance->statistic.put_recv_num += 1;
+	ctx->instance->error_code = CACHE_OK;
+
 	size = evbuffer_get_length(evbuf);
 	if(way == EVBUFFER_MOVE)
 	{
@@ -639,7 +644,6 @@ static size_t curl_get_response_body_cb(void *ptr, size_t size, size_t count, vo
 		ctx->fail_state = true;
 		ctx->error_code = CACHE_ERR_CURL;
 		ctx->get.state	= GET_STATE_DELETE;
-		ctx->instance->statistic.del_recv_num += 1;
 		promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, "cache Expires or x-amz-meta-lm not found");
 		return size*count;
 	}
@@ -679,7 +683,6 @@ static bool check_expires_header(struct tango_cache_ctx *ctx, const char *expire
 		ctx->fail_state = true;
 		ctx->error_code = 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>
-		ctx->instance->statistic.del_recv_num += 1;
 		easy_string_destroy(&ctx->response);
 		promise_failed(future_to_promise(ctx->future), FUTURE_ERROR_CANCEL, "cache not fresh");
 		return false;
@@ -791,6 +794,7 @@ 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);