| | |
| | | } |
| | | |
| | | queue = ( LockFreeQueue<shm_packet_t> *)tmp_ptr; |
| | | // hashtable_unlock(hashtable); |
| | | return queue; |
| | | } |
| | | |
| | |
| | | return 0; |
| | | } |
| | | |
| | | int shm_socket_bind_proc_id(shm_socket_t *sockt, const char *buf, int len) { |
| | | strncpy(sockt->proc_id, buf, len > MAX_STR_LEN ? MAX_STR_LEN : len); |
| | | |
| | | return 0; |
| | | } |
| | | |
| | | int shm_socket_get_key(shm_socket_t *sockt){ |
| | | return sockt->key; |
| | | } |
| | | |
| | | int shm_socket_get_procid(shm_socket_t *sockt, char *buf, int len) { |
| | | strncpy(buf, sockt->proc_id, len); |
| | | |
| | | return 0; |
| | | } |
| | | |
| | | // 短连接方式发送 |
| | |
| | | int shm_sendandrecv(shm_socket_t *sockt, const void *send_buf, |
| | | const int send_size, const int key, void **recv_buf, |
| | | int *recv_size, const struct timespec *timeout, int flags) { |
| | | // return _shm_sendandrecv_uuid(sockt, send_buf, send_size, key, recv_buf, recv_size, timeout, flags); |
| | | return _shm_sendandrecv_thread_local(sockt, send_buf, send_size, key, recv_buf, recv_size, timeout, flags); |
| | | } |
| | | |
| | |
| | | shm_packet_t sendpak; |
| | | shm_packet_t recvpak; |
| | | std::map<int, shm_packet_t>::iterator recvbufIter; |
| | | // 用thread local 保证每个线程用一个独占的socket接受对方返回的信息 |
| | | shm_socket_t *tmp_socket = NULL; |
| | | |
| | | rv = pthread_once(&_once_, _create_threadlocal_socket_key_); |
| | |
| | | tmp_socket = (shm_socket_t *)pthread_getspecific(_localthread_socket_key_); |
| | | if (tmp_socket == NULL) |
| | | { |
| | | /* If first call from this thread, allocate buffer for thread, and save its location */ |
| | | tmp_socket = shm_socket_open(SHM_SOCKET_DGRAM); |
| | | |
| | | } |
| | | |
| | | rv = pthread_setspecific(_localthread_socket_key_, tmp_socket); |
| | | if ( rv != 0) { |
| | | logger->error(rv, "shm_sendandrecv : pthread_setspecific"); |
| | | exit(1); |
| | | rv = pthread_setspecific(_localthread_socket_key_, tmp_socket); |
| | | if ( rv != 0) { |
| | | logger->error(rv, "shm_sendandrecv : pthread_setspecific"); |
| | | exit(1); |
| | | } |
| | | } |
| | | |
| | | sendpak.key = tmp_socket->key; |
| | |
| | | rv = shm_sendpakto(tmp_socket, &sendpak, key, timeout, flags); |
| | | |
| | | if(rv != 0) { |
| | | if(send_buf != NULL) { |
| | | mm_free(sendpak.buf); |
| | | } |
| | | |
| | | return rv; |
| | | } |
| | | |
| | |
| | | tryn--; |
| | | recvbufIter = tmp_socket->recvbuf2.find(key); |
| | | if(recvbufIter != tmp_socket->recvbuf2.end()) { |
| | | // 在缓存里查到了key匹配成功的 |
| | | recvpak = recvbufIter->second; |
| | | tmp_socket->recvbuf2.erase(recvbufIter); |
| | | tmp_socket->recvbuf2.erase(key); |
| | | goto LABLE_SUC; |
| | | } |
| | | |
| | |
| | | } |
| | | |
| | | if (key == recvpak.key) { |
| | | // 发送与接受的UUID匹配成功 |
| | | |
| | | goto LABLE_SUC; |
| | | |
| | | } else { |
| | | // 答非所问,放到缓存里 |
| | | |
| | | tmp_socket->recvbuf2.insert({recvpak.key, recvpak}); |
| | | exit(0); |
| | | continue; |
| | | } |
| | | } |
| | |
| | | sendpak->key = sockt->key; |
| | | } |
| | | rv = remoteQueue->push(*sendpak, timeout, flag); |
| | | |
| | | if(rv != 0) { |
| | | mm_free(sendpak->buf); |
| | | } |
| | | if(rv == ETIMEDOUT) { |
| | | return EBUS_TIMEOUT; |
| | | } |
| | | return rv; |
| | | |
| | | ERR_CLOSED: |