Passing Values To A Thread

[NathanOliver]

Hey Daniweb,

I have recently started working with threads in the c++11 library. I started with a simple dispatch mechanism where I have a queue and that would get passed to the threads and the threads would take turns emptying the queue. Since each thread needs to share the queue I am passing it by reference. My original code looks likes this and it wasn't working.

// what DoTransactions looks like
void DoTransaction(std::queue<int> & line, std::vector<int> & customersServed, int wait)
//...

//...
std::queue<int> line;
for (int i = 1; i <= 10; i++)
    line.push(i);

std::vector<std::vector<int>> returns(2);
std::vector<std::thread> threads;
for (int i = 0; i < 2; i++)
{
    threads.push_back(std::thread(DoTransaction, line, returns[i], i + 1));
}
//...

I was able to find that I needed to wrap objects that I want to be passed as a refernce with std::ref() and I got this

//...
for (int i = 0; i < 2; i++)
{
    threads.push_back(std::thread(DoTransaction, std::ref(line), std::ref(returns[i]), i + 1));
}
//...

My question is do I always need to wrap an object with std::ref() when I want to pass it by reference? Does this come from the fact that the call to std::thread is a constructor and it doesnt forwar the references to the function the thread runs?

[mike_2000_17]

The core issue is that the std::thread cannot guess whether you want to pass things by reference or by value, and it would not be safe or possible to assume that you want to pass things by reference, it must assume that all parameters are passed by value. The std::ref() and std::cref() functions allow you to wrap a reference into a small object that can be passed around by value, until there is a forced conversion to a reference (which happens when the actual function is called). This is the reason why you have to use these wrappers.

Internally, the std::thread class probably uses std::function<void()> class template to store the function pointer and its forwarded parameters. The rules are the same for constructing a std::function<void()> object.

Since each thread needs to share the queue I am passing it by reference.

Remember to also use a std::mutex to protect the accesses to the shared data. It implements a mutually exclusive lock such that multiple threads can access shared data without mutual conflicts. See docs on std::mutex.

[NathanOliver]

Thank you Mike. I always appreciate when you respond. I had a felling it was something along those lines. And I am using a mutex to protect the access to the queue. The following is the body of my thread function.

// global space
std::mutex mtx;

// function space
void DoTransaction(std::queue<int> & line, std::vector<int> & customersServed, int wait)
{
    while (!line.empty())
    {
        mtx.lock();
        std::cout << line.front() << "\t" << wait << std::endl;
        customersServed.push_back(line.front());
        line.pop();
        mtx.unlock();
        std::this_thread::sleep_for(std::chrono::seconds(wait));
    }
}

If you wouldnt mind answering a second question of mine I dont like the way I am using mtx. I have it declared globally and I hate having global variables. Should I instead declare it in main and pass it as a reference to the threads? Could it be declared statically in DoTransactions()?

[mike_2000_17]

Here is some food for thoughts, imagine you have this scenario:

std::queue<int> line1;
for (int i = 1; i <= 10; i++)
    line1.push(i);

std::vector<std::vector<int>> returns1(2);
std::vector<std::thread> threads1;
for (int i = 0; i < 2; i++)
    threads1.emplace_back(DoTransaction, std::ref(line1), std::ref(returns1[i]), i + 1);

std::queue<int> line2;
for (int i = 1; i <= 10; i++)
    line2.push(i);

std::vector<std::vector<int>> returns2(2);
std::vector<std::thread> threads2;
for (int i = 0; i < 2; i++)
    threads2.emplace_back(DoTransaction, std::ref(line2), std::ref(returns2[i]), i + 1);

Do you see a problem here?

The problem here is that whether the mutex is a global variable or a local static variable, the problem remains the same: both sets of threads will lock the same mutex even though they are using different queues.

The only viable solution is to associate the mutex with the queue, somehow. A cheap and easy solution is just to pass the mutex to the function along with the queue. A somewhat fancier solution would be to implement a thread-safe queue class, which could be as simple as just wrapping a mutex and a queue into a simple class that does what you need to use it for. Another solution might be to use something like this class I wrote a while back.

Another typical solution is just to use a functor class:

struct Transactioner {
  std::queue<int> line;
  std::mutex mtx;

  void serveCustomers(std::vector<int> & customersServed, int wait)
  {
    while (!line.empty())
    {
      mtx.lock();
      int val = line.front(); 
      line.pop();
      mtx.unlock();
      std::cout << val << "\t" << wait << std::endl;
      customersServed.push_back(val);

      std::this_thread::sleep_for( std::chrono::seconds(wait) );
    }
  }
}


// use-case:

Transactioner t1;
for (int i = 1; i <= 10; i++)
  t1.line.push(i);

std::vector<std::vector<int>> returns1(2);
std::vector<std::thread> threads1;
for (int i = 0; i < 2; i++)
  threads1.emplace_back(
    &Transaction::serveCustomers, &t1, 
    std::ref(returns1[i]), i + 1);

Transactioner t2;
for (int i = 1; i <= 10; i++)
  t2.line.push(i);

std::vector<std::vector<int>> returns2(2);
std::vector<std::thread> threads2;
for (int i = 0; i < 2; i++)
  threads2.emplace_back(
    &Transaction::serveCustomers, &t2,
    std::ref(returns2[i]), i + 1);

Basically, any way that associates the mutex with the queue is gonna work.

You might also want to take advantage of the std::async function. This is because your functions are returning a std::vector<int> through the pass-by-reference parameter. You don't need to do that. You can use std::async instead:

struct Transactioner {
  std::queue<int> line;
  std::mutex mtx;

  std::vector<int> serveCustomers(int wait)
  {
    std::vector<int> result;
    while (!line.empty())
    {
      mtx.lock();
      int val = line.front(); 
      line.pop();
      mtx.unlock();
      std::cout << val << "\t" << wait << std::endl;
      result.push_back(val);

      std::this_thread::sleep_for( std::chrono::seconds(wait) );
    }
    return result;
  }
}


// use-case:

Transactioner t1;
for (int i = 1; i <= 10; i++)
  t1.line.push(i);

std::vector< std::future< std::vector<int> > > results1;
for (int i = 0; i < 2; i++)
  results1.push_back( std::async(
    &Transaction::serveCustomers, &t1, i + 1));

Transactioner t2;
for (int i = 1; i <= 10; i++)
  t2.line.push(i);

std::vector< std::future< std::vector<int> > > results2;
for (int i = 0; i < 2; i++)
  results2.push_back( std::async(
    &Transaction::serveCustomers, &t2, i + 1));

You don't have to worry about excessive copies of the result vector, thanks to C++11 move-semantics!

[NathanOliver]

Mike thank you once again. You have given me a lot of food for thought. I will defenitley look into using std::future and std::async.