1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
/*
Copyright (c) 2009-2011 250bpm s.r.o.
Copyright (c) 2007-2009 iMatix Corporation
Copyright (c) 2011 VMware, Inc.
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
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
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_PIPE_HPP_INCLUDED__
#define __ZMQ_PIPE_HPP_INCLUDED__
#include "msg.hpp"
#include "ypipe.hpp"
#include "config.hpp"
#include "object.hpp"
#include "stdint.hpp"
#include "array.hpp"
#include "blob.hpp"
namespace zmq
{
class object_t;
class pipe_t;
// Create a pipepair for bi-directional transfer of messages.
// First HWM is for messages passed from first pipe to the second pipe.
// Second HWM is for messages passed from second pipe to the first pipe.
// Delay specifies how the pipe behaves when the peer terminates. If true
// pipe receives all the pending messages before terminating, otherwise it
// terminates straight away.
int pipepair (zmq::object_t *parents_ [2], zmq::pipe_t* pipes_ [2],
int hwms_ [2], bool delays_ [2]);
struct i_pipe_events
{
virtual ~i_pipe_events () {}
virtual void read_activated (zmq::pipe_t *pipe_) = 0;
virtual void write_activated (zmq::pipe_t *pipe_) = 0;
virtual void hiccuped (zmq::pipe_t *pipe_) = 0;
virtual void terminated (zmq::pipe_t *pipe_) = 0;
};
// Note that pipe can be stored in three different arrays.
// The array of inbound pipes (1), the array of outbound pipes (2) and
// the generic array of pipes to deallocate (3).
class pipe_t :
public object_t,
public array_item_t <1>,
public array_item_t <2>,
public array_item_t <3>
{
// This allows pipepair to create pipe objects.
friend int pipepair (zmq::object_t *parents_ [2],
zmq::pipe_t* pipes_ [2], int hwms_ [2], bool delays_ [2]);
public:
// Specifies the object to send events to.
void set_event_sink (i_pipe_events *sink_);
// Pipe endpoint can store an opaque ID to be used by its clients.
void set_identity (const blob_t &identity_);
blob_t get_identity ();
// Returns true if there is at least one message to read in the pipe.
bool check_read ();
// Reads a message to the underlying pipe.
bool read (msg_t *msg_);
// Checks whether messages can be written to the pipe. If writing
// the message would cause high watermark the function returns false.
bool check_write ();
// Writes a message to the underlying pipe. Returns false if the
// message cannot be written because high watermark was reached.
bool write (msg_t *msg_);
// Remove unfinished parts of the outbound message from the pipe.
void rollback ();
// Flush the messages downsteam.
void flush ();
// Temporaraily disconnects the inbound message stream and drops
// all the messages on the fly. Causes 'hiccuped' event to be generated
// in the peer.
void hiccup ();
// Ask pipe to terminate. The termination will happen asynchronously
// and user will be notified about actual deallocation by 'terminated'
// event. If delay is true, the pending messages will be processed
// before actual shutdown.
void terminate (bool delay_);
private:
// Type of the underlying lock-free pipe.
typedef ypipe_t <msg_t, message_pipe_granularity> upipe_t;
// Command handlers.
void process_activate_read ();
void process_activate_write (uint64_t msgs_read_);
void process_hiccup (void *pipe_);
void process_pipe_term ();
void process_pipe_term_ack ();
// Handler for delimiter read from the pipe.
void delimit ();
// Constructor is private. Pipe can only be created using
// pipepair function.
pipe_t (object_t *parent_, upipe_t *inpipe_, upipe_t *outpipe_,
int inhwm_, int outhwm_, bool delay_);
// Pipepair uses this function to let us know about
// the peer pipe object.
void set_peer (pipe_t *pipe_);
// Destructor is private. Pipe objects destroy themselves.
~pipe_t ();
// Underlying pipes for both directions.
upipe_t *inpipe;
upipe_t *outpipe;
// Can the pipe be read from / written to?
bool in_active;
bool out_active;
// High watermark for the outbound pipe.
int hwm;
// Low watermark for the inbound pipe.
int lwm;
// Number of messages read and written so far.
uint64_t msgs_read;
uint64_t msgs_written;
// Last received peer's msgs_read. The actual number in the peer
// can be higher at the moment.
uint64_t peers_msgs_read;
// The pipe object on the other side of the pipepair.
pipe_t *peer;
// Sink to send events to.
i_pipe_events *sink;
// State of the pipe endpoint. Active is common state before any
// termination begins. Delimited means that delimiter was read from
// pipe before term command was received. Pending means that term
// command was already received from the peer but there are still
// pending messages to read. Terminating means that all pending
// messages were already read and all we are waiting for is ack from
// the peer. Terminated means that 'terminate' was explicitly called
// by the user. Double_terminated means that user called 'terminate'
// and then we've got term command from the peer as well.
enum {
active,
delimited,
pending,
terminating,
terminated,
double_terminated
} state;
// If true, we receive all the pending inbound messages before
// terminating. If false, we terminate immediately when the peer
// asks us to.
bool delay;
// Identity of the writer. Used uniquely by the reader side.
blob_t identity;
// Returns true if the message is delimiter; false otherwise.
static bool is_delimiter (msg_t &msg_);
// Computes appropriate low watermark from the given high watermark.
static int compute_lwm (int hwm_);
// Disable copying.
pipe_t (const pipe_t&);
const pipe_t &operator = (const pipe_t&);
};
}
#endif
|
