boost::lockfree::queue with gcc-4.6 + optimization

[ygabay@…]

I am seeing a potential issue with lockfree queue when using gcc-4.6 with compilation optimization turned on.

The behavior i'm seeing is that call to push() which returns true does not guarantee that the pushed element is consumable immediately. (i.e. a subsequent call to pop() might return false. after further debugging it seems like the element _will become_ consumable but it might take few more cpu cycles till this happens. this issue is not happening when turning the compiler optimizations off or when using gcc-4.8.

here is the test program i am using to reproduce the issue, which is spawning 2 threads, each of them looping through push() and then pop(), and at one point, the call to pop() fails which should never fail.

compilation line:

g++-4.6 -O1 -g -fno-inline -pthread -m64 -I/usr/local/include -c lockfree_test.cpp -o lockfree_test.cpp.o

sample program:

struct SemTask
{
 int x;
};
boost::lockfree::queue<SemTask * > queue(1);

void* runner(void* arg)
{
    std::cout << '.' << std::endl;
    while (!t_exit)
    {
      SemTask * task = new SemTask();
      assert(queue.push(task));
      assert(queue.pop(task));
      delete task;
    }
    return NULL;
}

int main(int argc, char** argv)
{
  assert(queue.is_lock_free());
  #define NUM_THRD 2
  pthread_t inc_x_thread[NUM_THRD];

  for (int i = 0; i < NUM_THRD; ++i) {
    if(pthread_create(&inc_x_thread[i], NULL, runner, NULL)) {
      fprintf(stderr, "Error creating thread\n");
      return -1;
    }
  }

  for (int i = 0; i < NUM_THRD; ++i) {
    if(pthread_join(inc_x_thread[i], NULL)) {
      fprintf(stderr, "Error joining thread\n");
      return -1;
    }
  }

  return 0;
}

[anonymous]

minor compilation issue in the test program. here is updated version:

#include <string>
#include <deque>
#include <cstdio>

#include <boost/lockfree/queue.hpp>
#include <cassert>
#include <iostream>

struct SemTask
{
 int x;
};
boost::lockfree::queue<SemTask * > queue(1);

void* runner(void* arg)
{
    std::cout << '.' << std::endl;
    while (true)
    {
      SemTask * task = new SemTask();
      assert(queue.push(task));
      assert(queue.pop(task));
      delete task;
    }
    return NULL;
}

int main(int argc, char** argv)
{
  assert(queue.is_lock_free());
  #define NUM_THRD 2
  pthread_t inc_x_thread[NUM_THRD];

  for (int i = 0; i < NUM_THRD; ++i) {
    if(pthread_create(&inc_x_thread[i], NULL, runner, NULL)) {
      fprintf(stderr, "Error creating thread\n");
      return -1;
    }
  }

  for (int i = 0; i < NUM_THRD; ++i) {
    if(pthread_join(inc_x_thread[i], NULL)) {
      fprintf(stderr, "Error joining thread\n");
      return -1;
    }
  }

  return 0;
}