bug fixed ps_sub thread join

zhangmeng

2023-03-01 0f53806de20f07b0bc15877c7a9b293758acbfa7

 src/nng_wrap.cpp

@@ -68,7 +68,7 @@
}

void set_last_error(const std::string& emsg){
    verbose_info += emsg;
    // verbose_info += emsg;
}

void get_last_error(int* ec, void** emsg, int* emsg_len){
@@ -77,10 +77,12 @@

    *ec = nn_errno();
    const char* msg = nn_strerror(*ec);
    string strMsg(msg);
    strMsg = strMsg + "{" + verbose_info + "}";
    copy_memory(emsg, emsg_len, strMsg.data(), strMsg.size());
    verbose_info.clear();
    // string strMsg(msg);
    // strMsg = strMsg + "{" + verbose_info + "}";
    // copy_memory(emsg, emsg_len, strMsg.data(), strMsg.size());
    // verbose_info.clear();

    copy_memory(emsg, emsg_len, msg, strlen(msg));
}

///////////////////////////////////////////////////////
@@ -203,12 +205,13 @@
        PRNTVITAG("client_socket faild\n");
        return -1;
    }
    // set_socket_timeout(sock, timeout_req_rep);
    set_socket_timeout(sock, 300);
    sub->socket_ = sock;
    sub->t_ = get_thread([](const auto sub){
        while (!sub->t_quit_.load()) {
            char* m{};
            int m_len = nn_recv(sub->socket_, &m, NN_MSG, NN_DONTWAIT);
            // int m_len = nn_recv(sub->socket_, &m, NN_MSG, NN_DONTWAIT);
            int m_len = nn_recv(sub->socket_, &m, NN_MSG, 0);
            if (m_len > 0){
                string tmp_msg{m, (size_t)m_len};
                nn_freemsg(m);
@@ -587,20 +590,22 @@

    if (!w) return -1;

    *src = malloc(sizeof(uint64_t));
    *(uint64_t*)(*src) = key;
    *src = malloc(sizeof(key));
    memcpy(*src, &key, sizeof(key));

    return w->mode;
}

int send_reply(const void* src, const void* msg, const int msg_len, void* arg/*=NULL*/){
int send_reply(void* src, const void* msg, const int msg_len, void* arg/*=NULL*/){
    _rr* rep = (_rr*)arg;
    if (!rep) rep = singleton<_rr>();

    struct work* w{};
    {
        auto key = *(static_cast<uint64_t*>(const_cast<void*>(src)));
        free(const_cast<void*>(src));
        uint64_t key;
        memcpy(&key, src, sizeof(key));
        // auto key = *(static_cast<uint64_t*>(const_cast<void*>(src)));
        free(src);

        lock_guard<mutex> l{rep->mtx_msg_};
        auto iter = rep->works_.find(key);