From 2561a007b8d8999a4750046d0cfb3b1ad5af50ac Mon Sep 17 00:00:00 2001 From: zhangmeng <775834166@qq.com> Date: 星期二, 09 四月 2024 15:29:32 +0800 Subject: [PATCH] test for perf --- src/bh_api.cpp | 740 ++++++++++++++++++++++++++++++++++---------------------- 1 files changed, 452 insertions(+), 288 deletions(-) diff --git a/src/bh_api.cpp b/src/bh_api.cpp index 312b203..c450bc2 100644 --- a/src/bh_api.cpp +++ b/src/bh_api.cpp @@ -3,13 +3,13 @@ #include "bus_server_socket_wrapper.h" #include "shm_mm_wrapper.h" #include "proc_def.h" +#include "mm.h" #include "usg_common.h" #include "bh_api.h" #include <pthread.h> #include <getopt.h> -#include "bhome_msg_api.pb.h" -#include "bhome_msg.pb.h" -#include "error_msg.pb.h" +#include "msg_mgr.h" +#include "../proto/source/error_msg.pb.h" #include "../proto/source/bhome_msg.pb.h" #include "../proto/source/bhome_msg_api.pb.h" @@ -18,14 +18,15 @@ static Logger *logger = LoggerFactory::getLogger(); static int gRun_stat = 0; +static int gRun_flag = true; static void *gNetmod_socket = NULL; static pthread_mutex_t mutex; static pthread_t gTids; -static void *client_run_check(void *skptr) { - + +static void *client_run_check(void *skptr) { pthread_detach(pthread_self()); int data; @@ -38,19 +39,24 @@ sec = TIME_WAIT; nsec = 0; - sprintf(buf, "%s", "Success"); + sprintf(buf, "%s", STR_EXEC); data = net_mod_socket_int_get(gNetmod_socket); - while(true) { + while(gRun_flag == true) { rv = net_mod_socket_recvfrom(gNetmod_socket, &buf_temp, &size, &key, SVR_STR, data); if (rv == 0) { - BHFree(buf_temp, size); + if (strncmp((char *)buf_temp, STR_RET, strlen(STR_RET)) != 0) { - rv = net_mod_socket_sendto_timeout(gNetmod_socket, buf, strlen(buf), key, sec, nsec, SVR_STR, data); - if (rv != 0) { - logger->error("the process check response failed with error: %s!\n", bus_strerror(rv)); + rv = net_mod_socket_sendto_timeout(gNetmod_socket, buf, strlen(buf), key, sec, nsec, SVR_STR, data); + if (rv != 0) { + logger->error("the process check response failed with error: %s!\n", bus_strerror(rv)); + } + } else { + gRun_flag = false; } + + BHFree(buf_temp, size); } else { @@ -84,7 +90,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input.proc_id = input.proc_id().c_str(); @@ -97,13 +103,17 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif memset(&pData, 0x00, sizeof(ProcInfo)); if (gRun_stat == 0) { pthread_mutex_init(&mutex, NULL); + +#if defined(MSG_HANDLER) + msg_init(); +#endif } else { logger->error("the process has already registered!\n"); @@ -111,7 +121,7 @@ rv = EBUS_RES_BUSY; bus_errorset(rv); - return false; + goto exit_entry; } rv = pthread_mutex_trylock(&mutex); @@ -121,25 +131,25 @@ shm_mm_wrapper_init(SHM_RES_SIZE); #if defined(PRO_DE_SERIALIZE) - if (_input.proc_id != NULL) { + if (strlen(_input.proc_id) > 0) { count = strlen(_input.proc_id) + 1; min = count > (MAX_STR_LEN - 1) ? (MAX_STR_LEN - 1) : count; strncpy(pData.proc_id, _input.proc_id, min); } - if (_input.name != NULL) { + if (strlen(_input.name) > 0) { count = strlen(_input.name) + 1; min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN -1) : count; strncpy(pData.name, _input.name, min); } - if (_input.public_info != NULL) { + if (strlen(_input.public_info) > 0) { count = strlen(_input.public_info) + 1; min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN - 1) : count; strncpy(pData.public_info, _input.public_info, min); } - if (_input.private_info != NULL) { + if (strlen(_input.private_info) > 0) { count = strlen(_input.private_info) + 1; min = count > (MAX_STR_LEN - 1)? (MAX_STR_LEN - 1): count; strncpy(pData.private_info, _input.private_info, min); @@ -170,26 +180,27 @@ } #endif - if (pData.proc_id == NULL) { + if (strlen(pData.proc_id) == 0) { rv = EBUS_INVALID_PARA; bus_errorset(rv); + gRun_stat = 0; pthread_mutex_unlock(&mutex); - return false; + goto exit_entry; } gNetmod_socket = net_mod_socket_open(); hashtable_t *hashtable = mm_get_hashtable(); key = hashtable_alloc_key(hashtable); + net_mod_socket_bind(gNetmod_socket, key); count = hashtable_alloc_key(hashtable); rv = hashtable_alloc_key(hashtable); net_mod_socket_int_set(gNetmod_socket, count); net_mod_socket_svr_set(gNetmod_socket, rv); sprintf(pData.int_info, "%d", count); sprintf(pData.svr_info, "%d", rv); - net_mod_socket_bind(gNetmod_socket, key); rv = net_mod_socket_reg(gNetmod_socket, &pData, sizeof(ProcInfo), NULL, 0, timeout_ms, PROC_REG); @@ -202,11 +213,11 @@ rv = EBUS_RES_BUSY; bus_errorset(rv); - return false; + goto exit_entry; } +exit_entry: errString = bus_strerror(0, 1); - #if defined(PRO_DE_SERIALIZE) ::bhome_msg::MsgCommonReply mcr; mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); @@ -225,18 +236,25 @@ #endif - pthread_create(&gTids, NULL, client_run_check, NULL); + if (rv == 0) { + gRun_flag = true; + pthread_create(&gTids, NULL, client_run_check, NULL); - return true; + return true; + } + return false; } int BHUnregister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) { int rv; int min; + int data; + int diff; void *buf = NULL; char *errString = NULL; + struct timeval start, end; #if defined(PRO_DE_SERIALIZE) struct _ProcInfo_proto @@ -254,7 +272,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input.proc_id = input.proc_id().c_str(); @@ -266,7 +284,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -277,14 +295,30 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } rv = pthread_mutex_trylock(&mutex); if (rv == 0) { rv = net_mod_socket_reg(gNetmod_socket, NULL, 0, NULL, 0, timeout_ms, PROC_UNREG); if (rv == 0) { - + gettimeofday(&start, NULL); + data = net_mod_socket_int_get(gNetmod_socket); + rv = net_mod_socket_sendto_timeout(gNetmod_socket, STR_RET, strlen(STR_RET), data, 3, 0); + if (rv != 0) { + logger->error("the process check response failed with error: %s!\n", bus_strerror(rv)); + } + + while(gRun_flag == true) { + sleep(1); + + gettimeofday(&end, NULL); + + diff = end.tv_sec - start.tv_sec; + if (diff >= TIME_DUR) + break; + }; + net_mod_socket_close(gNetmod_socket); gNetmod_socket = NULL; @@ -301,9 +335,10 @@ rv = EBUS_RES_BUSY; bus_errorset(rv); - return false; + goto exit_entry; } +exit_entry: errString = bus_strerror(0, 1); #if defined(PRO_DE_SERIALIZE) ::bhome_msg::MsgCommonReply mcr; @@ -321,8 +356,12 @@ *reply = buf; *reply_len = min; #endif - - return true; + + if (rv == 0) { + return true; + } + + return false; } @@ -349,7 +388,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input.amount = input.topic_list_size(); @@ -368,7 +407,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } total = topics_len; @@ -380,7 +419,7 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } topics_buf = (char *)malloc(total); @@ -391,7 +430,7 @@ logger->error("in BHRegisterTopics: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, total); @@ -420,6 +459,8 @@ free(topics_buf); bus_errorset(rv); + +exit_entry: errString = bus_strerror(0, 1); #if defined(PRO_DE_SERIALIZE) @@ -439,8 +480,11 @@ *reply_len = len; #endif - return true; + if (rv == 0) { + return true; + } + return false; } int BHQueryTopicAddress(const void *remote, const int remote_len, const void *topic, const int topic_len, void **reply, int *reply_len, const int timeout_ms) @@ -464,6 +508,17 @@ }_input0; const char *_input1; + + struct _MsgQueryTopicReply + { + std::string proc_id; + + unsigned long long mq_id; + long long abs_addr; + std::string ip; + int port; + } mtr_list[128]; + int mtr_list_num = 0; ::bhome_msg::BHAddress input0; ::bhome_msg::MsgQueryTopic input1; @@ -471,7 +526,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input0.mq_id = input0.mq_id(); @@ -485,7 +540,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -495,7 +550,7 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } #if defined(PRO_DE_SERIALIZE) @@ -511,35 +566,33 @@ bus_errorset(rv); #if defined(PRO_DE_SERIALIZE) - struct _MsgQueryTopicReply - { - std::string proc_id; - - unsigned long long mq_id; - long long abs_addr; - std::string ip; - int port; - }mtr_list[128]; - int mtr_list_num = 0; - if (rv == 0) { - ptr = (ProcInfo_query *)((char *)buf + sizeof(int)); - mtr_list_num = ptr->num; + min = *(int *)buf; + if (min > 0) { + ptr = (ProcInfo_query *)((char *)buf + sizeof(int)); + mtr_list_num = ptr->num; + + if (mtr_list_num > sizeof(mtr_list) / sizeof(mtr_list[0])) { + mtr_list_num = sizeof(mtr_list) / sizeof(mtr_list[0]); + } - if (mtr_list_num > sizeof(mtr_list) / sizeof(mtr_list[0])) { - mtr_list_num = sizeof(mtr_list) / sizeof(mtr_list[0]); + Proc_ptr = &(ptr->procData); + for(int i = 0; i < mtr_list_num; i++) { + mtr_list[i].proc_id = (Proc_ptr + i)->proc_id; + mtr_list[i].mq_id = ID_RSV; + mtr_list[i].abs_addr = ABS_ID_RSV; + mtr_list[i].ip = "127.0.0.1"; + mtr_list[i].port = 5000; + } + } else { + mtr_list_num = 0; } - - for(int i = 0; i < mtr_list_num; i++) { - mtr_list[i].proc_id = ptr->procData.proc_id; - mtr_list[i].mq_id = ID_RSV; - mtr_list[i].abs_addr = ABS_ID_RSV; - mtr_list[i].ip = "127.0.0.1"; - mtr_list[i].port = 5000; - } + + free(buf); } +exit_entry: errString = bus_strerror(0, 1); ::bhome_msg::MsgQueryTopicReply mtr; @@ -575,8 +628,11 @@ #endif - return true; + if (rv == 0) { + return true; + } + return false; } int BHQueryProcs(const void *remote, const int remote_len, const void *query, const int query_len, void **reply, int *reply_len, const int timeout_ms) @@ -607,7 +663,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input0.mq_id = input0.mq_id(); @@ -620,7 +676,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -630,7 +686,7 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } if (query != NULL) { @@ -639,7 +695,8 @@ rv = net_mod_socket_reg(gNetmod_socket, data_buf, strlen(data_buf), &buf, &size, timeout_ms, PROC_QUE_ATCS); bus_errorset(rv); - + +exit_entry: #if defined(PRO_DE_SERIALIZE) struct _MsgQueryProcReply { @@ -663,52 +720,57 @@ if (mpr_list_num > (sizeof(mpr_list) / sizeof(mpr_list[0]))) { mpr_list_num = sizeof(mpr_list) / sizeof(mpr_list[0]); } - - Proc_ptr = (ProcInfo_sum *)((char *)buf + sizeof(int)); - for(int i = 0; i < mpr_list_num; i++) { - mpr_list[i].proc_id = (Proc_ptr + i)->procData.proc_id; - mpr_list[i].name = (Proc_ptr + i)->procData.name; - mpr_list[i].public_info = (Proc_ptr + i)->procData.public_info; - mpr_list[i].private_info = (Proc_ptr + i)->procData.private_info; - mpr_list[i].online = (Proc_ptr + i)->stat; - mpr_list[i].topic_list_num = (Proc_ptr + i)->list_num; - - for(int j = 0; j < mpr_list[i].topic_list_num; j++) - { - if (j == 0) { - mpr_list[i].topic_list[j] = (Proc_ptr + i)->reg_info; - } else if (j == 1) { - mpr_list[i].topic_list[j] = (Proc_ptr + i)->local_info; - } else if (j == 2) { - mpr_list[i].topic_list[j] = (Proc_ptr + i)->net_info; + + if (mpr_list_num > 0) { + Proc_ptr = (ProcInfo_sum *)((char *)buf + sizeof(int)); + for(int i = 0; i < mpr_list_num; i++) { + mpr_list[i].proc_id = (Proc_ptr + i)->procData.proc_id; + mpr_list[i].name = (Proc_ptr + i)->procData.name; + mpr_list[i].public_info = (Proc_ptr + i)->procData.public_info; + mpr_list[i].private_info = (Proc_ptr + i)->procData.private_info; + mpr_list[i].online = (Proc_ptr + i)->stat; + mpr_list[i].topic_list_num = (Proc_ptr + i)->list_num; + + for(int j = 0; j < mpr_list[i].topic_list_num; j++) + { + if (j == 0) { + mpr_list[i].topic_list[j] = (Proc_ptr + i)->reg_info; + } else if (j == 1) { + mpr_list[i].topic_list[j] = (Proc_ptr + i)->local_info; + } else if (j == 2) { + mpr_list[i].topic_list[j] = (Proc_ptr + i)->net_info; + } } } } - - errString = bus_strerror(0, 1); - ::bhome_msg::MsgQueryProcReply mpr; - mpr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); - mpr.mutable_errmsg()->set_errstring(errString); - - for(int i = 0; i < mpr_list_num; i++) - { - ::bhome_msg::MsgQueryProcReply_Info *mpri = mpr.add_proc_list(); - mpri->mutable_proc()->set_proc_id(mpr_list[i].proc_id); - mpri->mutable_proc()->set_name(mpr_list[i].name); - mpri->mutable_proc()->set_public_info(mpr_list[i].public_info); - mpri->mutable_proc()->set_private_info(mpr_list[i].private_info); - mpri->set_online(mpr_list[i].online); - for(int j = 0; j < mpr_list[i].topic_list_num; j++) - { - mpri->mutable_topics()->add_topic_list(mpr_list[i].topic_list[j]); - } - } - - *reply_len = mpr.ByteSizeLong(); - *reply = malloc(*reply_len); - mpr.SerializePartialToArray(*reply, *reply_len); + free(buf); } + + errString = bus_strerror(0, 1); + + ::bhome_msg::MsgQueryProcReply mpr; + mpr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); + mpr.mutable_errmsg()->set_errstring(errString); + + for(int i = 0; i < mpr_list_num; i++) + { + ::bhome_msg::MsgQueryProcReply_Info *mpri = mpr.add_proc_list(); + mpri->mutable_proc()->set_proc_id(mpr_list[i].proc_id); + mpri->mutable_proc()->set_name(mpr_list[i].name); + mpri->mutable_proc()->set_public_info(mpr_list[i].public_info); + mpri->mutable_proc()->set_private_info(mpr_list[i].private_info); + mpri->set_online(mpr_list[i].online); + for(int j = 0; j < mpr_list[i].topic_list_num; j++) + { + mpri->mutable_topics()->add_topic_list(mpr_list[i].topic_list[j]); + } + } + + *reply_len = mpr.ByteSizeLong(); + *reply = malloc(*reply_len); + mpr.SerializePartialToArray(*reply, *reply_len); + #else if (rv == 0) { *reply = buf; @@ -724,8 +786,11 @@ } #endif - return true; + if (rv == 0) { + return true; + } + return false; } int BHSubscribeTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) @@ -752,7 +817,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input.amount = input.topic_list_size(); @@ -772,7 +837,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -782,7 +847,7 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } topics_buf = (char *)malloc(total); @@ -793,7 +858,7 @@ logger->error("in BHSubscribeTopics: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, total); @@ -832,6 +897,7 @@ } +exit_entry: errString = bus_strerror(0, 1); free(topics_buf); @@ -840,8 +906,8 @@ ::bhome_msg::MsgCommonReply mcr; mcr.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); mcr.mutable_errmsg()->set_errstring(errString); - *reply_len=mcr.ByteSizeLong(); - *reply=malloc(*reply_len); + *reply_len = mcr.ByteSizeLong(); + *reply = malloc(*reply_len); mcr.SerializePartialToArray(*reply,*reply_len); #else len = strlen(errString) + 1; @@ -852,9 +918,12 @@ *reply = buf; *reply_len = len; #endif - - return true; + + if (rv == 0) { + return true; + } + return false; } int BHSubscribeNetTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) @@ -890,7 +959,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input.proc_id = input.proc_id().c_str(); @@ -900,6 +969,8 @@ rv = 0; bus_errorset(rv); + +exit_entry: errString = bus_strerror(0, 1); ::bhome_msg::MsgCommonReply mcr; @@ -910,7 +981,11 @@ mcr.SerializePartialToArray(*reply,*reply_len); #endif - return true; + if (rv == 0) { + return true; + } + + return false; } #if defined(PRO_DE_SERIALIZE) @@ -948,8 +1023,8 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - - return false; + + return false; } #endif @@ -986,9 +1061,7 @@ if (rv > 0) return true; - - bus_errorset(rv); - + return false; } @@ -997,7 +1070,7 @@ int rv; int len; void *buf; - int key; + int key = 0; int size; int sec, nsec; char buf_temp[100] = { 0x00 }; @@ -1017,14 +1090,14 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } if ((msgpub == NULL) || (msgpub_len == NULL)) { rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } if (timeout_ms > 0) { @@ -1055,9 +1128,9 @@ rv = EBUS_NO_MEM; bus_errorset(rv); - logger->error("in BHRequest: Out of memory!\n"); + logger->error("in BHReadSub: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, len + 10); @@ -1071,11 +1144,11 @@ rv = EBUS_NO_MEM; bus_errorset(rv); - logger->error("in BHRequest: Out of memory!\n"); + logger->error("in BHReadSub: Out of memory!\n"); free(topics_buf); - return false; + goto exit_entry; } memset(data_buf, 0x00, size - len + 10); @@ -1093,56 +1166,61 @@ rsr.data = topics_buf; } - sprintf(buf_temp, "%d", key); - - if ((proc_id != NULL) && (proc_id_len != NULL)) { - rsr.proc_id = buf_temp; - *proc_id_len = rsr.proc_id.size(); - *proc_id = malloc(*proc_id_len); - memcpy(*proc_id, rsr.proc_id.data(), *proc_id_len); - } - free(topics_buf); free(data_buf); - - ::bhome_msg::MsgPublish Mp; - Mp.set_topic(rsr.topic); - Mp.set_data(rsr.data.data()); - *msgpub_len = Mp.ByteSizeLong(); - *msgpub = malloc(*msgpub_len); - Mp.SerializePartialToArray(*msgpub, *msgpub_len); -#else - void *ptr; - if (len < size) { - ptr = malloc(size - len); - len = size - len; - memcpy(ptr, data_buf, len); - } else { - ptr = malloc(len); - memcpy(ptr, topics_buf, len); - } - *msgpub = ptr; - *msgpub_len = len; - - free(topics_buf); - free(data_buf); - - if ((proc_id != NULL) && (proc_id_len != NULL)) { - memset(buf_temp, 0x00, sizeof(buf_temp)); - sprintf(buf_temp, "%d", key); - - *proc_id_len = strlen(buf_temp); - *proc_id = malloc(*proc_id_len); - memcpy(*proc_id, buf_temp, *proc_id_len); - } - -#endif - - } else { - - bus_errorset(rv); - } + +exit_entry: + sprintf(buf_temp, "%d", key); + + if ((proc_id != NULL) && (proc_id_len != NULL)) { + rsr.proc_id = buf_temp; + *proc_id_len = rsr.proc_id.size(); + *proc_id = malloc(*proc_id_len); + memcpy(*proc_id, rsr.proc_id.data(), *proc_id_len); + } + + if (rv != 0) { + rsr.topic = STR_RSV; + if (data_buf != NULL) { + rsr.data = data_buf; + } else { + rsr.data = STR_RSV; + } + } + + ::bhome_msg::MsgPublish Mp; + Mp.set_topic(rsr.topic); + Mp.set_data(rsr.data.data()); + *msgpub_len = Mp.ByteSizeLong(); + *msgpub = malloc(*msgpub_len); + Mp.SerializePartialToArray(*msgpub, *msgpub_len); +#else + void *ptr; + if (len < size) { + ptr = malloc(size - len); + len = size - len; + memcpy(ptr, data_buf, len); + } else { + ptr = malloc(len); + memcpy(ptr, topics_buf, len); + } + *msgpub = ptr; + *msgpub_len = len; + + free(topics_buf); + free(data_buf); + + if ((proc_id != NULL) && (proc_id_len != NULL)) { + memset(buf_temp, 0x00, sizeof(buf_temp)); + sprintf(buf_temp, "%d", key); + + *proc_id_len = strlen(buf_temp); + *proc_id = malloc(*proc_id_len); + memcpy(*proc_id, buf_temp, *proc_id_len); + } + +#endif if (rv == 0) return true; @@ -1153,18 +1231,21 @@ int BHAsyncRequest(const void *remote, const int remote_len, const void *request, const int request_len, void **msg_id, int *msg_id_len) { int rv; - void *buf; + void *buf = NULL; int size; - int val; + int val = 0; int len; int min; int data; int sec, nsec; + std::string str; std::string MsgID; int timeout_ms = 3000; + char data_buf[MAX_STR_LEN] = { 0x00 }; char buf_temp[MAX_STR_LEN] = { 0x00 }; char *topics_buf = NULL; - + hashtable_t *hashtable = mm_get_hashtable(); + #if defined(PRO_DE_SERIALIZE) struct _BHAddress { @@ -1187,7 +1268,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } _input0.mq_id = input0.mq_id(); @@ -1203,7 +1284,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -1213,7 +1294,7 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } #if defined(PRO_DE_SERIALIZE) @@ -1221,14 +1302,35 @@ #else strncpy(buf_temp, (char *)request, (sizeof(buf_temp) - 1) > strlen((char *)request) ? strlen((char *)request) : (sizeof(buf_temp) - 1)); #endif - - rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + + str = buf_temp; + val = net_mod_socket_buf_data_get(gNetmod_socket, str); + if ((val > 0) && (hashtable_get(hashtable, val) != NULL)) { + + rv = 0; + + } else { + + if ((val > 0) && (hashtable_get(hashtable, val) == NULL)) { + net_mod_socket_buf_data_del(gNetmod_socket, str); + } + + rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + if (rv == 0) { + + len = size > (sizeof(data_buf) - 1) ? (sizeof(data_buf) - 1) : size; + memcpy(data_buf, (char *)buf, len); + val = atoi((char *)data_buf); + if (val > 0) { + str = buf_temp; + net_mod_socket_buf_data_set(gNetmod_socket, str, val); + } + + free(buf); + } + } + if (rv == 0) { - - val = atoi((char *)buf); - - free(buf); - if (val > 0) { len = strlen(buf_temp) + 1; @@ -1242,9 +1344,9 @@ rv = EBUS_NO_MEM; bus_errorset(rv); - logger->error("in BHRequest: Out of memory!\n"); + logger->error("in BHAsyncRequest: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, len); @@ -1280,6 +1382,8 @@ } bus_errorset(rv); + +exit_entry: if((msg_id == NULL) || (msg_id_len == NULL)) { if (rv == 0) return true; @@ -1287,18 +1391,16 @@ return false; } - if (rv == 0) { - - memset(buf_temp, 0x00, sizeof(buf_temp)); - sprintf(buf_temp, "%d", val); - MsgID = buf_temp; + memset(buf_temp, 0x00, sizeof(buf_temp)); + sprintf(buf_temp, "%d", val); + MsgID = buf_temp; - *msg_id_len = MsgID.size(); - *msg_id = malloc(*msg_id_len); - memcpy(*msg_id, MsgID.data(), *msg_id_len); + *msg_id_len = MsgID.size(); + *msg_id = malloc(*msg_id_len); + memcpy(*msg_id, MsgID.data(), *msg_id_len); + if (rv == 0) return true; - } return false; @@ -1308,21 +1410,24 @@ void **reply, int *reply_len, const int timeout_ms) { int rv; - void *buf; + void *buf = NULL; int size; int val; int min, len; int data; + std::string str; net_node_t node; int node_size; int recv_arr_size; + char data_buf[MAX_STR_LEN] = { 0x00 }; net_mod_recv_msg_t *recv_arr; net_mod_err_t *errarr; int errarr_size = 0; char *errString = NULL; char buf_temp[MAX_STR_LEN] = { 0x00 }; char *topics_buf = NULL; - + hashtable_t *hashtable = mm_get_hashtable(); + struct _RequestReply { std::string proc_id; @@ -1351,9 +1456,10 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } + memset(&node, 0x00, sizeof(node)); _input0.mq_id = input0.mq_id(); _input0.abs_addr = input0.abs_addr(); _input0.ip = input0.ip().c_str(); @@ -1367,7 +1473,7 @@ rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } #endif @@ -1377,7 +1483,7 @@ rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } #if defined(PRO_DE_SERIALIZE) @@ -1385,42 +1491,59 @@ #else strncpy(buf_temp, (char *)request, (sizeof(buf_temp) - 1) > request_len ? request_len : (sizeof(buf_temp) - 1)); #endif - - rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + + str = buf_temp; + val = net_mod_socket_buf_data_get(gNetmod_socket, str); + if ((val > 0) && (hashtable_get(hashtable, val) != NULL)) { + rv = 0; + } else { + if ((val > 0) && (hashtable_get(hashtable, val) == NULL)) { + net_mod_socket_buf_data_del(gNetmod_socket, str); + } + + rv = net_mod_socket_reg(gNetmod_socket, buf_temp, strlen(buf_temp), &buf, &size, timeout_ms, PROC_QUE_STCS); + if (rv == 0) { + + len = size > (sizeof(data_buf) - 1) ? (sizeof(data_buf) - 1) : size; + memcpy(data_buf, (char *)buf, len); + val = atoi((char *)data_buf); + if (val > 0) { + str = buf_temp; + net_mod_socket_buf_data_set(gNetmod_socket, str, val); + } + + free(buf); + } + } + if (rv == 0) { - val = atoi((char *)buf); - - free(buf); - if (val > 0) { memset(&node, 0x00, sizeof(node)); - + len = strlen(buf_temp) + 1; #if defined(PRO_DE_SERIALIZE) len += strlen(_input1.data); #endif - data = net_mod_socket_svr_get(gNetmod_socket); topics_buf = (char *)malloc(len); if (topics_buf == NULL) { - + rv = EBUS_NO_MEM; bus_errorset(rv); - + logger->error("in BHRequest: Out of memory!\n"); - - return false; + + goto exit_entry; } memset(topics_buf, 0x00, len); - + strncpy(topics_buf, buf_temp, strlen(buf_temp) + 1); #if defined(PRO_DE_SERIALIZE) strncpy(topics_buf + strlen(buf_temp) + 1, _input1.data, strlen(_input1.data)); -#endif +#endif node.key = val; - if (timeout_ms > 0) { rv = net_mod_socket_sendandrecv_timeout(gNetmod_socket, &node, 1, topics_buf, len, &recv_arr, &recv_arr_size, &errarr, &errarr_size, timeout_ms); @@ -1441,14 +1564,15 @@ size = recv_arr[0].content_length; buf = (char *)malloc(size); if (buf == NULL) { - printf("Out of memory\n"); + logger->error("in BHRequest: Out of memory!\n"); - exit(0); + free(topics_buf); + + goto exit_entry; } memset((char *)buf, 0x00, size); strncpy((char *)buf, (char *)recv_arr[0].content, size); - } net_mod_socket_free_recv_msg_arr(recv_arr, recv_arr_size); @@ -1458,30 +1582,31 @@ } rv = 0; - } else { + rv = EBUS_TIMEOUT; } } else { rv = EBUS_RES_UNSUPPORT; } - + free(topics_buf); } - + bus_errorset(rv); + + if ((proc_id != NULL) && (proc_id_len != NULL)) { + memset(buf_temp, 0x00, sizeof(buf_temp)); + sprintf(buf_temp, "%d", node.key); + + rr.proc_id = buf_temp; + *proc_id_len = rr.proc_id.size(); + *proc_id = malloc(*proc_id_len); + memcpy(*proc_id, rr.proc_id.c_str(), *proc_id_len); + } + if (rv == 0) { - if ((proc_id != NULL) && (proc_id_len != NULL)) { - memset(buf_temp, 0x00, sizeof(buf_temp)); - sprintf(buf_temp, "%d", node.key); - - rr.proc_id = buf_temp; - *proc_id_len = rr.proc_id.size(); - *proc_id = malloc(*proc_id_len); - memcpy(*proc_id, rr.proc_id.c_str(), *proc_id_len); - } - topics_buf = (char *)malloc(size + 10); if (topics_buf == NULL) { @@ -1490,7 +1615,7 @@ logger->error("in BHRequest: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, size + 10); @@ -1499,33 +1624,40 @@ free(buf); free(topics_buf); + } else { + + rr.data = STR_RSV; } +exit_entry: errString = bus_strerror(0, 1); + if (rv != 0) { + if ((proc_id != NULL) && (proc_id_len != NULL)) { + *proc_id_len = 0; + *proc_id = NULL; + } + } + #if defined(PRO_DE_SERIALIZE) - if (rv == 0) { - ::bhome_msg::MsgRequestTopicReply mrt; - mrt.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); - mrt.mutable_errmsg()->set_errstring(errString); - mrt.set_data(rr.data.data()); - *reply_len = mrt.ByteSizeLong(); - *reply = malloc(*reply_len); - mrt.SerializePartialToArray(*reply, *reply_len); - } + ::bhome_msg::MsgRequestTopicReply mrt; + mrt.mutable_errmsg()->set_errcode(::bhome_msg::ErrorCode(rv)); + mrt.mutable_errmsg()->set_errstring(errString); + mrt.set_data(rr.data.data()); + *reply_len = mrt.ByteSizeLong(); + *reply = malloc(*reply_len); + mrt.SerializePartialToArray(*reply, *reply_len); #else - if (rv == 0) { - min = strlen(errString) + 1; - buf = malloc(min); - memcpy(buf, errString, strlen(errString)); - *((char *)buf + min - 1) = '\0'; + min = strlen(errString) + 1; + buf = malloc(min); + memcpy(buf, errString, strlen(errString)); + *((char *)buf + min - 1) = '\0'; - *reply = buf; - *reply_len = min; - } + *reply = buf; + *reply_len = min; #endif - if (rv == 0) + if (rv == 0) return true; return false; @@ -1542,21 +1674,29 @@ int sec, nsec; char buf_temp[MAX_STR_LEN] = { 0x00 }; char *topics_buf = NULL; - + + struct _ReadRequestReply + { + std::string proc_id; + std::string topic; + std::string data; + void *src; + } rrr; + if (gRun_stat == 0) { logger->error("the process has not been registered yet!\n"); rv = EBUS_RES_NO; bus_errorset(rv); - return false; + goto exit_entry; } if ((request == NULL) || (request_len == 0) || (src == NULL)) { rv = EBUS_INVALID_PARA; bus_errorset(rv); - return false; + goto exit_entry; } data = net_mod_socket_svr_get(gNetmod_socket); @@ -1577,14 +1717,7 @@ } if (rv == 0) { - struct _ReadRequestReply - { - std::string proc_id; - std::string topic; - std::string data; - void *src; - } rrr; - + if ((proc_id != NULL) && (proc_id_len != NULL)) { sprintf(buf_temp, "%d", key); rrr.proc_id = buf_temp; @@ -1602,7 +1735,7 @@ logger->error("in BHReadRequest: Out of memory!\n"); - return false; + goto exit_entry; } memset(topics_buf, 0x00, size + MIN_STR_LEN); @@ -1620,28 +1753,40 @@ rrr.data = topics_buf + len + 1; free(topics_buf); - -#if defined(PRO_DE_SERIALIZE) - ::bhome_msg::MsgRequestTopic mrt; - mrt.set_topic(rrr.topic); - mrt.set_data(rrr.data.data()); - *request_len = mrt.ByteSizeLong(); - *request = malloc(*request_len); - mrt.SerializePartialToArray(*request,*request_len); -#else - *request = buf; - *request_len = size; -#endif - free(buf); - buf = malloc(sizeof(int)); - *(int *)buf = key; - *src = buf; + bus_errorset(rv); } - bus_errorset(rv); - +exit_entry: + + if (rv != 0) { + rrr.topic = STR_RSV; + rrr.data = STR_RSV; + + if ((proc_id != NULL) && (proc_id_len != NULL)) { + + *proc_id_len = 0; + *proc_id = NULL; + } + } + +#if defined(PRO_DE_SERIALIZE) + ::bhome_msg::MsgRequestTopic mrt; + mrt.set_topic(rrr.topic); + mrt.set_data(rrr.data.data()); + *request_len = mrt.ByteSizeLong(); + *request = malloc(*request_len); + mrt.SerializePartialToArray(*request,*request_len); +#else + *request = buf; + *request_len = size; +#endif + + buf = malloc(sizeof(int)); + *(int *)buf = key; + *src = buf; + if (rv == 0) return true; @@ -1730,3 +1875,22 @@ return false; } + +int inter_key_get(void) +{ + if (gNetmod_socket != NULL) + return net_mod_socket_get_key(gNetmod_socket); + + return SHM_BUS_KEY; +} + +void *socket_data_get(void) +{ + return gNetmod_socket; +} + +void inter_key_set(int key) +{ + net_mod_socket_bind(gNetmod_socket, key); +} + -- Gitblit v1.8.0