| | |
| | | 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; |
| | | } |
| | | } |
| | | |
| | |
| | | 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; |
| | | } |
| | | } |
| | | |
| | |
| | | const int key, const struct timespec *timeout, const int flags) { |
| | | |
| | | 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!"); |
| | |
| | | } else { |
| | | delete remoteQueue; |
| | | mm_free(dest.buf); |
| | | if(rv == EBUS_TIMEOUT) { |
| | | if(rv > EBUS_BASE) { |
| | | // bus_errno = EBUS_TIMEOUT; |
| | | // logger->error("sendto key %d failed, %s", key, bus_strerror(EBUS_TIMEOUT)); |
| | | return EBUS_TIMEOUT; |
| | | logger->debug("sendto key %d failed %s", key, bus_strerror(rv)); |
| | | } else { |
| | | //logger->error(errno, "sendto key %d failed!", key); |
| | | return rv; |
| | | logger->error(rv, "sendto key %d failed", key); |
| | | } |
| | | 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 rv; |
| | | |
| | | if (socket->socket_type != SHM_SOCKET_DGRAM) { |
| | | logger->error("shm_socket.shm_recvfrom: Can't invoke shm_recvfrom method in a %d type socket which " |
| | |
| | | } 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; |
| | | } |
| | | } |
| | | |
| | |
| | | rv = socket->queue->pop_nowait(src); |
| | | } else if(timeout != NULL) { |
| | | rv = socket->queue->pop_timeout(src, timeout); |
| | | // printf("0 shm_recvfrom====%d\n", rv); |
| | | } else { |
| | | rv = socket->queue->pop(src); |
| | | } |
| | |
| | | mm_free(src.buf); |
| | | return 0; |
| | | } else { |
| | | |
| | | if(rv == EBUS_TIMEOUT) { |
| | | // logger->error("shm_recvfrom failed, %s", bus_strerror(EBUS_TIMEOUT)); |
| | | return EBUS_TIMEOUT; |
| | | if(rv > EBUS_BASE) { |
| | | logger->debug("shm_recvfrom failed %s", bus_strerror(rv)); |
| | | } else { |
| | | // logger->error(rv, "shm_recvfrom failed!"); |
| | | return rv; |
| | | logger->error(rv, "shm_recvfrom failed"); |
| | | } |
| | | return 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, |