Possible inconsistency with xrep_t::current_in, xrep_t::more_in after pipe is terminated

[nirs]

It seems that after a reader invoke xrep_t::terminated() and the pipe is removed from the inpipes vector, current_in is not updated.

If the terminated pipe was the last in the list, the current_in was last index, it will be invalid at this point. The next time xrecv is called, inpipes[current_in] may be out of the vector bounds.

If the terminated pipe was before current_in, and more_in is true, current_in points now to the next pipe, and more_in is invalid.

I did not check that this situation is reproducible.

[sustrik]

Hi, can you point me to the version of 0MQ you are using, the file and the line? I'll have a look.

[nirs]

It was master branch at the time I started this issue.

More specific:
When a reader invokes xrep_t::terminated:
http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L90

The reader pipe is removed form the inpipes, without updating the current_in member.

current_in is initialized here:
http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L28

It is updated when iterating over readers here: http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L211

and here:
http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L225

Now if it is possible that a reader is terminated in the middle of this loop, for example, some other code is called, and the other code cause the reader to close, current_in wil be wrong, as described above.

Even if this is not possible now, it may be possible in the future. current_in should be update when a reader is terminated, just like current_out is when a writer is terminated, here:
http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L104

[sustrik]

I've checked the code and AFAICS the reader cannot be terminated in the middle of the loop. The list of pipes is local to this thread and cannot be modified in the background.

[sustrik]

Ah, but a different thing can happen: Termination of reader can leave current_in pointing past the end of the array and subsequent attempt to recv may thus segfault!

[sustrik]

Fixed in r98fa2fa. Thanks for spotting this!

[nirs]

This does not fix the inconsistency with more_in, prefetched and prefetched_msg members.

Possible issues:

1. Reader terminates when after you prefetched the single message and returned the "identity" message to the caller. current_in points now the next reader. When xrecv is called again, you will return the prefetched message from the terminated reader.
2. Reader terminates after you fetched the first message out of multipart message. and more_in is set to true. current_in points now to the next reader. When xrecv is called again, you will try to return the next message part from the next reader, which may be passive, and then you segfault here:
   http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L215

A better fix is to update all members related to reading a message from a reader: prefetched, prefetched_msg and more_in.

Something like:

void xrep_t::reset_reader()
{
    if (prefetched) {
        zmq_msg_close (&prefetched_msg);
        prefetched = false;
    }
    more_in = false;
}

The you invoke this when current reader is changed.

[sustrik]

The point is that reader cannot terminate in the middle of multipart message. Message is delivered either whole or not at all.

[nirs]

So the reader is responsible for checking xrep state before calling terminated? Where is this enforced?

[sustrik]

Nope. terminate only happens when reader reads the last message (the "delimiter") from the pipe. Pipe writer is responsible for sending only whole messages, so the delimiter may not occur in the middle of the message. If there's half written message and sender wants to send delimiter, it rollbacks the unfinished message from the pipe.

[nirs]

Ok, maybe some asserts are missing, blowing up if the reader breaks the contract you described.

Add above http://github.com/zeromq/zeromq2/blob/master/src/xrep.cpp#L104:

if (pipe_ == inpipes [current_in].reader) {
    // Reader is not allowed to terminate in the middle of multipart message
    assert (!prefetched);
    assert (!more_in);
}

[sustrik]

Sure. Feel free to send it as a patch to the mailing list.

[nirs]

I guess we can close this issue.