Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.programming.threads > #578 > unrolled thread

usage of vyukov's mpsc queue

Back to article view | Back to comp.programming.threads

Contents

  usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-20 06:55 -0700
    Re: usage of vyukov's mpsc queue Joshua Maurice <joshuamaurice@gmail.com> - 2011-09-20 17:19 -0700
      Re: usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-20 18:35 -0700
        Re: usage of vyukov's mpsc queue Ephraim Sinowitz <ephisinowitz@yahoo.com> - 2011-09-21 06:42 -0400
    Re: usage of vyukov's mpsc queue Joshua Maurice <joshuamaurice@gmail.com> - 2011-09-21 00:26 -0700
    Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 12:19 +0200
      Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 12:34 +0200
      Re: usage of vyukov's mpsc queue Ephi Sinowitz <ephisinowitz@yahoo.com> - 2011-09-21 06:58 -0400
        Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 14:02 +0200
          Re: usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-21 06:08 -0700
            Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 15:18 +0200
              Re: usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-21 06:56 -0700
                Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 17:06 +0200
                  Re: usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-21 09:16 -0700
                    Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 18:20 +0200
    Re: usage of vyukov's mpsc queue Dmitriy Vyukov <dvyukov@gmail.com> - 2011-09-21 09:58 -0700
      Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-09-21 19:33 +0200
        Re: usage of vyukov's mpsc queue dvyukov@google.com - 2011-10-02 04:13 -0700
          Re: usage of vyukov's mpsc queue Alexander Terekhov <terekhov@web.de> - 2011-10-04 12:15 +0200
      Re: usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-21 11:04 -0700
        Re: usage of vyukov's mpsc queue Ephi Sinowitz <esinowitz.public@gmail.com> - 2011-09-21 11:17 -0700
        Re: usage of vyukov's mpsc queue "Chris M. Thomasson" <cristom@charter.net> - 2011-09-24 22:44 -0700

Page 1 of 2  [1] 2  Next page →

#578 — usage of vyukov's mpsc queue

Hi,
I implemented the queue http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
with a signaling mechanism for the consumer.
Do I need to ensure visibility somehow before setting the event since
I'm not spinning in the consumer? How?
Producers:
----------------
    vyukov_queue.push(item);
    SetEvent(m_queueEvent);
Consumer:
---------------
    for(;;)
    {
        WaitForSingleObject(m_queueEvent, INFINITE);
        while (void* item = vyukov_queue.pop())
        {
            // do work
        }
    }
For reference here is the push()
void MPSC_Queue::push(void* item)
{
    item->next = NULL;
    void* old_head = InterlockedExchangePointer((void *volatile
*)&m_head, item);
    // BLOCK POINT
    old_head->next = item;  // Do I need to ensure this is visible to
the consumer thread?
}

Any other critique or comments welcome as well :)
Thanks,
-ephi

[toc] | [next] | [standalone]

#579

On Sep 20, 6:55 am, Ephi Sinowitz <esinowitz.pub...@gmail.com> wrote:
> Hi,
> I implemented the queuehttp://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
> with a signaling mechanism for the consumer.
> Do I need to ensure visibility somehow before setting the event since
> I'm not spinning in the consumer? How?
> Producers:
> ----------------
>     vyukov_queue.push(item);
>     SetEvent(m_queueEvent);
> Consumer:
> ---------------
>     for(;;)
>     {
>         WaitForSingleObject(m_queueEvent, INFINITE);
>         while (void* item = vyukov_queue.pop())
>         {
>             // do work
>         }
>     }
> For reference here is the push()
> void MPSC_Queue::push(void* item)
> {
>     item->next = NULL;
>     void* old_head = InterlockedExchangePointer((void *volatile
> *)&m_head, item);
>     // BLOCK POINT
>     old_head->next = item;  // Do I need to ensure this is visible to
> the consumer thread?
>
> }
>
> Any other critique or comments welcome as well :)
> Thanks,
> -ephi

If you posted actual code, we might be better able to help you. As
such, that code cannot be compilable. Specifically,

> void MPSC_Queue::push(void* item)
> {
>     item->next = NULL;

There's no way that can compile (barring some really ugly and bad
macro hackery).

If you provide more complete code, we can help you out. However, it's
not looking good in terms of correctness. Still, I'll withhold
judgment until I see more of the code.

[toc] | [prev] | [next] | [standalone]

#580

On Sep 20, 8:19 pm, Joshua Maurice <joshuamaur...@gmail.com> wrote:
>
> If you posted actual code, we might be better able to help you. As
> such, that code cannot be compilable. Specifically,
>
> > void MPSC_Queue::push(void* item)
> > {
> >     item->next = NULL;
>
> There's no way that can compile (barring some really ugly and bad
> macro hackery).
>
> If you provide more complete code, we can help you out. However, it's
> not looking good in terms of correctness. Still, I'll withhold
> judgment until I see more of the code.

here you go.

void MPSC_Queue::push(entry_t* item)
{
    item->next = NULL;
    entry_t* old_head = (entry_t*)InterlockedExchangePointer((void
*volatile
*)&m_head, item);
    // BLOCK POINT
    old_head->next = item;  // Do I need to ensure this is visible to
the consumer thread?
}

[toc] | [prev] | [next] | [standalone]

#584

Sorry I though you followed the link on the original post but it is a 
little broken. Here are the full details.

Intrusive MPSC node-based queue
-------------------------------
Advantages:
+ Intrusive. No need for additional internal nodes.
+ Wait-free and fast producers. One XCHG is maximum what one can get 
with multi-producer non-distributed queue.
+ Extremely fast consumer. On fast-path it's atomic-free, XCHG executed 
per node batch, in order to grab 'last item'.
+ No need for node order reversion. So pop operation is always O(1).
+ ABA-free.
+ No need for PDR. That is, one can use this algorithm out-of-the-box. 
No need for thread registration/deregistration, periodic activity, 
deferred garbage etc.

Disadvantages:
- Push function is blocking wrt consumer. I.e. if producer blocked in 
(*), then consumer is blocked too. Fortunately 'window of 
inconsistency' is extremely small - producer must be blocked exactly in 
(*). Actually it's disadvantage only as compared with totally lockfree 
algorithm. It's still much better lockbased algorithm.

struct mpscq_node_t
{
    mpscq_node_t* volatile  next;

};

struct mpscq_t
{
    mpscq_node_t* volatile  head;
    mpscq_node_t*           tail;
    mpscq_node_t            stub;

};

#define MPSCQ_STATIC_INIT(self) {&self.stub, &self.stub, {0}}

void mpscq_create(mpscq_t* self)
{
    self->head = &self->stub;
    self->tail = &self->stub;
    self->stub.next = 0;

}

void mpscq_push(mpscq_t* self, mpscq_node_t* n)
{
    n->next = 0;
    mpscq_node_t* prev = XCHG(&self->head, n);
    //(*)
    prev->next = n;

}

mpscq_node_t* mpscq_pop(mpscq_t* self)
{
    mpscq_node_t* tail = self->tail;
    mpscq_node_t* next = tail->next;
    if (tail == &self->stub)
    {
        if (0 == next)
            return 0;
        self->tail = next;
        tail = next;
        next = next->next;
    }
    if (next)
    {
        self->tail = next;
        return tail;
    }
    mpscq_node_t* head = self->head;
    if (tail != head)
        return 0;
    mpscq_push(self, &self->stub);
    next = tail->next;
    if (next)
    {
        self->tail = next;
        return tail;
    }
    return 0;
} 

Now if my consumer spins I can use this as is. But what if I am using 
signaling with an event as described in the original post. Can I just 
use the line     
  prev->next = n;
as is or do I need something else to ensure visiblity to the other thread?

Thanks,
-ephi
 

[toc] | [prev] | [next] | [standalone]

#581

On Sep 20, 6:55 am, Ephi Sinowitz <esinowitz.pub...@gmail.com> wrote:
> Hi,
> I implemented the queuehttp://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
> with a signaling mechanism for the consumer.
> Do I need to ensure visibility somehow before setting the event since
> I'm not spinning in the consumer? How?
[...]
> Consumer:
> ---------------
>     for(;;)
>     {
>         WaitForSingleObject(m_queueEvent, INFINITE);
>         while (void* item = vyukov_queue.pop())
>         {
>             // do work
>         }
>     }
[...]

You need to provide more code for me to know for sure. For example,
what is the type of "vyukov_queue"? You need to provide some
compilable complete code. At the very least the full class definition
for this FIFO concurrent queue.

If that queue object is a simple unsynchronized queue, then I guess
that your code is broken. The consumer does not acquire a lock nor do
an interlocked function, and thus it is broken. There is no
synchronization to prevent the race condition on vyukov_queue nor
*item.

[toc] | [prev] | [next] | [standalone]

#582

Ephi Sinowitz wrote:
> 
> Hi,
> I implemented the queue http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
> with a signaling mechanism for the consumer.
> Do I need to ensure visibility somehow before setting the event since
> I'm not spinning in the consumer? How?
> Producers:
> ----------------
>     vyukov_queue.push(item);
>     SetEvent(m_queueEvent);
> Consumer:
> ---------------
>     for(;;)
>     {
>         WaitForSingleObject(m_queueEvent, INFINITE);
>         while (void* item = vyukov_queue.pop())
>         {
>             // do work
>         }
>     }
> For reference here is the push()
> void MPSC_Queue::push(void* item)
> {
>     item->next = NULL;
>     void* old_head = InterlockedExchangePointer((void *volatile
> *)&m_head, item);
>     // BLOCK POINT
>     old_head->next = item;  // Do I need to ensure this is visible to
> the consumer thread?
> }

Yes, you need to make old_head->next = item store atomic-release store.

Consumer's m_tail->next load must be atomic-acquire load.

Producer's RMW can be totally relaxed with respect to memory barriers.

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#583

Alexander Terekhov wrote:
> 
> Ephi Sinowitz wrote:
> >
> > Hi,
> > I implemented the queue http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
> > with a signaling mechanism for the consumer.
> > Do I need to ensure visibility somehow before setting the event since
> > I'm not spinning in the consumer? How?
> > Producers:
> > ----------------
> >     vyukov_queue.push(item);
> >     SetEvent(m_queueEvent);
> > Consumer:
> > ---------------
> >     for(;;)
> >     {
> >         WaitForSingleObject(m_queueEvent, INFINITE);
> >         while (void* item = vyukov_queue.pop())
> >         {
> >             // do work
> >         }
> >     }
> > For reference here is the push()
> > void MPSC_Queue::push(void* item)
> > {
> >     item->next = NULL;
> >     void* old_head = InterlockedExchangePointer((void *volatile
> > *)&m_head, item);
> >     // BLOCK POINT
> >     old_head->next = item;  // Do I need to ensure this is visible to
> > the consumer thread?
> > }
> 
> Yes, you need to make old_head->next = item store atomic-release store.
> 
> Consumer's m_tail->next load must be atomic-acquire load.
> 
> Producer's RMW can be totally relaxed with respect to memory barriers.

Producer's RMW can be totally relaxed with respect to memory barriers if
implemented as load-reserve/store-conditional or CAS loop (making
subsequent stores control-dependent).

OTOH, for loopless XCHG-style RMW you need to ensure that it has acquire
semantics (InterlockedExchangePointerAcquire in MS terms).

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#585

On 2011-09-21 10:19:49 +0000, Alexander Terekhov said:

> Ephi Sinowitz wrote:
>> 
>> Hi,
>> I implemented the queue 
>> http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
>> with a signaling mechanism for the consumer.
>> Do I need to ensure visibility somehow before setting the event since
>> I'm not spinning in the consumer? How?
>> Producers:
>> ----------------
>> vyukov_queue.push(item);
>> SetEvent(m_queueEvent);
>> Consumer:
>> ---------------
>> for(;;)
>> {
>> WaitForSingleObject(m_queueEvent, INFINITE);
>> while (void* item = vyukov_queue.pop())
>> {
>> // do work
>> }
>> }
>> For reference here is the push()
>> void MPSC_Queue::push(void* item)
>> {
>> item->next = NULL;
>> void* old_head = InterlockedExchangePointer((void *volatile
>> *)&m_head, item);
>> // BLOCK POINT
>> old_head->next = item;  // Do I need to ensure this is visible to
>> the consumer thread?
>> }
> 
> Yes, you need to make old_head->next = item store atomic-release store.

So another interlocked exchange should take care of it.

> 
> Consumer's m_tail->next load must be atomic-acquire load.

Why I don't see any contention here. Just to ensure we are talking 
about the same thing. (The link in the original post was a little off.)
http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue


Producer's 

> RMW can be totally relaxed with respect to memory barriers.
> 
> regards,
> alexander.

[toc] | [prev] | [next] | [standalone]

#586

Ephi Sinowitz wrote:
[...]
> Why I don't see any contention here. Just to ensure we are talking
> about the same thing. (The link in the original post was a little off.)
> http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue

You posted

<quote>

void MPSC_Queue::push(void* item)
{
    item->next = NULL;
    void* old_head = InterlockedExchangePointer((void *volatile
*)&m_head, item);
    // BLOCK POINT
    old_head->next = item;  // Do I need to ensure this is visible to
the consumer thread?
}

</quote>

which looks like mpscq_push() of non-intrusive version posted at

http://groups.google.com/group/lock-free/browse_thread/thread/55df71b87acb8201

<quote>

struct mpscq_node_t 
{ 
    mpscq_node_t* volatile  next; 
    void*                   state; 

}; 

struct mpscq_t 
{ 
    mpscq_node_t* volatile  head; 
    mpscq_node_t*           tail; 

}; 

void mpscq_create(mpscq_t* self, mpscq_node_t* stub) 
{ 
    stub->next = 0; 
    self->head = stub; 
    self->tail = stub; 

} 

void mpscq_push(mpscq_t* self, mpscq_node_t* n) 
{ 
    n->next = 0; 
    mpscq_node_t* prev = XCHG(&self->head, n); // serialization-point 
wrt producers 
    prev->next = n; // serialization-point wrt consumer 

} 

mpscq_node_t* mpscq_pop(mpscq_t* self) 
{ 
    mpscq_node_t* tail = self->tail; 
    mpscq_node_t* next = tail->next; // serialization-point wrt 
producers 
    if (next) 
    { 
        self->tail = next; 
        tail->state = next->state; 
        return tail; 
    } 
    return 0; 

} 

</quote>

Intrusive version aside for a moment, lets start with non-intrusive
version....

To repeat:

You need to make prev->next = n store atomic-release store.

Consumer's next = tail->next load must be atomic-acquire load.

Producer's RMW can be totally relaxed with respect to memory barriers if
implemented as load-reserve/store-conditional or CAS loop (making
subsequent stores control-dependent).

OTOH, for loopless XCHG-style RMW you need to ensure that it has acquire
semantics (InterlockedExchangePointerAcquire in MS terms).

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#587

On Sep 21, 8:02 am, Alexander Terekhov <terek...@web.de> wrote:
> Ephi Sinowitz wrote:
>
> [...]
>
> > Why I don't see any contention here. Just to ensure we are talking
> > about the same thing. (The link in the original post was a little off.)
> >http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-m...
>
> You posted
>
> <quote>
>
> void MPSC_Queue::push(void* item)
> {
>     item->next = NULL;
>     void* old_head = InterlockedExchangePointer((void *volatile
> *)&m_head, item);
>     // BLOCK POINT
>     old_head->next = item;  // Do I need to ensure this is visible to
> the consumer thread?
>
> }
>
> </quote>
>
> which looks like mpscq_push() of non-intrusive version posted at
>
> http://groups.google.com/group/lock-free/browse_thread/thread/55df71b...
>
> <quote>
>
> struct mpscq_node_t
> {
>     mpscq_node_t* volatile  next;
>     void*                   state;
>
> };
>
> struct mpscq_t
> {
>     mpscq_node_t* volatile  head;
>     mpscq_node_t*           tail;
>
> };
>
> void mpscq_create(mpscq_t* self, mpscq_node_t* stub)
> {
>     stub->next = 0;
>     self->head = stub;
>     self->tail = stub;
>
> }
>
> void mpscq_push(mpscq_t* self, mpscq_node_t* n)
> {
>     n->next = 0;
>     mpscq_node_t* prev = XCHG(&self->head, n); // serialization-point
> wrt producers
>     prev->next = n; // serialization-point wrt consumer
>
> }
>
> mpscq_node_t* mpscq_pop(mpscq_t* self)
> {
>     mpscq_node_t* tail = self->tail;
>     mpscq_node_t* next = tail->next; // serialization-point wrt
> producers
>     if (next)
>     {
>         self->tail = next;
>         tail->state = next->state;
>         return tail;
>     }
>     return 0;
>
> }
>
> </quote>
>
> Intrusive version aside for a moment, lets start with non-intrusive
> version....
>
> To repeat:
>
> You need to make prev->next = n store atomic-release store.
>
> Consumer's next = tail->next load must be atomic-acquire load.
>
> Producer's RMW can be totally relaxed with respect to memory barriers if
> implemented as load-reserve/store-conditional or CAS loop (making
> subsequent stores control-dependent).
>
> OTOH, for loopless XCHG-style RMW you need to ensure that it has acquire
> semantics (InterlockedExchangePointerAcquire in MS terms).
>
> regards,
> alexander.

OK I think I get it. Even though I am setting an event after each push
there is no guarantee of visiblity on the next pointer. However if I
am correct, looping on the consumer should remove the need for the
store-release / load-acquire on the next pointer. Like so. Does that
make sense?
for(;;)
    {
        WaitForSingleObject(m_queueEvent, 30000 /*30 secs*/);
        while (void* item = vyukov_queue.pop())
        {
            // do work
        }
    }

[toc] | [prev] | [next] | [standalone]

#588

Ephi Sinowitz wrote:
[...]
> OK I think I get it. Even though I am setting an event after each push
> there is no guarantee of visiblity on the next pointer. However if I

The whole point of "vyukov's mpsc queue" is to get rid of blocking
syncronization objects.

> am correct, looping on the consumer should remove the need for the
> store-release / load-acquire on the next pointer. Like so. Does that
> make sense?

You'd need a counting semaphore, not auto-reset event (which is binary
semaphore).

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#589

On Sep 21, 9:18 am, Alexander Terekhov <terek...@web.de> wrote:
> Ephi Sinowitz wrote:
>
> [...]
>
> > OK I think I get it. Even though I am setting an event after each push
> > there is no guarantee of visiblity on the next pointer. However if I
>
> The whole point of "vyukov's mpsc queue" is to get rid of blocking
> syncronization objects.
>
> > am correct, looping on the consumer should remove the need for the
> > store-release / load-acquire on the next pointer. Like so. Does that
> > make sense?
>
> You'd need a counting semaphore, not auto-reset event (which is binary
> semaphore).
>
> regards,
> alexander.

Not sure why.
Producers set event after the push(). Event only transitions from
unset to set when producer is inside the while loop.
Now when Consumer goes back to Wait he should see the set event. The
only issue is that he may still see a NULL next pointer because there
are no memory barriers in the push(). In that case the consumer will
go back to wait and the event will stay unset forever. But the 30
second time limit should take care of that.
for(;;)
       WaitForSingleObject(m_queueEvent, 30000);
        while (void* item = vyukov_queue.pop())
        {
            // do work
        }
}

[toc] | [prev] | [next] | [standalone]

#590

Ephi Sinowitz wrote:
[...]
> Producers set event after the push(). Event only transitions from
> unset to set when producer is inside the while loop.
> Now when Consumer goes back to Wait he should see the set event. The
> only issue is that he may still see a NULL next pointer because there
> are no memory barriers in the push(). In that case the consumer will
> go back to wait and the event will stay unset forever. But the 30
> second time limit should take care of that.
> for(;;)
>        WaitForSingleObject(m_queueEvent, 30000);
>         while (void* item = vyukov_queue.pop())
>         {
>             // do work
>         }
> }

With *timed* wait on <whatever> in consumer, push() and pop() need the
same barriers as without <whatever> blocking synchronization object.

Think of it as sleep(30) instead of WaitForSingleObject(m_queueEvent,
30000).

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#591

On Sep 21, 11:06 am, Alexander Terekhov <terek...@web.de> wrote:
> Ephi Sinowitz wrote:
>
> [...]
>
> > Producers set event after the push(). Event only transitions from
> > unset to set when producer is inside the while loop.
> > Now when Consumer goes back to Wait he should see the set event. The
> > only issue is that he may still see a NULL next pointer because there
> > are no memory barriers in the push(). In that case the consumer will
> > go back to wait and the event will stay unset forever. But the 30
> > second time limit should take care of that.
> > for(;;)
> >        WaitForSingleObject(m_queueEvent, 30000);
> >         while (void* item = vyukov_queue.pop())
> >         {
> >             // do work
> >         }
> > }
>
> With *timed* wait on <whatever> in consumer, push() and pop() need the
> same barriers as without <whatever> blocking synchronization object.
>
> Think of it as sleep(30) instead of WaitForSingleObject(m_queueEvent,
> 30000).
>
> regards,
> alexander.

But that's the point. This queue does not need any memory barriers if
the consumer is just spinning with a sleep.

[toc] | [prev] | [next] | [standalone]

#592

Ephi Sinowitz wrote:
> 
> On Sep 21, 11:06 am, Alexander Terekhov <terek...@web.de> wrote:
> > Ephi Sinowitz wrote:
> >
> > [...]
> >
> > > Producers set event after the push(). Event only transitions from
> > > unset to set when producer is inside the while loop.
> > > Now when Consumer goes back to Wait he should see the set event. The
> > > only issue is that he may still see a NULL next pointer because there
> > > are no memory barriers in the push(). In that case the consumer will
> > > go back to wait and the event will stay unset forever. But the 30
> > > second time limit should take care of that.
> > > for(;;)
> > >        WaitForSingleObject(m_queueEvent, 30000);
> > >         while (void* item = vyukov_queue.pop())
> > >         {
> > >             // do work
> > >         }
> > > }
> >
> > With *timed* wait on <whatever> in consumer, push() and pop() need the
> > same barriers as without <whatever> blocking synchronization object.
> >
> > Think of it as sleep(30) instead of WaitForSingleObject(m_queueEvent,
> > 30000).
> >
> > regards,
> > alexander.
> 
> But that's the point. This queue does not need any memory barriers if
> the consumer is just spinning with a sleep.

Who told you that, Ephi?

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#593

Since you are doing signaling (SetEvent) after each push, I think you
don't need anything special for it to work.
However of course you need correct memory fences in the queue
algorithm itself, that is I think as follows:

void mpscq_push(mpscq_t* self, mpscq_node_t* n)
{
    n->next = 0;
    mpscq_node_t* prev = XCHG(&self->head, n); // ACQUIRE_RELEASE
    prev->next = n; // RELEASE
}

mpscq_node_t* mpscq_pop(mpscq_t* self)
{
    mpscq_node_t* tail = self->tail;
    mpscq_node_t* next = tail->next; // ACQUIRE
    if (next)
    {
        self->tail = next;
        tail->state = next->state;
        return tail;
    }
    return 0;
}

XCHG-RELEASE in push synchronizes-with XCHG-ACQUIRE in push executed
by subsequent producer (that is to properly handoff a node from one
producer to another in order to fill in 'next').
STORE-RELEASE in push synchronizes-with LOAD-ACQUIRE in pop (that is
to properly hand off nodes from producers to consumers).

However, that frequent signaling makes the whole algorithm sort of
senseless, because SetEvent is heavy and most likely involved locking.
Ideally you want to signal only when there is a waiting consumer.
Simple CRITICAL_SECTION+CondVar will be most likely faster than this.

[toc] | [prev] | [next] | [standalone]

#594

Dmitriy Vyukov wrote:
> 
> Since you are doing signaling (SetEvent) after each push, I think you
> don't need anything special for it to work.
> However of course you need correct memory fences in the queue
> algorithm itself, that is I think as follows:
> 
> void mpscq_push(mpscq_t* self, mpscq_node_t* n)
> {
>     n->next = 0;
>     mpscq_node_t* prev = XCHG(&self->head, n); // ACQUIRE_RELEASE

"// ACQUIRE" (without _RELEASE) is sufficient here.

But XCHG(...) can be totally relaxed with respect to memory barriers if
implemented as load-reserve/store-conditional or CAS loop (making
subsequent stores control-dependent).

X86 aside, try to unfold XCHG() using load-reserve/store-conditional or
CAS loop.

>     prev->next = n; // RELEASE

Yes.

> }
> 
> mpscq_node_t* mpscq_pop(mpscq_t* self)
> {
>     mpscq_node_t* tail = self->tail;
>     mpscq_node_t* next = tail->next; // ACQUIRE

Yes.

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#619

On Wednesday, September 21, 2011 9:33:42 PM UTC+4, Alexander Terekhov wrote:
> Dmitriy Vyukov wrote:
> > 
> > Since you are doing signaling (SetEvent) after each push, I think you
> > don't need anything special for it to work.
> > However of course you need correct memory fences in the queue
> > algorithm itself, that is I think as follows:
> > 
> > void mpscq_push(mpscq_t* self, mpscq_node_t* n)
> > {
> >     n->next = 0;
> >     mpscq_node_t* prev = XCHG(&self->head, n); // ACQUIRE_RELEASE
> 
> "// ACQUIRE" (without _RELEASE) is sufficient here.
> 
> But XCHG(...) can be totally relaxed with respect to memory barriers if
> implemented as load-reserve/store-conditional or CAS loop (making
> subsequent stores control-dependent).

In my model there is no such thing as "control-dependent", each acquire simply must pair with release :)
But I am curious about your model, how is n->next=0 prevented from sinking below XCHG (which would be a disaster)? It seems that even a compiler can easily do that.

[toc] | [prev] | [next] | [standalone]

#623

dvyukov@google.com wrote:
> 
> On Wednesday, September 21, 2011 9:33:42 PM UTC+4, Alexander Terekhov wrote:
> > Dmitriy Vyukov wrote:
> > >
> > > Since you are doing signaling (SetEvent) after each push, I think you
> > > don't need anything special for it to work.
> > > However of course you need correct memory fences in the queue
> > > algorithm itself, that is I think as follows:
> > >
> > > void mpscq_push(mpscq_t* self, mpscq_node_t* n)
> > > {
> > >     n->next = 0;
> > >     mpscq_node_t* prev = XCHG(&self->head, n); // ACQUIRE_RELEASE
> >
> > "// ACQUIRE" (without _RELEASE) is sufficient here.
> >
> > But XCHG(...) can be totally relaxed with respect to memory barriers if
> > implemented as load-reserve/store-conditional or CAS loop (making
> > subsequent stores control-dependent).
> 
> In my model there is no such thing as "control-dependent", 
> each acquire simply must pair with release :)
> But I am curious about your model, how is n->next=0 prevented 
> from sinking below XCHG (which would be a disaster)? 
> It seems that even a compiler can easily do that.

My mistake. The XCHG's store is not control dependent on 
n->next = 0 and hence must have either release or sink-store 
(relaxed release affecting only preceding stores) label.

regards,
alexander.

[toc] | [prev] | [next] | [standalone]

#595

On Sep 21, 12:58 pm, Dmitriy Vyukov <dvyu...@gmail.com> wrote:
> Since you are doing signaling (SetEvent) after each push, I think you
> don't need anything special for it to work.
> However of course you need correct memory fences in the queue
> algorithm itself, that is I think as follows:
>
> void mpscq_push(mpscq_t* self, mpscq_node_t* n)
> {
>     n->next = 0;
>     mpscq_node_t* prev = XCHG(&self->head, n); // ACQUIRE_RELEASE
>     prev->next = n; // RELEASE
>
> }
>
> mpscq_node_t* mpscq_pop(mpscq_t* self)
> {
>     mpscq_node_t* tail = self->tail;
>     mpscq_node_t* next = tail->next; // ACQUIRE
>     if (next)
>     {
>         self->tail = next;
>         tail->state = next->state;
>         return tail;
>     }
>     return 0;
>
> }
>
> XCHG-RELEASE in push synchronizes-with XCHG-ACQUIRE in push executed
> by subsequent producer (that is to properly handoff a node from one
> producer to another in order to fill in 'next').
> STORE-RELEASE in push synchronizes-with LOAD-ACQUIRE in pop (that is
> to properly hand off nodes from producers to consumers).

But I am on x64. Shouldn't the STORE-RELEASE /LOAD-ACQUIRE come
automatically.
http://www.justsoftwaresolutions.co.uk/threading/intel-memory-ordering-and-c++-memory-model.html

>
> However, that frequent signaling makes the whole algorithm sort of
> senseless, because SetEvent is heavy and most likely involved locking.
> Ideally you want to signal only when there is a waiting consumer.
> Simple CRITICAL_SECTION+CondVar will be most likely faster than this.

OK. So like this?
Producers
---------------
push()
if (TryEnterCriticalSection(&cs)) WakeConditionVariable(&cond);

Consumer
---------------
EnterCriticalSection(&cs);
for (;;)
   SleepConditionVariableCS (&cond, &cs, INFINITE);
    while (void* item = vyukov_queue.pop())
    {
          // do work
     }
}

[toc] | [prev] | [next] | [standalone]

Page 1 of 2  [1] 2  Next page →

Back to top | Article view | comp.programming.threads

csiph-web