wangzhengquan
2021-02-04 83956b12d863924936a98c9dfbece37feb0cce9c
update
6个文件已修改
79 ■■■■■ 已修改文件
CMakeLists.txt 2 ●●● 补丁 | 查看 | 原始文档 | blame | 历史
build.sh 1 ●●●● 补丁 | 查看 | 原始文档 | blame | 历史
src/CMakeLists.txt 4 ●●●● 补丁 | 查看 | 原始文档 | blame | 历史
src/net/net_mod_socket.cpp 32 ●●●● 补丁 | 查看 | 原始文档 | blame | 历史
src/net/net_mod_socket.h 4 ●●● 补丁 | 查看 | 原始文档 | blame | 历史
src/socket/shm_socket.cpp 36 ●●●●● 补丁 | 查看 | 原始文档 | blame | 历史
CMakeLists.txt
@@ -13,7 +13,7 @@
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY "${PROJECT_BINARY_DIR}/lib")
# set(CMAKE_RUNTIME_OUTPUT_DIRECTORY "${PROJECT_BINARY_DIR}")
#option(BUILD_SHARED_LIBS "Build using shared libraries" ON)
option(BUILD_SHARED_LIBS "Build using shared libraries" ON)
option(BUILD_DOC "Build doc" OFF)
build.sh
@@ -16,6 +16,7 @@
  
  "debug")
  BUILD_TYPE="Debug"
  BUILD_SHARED_LIBS="ON"
  ;;
  "doc")
src/CMakeLists.txt
@@ -29,7 +29,7 @@
)
add_library(shm_queue ${_SOURCES_})
add_library(shm_queue SHARED ${_SOURCES_})
target_include_directories(shm_queue PUBLIC ${EXTRA_INCLUDES} )
@@ -46,7 +46,7 @@
target_link_libraries(shm_queue PUBLIC  ${EXTRA_LIBS} )
# install rules
install(TARGETS shm_queue DESTINATION lib)
install(TARGETS shm_queue  DESTINATION lib)
install(FILES 
  ./socket/socket_def.h
  ./socket/bus_server_socket.h
src/net/net_mod_socket.cpp
@@ -19,30 +19,30 @@
  if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)
    logger->error(errno, "NetModSocket::NetModSocket signal");
  gpool = new NetConnPool();
  // gpool = new NetConnPool();
  pthread_mutexattr_t mtxAttr;
  s = pthread_mutexattr_init(&mtxAttr);
  if (s != 0)
    err_exit(s, "pthread_mutexattr_init");
  s = pthread_mutexattr_settype(&mtxAttr, PTHREAD_MUTEX_ERRORCHECK);
  if (s != 0)
    err_exit(s, "pthread_mutexattr_settype");
  s = pthread_mutex_init(&sendMutex, &mtxAttr);
  if (s != 0)
    err_exit(s, "pthread_mutex_init");
  // pthread_mutexattr_t mtxAttr;
  // s = pthread_mutexattr_init(&mtxAttr);
  // if (s != 0)
  //   err_exit(s, "pthread_mutexattr_init");
  // s = pthread_mutexattr_settype(&mtxAttr, PTHREAD_MUTEX_ERRORCHECK);
  // if (s != 0)
  //   err_exit(s, "pthread_mutexattr_settype");
  // s = pthread_mutex_init(&sendMutex, &mtxAttr);
  // if (s != 0)
  //   err_exit(s, "pthread_mutex_init");
  s = pthread_mutexattr_destroy(&mtxAttr);
  if (s != 0)
    err_exit(s, "pthread_mutexattr_destroy");
  // s = pthread_mutexattr_destroy(&mtxAttr);
  // if (s != 0)
  //   err_exit(s, "pthread_mutexattr_destroy");
}
NetModSocket::~NetModSocket() {
  int s;
  delete gpool;
  s =  pthread_mutex_destroy(&sendMutex);
  // delete gpool;
  // s =  pthread_mutex_destroy(&sendMutex);
  if(s != 0) {
    err_exit(s, "shm_close_socket");
  }
src/net/net_mod_socket.h
@@ -62,10 +62,8 @@
private:
   
  ShmModSocket shmModSocket;
  // pool req_resp_pool;
  NetConnPool *gpool;
  pthread_mutex_t sendMutex;
  // pthread_mutex_t sendMutex;
  // request header 编码为网络传输的字节
  static void * encode_request_head(net_mod_request_head_t & request);
src/socket/shm_socket.cpp
@@ -156,8 +156,11 @@
  shm_packet_t sendpak = {0};
  sendpak.key = sockt->key;
  sendpak.size = size;
  sendpak.buf = mm_malloc(size);
  memcpy(sendpak.buf, buf, size);
  if(buf != NULL) {
    sendpak.buf = mm_malloc(size);
    memcpy(sendpak.buf, buf, size);
  }
  rv = shm_sendpakto(sockt, &sendpak, key, timeout, flag);
  return rv;
}
@@ -176,8 +179,10 @@
  
  sendpak.key = sockt->key;
  sendpak.size = send_size;
  sendpak.buf = mm_malloc(send_size);
  memcpy(sendpak.buf, send_buf, send_size);
  if(send_buf != NULL) {
    sendpak.buf = mm_malloc(send_size);
    memcpy(sendpak.buf, send_buf, send_size);
  }
  memcpy(sendpak.uuid, uuid.c_str(), uuid.length() + 1);
  // uuid.copy(sendpak.uuid, sizeof sendpak.uuid);
  rv = shm_sendpakto(sockt, &sendpak, key, timeout, flags);
@@ -251,7 +256,7 @@
  
  int rv;
 
  void *sendbuf = NULL;
  void *sendbuf, *recvbuf = NULL;
  int sendsize = 0;
  shm_packet_t recvpak = {0};
@@ -259,18 +264,20 @@
  if (rv != 0) {
    if(rv == ETIMEDOUT){
      logger->debug("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(EBUS_TIMEOUT));
      logger->debug("%d shm_recvandsend failed %s", shm_socket_get_key(sockt), bus_strerror(EBUS_TIMEOUT));
      return EBUS_TIMEOUT;
    }
    
    logger->error("%d shm_recvfrom failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
    logger->error("%d shm_recvandsend failed %s", shm_socket_get_key(sockt), bus_strerror(rv));
    return rv;
  }
   
  void *recvbuf = malloc(recvpak.size);
  memcpy(recvbuf, recvpak.buf, recvpak.size);
  mm_free(recvpak.buf);
  if(recvpak.buf != NULL) {
    recvbuf = malloc(recvpak.size);
    memcpy(recvbuf, recvpak.buf, recvpak.size);
    mm_free(recvpak.buf);
  }
  callback(recvbuf, recvpak.size, recvpak.key, &sendbuf, &sendsize, user_data);
  shm_packet_t sendpak = {0};
@@ -280,6 +287,10 @@
  if(sendbuf !=NULL && sendsize > 0) {
    sendpak.buf = mm_malloc(sendsize);
    memcpy(sendpak.buf, sendbuf, sendsize);
  } else {
    logger->warn("%d shm_recvandsend : sendbuf is null", shm_socket_get_key(sockt));
    // return -1;
  }
 
  rv = shm_sendpakto(sockt, &sendpak, recvpak.key, timeout, flag);
@@ -307,7 +318,7 @@
  } 
 if(buf != NULL) {
 if(buf != NULL && recvpak.buf != NULL) {
    void *_buf = malloc(recvpak.size);
    memcpy(_buf, recvpak.buf, recvpak.size);
    *buf = _buf; 
@@ -544,6 +555,7 @@
LABEL_POP:
 
  printf("%p start recv.....\n", sockt);
 
  rv = sockt->queue->pop(*recvpak, timeout, flag);