threads are halted by function send() or recv() of ACE_SOCK_Stream?

Question

I am using pthread + ACE to write a fake client.

this client has 3 threads, and each thread could send and receive message endlessly by using ACE. however, these threads always be halted by the function send() or recv(). that is, the thread would exit if there is something wrong with sending or receiving, unfortunately, I don't know what is the error, and I could not catch it, too. the codes are:

struct thread_data {
    int  thread_id;
    string ip;
    uint32_t port;
    uint32_t timeout;
};
std::vector<struct thread_data> m_thread;

void * test_fun1(void * threadid)
{
    struct thread_data * tmp_thread_data = (struct thread_data *)threadid;
    long tmp_threadid = (long)tmp_thread_data->thread_id;
    string tmp_ip = tmp_thread_data->ip;
    uint32_t tmp_port = tmp_thread_data->port;
    uint32_t tmp_timeout = tmp_thread_data->timeout;

    ACE_INET_Addr addr(tmp_port, tmp_ip.c_str());
    ACE_Time_Value timeout(0, tmp_timeout * 1000);
    ACE_SOCK_Connector connector;
    ACE_SOCK_Stream peer;

    // connect
    if(connector.connect(peer, addr, &timeout) != 0)
            pthread_exit((void *) threadid);

    // send msg
    while (1)
    {
            ssize_t tmp_ret1 = peer.send("hello world", 12);
            if (tmp_ret1 <= 0)
                    continue;

            char tmp_buf[1024] = '\0';
            ssize_t tmp_ret2 = peer.recv(tmp_buf, 1024, &timeout);
            if (tmp_ret2 <= 0)
                     continue;
            else
                     fprintf(stderr, "recv:%s\n", tmp_buf);
    }

    // close
    peer.close();
    pthread_exit((void *) threadid);
}

int main(int argc, char *argv[]) 
{
    std::vector<pthread_t> threads;
    pthread_attr_t attr;
    int rc;
    int i = 0;
    void * status;

    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    // thread create
    int tmp_num = 3;
    for(i = 0; i < tmp_num; i++)
    {
            pthread_t tmp_thread_handler;
            struct thread_data tmp_thread_info;
            tmp_thread_info.thread_id = i;
            tmp_thread_info.ip = "127.0.0.1";
            tmp_thread_info.port = 8001;
            tmp_thread_info.timeout = 100;
            rc = pthread_create(&tmp_thread_handler, NULL, test_fun1, (void *)&tmp_thread_info);
            if (rc != 0)
                    return -1;

            threads.push_back(tmp_thread_handler);
            m_thread.push_back(tmp_thread_info);
    }

    // thread start
    pthread_attr_destroy(&attr);
    for(i = 0; i < tmp_num; i++) 
    {

            rc = pthread_join(threads[i], &status);
            if (rc != 0) 
                    return -1;
    }

    pthread_exit(NULL);
    return 0; 
}

what should I do if I want to send and receive message endlessly? any help would be appreciated!

Answer 1

You have a significant scope-lifetime issue in this code. You're sending a data block to your threads that is created/destroyed with each loop iteration. Leaving out the side-lines:

// thread create
int tmp_num = 3;
for(i = 0; i < tmp_num; i++)
{
    pthread_t tmp_thread_handler;
    struct thread_data tmp_thread_info;
    // --- snip ----
    rc = pthread_create(&tmp_thread_handler, NULL, test_fun1, (void *)&tmp_thread_info);
    // --- snip ----
}

Whether the resulting block is copied into the underlying info vectors or not is not relevant. Once the loop slings around, the object is destroyed, and with it the std::string member variable ip. It may reside in the same memory space (the struct), but the reallocation of the string almost certainly will not. In other words, you're invoking undefined behavior.

A possible resolution to this is to use a smart pointer for the structure, passing its get() member to the thread, and pushing said-same smart pointer into your info vector ( retooled for smart pointers). I personally prefer a simpler approach, namely allocating both vectors before the threads are kicked off, thereby fixing their "innards" for by-address usage. From there you can send the structure address since the vector holds it for you:

// thread create
size_t tmp_num = 3;
m_thread.resize(tmp_num);
threads.resize(tmp_num);
for(size_t i = 0; i < tmp_num; i++)
{
    m_thread[i].thread_id = i;
    m_thread[i].ip = "127.0.0.1";
    m_thread[i].port = 8001;
    m_thread[i].timeout = 100;

    rc = pthread_create(&threads[i], NULL, test_fun1, &m_thread[i]);
    if (rc != 0)
        return -1;
}

Regarding other issues, in your thread proc this doesn't compile:

char tmp_buf[1024] = '\0';

But that's pretty minor compared to the undefined behavior you're having.