Implement flow control

This commit introduces the necessary changes necessary
for implementing flow control. None of the socket types
implements the flow control yet. The code will crash when
the flow control is enabled and the thw lwm is reached.

The following commits will add flow-control support for
individual socket types.
This commit is contained in:
Martin Hurton
2010-03-01 10:13:26 +01:00
parent 31d36104aa
commit 61ee6fae53
40 changed files with 243 additions and 78 deletions

View File

@@ -68,10 +68,8 @@ namespace zmq
uint64_t hwm;
uint64_t lwm;
// Positions of head and tail of the pipe (in bytes).
uint64_t head;
uint64_t tail;
uint64_t last_sent_head;
// Number of messages read so far.
uint64_t msgs_read;
// Endpoint (either session or socket) the pipe is attached to.
i_endpoint *endpoint;
@@ -91,10 +89,10 @@ namespace zmq
void set_pipe (class pipe_t *pipe_);
void set_endpoint (i_endpoint *endpoint_);
// Checks whether message with specified size can be written to the
// pipe. If writing the message would cause high watermark to be
// Checks whether a message can be written to the pipe.
// If writing the message would cause high watermark to be
// exceeded, the function returns false.
bool check_write (uint64_t size_);
bool check_write ();
// Writes a message to the underlying pipe. Returns false if the
// message cannot be written because high watermark was reached.
@@ -111,9 +109,14 @@ namespace zmq
private:
void process_reader_info (uint64_t msgs_read_);
// Command handlers.
void process_pipe_term ();
// Tests whether the pipe is already full.
bool pipe_full ();
// The underlying pipe.
class pipe_t *pipe;
@@ -124,9 +127,15 @@ namespace zmq
uint64_t hwm;
uint64_t lwm;
// Positions of head and tail of the pipe (in bytes).
uint64_t head;
uint64_t tail;
// Last confirmed number of messages read from the pipe.
// The actual number can be higher.
uint64_t msgs_read;
// Number of messages we have written so far.
uint64_t msgs_written;
// True iff the last attempt to write a message has failed.
bool stalled;
// Endpoint (either session or socket) the pipe is attached to.
i_endpoint *endpoint;