Can you please take a look at the following x86-64 C++ code which should implement a multi consumer/produce lockfree stack? Do you think I have missed anything?

namespace lockfree {
    // NOTE: this lock-free stack uses the highest 2 bytes
    // of each pointer to store a 'counter' (an "id") to
    // ensure that the ABA problem doesn't happen in case
    // of multiple consumers.
    // If the stack was to be read by _one only_ consumer
    // the ABA problem wouldn't manifest and the code would
    // not need the 'counter' escamotage.
    // Using an unsigned short is ok-ish because 65536 values
    // might not be enough in some cases. Ideally we should
    // be using 32+ bits.
    template<typename T>
    class stack {

        struct element {
            element *next;
            const T data;

            element(const T& data_) : data(data_) {
                element(const element&);
                element& operator=(const element&);

        // NOTE: we use the highest 16 bits as counter
        // this trick should perform ok for quite
        // _a while_ on Linux 64-bit...
        static const unsigned short get_ptr_cnt(element** p) {
            unsigned short *us_p = (unsigned short *)p;
            return us_p[3];

        // NOTE: same as above!
        static void set_ptr_cnt(element** p, const unsigned short cnt) {
        unsigned short *us_p = (unsigned short *)p;
            us_p[3] = cnt;

        element *_head;

        // this lockfree stack can only be initialised and destroyed
        // in single threaded contexts
        stack() : _head(0) {

        // clean the stack (only in single threaded context!)
        ~stack() {
            element *head_ptr = _head;
            set_ptr_cnt(&head_ptr, 0);
            while(head_ptr) {
                element *next = head_ptr->next;
                delete head_ptr;
                head_ptr = next;
                set_ptr_cnt(&head_ptr, 0);
            _head = 0;

        void push(const T& data) {
            // first create an element to be pushed
            element *to_be_pushed = new element(data);
            // make sure the highest 16 bits are always
            // 0 and not part of the address. This check
            // should always return _false_ on modern x86-64
            if(0 != get_ptr_cnt(&to_be_pushed)) {
                int* p = 0;
                *p = 1;
            while(1) {
                // get current head in local variable
                element *local_head = _head;
                to_be_pushed->next = local_head;
                // adjust the pointers and add counter
                const unsigned short cur_cnt = get_ptr_cnt(&local_head) + 1;
                set_ptr_cnt(&to_be_pushed, cur_cnt);
                // compare and swap
                if(__sync_bool_compare_and_swap(&_head, local_head, to_be_pushed))
                // otherwise reset the counter to 0 and prepare for another round
                set_ptr_cnt(&to_be_pushed, 0);

        bool pop(T& data) {
            while(1) {
                // get the current head and next head
                // important to get from local_head variable!
                element *local_head = _head;
                // get the counter
                const unsigned short head_cnt = get_ptr_cnt(&local_head);
                // reset the counter to get real pointer
                set_ptr_cnt(&local_head, 0);
                element *next_head = local_head->next;
                // set again original counter and next head
                set_ptr_cnt(&local_head, head_cnt);
                set_ptr_cnt(&next_head, head_cnt+1);
                // compare and swap the local head with head
                if(__sync_bool_compare_and_swap(&_head, local_head, next_head)) {
                    // reset the counter once more
                    set_ptr_cnt(&local_head, 0);
                    data = local_head->data;
                    delete local_head;
                    return true;
            return false;

I've written even the version which stores the counter in another 64 bit size_t and implicitly uses cmpxchg16b.


I don't see any problem with the locking, although it is usually quite difficult to see race conditions (which is why you posted it I guess). The code is clearly non-portable (but maybe that is not important), with regard to endian-ness and because unsigned short is not guaranteed to be 16 bits (uint16_t is).

I did wonder about two lines that seem unnecessary. Firstly, setting the count after the swap in push:

if(__sync_bool_compare_and_swap(&_head, local_head, to_be_pushed))
set_ptr_cnt(&to_be_pushed, 0); // this line

Secondly setting the count on next_head in pop

set_ptr_cnt(&local_head, head_cnt);
set_ptr_cnt(&next_head, head_cnt+1);  // this line
if(__sync_bool_compare_and_swap(&_head, local_head, next_head)) {

EDIT - after discussion with Emanuele, it is clear that the first of these (resetting the to_be_pushed pointer's counter) is absolutely necessary. The second however (set_ptr_cnt(&next_head, head_cnt+1);) is not, but instead is currently (but not reliably) causing the compiler to generate code that 'works'. The reason, I think, is that the stack is really volatile - meaning that the compiler cannot assume that pointer or data values do not change and cannot optimize access to these locations.

So _head maybe should be declared

volatile element *_head;

and other references to the stack maybe too. Having said that, I have modified the code to use volatile for _head and elsewhere (see previous edit) but have been unable to make the compiler take any notice - in other words the assembler code generated with volatile has been the same as that without (gcc -S -O3 using gcc 4.2.1 and 4.4.5). So maybe volatile is not the answer...

As regards coding style, I think the code is heavily over-commented. Most of the comments are not very useful and could be omitted. Also the variable names are verbose or cryptic. To make the code more readable, I did a global deletion of _ptr, and a global replace of:

  • local_head -> head
  • next_head -> next
  • _head -> stack_head (leading underscores should be avoided as they are, I think, reserved, and in any case add nothing).
  • head_cnt -> count
  • *_cnt -> *_count

Finally, the compiler warns (-Wignored-qualifiers) against the const in:

static const unsigned short get_ptr_cnt(element** p) {
| improve this answer | |
  • \$\begingroup\$ About first point, agreed, might remove it. Not performing set_ptr_cnt(&next_head, head_cnt+1); was making ABA problem (or similar) appear because probably of an optimisation? Should look at generated code though... \$\endgroup\$ – Emanuele Nov 20 '13 at 20:39
  • \$\begingroup\$ No entry can get onto the stack without its counter value being set. And it is the fact that there is a unique count that prevents the ABA problem. So changing the count in next_head during pop seems to me to have no purpose - but I could easily believe that I am mistaken :-) \$\endgroup\$ – William Morris Nov 20 '13 at 21:28
  • \$\begingroup\$ Was forgetting, as for the first point set_ptr_cnt(&to_be_pushed, 0); you need to reset to 0, because you need to use the pointer at each loop! \$\endgroup\$ – Emanuele Nov 20 '13 at 23:48
  • \$\begingroup\$ Oh, yes! It is obvious now you point it out... \$\endgroup\$ – William Morris Nov 21 '13 at 0:36
  • \$\begingroup\$ I've got an explanation for set_ptr_cnt(&next_head, head_cnt+1);. I've generated asm (with -S option); seems like this is forcing g++ 4.6.3 to store on the local stack the next_head value (which is what I wanted), so when cas occurs the generated code won't go and fetch it from memory (which could be changed by another thread), but will use what was stored on the stack as first read. Does it make sense? \$\endgroup\$ – Emanuele Nov 21 '13 at 0:42

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.