fq.cpp 4.04 KB
Newer Older
1
/*
2
    Copyright (c) 2007-2014 Contributors as noted in the AUTHORS file
3 4 5 6

    This file is part of 0MQ.

    0MQ is free software; you can redistribute it and/or modify it under
7
    the terms of the GNU Lesser General Public License as published by
8 9 10 11 12 13
    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
14
    GNU Lesser General Public License for more details.
15

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

#include "fq.hpp"
#include "pipe.hpp"
#include "err.hpp"
23
#include "msg.hpp"
24

25
zmq::fq_t::fq_t () :
26
    active (0),
27
    last_in (NULL),
28
    current (0),
29
    more (false)
30 31 32 33 34
{
}

zmq::fq_t::~fq_t ()
{
35
    zmq_assert (pipes.empty ());
36 37
}

38
void zmq::fq_t::attach (pipe_t *pipe_)
39 40 41 42 43 44
{
    pipes.push_back (pipe_);
    pipes.swap (active, pipes.size () - 1);
    active++;
}

45
void zmq::fq_t::pipe_terminated (pipe_t *pipe_)
46
{
47 48
    const pipes_t::size_type index = pipes.index (pipe_);

49 50
    //  Remove the pipe from the list; adjust number of active pipes
    //  accordingly.
51
    if (index < active) {
52
        active--;
53
        pipes.swap (index, active);
54 55 56
        if (current == active)
            current = 0;
    }
57
    pipes.erase (pipe_);
58 59 60 61 62

    if (last_in == pipe_) {
        saved_credential = last_in->get_credential ();
        last_in = NULL;
    }
63 64
}

65
void zmq::fq_t::activated (pipe_t *pipe_)
66 67 68 69 70 71
{
    //  Move the pipe to the list of active pipes.
    pipes.swap (pipes.index (pipe_), active);
    active++;
}

72
int zmq::fq_t::recv (msg_t *msg_)
73
{
74
    return recvpipe (msg_, NULL);
75 76
}

77
int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
78 79
{
    //  Deallocate old content of the message.
80 81
    int rc = msg_->close ();
    errno_assert (rc == 0);
82

83
    //  Round-robin over the pipes to get the next message.
Martin Hurton's avatar
Martin Hurton committed
84
    while (active > 0) {
85

86 87
        //  Try to fetch new message. If we've already read part of the message
        //  subsequent part should be immediately available.
88
        bool fetched = pipes [current]->read (msg_);
89

90 91
        //  Note that when message is not fetched, current pipe is deactivated
        //  and replaced by another active pipe. Thus we don't have to increase
92 93
        //  the 'current' pointer.
        if (fetched) {
94 95
            if (pipe_)
                *pipe_ = pipes [current];
Martin Hurton's avatar
Martin Hurton committed
96
            more = msg_->flags () & msg_t::more? true: false;
97 98
            if (!more) {
                last_in = pipes [current];
Martin Hurton's avatar
Martin Hurton committed
99
                current = (current + 1) % active;
100
            }
101
            return 0;
102
        }
Martin Hurton's avatar
Martin Hurton committed
103

104 105 106 107 108
        //  Check the atomicity of the message.
        //  If we've already received the first part of the message
        //  we should get the remaining parts without blocking.
        zmq_assert (!more);

Martin Hurton's avatar
Martin Hurton committed
109 110 111 112
        active--;
        pipes.swap (current, active);
        if (current == active)
            current = 0;
113 114 115 116
    }

    //  No message is available. Initialise the output parameter
    //  to be a 0-byte message.
117 118
    rc = msg_->init ();
    errno_assert (rc == 0);
119 120 121 122 123 124
    errno = EAGAIN;
    return -1;
}

bool zmq::fq_t::has_in ()
{
125
    //  There are subsequent parts of the partly-read message available.
126
    if (more)
127 128
        return true;

129 130 131 132
    //  Note that messing with current doesn't break the fairness of fair
    //  queueing algorithm. If there are no messages available current will
    //  get back to its original value. Otherwise it'll point to the first
    //  pipe holding messages, skipping only pipes with no messages available.
Martin Hurton's avatar
Martin Hurton committed
133
    while (active > 0) {
134 135
        if (pipes [current]->check_read ())
            return true;
136 137 138 139 140

        //  Deactivate the pipe.
        active--;
        pipes.swap (current, active);
        if (current == active)
141 142 143 144 145 146
            current = 0;
    }

    return false;
}

147 148 149 150 151 152
zmq::blob_t zmq::fq_t::get_credential () const
{
    return last_in?
        last_in->get_credential (): saved_credential;
}