fix:解决并行锁管理问题

Change-Id: I2bab6f8d258c9980fe36cf4b1b2ba2c3c5e3fe0e

Author: hongchunhua

CMakeLists.txt

@@ -38,7 +38,7 @@ endif()
 set(VERSION_FILE_PATH ${XIO_ROOT_PATH})
 
 #版本
-set(VERSION_NUM "xio_1.7.0_release")
+set(VERSION_NUM "v1.0.0")
 
 set(VERSION_MSG "const char XIO_VERSION_STRING[] = \"${VERSION_NUM}\"\;")
 file(WRITE ${VERSION_FILE_PATH}/version.c ${VERSION_MSG})

inc/arpc_api.h

@@ -58,6 +58,14 @@ enum arpc_ctrl_attr{
 	ARPC_E_CTRL_MAX = (1<<31), 		/*! @brief 最大标记位*/
 };
 
+/*!
+ *  @brief  arpc版本
+ *
+ *  @return  char, version
+ *
+ */
+const char *arpc_version();
+
 /*!
  *  @brief  arpc消息框架全局初始化，支持参数设置
  *

src/session/arpc_client.c

@@ -141,7 +141,7 @@ arpc_session_handle_t arpc_client_create_session(const struct arpc_client_sessio
 		ret = arpc_session_connect_for_client(session, 3*1000);//等待至少一条链路可用
 		LOG_THEN_GOTO_TAG_IF_VAL_TRUE(ret, error_2, "client session connect server[%s] fail", client_ctx->uri);
 	}
-
+	ARPC_LOG_NOTICE("ARPC version[%s].", arpc_version());
 	ARPC_LOG_NOTICE("Create session[%p] success, work thread num[%u], rx num[%u]!!", session, idle_thread_num, rx_con_num);
 	return (arpc_session_handle_t)session;
 

src/session/arpc_com.c

@@ -26,6 +26,7 @@
 #include "arpc_response.h"
 #include "crc64.h"
 
+static const char *version = "v1.0.0";
 struct aprc_paramter{
 	uint16_t is_init;
 	uint16_t pad;
@@ -34,6 +35,11 @@ struct aprc_paramter{
 	struct aprc_option opt;
 };
 
+const char *arpc_version()
+{
+	return xio_version();
+}
+
 static struct aprc_paramter g_param= {
 	.is_init        = 0,
 	.thread_pool	= NULL,

src/session/arpc_connection.c

@@ -593,6 +593,7 @@ static int arpc_client_run_loop(void * thread_ctx)
 		ctx->xio_con_ctx = NULL;
 	}
 	ctx->flags = 0;
+	SET_FLAG(ctx->flags, ARPC_CONN_ATTR_TELL_LIVE);
 	prctl(PR_SET_NAME, "share_thread");
 	arpc_cond_notify(&ctx->cond);
 	ARPC_LOG_NOTICE("xio connection[%u] on thread[%lu] exit now.", con->id,  pthread_self());
@@ -677,6 +678,7 @@ static void arpc_tx_event_callback(struct arpc_connection *usr_conn)
 		}
 		arpc_mutex_unlock(&con->msg_lock);
 		ARPC_LOG_TRACE("xio send msg on client, msg type:%d", msg->type);
+		arpc_cond_lock(&msg->cond);
 		switch (msg->type)
 		{
 			case ARPC_MSG_TYPE_REQ:
@@ -687,18 +689,21 @@ static void arpc_tx_event_callback(struct arpc_connection *usr_conn)
 			case ARPC_MSG_TYPE_RSP:
 				usr_conn->tx_rsp_count++;
 				ret = xio_send_response(msg->tx_msg);
+				arpc_cond_notify(&msg->cond);
 				statistics_per_time(&msg->now, &usr_conn->tx_rsp_cmit, 3);
 				break;
 			case ARPC_MSG_TYPE_OW:
 				usr_conn->tx_ow_count++;
 				ret = xio_send_msg(con->xio_con, msg->tx_msg);
+				arpc_cond_notify(&msg->cond);
 				statistics_per_time(&msg->now, &usr_conn->tx_ow_cmit, 3);
 				break;
 			default:
 				ret = ARPC_ERROR;
 				ARPC_LOG_ERROR("unkown msg type[%d]", msg->type);
 				break;
 		}
+		arpc_cond_unlock(&msg->cond);
 		ARPC_LOG_TRACE("xio send msg end, msg type:%d", msg->type);
 		if(ret){
 			ARPC_LOG_ERROR("send msg[%d] fail, errno code[%u], err msg[%s].", msg->type, xio_errno(), xio_strerror(xio_errno()));

src/session/arpc_process_request.c

@@ -26,6 +26,7 @@
 #include "arpc_request.h"
 #include "arpc_response.h"
 
+static const int32_t SEND_RSP_END_MAX_TIME_MS = 2*1000;
 static int request_msg_async_deal(void *usr_ctx);
 
 int process_request_header(struct arpc_connection *con, struct xio_msg *msg)
@@ -60,7 +61,7 @@ int process_request_data(struct arpc_connection *con, struct xio_msg *req, int l
 	uint32_t			i;
 	struct arpc_vmsg 	rev_iov;
 	struct arpc_rsp 	usr_rsp_param;
-	struct arpc_common_msg *rsp_hanlde;
+	struct arpc_common_msg *rsp_msg;
 	struct arpc_rsp_handle	*rsp_fd_ex;
 	struct arpc_thread_param *async_param;
 	int						ret;
@@ -83,14 +84,14 @@ int process_request_data(struct arpc_connection *con, struct xio_msg *req, int l
 	ret = destroy_xio_msg_usr_buf(req, ops->free_cb, usr_ctx);
 	LOG_THEN_RETURN_VAL_IF_TRUE((ret), ARPC_ERROR, "destroy_xio_msg_usr_buf fail.");
 
-	rsp_hanlde = get_common_msg(con, ARPC_MSG_TYPE_RSP);
-	LOG_THEN_RETURN_VAL_IF_TRUE(!rsp_hanlde, ARPC_ERROR, "rsp_hanlde alloc null.");
-	rsp_fd_ex = (struct arpc_rsp_handle*)rsp_hanlde->ex_data;
+	rsp_msg = get_common_msg(con, ARPC_MSG_TYPE_RSP);
+	LOG_THEN_RETURN_VAL_IF_TRUE(!rsp_msg, ARPC_ERROR, "rsp_msg alloc null.");
+	rsp_fd_ex = (struct arpc_rsp_handle*)rsp_msg->ex_data;
 	rsp_fd_ex->x_rsp_msg = req;//保存回复的结构体
-	rsp_hanlde->attr.rsp_crc = attr.req_crc;//请求保存在回复体里
-	rsp_hanlde->attr.req_crc = 0;
+	rsp_msg->attr.rsp_crc = attr.req_crc;//请求保存在回复体里
+	rsp_msg->attr.req_crc = 0;
 	memset(&usr_rsp_param, 0, sizeof(struct arpc_rsp));
-	usr_rsp_param.rsp_fd = (void *)rsp_hanlde;
+	usr_rsp_param.rsp_fd = (void *)rsp_msg;
 
 	if(IS_SET(req->usr_flags, METHOD_ARPC_PROC_SYNC) && ops->proc_data_cb){
 		ARPC_LOG_TRACE("process rx request msg with async.");
@@ -122,7 +123,7 @@ int process_request_data(struct arpc_connection *con, struct xio_msg *req, int l
 		async_param->ops.proc_async_cb = ops->proc_async_cb;
 		async_param->ops.release_rsp_cb = ops->release_rsp_cb;
 		async_param->ops.proc_oneway_async_cb = NULL;
-		async_param->rsp_ctx = rsp_hanlde;
+		async_param->rsp_ctx = rsp_msg;
 		async_param->rev_iov = rev_iov;
 		async_param->req_msg = NULL;
 		async_param->usr_ctx = usr_ctx;
@@ -140,20 +141,19 @@ int process_request_data(struct arpc_connection *con, struct xio_msg *req, int l
 	free_msg_xio2arpc(&rev_iov, ops->free_cb, usr_ctx);
 
 do_respone:	
-	/* attach request to response */
+	/* 框架内回复，则不需要通知模式，也不需要加锁 */
 	ARPC_LOG_TRACE("do respone request msg.");
 	gettimeofday(&tx_time, NULL);	// 线程安全
 
-	rsp_hanlde->attr.tx_sec= tx_time.tv_sec;
-	rsp_hanlde->attr.tx_usec= tx_time.tv_usec;
-	rsp_hanlde->attr.conn_id = con->id;
+	rsp_msg->attr.tx_sec= tx_time.tv_sec;
+	rsp_msg->attr.tx_usec= tx_time.tv_usec;
+	rsp_msg->attr.conn_id = con->id;
 
-	ret = arpc_init_response(rsp_hanlde);
+	ret = arpc_init_response(rsp_msg);
 	LOG_ERROR_IF_VAL_TRUE(ret, "arpc_init_response fail.");
-	if(!ret){
-		ret = arpc_connection_async_send(rsp_hanlde->conn, rsp_hanlde);
-		LOG_ERROR_IF_VAL_TRUE(ret, "arpc_connection_async_send fail.");
-	}
+	ret = xio_send_response(rsp_msg->tx_msg);
+	LOG_ERROR_IF_VAL_TRUE(ret, "xio_send_response fail.");
+
 	return ret;
 }
 
@@ -162,17 +162,17 @@ static int request_msg_async_deal(void *usr_ctx)
 	struct arpc_thread_param *async = (struct arpc_thread_param *)usr_ctx;
 	struct arpc_rsp rsp;
 	int ret;
-	struct arpc_common_msg *rsp_fd;
+	struct arpc_common_msg *rsp_msg;
 	struct arpc_rsp_handle	*rsp_fd_ex;
 	struct timeval 		tx_time;
 
 	LOG_THEN_RETURN_VAL_IF_TRUE(!async, ARPC_ERROR, "async null.");
 	LOG_THEN_RETURN_VAL_IF_TRUE(!async->ops.proc_async_cb, ARPC_ERROR, "request proc_async_cb null.");
 	LOG_THEN_RETURN_VAL_IF_TRUE(!async->rsp_ctx, ARPC_ERROR, "request rsp context is null.");
 
-	rsp_fd  = (struct arpc_common_msg *)async->rsp_ctx;
+	rsp_msg  = (struct arpc_common_msg *)async->rsp_ctx;
 	memset(&rsp, 0, sizeof (struct arpc_rsp));						
-	rsp.rsp_fd = (void*)rsp_fd;
+	rsp.rsp_fd = (void*)rsp_msg;
 
 	ARPC_LOG_TRACE("process request msg with async.");
 	ret = async->ops.proc_async_cb(&async->rev_iov, &rsp, async->usr_ctx);
@@ -185,21 +185,26 @@ static int request_msg_async_deal(void *usr_ctx)
 		async->rev_iov.vec = NULL;
 	}
 
-	rsp_fd_ex = (struct arpc_rsp_handle*)rsp_fd->ex_data;
+	rsp_fd_ex = (struct arpc_rsp_handle*)rsp_msg->ex_data;
 	rsp_fd_ex->rsp_usr_iov = rsp.rsp_iov;
 	rsp_fd_ex->rsp_usr_ctx = rsp.rsp_ctx;
 	rsp_fd_ex->release_rsp_cb = async->ops.release_rsp_cb;
 
 	if (!IS_SET(rsp.flags, METHOD_CALLER_ASYNC)) {
 		gettimeofday(&tx_time, NULL);	// 线程安全
-		rsp_fd->attr.tx_sec= tx_time.tv_sec;
-		rsp_fd->attr.tx_usec= tx_time.tv_usec;
-		ret = arpc_init_response(rsp_fd);
-		LOG_ERROR_IF_VAL_TRUE(ret, "arpc_do_respone fail.");
-		if(!ret){
-			ret = arpc_connection_async_send(rsp_fd->conn, rsp_fd);
-			LOG_ERROR_IF_VAL_TRUE(ret, "arpc_do_respone fail.");
+		arpc_cond_lock(&rsp_msg->cond);
+		rsp_msg->attr.tx_sec= tx_time.tv_sec;
+		rsp_msg->attr.tx_usec= tx_time.tv_usec;
+		ret = arpc_init_response(rsp_msg);
+		LOG_ERROR_IF_VAL_TRUE(ret, "arpc_init_response fail.");
+		ret = arpc_connection_async_send(rsp_msg->conn, rsp_msg);
+		if(!ret) {
+			ret = arpc_cond_wait_timeout(&rsp_msg->cond, SEND_RSP_END_MAX_TIME_MS); // 默认等待
+			LOG_ERROR_IF_VAL_TRUE(ret, "rsp send timeout fail, conn[%u], msg[%p].", rsp_msg->conn->id, rsp_msg);
+		}else{
+			ARPC_LOG_ERROR("arpc_connection_async_send fail, conn[%u], msg[%p].", rsp_msg->conn->id, rsp_msg);
 		}
+		arpc_cond_unlock(&rsp_msg->cond);
 	}
 	SAFE_FREE_MEM(async->rev_iov.head);
 	SAFE_FREE_MEM(async);

src/session/arpc_request.c

@@ -204,10 +204,9 @@ int arpc_send_oneway_msg(const arpc_session_handle_t fd, struct arpc_vmsg *send,
 	ret = convert_msg_arpc2xio(send, &req->out, &req_msg->attr);
 	LOG_THEN_GOTO_TAG_IF_VAL_TRUE(ret, free_common_msg, "convert xio msg fail.");
 	req->user_context = req_msg;
-	if (!ow_msg->clean_send_cb){
+	/*if (!ow_msg->clean_send_cb){
 		req->flags |= XIO_MSG_FLAG_IMM_SEND_COMP;
-	}
-
+	}*/
 	arpc_cond_lock(&req_msg->cond);
 	while(send_cnt < 1) {
 		send_cnt++;
@@ -224,11 +223,8 @@ int arpc_send_oneway_msg(const arpc_session_handle_t fd, struct arpc_vmsg *send,
 	if (!ow_msg->clean_send_cb){
 		ret = arpc_cond_wait_timeout(&req_msg->cond, SEND_ONEWAY_END_MAX_TIME); // 默认等待
 		arpc_cond_unlock(&req_msg->cond);
-		if (!ret){
-			free_msg_arpc2xio(&req->out);
-			put_common_msg(req_msg);	//un lock
-		}else{
-			free_msg_arpc2xio(&req->out);
+		free_msg_arpc2xio(&req->out);
+		if (ret){
 			ARPC_LOG_ERROR("wait oneway msg send complete timeout fail, msg keep."); // TODO 释放超时的资源
 		}
 	}else{
@@ -263,7 +259,6 @@ int arpc_oneway_send_complete(struct arpc_common_msg *ow_msg)
 		free_msg_arpc2xio(&ow_msg->xio_msg.out);
 		put_common_msg(ow_msg);	//un lock
 	}else{
-		arpc_cond_notify(&ow_msg->cond);
 		arpc_cond_unlock(&ow_msg->cond);
 	}
 	ARPC_LOG_DEBUG("send end complete.");