| | |
| | | static inline int _shm_socket_check_key(shm_socket_t *socket) { |
| | | void *tmp_ptr = mm_get_by_key(socket->key); |
| | | if (tmp_ptr!= NULL && tmp_ptr != (void *)1 && !socket->force_bind ) { |
| | | bus_errno = ESHM_BUS_KEY_INUSED; |
| | | logger->error("%s. key = %d ", bus_strerror(bus_errno), socket->key); |
| | | bus_errno = EBUS_KEY_INUSED; |
| | | logger->error("%s. key = %d ", bus_strerror(EBUS_KEY_INUSED), socket->key); |
| | | return 0; |
| | | } |
| | | return 1; |
| | |
| | | } else { |
| | | |
| | | if(!_shm_socket_check_key(socket)) { |
| | | bus_errno = ESHM_BUS_KEY_INUSED; |
| | | return ESHM_BUS_KEY_INUSED; |
| | | bus_errno = EBUS_KEY_INUSED; |
| | | return EBUS_KEY_INUSED; |
| | | } |
| | | } |
| | | |
| | |
| | | msg.size = 0; |
| | | msg.type = SHM_SOCKET_OPEN_REPLY; |
| | | |
| | | if (client_socket->remoteQueue->push_timeout(msg, &timeout) == 0) { |
| | | if (client_socket->remoteQueue->push(msg, &timeout, BUS_TIMEOUT_FLAG) == 0) { |
| | | client_socket->status = SHM_CONN_ESTABLISHED; |
| | | return client_socket; |
| | | } else { |
| | |
| | | socket->key = hashtable_alloc_key(hashtable); |
| | | } else { |
| | | if(!_shm_socket_check_key(socket)) { |
| | | bus_errno = ESHM_BUS_KEY_INUSED; |
| | | return ESHM_BUS_KEY_INUSED; |
| | | bus_errno = EBUS_KEY_INUSED; |
| | | return EBUS_KEY_INUSED; |
| | | } |
| | | } |
| | | |
| | |
| | | msg.key = socket->key; |
| | | msg.size = 0; |
| | | msg.type = SHM_SOCKET_OPEN; |
| | | socket->remoteQueue->push_timeout(msg, &timeout); |
| | | socket->remoteQueue->push(msg, &timeout, BUS_TIMEOUT_FLAG); |
| | | |
| | | //接受open reply |
| | | if (socket->queue->pop(msg) == 0) { |
| | |
| | | |
| | | // 短连接方式发送 |
| | | int shm_sendto(shm_socket_t *socket, const void *buf, const int size, |
| | | const int key, const struct timespec *timeout, const int flags) { |
| | | const int key, const struct timespec *timeout, const int flag) { |
| | | |
| | | int s; |
| | | bool rv; |
| | | int rv; |
| | | |
| | | if (socket->socket_type != SHM_SOCKET_DGRAM) { |
| | | logger->error( "shm_socket.shm_sendto: Can't invoke shm_sendto method in a %d type socket which is " |
| | |
| | | } else { |
| | | |
| | | if(!_shm_socket_check_key(socket)) { |
| | | bus_errno = ESHM_BUS_KEY_INUSED; |
| | | return ESHM_BUS_KEY_INUSED; |
| | | bus_errno = EBUS_KEY_INUSED; |
| | | return EBUS_KEY_INUSED; |
| | | } |
| | | |
| | | } |
| | |
| | | |
| | | if ((s = pthread_mutex_unlock(&(socket->mutex))) != 0) |
| | | err_exit(s, "shm_sendto : pthread_mutex_unlock"); |
| | | |
| | | // There is some case where a socket need to send to himeself, for example when bus server need to stop, he need to send himself |
| | | // a top message. |
| | | |
| | | // if (key == socket->key) { |
| | | // logger->error( "can not send to your self!"); |
| | |
| | | dest.buf = mm_malloc(size); |
| | | memcpy(dest.buf, buf, size); |
| | | |
| | | |
| | | if(flags & SHM_MSG_NOWAIT != 0) { |
| | | rv = remoteQueue->push_nowait(dest); |
| | | } else if(timeout != NULL) { |
| | | rv = remoteQueue->push_timeout(dest, timeout); |
| | | } else { |
| | | rv = remoteQueue->push(dest); |
| | | } |
| | | rv = remoteQueue->push(dest, timeout, flag); |
| | | |
| | | if (rv == 0) { |
| | | // printf("shm_sendto push after\n"); |
| | | delete remoteQueue; |
| | | return 0; |
| | | } else { |
| | | delete remoteQueue; |
| | | mm_free(dest.buf); |
| | | if(rv == EBUS_TIMEOUT) { |
| | | // bus_errno = EBUS_TIMEOUT; |
| | | // logger->error("sendto key %d failed, %s", key, bus_strerror(EBUS_TIMEOUT)); |
| | | return EBUS_TIMEOUT; |
| | | } else { |
| | | //logger->error(errno, "sendto key %d failed!", key); |
| | | return rv; |
| | | } |
| | | |
| | | |
| | | logger->debug("sendto key %d failed %s", key, bus_strerror(rv)); |
| | | return rv; |
| | | } |
| | | } |
| | | |
| | | // 短连接方式接受 |
| | | int shm_recvfrom(shm_socket_t *socket, void **buf, int *size, int *key, struct timespec *timeout, int flags) { |
| | | int s; |
| | | bool rv; |
| | | int shm_recvfrom2(shm_socket_t *socket, void **buf, int *size, int *key, const struct timespec *timeout, int flag) { |
| | | return 501; |
| | | } |
| | | |
| | | if (socket->socket_type != SHM_SOCKET_DGRAM) { |
| | | // 短连接方式接受 |
| | | int shm_recvfrom(shm_socket_t *sokt, void **buf, int *size, int *key, const struct timespec *timeout, int flag) { |
| | | int s; |
| | | int rv; |
| | | |
| | | if (sokt->socket_type != SHM_SOCKET_DGRAM) { |
| | | logger->error("shm_socket.shm_recvfrom: Can't invoke shm_recvfrom method in a %d type socket which " |
| | | "is not a SHM_SOCKET_DGRAM socket ", |
| | | socket->socket_type); |
| | | sokt->socket_type); |
| | | exit(1); |
| | | } |
| | | hashtable_t *hashtable = mm_get_hashtable(); |
| | | |
| | | if ((s = pthread_mutex_lock(&(socket->mutex))) != 0) |
| | | if ((s = pthread_mutex_lock(&(sokt->mutex))) != 0) |
| | | err_exit(s, "shm_recvfrom : pthread_mutex_lock"); |
| | | |
| | | if (socket->queue == NULL) { |
| | | if (socket->key == 0) { |
| | | socket->key = hashtable_alloc_key(hashtable); |
| | | if (sokt->queue == NULL) { |
| | | if (sokt->key == 0) { |
| | | sokt->key = hashtable_alloc_key(hashtable); |
| | | } else { |
| | | |
| | | if(!_shm_socket_check_key(socket)) { |
| | | bus_errno = ESHM_BUS_KEY_INUSED; |
| | | return ESHM_BUS_KEY_INUSED; |
| | | if(!_shm_socket_check_key(sokt)) { |
| | | bus_errno = EBUS_KEY_INUSED; |
| | | return EBUS_KEY_INUSED; |
| | | } |
| | | } |
| | | |
| | | socket->queue = new SHMQueue<shm_msg_t>(socket->key, 16); |
| | | sokt->queue = new SHMQueue<shm_msg_t>(sokt->key, 16); |
| | | } |
| | | |
| | | if ((s = pthread_mutex_unlock(&(socket->mutex))) != 0) |
| | | if ((s = pthread_mutex_unlock(&(sokt->mutex))) != 0) |
| | | err_exit(s, "shm_recvfrom : pthread_mutex_unlock"); |
| | | |
| | | shm_msg_t src; |
| | | |
| | | if(flags & SHM_MSG_NOWAIT != 0) { |
| | | rv = socket->queue->pop_nowait(src); |
| | | } else if(timeout != NULL) { |
| | | rv = socket->queue->pop_timeout(src, timeout); |
| | | } else { |
| | | rv = socket->queue->pop(src); |
| | | } |
| | | printf ("====== before ======\n"); |
| | | rv = sokt->queue->pop(src, timeout, flag); |
| | | printf ("====== after ======\n %d", rv); |
| | | |
| | | if (rv == 0) { |
| | | if(buf != NULL) { |
| | |
| | | mm_free(src.buf); |
| | | return 0; |
| | | } else { |
| | | |
| | | if(rv == EBUS_TIMEOUT) { |
| | | // logger->error("shm_recvfrom failed, %s", bus_strerror(EBUS_TIMEOUT)); |
| | | return EBUS_TIMEOUT; |
| | | } else { |
| | | // logger->error(rv, "shm_recvfrom failed!"); |
| | | return rv; |
| | | } |
| | | logger->debug("shm_recvfrom failed %s", bus_strerror(rv)); |
| | | return rv; |
| | | |
| | | } |
| | | } |
| | |
| | | // use thread local |
| | | int _shm_sendandrecv_thread_local(shm_socket_t *socket, const void *send_buf, |
| | | const int send_size, const int send_key, void **recv_buf, |
| | | int *recv_size, struct timespec *timeout, int flags) { |
| | | int *recv_size, const struct timespec *timeout, int flags) { |
| | | int recv_key; |
| | | int rv; |
| | | |
| | |
| | | if (tmp_socket == NULL) |
| | | { |
| | | /* If first call from this thread, allocate buffer for thread, and save its location */ |
| | | logger->debug("%d create tmp socket\n", pthread_self() ); |
| | | logger->debug("%ld create tmp socket\n", (long)pthread_self() ); |
| | | tmp_socket = shm_open_socket(SHM_SOCKET_DGRAM); |
| | | |
| | | rv = pthread_setspecific(_tmp_recv_socket_key_, tmp_socket); |
| | |
| | | |
| | | if ((rv = shm_sendto(tmp_socket, send_buf, send_size, send_key, timeout, flags)) == 0) { |
| | | rv = shm_recvfrom(tmp_socket, recv_buf, recv_size, &recv_key, timeout, flags); |
| | | |
| | | return rv; |
| | | } else { |
| | | |
| | | return rv; |
| | | } |
| | | return -1; |
| | | } |
| | | |
| | | int _shm_sendandrecv_alloc_new(shm_socket_t *socket, const void *send_buf, |
| | | const int send_size, const int send_key, void **recv_buf, |
| | | int *recv_size, struct timespec *timeout, int flags) { |
| | | int *recv_size, const struct timespec *timeout, int flags) { |
| | | int recv_key; |
| | | int rv; |
| | | |
| | |
| | | |
| | | int shm_sendandrecv_unsafe(shm_socket_t *socket, const void *send_buf, |
| | | const int send_size, const int send_key, void **recv_buf, |
| | | int *recv_size, struct timespec *timeout, int flags) { |
| | | int *recv_size, const struct timespec *timeout, int flags) { |
| | | if (socket->socket_type != SHM_SOCKET_DGRAM) { |
| | | logger->error( "shm_socket.shm_sendandrecv_unsafe : Can't invoke shm_sendandrecv method in a %d type socket " |
| | | "which is not a SHM_SOCKET_DGRAM socket ", |
| | |
| | | |
| | | int shm_sendandrecv(shm_socket_t *socket, const void *send_buf, |
| | | const int send_size, const int send_key, void **recv_buf, |
| | | int *recv_size, struct timespec *timeout, int flags) { |
| | | int *recv_size, const struct timespec *timeout, int flags) { |
| | | return _shm_sendandrecv_thread_local(socket, send_buf, send_size, send_key,recv_buf, recv_size, timeout, flags); |
| | | } |
| | | |
| | |
| | | |
| | | switch (src.type) { |
| | | case SHM_SOCKET_OPEN: |
| | | socket->acceptQueue->push_timeout(src, &timeout); |
| | | socket->acceptQueue->push(src, &timeout, BUS_TIMEOUT_FLAG); |
| | | break; |
| | | case SHM_SOCKET_CLOSE: |
| | | _server_close_conn_to_client(socket, src.key); |
| | |
| | | if (iter != socket->clientSocketMap->end()) { |
| | | client_socket = iter->second; |
| | | // print_msg("_server_run_msg_rev push before", src); |
| | | client_socket->messageQueue->push_timeout(src, &timeout); |
| | | client_socket->messageQueue->push(src, &timeout, BUS_TIMEOUT_FLAG); |
| | | // print_msg("_server_run_msg_rev push after", src); |
| | | } |
| | | |
| | |
| | | _client_close_conn_to_server(socket); |
| | | break; |
| | | case SHM_COMMON_MSG: |
| | | socket->messageQueue->push_timeout(src, &timeout); |
| | | socket->messageQueue->push(src, &timeout, BUS_TIMEOUT_FLAG); |
| | | break; |
| | | default: |
| | | logger->error( "shm_socket._client_run_msg_rev: undefined message type."); |
| | |
| | | close_msg.size = 0; |
| | | close_msg.type = SHM_SOCKET_CLOSE; |
| | | if (notifyRemote && socket->remoteQueue != NULL) { |
| | | socket->remoteQueue->push_timeout(close_msg, &timeout); |
| | | socket->remoteQueue->push(close_msg, &timeout, BUS_TIMEOUT_FLAG); |
| | | } |
| | | |
| | | if (socket->queue != NULL) { |
| | |
| | | socket->queue = NULL; |
| | | } |
| | | |
| | | if (socket->remoteQueue != NULL) { |
| | | delete socket->remoteQueue; |
| | | socket->remoteQueue = NULL; |
| | | } |
| | | |
| | | |
| | | if (socket->messageQueue != NULL) { |
| | | delete socket->messageQueue; |
| | |
| | | iter != socket->clientSocketMap->end(); iter++) { |
| | | client_socket = iter->second; |
| | | |
| | | client_socket->remoteQueue->push_timeout(close_msg, &timeout); |
| | | delete client_socket->remoteQueue; |
| | | client_socket->remoteQueue->push(close_msg, &timeout, BUS_TIMEOUT_FLAG); |
| | | client_socket->remoteQueue = NULL; |
| | | |
| | | delete client_socket->messageQueue; |