Merge pull request #1479 from jens-auer/issue1477

Fixed #1477  corruption in "zero-copy" raw_decoder for payloads large…

Makefile.am

@@ -364,7 +364,8 @@ test_apps = \
 	tests/test_client_drop_more \
 	tests/test_thread_safe \
 	tests/test_socketopt_hwm \
-	tests/test_heartbeats
+	tests/test_heartbeats \
+	tests/test_stream_exceeds_buffer
 
 tests_test_system_SOURCES = tests/test_system.cpp
 tests_test_system_LDADD = src/libzmq.la
@@ -569,6 +570,9 @@ tests_test_socketopt_hwm_LDADD = src/libzmq.la
 tests_test_heartbeats_SOURCES = tests/test_heartbeats.cpp
 tests_test_heartbeats_LDADD = src/libzmq.la
 
+tests_test_stream_exceeds_buffer_SOURCES = tests/test_stream_exceeds_buffer.cpp
+tests_test_stream_exceeds_buffer_LDADD = src/libzmq.la
+
 if !ON_MINGW
 if !ON_CYGWIN
 test_apps += \

src/raw_decoder.cpp

@@ -62,6 +62,7 @@ int zmq::raw_decoder_t::decode (const uint8_t *data_, size_t size_,
 {
     int rc = in_progress.init ((unsigned char*)data_, size_,
                                shared_message_memory_allocator::call_dec_ref,
+                               allocator.buffer(),
                                allocator.create_refcnt() );
 
     // if the buffer serves as memory for a zero-copy message, release it

tests/test_stream_exceeds_buffer.cpp

@@ -0,0 +1,83 @@
+#include <assert.h>
+#include <string.h>
+#include <stdio.h>
+
+#include <netinet/in.h>
+#include <unistd.h>
+
+#include <zmq.h>
+
+int main()
+{
+    const int msgsize = 8193;
+    char sndbuf[msgsize] = "\xde\xad\xbe\xef";
+    unsigned char rcvbuf[msgsize];
+
+    int server_sock = socket(AF_INET, SOCK_STREAM, 0);
+    assert(server_sock!=-1);
+    int enable = 1;
+    int rc = setsockopt(server_sock, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
+    assert(rc!=-1);
+
+    struct sockaddr_in saddr;
+    memset(&saddr, 0, sizeof(saddr));
+    saddr.sin_family = AF_INET;
+    saddr.sin_addr.s_addr = INADDR_ANY;
+    saddr.sin_port = htons(12345);
+
+    rc = bind(server_sock, (struct sockaddr *)&saddr, sizeof(saddr));
+    assert(rc!=-1);
+    rc = listen(server_sock, 1);
+    assert(rc!=-1);
+
+    void *zctx = zmq_ctx_new();
+    assert(zctx);
+    void *zsock = zmq_socket(zctx, ZMQ_STREAM);
+    assert(zsock);
+    rc = zmq_connect(zsock, "tcp://127.0.0.1:12345");
+    assert(rc!=-1);
+
+    int client_sock = accept(server_sock, NULL, NULL);
+    assert(client_sock!=-1);
+
+    rc = close(server_sock);
+    assert(rc!=-1);
+
+    rc = send(client_sock, sndbuf, msgsize, 0);
+    assert(rc==msgsize);
+
+    zmq_msg_t msg;
+    zmq_msg_init(&msg);
+
+    int rcvbytes = 0;
+    while (rcvbytes==0) // skip connection notification, if any
+    {
+        rc = zmq_msg_recv(&msg, zsock, 0);  // peerid
+        assert(rc!=-1);
+        assert(zmq_msg_more(&msg));
+        rcvbytes = zmq_msg_recv(&msg, zsock, 0);
+        assert(rcvbytes!=-1);
+        assert(!zmq_msg_more(&msg));
+        printf("got %d bytes\n", rcvbytes);
+    }
+
+    // for this test, we only collect the first chunk
+    // since the corruption already occurs in the first chunk
+    memcpy(rcvbuf, zmq_msg_data(&msg), zmq_msg_size(&msg));
+
+    zmq_msg_close(&msg);
+    zmq_close(zsock);
+    close(client_sock);
+
+    zmq_ctx_destroy(zctx);
+
+    assert(rcvbytes >= 4);
+    printf("%x %x %x %x\n", rcvbuf[0], rcvbuf[1], rcvbuf[2], rcvbuf[3]);
+
+    // notice that only the 1st byte gets corrupted
+    assert(rcvbuf[3]==0xef);
+    assert(rcvbuf[2]==0xbe);
+    assert(rcvbuf[1]==0xad);
+    assert(rcvbuf[0]==0xde);
+}
+