CODEWITHSUNDEEP
c++multithreadingc++11concurrencyzeromq
JACK M
JACK M

Reputation: 2841

More threads created than expected

You could find the program here

I am building a program in message passing framework 0MQ. I try to implement what I posted in here

Program compiled with g++ -std=c++11 test.cpp -o test -lzmq -lpthread.

To run the program, pass one parameter as the thread number you would like to have. That parameter is then assigned to variable worker_num.

In main thread, I setup thread with:

  vector<thread> pool;
  for(int i = 0; i < worker_num; i++)
  {
    cout << "main() : creating thread, " << i << endl;
    pool.push_back(thread(task1, (void *)&context, i));
  }

I would like to make sure all worker threads have successful connection to main thread before main thread distributes jobs to them. 

  while(true)
  {
    if(sync_done)
    {
      cout << "sync done in main thread" << endl;
      break;
    }

    zmq::message_t sync_msg(4);
    memcpy((void *)sync_msg.data(), SYNC_MSG, SYNC_MSGLEN);
    for(int i = 0; i < worker_num; i++)
      distask_socket.send(sync_msg);

    for(int i = 0; i < worker_num; i++)
    {
      if(sync_done)
        break;
      if(i != 0)
        this_thread::sleep_for(chrono::milliseconds(500));

      zmq::message_t res_msg;
      int ret = getres_socket.recv(&res_msg, ZMQ_DONTWAIT);

      if(ret == -1 && errno == EAGAIN)
        continue;

      int threadID = stoi(string((char *)res_msg.data()));
      sync_done = if_sync_done(threadID, sync_array, worker_num);
    }
  }

So what main thread does is: pushing #worker_num of sync msgs with its PUSH endpoint to worker threads each time and then reads confirmation msg from its PULL endpoint. If main thread retrieves #worker_num of confirmation msgs, then sync done. Format of the sync msg from worker is: the worker thread's ID in a string. So thread 0 would pass a 0 in string back to main thread.

But running the program I have:

$ ./test 1
main() : creating thread, 0
thread id:0
thread 0 receives: sync
thread 0 sends: 0
thread 0 sync done
main thread receives sync msg from thread 1 # you may get many copies of this msg
terminate called after throwing an instance of 'std::invalid_argument'
  what():  stoi
Aborted

main thread receives sync msg from thread 1 means thread are 2 threads created: thread 0 and thread 1. Any idea why? I did pass 1 as parameter. Noted that if you run the program yourself you may get other outputs.

UPDATE:

Program updated: here.

Finally I figured out what's wrong.

expected output, you see thread 0 pass a 0 to main thread to notify sync done:

$ ./test 1
input parameter is: 1
main() : creating thread, 0
thread 0 receives: sync
to_string 0
thread 0 sends: 0, with size: 1
thread 0 sync done
pass 0 to if_sync_done
main thread receives sync msg from thread 0
sync done in main thread

unexpected output, you see unprintable char is passed to stoi():

$ ./test 1
input parameter is: 1
main() : creating thread, 0
thread 0 receives: sync
to_string 0
thread 0 sends: 0, with size: 1
thread 0 sync done
pass  to if_sync_done  # !!!!!
terminate called after throwing an instance of 'std::invalid_argument'
  what():  stoi
Aborted

So it seems that I use message_t incorrectly. So I need to ensure that before main thread passes the content to stoi(), the buffer still exists.

I will add an answer myself.

Upvotes: 3

Views: 171

Answers (1)

Igor Tandetnik
Igor Tandetnik

Reputation: 52471

zmq::message_t msg_back((void *)to_string(id).c_str(), to_string(id).size() + 1, NULL);

zmq::message_t constructor you use does not make a copy of the buffer, if [1] and [2] are to be believed. Instead, it takes ownership of the buffer.

However, you are passing a buffer managed by a temporary; that buffer is destroyed as soon as the constructor returns. You have msg_back store a dangling pointer. Any attempt to use that pointer - e.g. trying to read the message on the receiving end - exhibits undefined behavior.

Upvotes: 2

Related Questions