yqueue.hpp 6.41 KB
Newer Older
Martin Sustrik's avatar
Martin Sustrik committed
1
/*
2 3
    Copyright (c) 2007-2011 iMatix Corporation
    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
Martin Sustrik's avatar
Martin Sustrik committed
4 5 6 7

    This file is part of 0MQ.

    0MQ is free software; you can redistribute it and/or modify it under
8
    the terms of the GNU Lesser General Public License as published by
Martin Sustrik's avatar
Martin Sustrik committed
9 10 11 12 13 14
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.

    0MQ is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15
    GNU Lesser General Public License for more details.
Martin Sustrik's avatar
Martin Sustrik committed
16

17
    You should have received a copy of the GNU Lesser General Public License
Martin Sustrik's avatar
Martin Sustrik committed
18 19 20
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

Martin Sustrik's avatar
Martin Sustrik committed
21 22
#ifndef __ZMQ_YQUEUE_HPP_INCLUDED__
#define __ZMQ_YQUEUE_HPP_INCLUDED__
Martin Sustrik's avatar
Martin Sustrik committed
23

24
#include <stdlib.h>
Martin Sustrik's avatar
Martin Sustrik committed
25 26 27
#include <stddef.h>

#include "err.hpp"
28
#include "atomic_ptr.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
29

Martin Sustrik's avatar
Martin Sustrik committed
30
namespace zmq
Martin Sustrik's avatar
Martin Sustrik committed
31 32 33 34 35 36 37 38 39 40 41
{

    //  yqueue is an efficient queue implementation. The main goal is
    //  to minimise number of allocations/deallocations needed. Thus yqueue
    //  allocates/deallocates elements in batches of N.
    //
    //  yqueue allows one thread to use push/back function and another one 
    //  to use pop/front functions. However, user must ensure that there's no
    //  pop on the empty queue and that both threads don't access the same
    //  element in unsynchronised manner.
    //
42
    //  T is the type of the object in the queue.
Martin Sustrik's avatar
Martin Sustrik committed
43
    //  N is granularity of the queue (how many pushes have to be done till
44
    //  actual memory allocation is required).
Martin Sustrik's avatar
Martin Sustrik committed
45 46 47 48 49 50 51 52

    template <typename T, int N> class yqueue_t
    {
    public:

        //  Create the queue.
        inline yqueue_t ()
        {
53
             begin_chunk = (chunk_t*) malloc (sizeof (chunk_t));
54
             alloc_assert (begin_chunk);
Martin Sustrik's avatar
Martin Sustrik committed
55 56 57 58 59 60 61 62 63 64 65
             begin_pos = 0;
             back_chunk = NULL;
             back_pos = 0;
             end_chunk = begin_chunk;
             end_pos = 0;
        }

        //  Destroy the queue.
        inline ~yqueue_t ()
        {
            while (true) {
66
                if (begin_chunk == end_chunk) {
67
                    free (begin_chunk);
68 69
                    break;
                } 
Martin Sustrik's avatar
Martin Sustrik committed
70 71
                chunk_t *o = begin_chunk;
                begin_chunk = begin_chunk->next;
72
                free (o);
Martin Sustrik's avatar
Martin Sustrik committed
73
            }
74 75 76

            chunk_t *sc = spare_chunk.xchg (NULL);
            if (sc)
77
                free (sc);
Martin Sustrik's avatar
Martin Sustrik committed
78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99
        }

        //  Returns reference to the front element of the queue.
        //  If the queue is empty, behaviour is undefined.
        inline T &front ()
        {
             return begin_chunk->values [begin_pos];
        }

        //  Returns reference to the back element of the queue.
        //  If the queue is empty, behaviour is undefined.
        inline T &back ()
        {
            return back_chunk->values [back_pos];
        }

        //  Adds an element to the back end of the queue.
        inline void push ()
        {
            back_chunk = end_chunk;
            back_pos = end_pos;

Martin Sustrik's avatar
Martin Sustrik committed
100
            if (++end_pos != N)
Martin Sustrik's avatar
Martin Sustrik committed
101 102
                return;

103 104 105
            chunk_t *sc = spare_chunk.xchg (NULL);
            if (sc) {
                end_chunk->next = sc;
106
                sc->prev = end_chunk;
107
            } else {
108
                end_chunk->next = (chunk_t*) malloc (sizeof (chunk_t));
109
                alloc_assert (end_chunk->next);
110
                end_chunk->next->prev = end_chunk;
111
            }
Martin Sustrik's avatar
Martin Sustrik committed
112 113 114 115
            end_chunk = end_chunk->next;
            end_pos = 0;
        }

116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146
        //  Removes element from the back end of the queue. In other words
        //  it rollbacks last push to the queue. Take care: Caller is
        //  responsible for destroying the object being unpushed.
        //  The caller must also guarantee that the queue isn't empty when
        //  unpush is called. It cannot be done automatically as the read
        //  side of the queue can be managed by different, completely
        //  unsynchronised thread.
        inline void unpush ()
        {
            //  First, move 'back' one position backwards.
            if (back_pos)
                --back_pos;
            else {
                back_pos = N - 1;
                back_chunk = back_chunk->prev;
            }

            //  Now, move 'end' position backwards. Note that obsolete end chunk
            //  is not used as a spare chunk. The analysis shows that doing so
            //  would require free and atomic operation per chunk deallocated
            //  instead of a simple free.
            if (end_pos)
                --end_pos;
            else {
                end_pos = N - 1;
                end_chunk = end_chunk->prev;
                free (end_chunk->next);
                end_chunk->next = NULL;
            }
        }

Martin Sustrik's avatar
Martin Sustrik committed
147 148 149 150 151 152
        //  Removes an element from the front end of the queue.
        inline void pop ()
        {
            if (++ begin_pos == N) {
                chunk_t *o = begin_chunk;
                begin_chunk = begin_chunk->next;
153
                begin_chunk->prev = NULL;
Martin Sustrik's avatar
Martin Sustrik committed
154
                begin_pos = 0;
155 156 157 158 159 160

                //  'o' has been more recently used than spare_chunk,
                //  so for cache reasons we'll get rid of the spare and
                //  use 'o' as the spare.
                chunk_t *cs = spare_chunk.xchg (o);
                if (cs)
161
                    free (cs);
Martin Sustrik's avatar
Martin Sustrik committed
162 163 164 165 166 167 168 169 170
            }
        }

    private:

        //  Individual memory chunk to hold N elements.
        struct chunk_t
        {
             T values [N];
171
             chunk_t *prev;
Martin Sustrik's avatar
Martin Sustrik committed
172 173 174 175 176 177 178 179 180 181 182 183 184 185
             chunk_t *next;
        };

        //  Back position may point to invalid memory if the queue is empty,
        //  while begin & end positions are always valid. Begin position is
        //  accessed exclusively be queue reader (front/pop), while back and
        //  end positions are accessed exclusively by queue writer (back/push).
        chunk_t *begin_chunk;
        int begin_pos;
        chunk_t *back_chunk;
        int back_pos;
        chunk_t *end_chunk;
        int end_pos;

186 187
        //  People are likely to produce and consume at similar rates.  In
        //  this scenario holding onto the most recently freed chunk saves
188
        //  us from having to call malloc/free.
189 190
        atomic_ptr_t<chunk_t> spare_chunk;

Martin Sustrik's avatar
Martin Sustrik committed
191 192
        //  Disable copying of yqueue.
        yqueue_t (const yqueue_t&);
193
        const yqueue_t &operator = (const yqueue_t&);
Martin Sustrik's avatar
Martin Sustrik committed
194 195 196 197 198
    };

}

#endif