#include "usg_common.h" #include "dgram_mod_socket.h" #include "shm_socket.h" #include "shm_allocator.h" #include "mem_pool.h" #include "hashtable.h" #include "sem_util.h" #include "logger_factory.h" typedef struct dgram_mod_socket_t { socket_mod_t mod; shm_socket_t *shm_socket; pthread_t recv_thread; std::map * > *recv_queue_map; } dgram_mod_socket_t; void *dgram_mod_open_socket(int mod) { dgram_mod_socket_t * socket = (dgram_mod_socket_t *)calloc(1, sizeof(dgram_mod_socket_t)); socket->mod = (socket_mod_t)mod; socket->recv_thread = 0; socket->recv_queue_map = NULL; socket->shm_socket = shm_open_socket(SHM_SOCKET_DGRAM); printf("socket->shm_socket = %p \n" , socket->shm_socket); return (void *)socket; } int dgram_mod_close_socket(void * _socket) { dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket; shm_close_socket(socket->shm_socket); if(socket->recv_queue_map != NULL) { for(auto iter = socket->recv_queue_map->begin(); iter != socket->recv_queue_map->end(); iter++) { delete iter->second; socket->recv_queue_map->erase(iter); } delete socket->recv_queue_map; } if(socket->recv_thread != 0) pthread_cancel(socket->recv_thread); free(_socket); } int dgram_mod_bind(void * _socket, int port){ dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket; return shm_socket_bind(socket->shm_socket, port); } int dgram_mod_sendto(void *_socket, const void *buf, const int size, const int port) { dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket; return shm_sendto(socket->shm_socket, buf, size, port); } int dgram_mod_recvfrom(void *_socket, void **buf, int *size, int *port) { dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket; if(socket->mod == REQ_REP && socket->recv_thread != 0) { err_exit(0, "you have used sendandrecv method os you can not use recvfrom method any more. these two method can not be used at the same time."); return -1; } return shm_recvfrom(socket->shm_socket, buf, size, port); } void *_dgram_mod_run_recv(void * _socket) { pthread_detach(pthread_self()); dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket; void *buf; int size; int port; shm_msg_t msg; LockFreeQueue *queue; std::map * >::iterator iter; // printf("==============_dgram_mod_run_recv recv before\n"); while(shm_recvfrom(socket->shm_socket, &buf, &size, &port) == 0) { if( (iter = socket->recv_queue_map->find(port) ) != socket->recv_queue_map->end()) { queue = iter->second; } else { queue = new LockFreeQueue(16); socket->recv_queue_map->insert({port, queue}); } msg.buf = buf; msg.size = size; msg.port = port; // printf("==============_dgram_mod_run_recv push before\n"); queue->push(msg); // printf("==============_dgram_mod_run_recv push after\n"); } return NULL; } int dgram_mod_sendandrecv(void * _socket, const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size) { dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket; if(socket->mod != REQ_REP) { err_exit(0, "you can't use this method other than REQ_REP mod!"); } if(socket->recv_queue_map == NULL) { socket->recv_queue_map = new std::map * >; } std::map * >::iterator iter; LockFreeQueue *queue; if( (iter = socket->recv_queue_map->find(port) ) != socket->recv_queue_map->end()) { queue = iter->second; } else { queue = new LockFreeQueue(16); socket->recv_queue_map->insert({port, queue}); } if (socket->recv_thread == 0) { pthread_create(&(socket->recv_thread ), NULL, _dgram_mod_run_recv , _socket); } shm_sendto(socket->shm_socket, send_buf, send_size, port); shm_msg_t msg; // printf("==============dgram_mod_sendandrecv pop before\n"); if(queue->pop(msg)) { *recv_buf = msg.buf; *recv_size = msg.size; // printf("==============dgram_mod_sendandrecv pop after\n"); return 0; } return -1; }