I'm seeing this assertion sporadically with 2.0.7. Will provide more details when I manage to catch it in the debugger.

I found some queer behaviour demonstrated in the code posted below.
Description in short:
I used the "multithreaded server" example and explicitly set the socket identifier for being able to relaunch a crashed service. During testing, I mostly had the services started, then the clients and then killed and restarted some service and all was fine ;o)

But if I start the Clients first (errorfree trying a connect to still closed socket(s)) and AFTERWARDS start the service(s) (including the inproc-queue-device from the mentioned example), the request, send by the clients while there was no service, is gone. The clients get no reply and no service worker ever gets the request. Result: the client waits endlessly for an answer.

I stripped it down to the (ugly) snipped below.
This demo code runs fine when started without param (delaying the client thread) and shows a client deadlock when started with some parameter (delaying the worker until the client has finisched sending his request and is allready waiting for an answer).

When I do not use the device in between (clients connecting to several sockets and afterward launching the service workers binding directly the socket(s)), it works fine, so I assume, it's something to do with the queue device.

#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <zmq.hpp>
#include <iostream>

using namespace std;

// if true, the worker threads will take 5 seconds to bind the socket,
// so the clients will do the request while there's no worker.
bool DELAY_WORKER;

void* internal_client(void *arg)
{
    if (! DELAY_WORKER) sleep (5);
    zmq::context_t *ctx = (zmq::context_t*) arg;
    zmq::socket_t s (*ctx, ZMQ_REQ);
    s.connect ("tcp://localhost:12345");
    for (int i = 0; i != 10; i++) {
        zmq::message_t request (10);
        memset (request.data (), i+65, request.size ()-1);
        ((char*)request.data ())[request.size ()]=0;
        cout<<"client sending message Nr."<<i<<"with content"<<
                (char*)request.data()<<endl;
        s.send (request);
        //  Get the reply.
        zmq::message_t reply;
        cout<<"client waiting for reply ..."<<endl;
        s.recv (&reply);
        cout<<"client got reply: "<<(char*)reply.data()<<endl;
    }

    return 0;
}

void *worker_routine (void *arg)
{
    if (DELAY_WORKER) sleep (5);
    zmq::context_t *ctx = (zmq::context_t*) arg;
    zmq::socket_t s (*ctx, ZMQ_REP);
    //  Connect to the dispatcher (queue) running in the main thread.
    s.connect ("inproc://workers");
    cout<<"worker connected to queue device now"<<endl;
    while (true) {
        //  Get a request from the dispatcher.
        zmq::message_t request;
        cout << "worker waiting for request"<<endl;
        s.recv (&request);
        cout << "worker received request: " << (char*)request.data() << endl;
        sleep (1);//  to simulate actual processing.
        zmq::message_t reply (10);
        memset (reply.data (), ((char*)request.data())[0], reply.size ()-1);
        ((char*)reply.data ())[reply.size ()]=0;
        s.send (reply);
        cout << "worker sent a reply" <<endl;
    }
}

int main (int argc, char**argv)
{
    if (argc>1) {
            DELAY_WORKER = true;
            cout<<"worker thread will be delayed ==> deadlock"<<endl;
    } else {
            DELAY_WORKER = false;
            cout<<"client thread will be delayed ==> runs fine."<<endl;
    }
    zmq::context_t ctx (1);

    // worker side of queue listening for worker to connect
    zmq::socket_t workers (ctx, ZMQ_XREQ);
    workers.bind ("inproc://workers");
    // client side listening on socke5t for real clients to connect
    zmq::socket_t clients (ctx, ZMQ_XREP);
    clients.bind ("tcp://127.0.0.1:12345");

    // launch worker
    pthread_t worker;
    int rc = pthread_create (&worker, NULL, worker_routine, (void*) &ctx);
    assert (rc == 0);

    //  Launch client threads.
    rc = pthread_create (&worker, NULL, internal_client, (void*) &ctx);
    assert (rc == 0);

    cout<<"QUEUE device gets started"<<endl;
    zmq::device (ZMQ_QUEUE, clients, workers);
    return 0;
}

Per a discussion in the comments here: http://www.zeromq.org/blog:multithreaded-server

Expand the 0mq socket api to allow for attaching callbacks to REQ sockets on send. The callback should be invoked when its matching response arrives.

Similarly, allow for attaching callbacks to SUB sockets. All incoming messages should be passed to the callback for processing.

While these features can currently be built on top of the existing api via zmq_poll, it would be cleaner to have it available as part of the supported api.

I experienced some semi-reproducable segfaults on a socket which uses poll. I could track it down to a buffer that is allocated one byte so small due to empty reads or zero-sized packets (didn't dig too far, though). I also have no easy way to reproduce this outside of our embedded system.

Here's a valgrind report of the problem:

==3958== Thread 3:
==3958== Invalid write of size 4
==3958==    at 0x407F570: zmq_msg_init_size (atomic_counter.hpp:59)
==3958==    by 0x40806C4: zmq::zmq_decoder_t::one_byte_size_ready() (zmq_decoder.cpp:62)
==3958==    by 0x40812EE: zmq::zmq_engine_t::in_event() (decoder.hpp:109)
==3958==    by 0x40699BA: zmq::epoll_t::loop() (epoll.cpp:197)
==3958==    by 0x4069A8C: zmq::epoll_t::worker_routine(void*) (epoll.cpp:210)
==3958==    by 0x407BB24: zmq::thread_t::thread_routine(void*) (thread.cpp:99)
==3958==    by 0x4246F3A: start_thread (in /lib/libpthread-2.7.so)
==3958==    by 0x41CDBED: clone (in /lib/libc-2.7.so)
==3958==  Address 0x449d068 is 16 bytes inside a block of size 19 alloc'd
==3958==    at 0x4023D6E: malloc (vg_replace_malloc.c:207)
==3958==    by 0x407F54D: zmq_msg_init_size (zmq.cpp:122)
==3958==    by 0x40806C4: zmq::zmq_decoder_t::one_byte_size_ready() (zmq_decoder.cpp:62)
==3958==    by 0x40812EE: zmq::zmq_engine_t::in_event() (decoder.hpp:109)
==3958==    by 0x40699BA: zmq::epoll_t::loop() (epoll.cpp:197)
==3958==    by 0x4069A8C: zmq::epoll_t::worker_routine(void*) (epoll.cpp:210)
==3958==    by 0x407BB24: zmq::thread_t::thread_routine(void*) (thread.cpp:99)
==3958==    by 0x4246F3A: start_thread (in /lib/libpthread-2.7.so)
==3958==    by 0x41CDBED: clone (in /lib/libc-2.7.so)

This little patch is enough to at least mask the problem:

Index: zmq_decoder.cpp
===================================================================
--- zmq_decoder.cpp     (revision 6578)
+++ zmq_decoder.cpp     (revision 6579)
@@ -52,7 +52,9 @@
     //  message data into it.
     if (*tmpbuf == 0xff)
         next_step (tmpbuf, 8, &zmq_decoder_t::eight_byte_size_ready);
-    else {
+    else if (*tmpbuf == 0x00) {
+        return false;
+    } else {

         //  TODO:  Handle over-sized message decently.

I hope this is enough information for a real fix.

The existing IPv6 support was disabled because it cased problems on boxes with both IPv6 and IPv4 configured. Also, the IPv6 hostnames were resolved while IPv6 NIC names weren't.

We'll need to support IPv6 In a systematic manner.

When specifying an illegal number in zmq_init() will lead to a crash, this should be correctly handled as the reference specifies.

ie, using zmq_init(1, -1, 0) gives the following:

Too many open files
rc == 0 (signaler.cpp:236)
Aborted

Also, zmq_init(-1, 1, 0) just works, according to the reference this should also lead to the EINVAL, zmq_init(1,0,0) gives:

Assertion failed: io_threads.size () > 0 (ctx.cpp:228)
Aborted

The QNX C++ compiler, which is a bit more restrictive than GCC emits the following warning:
ctx.cpp:153: note: 'info.zmq::ctx_t::app_thread_info_t::tid' was declared here.

While this caused by the call to app_threads.push_back (info); (line 160) because uninitialized value is stored in the container. This value is assigned just few lines below (line169), app_threads [current].tid = thread_t::id ();, so probably it is not a problem.

This can be however be an issue if incidentally some exception would be thrown in the meantime - we would end up with uninitialized data in the container.

I think the solution is to:

1. provide a initializers to default value in constructor of app_thread_info_t struct
2. or to initialize tid to some value before storing the data in a container.
   The problem is that this warning (together with -Werror) does not allow for a clean build under QNX.

Ouput of configure script:
checking for a BSD-compatible install... config/install-sh -c
checking whether build environment is sane... yes
checking for a thread-safe mkdir -p... config/install-sh -c -d
checking for gawk... gawk
checking whether make sets $(MAKE)... yes
checking how to create a ustar tar archive... gnutar
checking for style of include used by make... GNU
checking for gcc... gcc
checking for C compiler default output file name... a.out
checking whether the C compiler works... yes
checking whether we are cross compiling... no
checking for suffix of executables...
checking for suffix of object files... o
checking whether we are using the GNU C compiler... yes
checking whether gcc accepts -g... yes
checking for gcc option to accept ISO C89... none needed
checking dependency style of gcc... gcc3
checking whether gcc and cc understand -c and -o together... yes
checking for g++... g++
checking whether we are using the GNU C++ compiler... yes
checking whether g++ accepts -g... yes
checking dependency style of g++... gcc3
checking build system type... sparc-sun-solaris2.10
checking host system type... sparc-sun-solaris2.10
checking for a sed that does not truncate output... /bin/sed
checking for grep that handles long lines and -e... /usr/xpg4/bin/grep
checking for egrep... /usr/xpg4/bin/grep -E
checking for fgrep... /usr/xpg4/bin/grep -F
checking for ld used by gcc... /usr/ccs/bin/ld
checking if the linker (/usr/ccs/bin/ld) is GNU ld... no
checking for BSD- or MS-compatible name lister (nm)... /usr/ccs/bin/nm -p
checking the name lister (/usr/ccs/bin/nm -p) interface... BSD nm
checking whether ln -s works... yes
checking the maximum length of command line arguments... 786240
checking whether the shell understands some XSI constructs... yes
checking whether the shell understands "+="... no
checking for /usr/ccs/bin/ld option to reload object files... -r
checking for objdump... objdump
checking how to recognize dependent libraries... pass_all
checking for ar... ar
checking for strip... strip
checking for ranlib... ranlib
checking command to parse /usr/ccs/bin/nm -p output from gcc object... ok
checking how to run the C preprocessor... gcc -E
checking for ANSI C header files... yes
checking for sys/types.h... yes
checking for sys/stat.h... yes
checking for stdlib.h... yes
checking for string.h... yes
checking for memory.h... yes
checking for strings.h... yes
checking for inttypes.h... yes
checking for stdint.h... yes
checking for unistd.h... yes
checking for dlfcn.h... yes
checking whether we are using the GNU C++ compiler... (cached) yes
checking whether g++ accepts -g... (cached) yes
checking dependency style of g++... (cached) gcc3
checking how to run the C++ preprocessor... g++ -E
checking for objdir... .libs
checking if gcc supports -fno-rtti -fno-exceptions... no
checking for gcc option to produce PIC... -fPIC -DPIC
checking if gcc PIC flag -fPIC -DPIC works... yes
checking if gcc static flag -static works... no
checking if gcc supports -c -o file.o... yes
checking if gcc supports -c -o file.o... (cached) yes
checking whether the gcc linker (/usr/ccs/bin/ld) supports shared libraries... yes
checking whether -lc should be explicitly linked in... yes
checking dynamic linker characteristics... solaris2.10 ld.so
checking how to hardcode library paths into programs... immediate
checking whether stripping libraries is possible... no
checking if libtool supports shared libraries... yes
checking whether to build shared libraries... yes
checking whether to build static libraries... yes
checking for ld used by g++... /usr/ccs/bin/ld
checking if the linker (/usr/ccs/bin/ld) is GNU ld... no
checking whether the g++ linker (/usr/ccs/bin/ld) supports shared libraries... yes
checking for g++ option to produce PIC... -fPIC -DPIC
checking if g++ PIC flag -fPIC -DPIC works... yes
checking if g++ static flag -static works... no
checking if g++ supports -c -o file.o... yes
checking if g++ supports -c -o file.o... (cached) yes
checking whether the g++ linker (/usr/ccs/bin/ld) supports shared libraries... yes
checking dynamic linker characteristics... solaris2.10 ld.so
checking how to hardcode library paths into programs... immediate
checking for a sed that does not truncate output... (cached) /bin/sed
checking for gawk... (cached) gawk
checking for pthread_create in -lpthread... yes
checking for main in -lm... yes
checking for main in -lstdc++... yes
checking whether to build documentation... no
checking whether to install manpages... yes
checking for main in -lsocket... yes
checking for main in -lnsl... yes
checking for main in -lrt... yes
checking for main in -luuid... yes
checking whether atomic operations can be used... no
checking for ANSI C header files... (cached) yes
checking errno.h usability... no
checking errno.h presence... yes
configure: WARNING: errno.h: present but cannot be compiled
configure: WARNING: errno.h: check for missing prerequisite headers?
configure: WARNING: errno.h: see the Autoconf documentation
configure: WARNING: errno.h: section "Present But Cannot Be Compiled"
configure: WARNING: errno.h: proceeding with the compiler's result
configure: WARNING: ## ------------------------------------------ ##
configure: WARNING: ## Report this to [email protected] ##
configure: WARNING: ## ------------------------------------------ ##
checking for errno.h... no
checking arpa/inet.h usability... no
checking arpa/inet.h presence... no
checking for arpa/inet.h... no
checking netinet/tcp.h usability... no
checking netinet/tcp.h presence... no
checking for netinet/tcp.h... no
checking netinet/in.h usability... no
checking netinet/in.h presence... no
checking for netinet/in.h... no
checking stddef.h usability... no
checking stddef.h presence... yes
configure: WARNING: stddef.h: present but cannot be compiled
configure: WARNING: stddef.h: check for missing prerequisite headers?
configure: WARNING: stddef.h: see the Autoconf documentation
configure: WARNING: stddef.h: section "Present But Cannot Be Compiled"
configure: WARNING: stddef.h: proceeding with the compiler's result
configure: WARNING: ## ------------------------------------------ ##
configure: WARNING: ## Report this to [email protected] ##
configure: WARNING: ## ------------------------------------------ ##
checking for stddef.h... no
checking for stdlib.h... (cached) yes
checking for string.h... (cached) yes
checking sys/socket.h usability... no
checking sys/socket.h presence... no
checking for sys/socket.h... no
checking sys/time.h usability... no
checking sys/time.h presence... no
checking for sys/time.h... no
checking for unistd.h... (cached) yes
checking limits.h usability... no
checking limits.h presence... no
checking for limits.h... no
checking sys/eventfd.h usability... no
checking sys/eventfd.h presence... no
checking for sys/eventfd.h... no
checking for sys/eventfd.h functionality... no
checking ifaddrs.h usability... no
checking ifaddrs.h presence... no
checking for ifaddrs.h... no
checking for pkg-config... yes
checking for stdbool.h that conforms to C99... yes
checking for _Bool... no
checking for an ANSI C-conforming const... yes
checking for inline... inline
checking for size_t... yes
checking for ssize_t... yes
checking whether time.h and sys/time.h may both be included... yes
checking for uint32_t... yes
checking for working volatile... yes
checking return type of signal handlers... void
checking for perror... yes
checking for gettimeofday... yes
checking for memset... yes
checking for socket... yes
checking for getifaddrs... no
checking for freeifaddrs... no
configure: creating ./config.status
config.status: creating Makefile
config.status: creating src/Makefile
config.status: creating doc/Makefile
config.status: creating perf/Makefile
config.status: creating src/libzmq.pc
config.status: creating devices/Makefile
config.status: creating devices/zmq_forwarder/Makefile
config.status: creating devices/zmq_streamer/Makefile
config.status: creating devices/zmq_queue/Makefile
config.status: creating builds/msvc/Makefile
config.status: creating src/platform.hpp
config.status: executing depfiles commands
config.status: executing libtool commands

GCC version:
Reading specs from /usr/local/lib/gcc/sparc-sun-solaris2.10/3.4.6/specs
Configured with: ../configure --with-as=/usr/ccs/bin/as --with-ld=/usr/ccs/bin/ld --enable-shared --enable-languages=c,c++,f77
Thread model: posix
gcc version 3.4.6

Makefile seems to be corrupted. Output of gmake:
Makefile:1: *** missing separator. Stop.

Hi,

I have compiled zeromq with the ./configure; make; make install, but the binaries are missing some libzmq.so.0:

zoobab@buzek /home/zoobab/soft/zeromq-2.0.6 [83]$ zmq_forwarder
zmq_forwarder: error while loading shared libraries: libzmq.so.0: cannot open shared object file: No such file or directory
zoobab@buzek /home/zoobab/soft/zeromq-2.0.6 [84]$ zmq_streamer
zmq_streamer: error while loading shared libraries: libzmq.so.0: cannot open shared object file: No such file or directory

Where do I find/compile such lib?

include <zmq.h>

include <stdio.h>

include <stdlib.h>

int main (int argc, char *argv [])
{
const char *bind_to;
int message_count;
int message_size;
void *ctx;
void *s;
int rc;
int i;
zmq_msg_t msg;
void *watch;
unsigned long elapsed;
unsigned long throughput;
double megabits;

if (argc != 4) {
    printf ("usage: local_thr <bind-to> <message-size> <message-count>\n");
    return 1;
}
bind_to = argv [1];
message_size = atoi (argv [2]);
message_count = atoi (argv [3]);

ctx = zmq_init (1, 1, 0);
if (!ctx) {
    printf ("error in zmq_send: %s\n", zmq_strerror (errno));
    return -1;
}

s = zmq_socket (ctx, ZMQ_P2P);
if (!s) {
    printf ("error in zmq_socket: %s\n", zmq_strerror (errno));
    return -1;
}

//rc = zmq_setsockopt (s, ZMQ_SUBSCRIBE , "", 0);
//if (rc != 0) {
//    printf ("error in zmq_setsockopt: %s\n", zmq_strerror (errno));
//    return -1;
//}

//  Add your socket options here.
//  For example ZMQ_RATE, ZMQ_RECOVERY_IVL and ZMQ_MCAST_LOOP for PGM.

rc = zmq_bind (s, bind_to);
if (rc != 0) {
    printf ("error in zmq_bind: %s\n", zmq_strerror (errno));
    return -1;
}

rc = zmq_msg_init (&msg);
if (rc != 0) {
    printf ("error in zmq_msg_init: %s\n", zmq_strerror (errno));
    return -1;
}

rc = zmq_recv (s, &msg, 0);
if (rc != 0) {
    printf ("error in zmq_recv: %s\n", zmq_strerror (errno));
    return -1;
}
if (zmq_msg_size (&msg) != message_size) {
    printf ("message of incorrect size received\n");
    return -1;
}

watch = zmq_stopwatch_start ();

for (i = 0; i != message_count - 1; i++) {
    rc = zmq_recv (s, &msg, 0);
    if (rc != 0) {
        printf ("error in zmq_recv: %s\n", zmq_strerror (errno));
        return -1;
    }
    if (zmq_msg_size (&msg) != message_size) {
        printf ("message of incorrect size received\n");
        return -1;
    }
}

elapsed = zmq_stopwatch_stop (watch);
if (elapsed == 0)
    elapsed = 1;

rc = zmq_msg_close (&msg);
if (rc != 0) {
    printf ("error in zmq_msg_close: %s\n", zmq_strerror (errno));
    return -1;
}
//rc = zmq_msg_init_size(&msg, 2);
//rc = zmq_send(s, &msg, 0);
//rc = zmq_msg_close(&msg);

throughput = (unsigned long)
    ((double) message_count / (double) elapsed * 1000000);
megabits = (double) (throughput * message_size * 8) / 1000000;

printf ("message size: %d [B]\n", (int) message_size);
printf ("message count: %d\n", (int) message_count);
printf ("mean throughput: %d [msg/s]\n", (int) throughput);
printf ("mean throughput: %.3f [Mb/s]\n", (double) megabits);

rc = zmq_close (s);
if (rc != 0) {
    printf ("error in zmq_close: %s\n", zmq_strerror (errno));
    return -1;
}

rc = zmq_term (ctx);
if (rc != 0) {
    printf ("error in zmq_term: %s\n", zmq_strerror (errno));
    return -1;
}

return 0;

}

output:
(gdb) run
Starting program: /home/tcip/src/zeromq/foo/local_thr tcp://192.168.2.2:20000 1000000 100
[Thread debugging using libthread_db enabled]
[New Thread 0x7fefeca12750 (LWP 9184)]
[New Thread 0x7fefeb68d950 (LWP 9187)]

Program received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0x7fefeca12750 (LWP 9184)]
zmq_msg_close (msg_=0x7fffb76ff780) at zmq.cpp:168
168 if (content->ffn)
Current language: auto; currently c++
(gdb) bt
#0 zmq_msg_close (msg_=0x7fffb76ff780) at zmq.cpp:168
#1 0x00007fefec5e9fec in zmq::p2p_t::xrecv (this=0x1782fe0,

msg_=0x7fffb76ff780, flags_=0) at p2p.cpp:115

#2 0x00007fefec5f208f in zmq::socket_base_t::recv (this=0x1782fe0,

msg_=0x7fffb76ff780, flags_=0) at socket_base.cpp:393

#3 0x0000000000400dc7 in main ()

(gdb)

I'm trying to compile i386 libraries on x86_64 platform (Centos5.4 64-bit) by executing:

./configure --build=i686-pc-linux-gnu

however build process still produces 64-bit code.

There's still one possibility for optimisation though: If, at any particular place of the tree there's a single-branch node (currently the memory footprint optimisation you've noticed), the associated structure could possibly hold up to say 8 subsequent characters. The idea is that single-branch nodes tend to appear in strings. Consider following example. Plus signs correspond to real branchings while minus signs denote strings of single-branch nodes:

animals.dogs
--------+++-

animals.donkeys
--------+++----

animals.ducks
--------++---

animals.cats
--------+---

The whole point is to significantly lower memory footprint which in turn should have severe impact on performance. Specifically with large search trees it may happen that some parts of the tree are eliminted from the CPU cache. In such case, access to physical memory can slow down the algorithm by an order of magnitude. Thus keeping the data structure as compact as possible will make it much faster, even for sparsely used topics.

Currently, when you try to get the value of ZMQ_IDENTITY using getsockopt you see:

• Invalid argument error upon getsockopt.
  *Segfault if ZMQ_IDENTITY has not been setsockopt'd

Application crashes periodically with

Connection timed out
nbytes != -1 (tcp_socket.cpp:216)

Reported by Michael Frumin [email protected], also seen by myself:

I am running zmq_forwarder for a pub/sub situation.  My config file is below:

[...]

when my machine is connected to the internet it runs no problem.  when I am
offline, it terminates immediately with this as output:

terminate called after throwing an instance of 'zmq::error_t'
  what():  Invalid argument
Aborted

I know about this and have reported the issue to the bug tracker.

There is a problem at least on Linux when the host only has a loopback
address configured, getaddrinfo() will not resolve anything. AFAICT this is
due to the AI_ADDRCONFIG flag being used when resolving names/addresses.

Any thoughts?

AI_ADDRCONFIG can probably go while we're ignoring IPv6, but this does look
like a getaddrinfo() bug to me, since the POSIX wording is:

   "If the AI_ADDRCONFIG flag is specified, IPv4 addresses shall be
   returned only if an  IPv4  address is configured on the local
   system,    and IPv6 addresses shall be returned only if an IPv6
   address is configured on the local
   system."

Currently, the bind side of P2P communication drops the messages until the other peer connects. Instead, it should queue them and send them once the peer is available.

Here is a test case in C:

http://gist.github.com/540065

It consists of three steps. Step 1 sends a message to step 2, which sends a message to step 3. I'm using PUSH/PULL over tcp between each step. Step 2 does not manage to connect to step 3, and hangs. Adding pauses does not make any difference.

The functionality is missing. It should be implemented.

While read auto-reconnect is meaningless for inproc transport (both peers fail simultaneously), allowing connect before bind as TCP transport does would be useful.

If peer A connects to peer B, then loses the connection and reconnects to peer C (which is now using peer B's address), then peer A will crash with an assertion failure due to the mismatch of peer identities.

I ran into this when I changed a daemon so that it was using an identity on one of its 0MQ sockets instead of being anonymous. When I restarted the daemon, all of its clients crashed.

In session_t::process_attach():

    if (peer_identity.empty () || peer_identity [0] != 0 ||
        peer_identity_.empty () || peer_identity_ [0] != 0) {

        //  If we already know the peer name do nothing, just check whether
        //  it haven't changed.
        zmq_assert (peer_identity == peer_identity_);
    }

I'm not sure what the right thing to do here is, but I think that crashing an innocent process is not the best option.

This is an issue to replace the existing stand alone devices with a single device daemon that allows various kinds of configuration. Will allow configuration of all socket options, and simple command-line execution.

zmq.h defines few POSIX errnos that Windows does not define. However, zmq_strerror doesn't translate those to human-readable strings.

No route to host
nbytes != -1 (tcp_socket.cpp:216)
Aborted

Similar to issue #40, except that EHOSTUNREACH needs to be added as an expected value of errno.

Test program:

int main ()
{
    zmq::context_t ctx (1,1,0);
    zmq::socket_t p (ctx, ZMQ_PUB);
    p.bind ("inproc://x");
    p.connect ("inproc://x");
    return 0;
}

Here's the test program:

include <assert.h>

include <stdio.h>

include <string.h>

include <stdlib.h>

include <zmq.h>

int main ()
{
zmq_pollitem_t items[1];

int rc;
void *ctx, *s;
zmq_msg_t query, resultset;
const char *query_string, *resultset_string = "OK";

/* Initialise 0MQ context, requesting a single application thread
   and a single I/O thread */
ctx = zmq_init (1, 1, 1);
assert (ctx);
/* Create a ZMQ_REP socket to receive requests and send replies */
s = zmq_socket (ctx, ZMQ_REP);
assert (s);
/* Bind to the TCP transport and port 5555 on the 'lo' interface */
rc = zmq_bind (s, "tcp://lo:5555");
assert (rc == 0);

items[0].events = ZMQ_POLLOUT;
items[0].socket = s;

zmq_close(s);
zmq_term(ctx);

}

Hello,

I noticed something which I think is a bug. The erorr seems to occur if a socket is created, connected but no send/recv happens. The following code seems to trigger the bug on github master version:

#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <zmq.h>

/*
    Adapted from ZMQ examples. Compile with 
    gcc test2.c -lzmq -o test2

    Resulting output:
    # ./test2
    Assertion failed: unnamed_sessions.empty () (socket_base.cpp:439)
    Aborted
*/

int main (int argc, char *argv[]) 
{
    int rc;
    void *ctx, *s;

    /* Initialise 0MQ context, requesting a single application thread
       and a single I/O thread */
    ctx = zmq_init (1, 1, 0);
    assert (ctx);
    /* Create a ZMQ_REP socket to receive requests and send replies */
    s = zmq_socket (ctx, ZMQ_REQ);
    assert (s);
    rc = zmq_connect (s, "tcp://127.0.0.1:5555");
    assert (rc == 0);

    zmq_close(s);
    zmq_term(ctx);
}

Had an application crash with this assertion today. Not entirely sure how this happened--perhaps the I/O thread closed a socket while the app thread was waiting in epoll? Looks like this condition should be silently ignored, instead of resulting in a crash. Assuming, that is, that the EBADF is a result of the attempt to delete a closed file descriptor, and not that the epoll fd itself has somehow become invalid!

When a PULL socket connects to a PULL socket, or a SUB socket connects to a SUB socket over inproc:, reading from either socket causes an assertion failure.

I haven't tried yet to reproduce this using c/c++ bindings, but using Lua bindings, the following code will drop every other message. I could not find any issue in the Lua bindings that would cause this.

-- file 1
require 'zmq';

local zmq_c = zmq.init(1, 1);
local zmq_dbs = zmq_c:socket(zmq.REQ);
zmq_dbs:connect("ipc:///tmp/zmq");

zmq_dbs:send("SELECT * FROM `MyTable`");
local d = {};
d[#d+1] = zmq_dbs:recv();
print(#d..') '..d[#d]);
while zmq_dbs:getopt(zmq.RCVMORE) == 1 do
  d[#d+1] = zmq_dbs:recv();
  print(#d..') '..d[#d]);
end

zmq_dbs:close();
zmq_c:term();


-- file 2
require 'zmq';
require 'json';
require 'luasql.mysql';

local zmq_c = zmq.init(1, 1);
local zmq_dbs = zmq_c:socket(zmq.REP);
zmq_dbs:bind("ipc:///tmp/zmq");

local mysql = luasql.mysql();
local con = mysql:connect('mydb', 'root', 'password', 'localhost');

local sql = zmq_dbs:recv();
if sql then
  local res = con:execute(sql)
  if res then
    local count = 0;
    local result = res:fetch({}, 'a');
    while result do
      count = count + 1;
      print(count..json.encode(result));
      zmq_dbs:send(count..json.encode(result), zmq.SNDMORE);
      result = res:fetch({}, 'a');
    end
    print(count+1);
    zmq_dbs:send(count+1);
    res:close();
  end
end

con:close();

zmq_dbs:close();
zmq_c:term();

Example Output:

REP:Send
1{"Color":"red"}
2{"Color":"blue"}
3{"Color":"green"}
4{"Color":"orange"}
5{"Color":"yellow"}
6{"Color":"purple"}
7

REQ:Recv
1) 1{"Color":"red"}
2) 3{"Color":"green"}
3) 5{"Color":"yellow"}
4) 7

When you poll on a PAIR socket that has been bound to an address, zmq_poll hangs. A PAIR socket that has connected works fine though. I realize that the PAIR sockets are not finished, but this used to work but now doesn't.

1. Download Source from github (.tar.gz format)
2. Unpack
3. ./autogen
4. ./configure --with-pgm

checking whether to install manpages... configure: error: configure thinks we want to install manpages but they're not present. Help!

When a client of an XREP socket disconnects, this assertion fails. Code to reproduce to follow. According to git-bisect, the bug was introduced by commit 10c28c1:

Author: Martin Hurton <[email protected]>
Date:   Sat Jun 19 20:46:16 2010 +0200

    Revive reader on pipe termination

The PAIR socket type is designed to be a one-to-one peer connection. Thus, when one PAIR socket binds and then another connects, further attemps by other PAIR sockets to connect to the bound PAIR socket should fail. Currently, the connection seems to happen, but then once everyone starts to send, you get seg faults and other failures, like:

Assertion failed: !inpipe && !outpipe (pair.cpp:47)

We need to think carefully about what error should be raised when two PAIR sockets try to connect to the same bound PAIR socket.

When trying the performance tools remote_thr and local_thr in combination with (e)PGM it is possible to raise an assertion error.

I did the following on a single machine:

./local_thr "epgm://127.0.0.1;224.0.0.1:5555" 1024 100000

This will subscribe to multicast messages at the group 244.0.0.1, then on the same machine (not sure if that matters) i send the messages as follows:

./remote_thr "epgm://127.0.0.1;224.0.0.1:5555" 1024 10000

This goes well, but local_thr is still waiting for messages as MULTICAST send rate is effectively limiting the send rate. Now when we try to continue sending messages by firing up remote_thr one after another, we should expect that local_thr eventually receives the requested number of messages.

However, after firing up remote_thr for the 5th (orso) time, local_thr dies with the following message:

(process:8563): Pgm-WARNING **: peer expired, tsi 142.143.27.153.192.188.31119
Assertion failed: it != peers.end () (pgm_receiver.cpp:161)
Aborted

There are bugs in the v2.0.7 code that result in problems with receiving multi-part PUB/SUB messages as described in the article. The simplest case is a two-part message with the first part containing a topic and the second part containing a payload.

1. First received message becomes lost since the topic part is skipped over in the code that accounts for a new peer (pgm_receiver.cpp),

// We have to move data to the beginning of the first message.
data += offset;
received -= offset;

Payload part is then treated as a topic part and gets discarded by the filter.

1. If the publisher terminates without waiting for zmq thread to drain its send queue and then restarts, this may result in concatenation of the topic part belonging to last message from the old instance and payload part of first message from the new instance. Topic part of the new message gets ignored and the payload part gets attached to the stale topic part, which, apparently, can linger in the decoder. Note that this may assign the payload to a wrong topic if the publisher restarts with a message to a topic different from the previous.

Workarounds have been suggested to have zmq create the files used for the UDS sockets in the file system with the appropriate permissions. My suggestion is to instead use anonymous sockets, created by ensuring the first character of the char* path passed in is zero. Example below, possibly could be added to zmq::resolve_local_path in ipc.cpp

struct sockaddr_un address;
bzero(&address, sizeof(address));
address.sun_family = AF_UNIX;
strncpy(address.sun_path, localSocketName.c_str(), sizeof(address.sun_path));

int length = sizeof(address.sun_family) + strlen(address.sun_path);

// make sure we use an abstract name for the socket
address.sun_path[0] = 0;

// call bind / connect as normal

If TCP engine happens to be closed because writing to the socket produced error (disconnection), there still may be inbound data in the socket. We should read these to improve reliability.

Following program shows the problem, first call to zmq_poll timeouts even though there's a message available. The behaviour is not completely deterministic:

#include "zmq.hpp"
#include "stdio.h"

int main(int argc, char* argv[])
{
    printf ("starting\n");

    int rc;
    zmq::context_t ctx(1, 1, ZMQ_POLL);
    zmq::socket_t s(ctx, ZMQ_REP);
    s.bind("tcp://lo:5555");
    zmq::socket_t r (ctx, ZMQ_REQ);
    r.connect("tcp://localhost:5555");

    zmq::message_t msg (1);
    r.send (msg);

    zmq::pollitem_t items[1];
    while(1) {
        items[0].socket = s;
        items[0].fd = -1;
        items[0].events = ZMQ_POLLIN;
        items[0].revents = 0;
        printf ("polling\n");
        rc = zmq_poll(items, 1, 5000000);
        printf ("done\n");
        printf ("%d, %d\n", rc, (int) items[0].revents);
        assert(rc != -1);

        zmq::message_t msgg;
        printf ("getting it\n");
        s.recv (&msgg);
        printf ("got it\n");
    }
    return 0;
}

Fedora13 provides an older version of libuuid (2.17.2) and uuid-devel (1.6.1), which zeromq has trouble compiling against.

1. Fedora13 installs libuuid at /lib/libuuid.so.1 and does not create a libuuid.so symlink. This causes zeromq's ./configure script to fail.
2. uuid-devel installs only one header file at /usr/include/uuid,h. ZeroMQ fails to compile since it requires uuid/uuid.h.

When running the following code:

#include 
#include 
#include 

int main(int argc, char* argv[])
{

    int rc;
    zmq::context_t ctx(1, 1, ZMQ_POLL);
    zmq::socket_t s(ctx, ZMQ_PUB);
    zmq::socket_t r(ctx, ZMQ_PUB);

    s.bind("tcp://lo:5555");
    sleep(1);
    //r.connect("tcp://localhost:5555");
    r.bind("tcp://lo:5556");

    zmq::pollitem_t items[2];
    items[0].socket = s;
    items[0].fd = -1;
    items[0].events = ZMQ_POLLOUT | ZMQ_POLLIN;
    items[0].revents = 0;

    items[1].socket = r;
    items[1].fd = -1;
    items[1].events = ZMQ_POLLOUT | ZMQ_POLLIN;
    items[1].revents = 0;

    rc = zmq_poll(items, 1, 5000000);
    printf ("rc: %d\n", rc);
    printf ("s:, %d\n", (int) items[0].revents);
    printf ("r:, %d\n", (int) items[1].revents);

    return 0;
}

I am getting the following results:

rc: 1
s:, 2
r:, 0

I would expect: 2,2,2. Occasionally i get 0,0,0 This issue could be related with issue 7.

The following issues have been discovered by cppcheck static checker:

[./src/yarray_item.hpp:36]: (error) Class yarray_item_t which is inherited by class reader_t does not have a virtual destructor
[./src/ypipe.hpp:37]: (error) Class ypipe_t which is inherited by class pipe_t does not have a virtual destructor
[./src/decoder.hpp:56]: (error) Class decoder_t which is inherited by class zmq_decoder_t does not have a virtual destructor
[./src/encoder.hpp:51]: (error) Class encoder_t which is inherited by class zmq_encoder_t does not have a virtual destructor

I am not sure, if this is a problem in case of ZeroMQ:
http://www.parashift.com/c++-faq-lite/virtual-functions.html#faq-20.7

After adding '#define ZMQ_FORCE_POLL 1' to src/platform.hpp under LINUX local_lat fails with

1. ./local_lat tcp://192.168.2.100:5555 10 10
   Assertion failed: !(it->revents & POLLNVAL) (poll.cpp:171)
   Aborted

or

LD_LIBRARY_PATH=../src/.libs gdb ./.libs/local_lat
GNU gdb (Gentoo 7.0 p1) 7.0
Copyright (C) 2009 Free Software Foundation, Inc.
License GPLv3+: GNU GPL version 3 or later http://gnu.org/licenses/gpl.html
This is free software: you are free to change and redistribute it.
There is NO WARRANTY, to the extent permitted by law. Type "show copying"
and "show warranty" for details.
This GDB was configured as "i686-pc-linux-gnu".
For bug reporting instructions, please see:
http://bugs.gentoo.org/...
Reading symbols from /home/ptroja/ports/zeromq-2.0.7/perf/.libs/local_lat...done.
(gdb) r tcp://192.168.2.100:5555 10 10
Starting program: /home/ptroja/ports/zeromq-2.0.7/perf/.libs/local_lat tcp://192.168.2.100:5555 10 10
[Thread debugging using libthread_db enabled]
[New Thread 0xb7c8db70 (LWP 8254)]

Program received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0xb7c8db70 (LWP 8254)]
0xb7fb192a in zmq::poll_t::loop (this=0x804c8f8) at poll.cpp:179
179 fd_table [it->fd].events->out_event ();
(gdb) bt
#0 0xb7fb192a in zmq::poll_t::loop (this=0x804c8f8) at poll.cpp:179
#1 0xb7fb1b03 in zmq::poll_t::worker_routine (arg_=0x804c8f8) at poll.cpp:204
#2 0xb7fc3250 in zmq::thread_t::thread_routine (arg_=0x804c924) at thread.cpp:99
#3 0xb7f1f8cf in start_thread () from /lib/libpthread.so.0

While this is not a big problem under Linux (since poll_t is not the default poller) it is a problem under QNX (and other platforms, i.e. AIX), where this is a default poller...
While it is possible to simply change the default poller for the QNX I think the real solution is to find a and fix this bug under Linux, which has better tools for that (i.e. strace).

When a ZMQ_REP socket is in non-blocking receive mode, the very first message it receives generates a POLLOUT revent instead of a POLLIN event. All subsequent received messages generate the proper POLLIN, so this only affects the first message detected via zmq_poll.

The REQ socket behaves correctly. It generates only POLLIN revents.

Currently, zmq_term only decrements the refcount. 0MQ infrastructure is physically deallocated only when zmq_term was called and all the sockets were closed.

Instead, zmq_term should cause all the currently executed blocking functions to return with EZOMBIE. Also, any further attempt to call anything else except zmq_close should result in the same error.

In rare cases, the ZMQ_RCVMORE socket option is not set when an incoming message has the ZMQ_MSG_MORE flag. This is difficult to reproduce, but I believe I have found the problem. Patch follows.

diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index eddb297..6a309a0 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -424,7 +424,14 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_)
             return -1;
         }
         ticks = 0;
-        return xrecv (msg_, flags_);
+        rc = xrecv (msg_, flags_);
+
+        if (rc == 0) {
+            rcvmore = msg_->flags & ZMQ_MSG_MORE;
+            if (rcvmore)
+                msg_->flags &= ~ZMQ_MSG_MORE;
+        }
+        return rc;
     }

     //  In blocking scenario, commands are processed over and over again until

When set_address in socket_base_t::connect fails, the unnamed session created beforehand is zombified.

Test program:

int main ()
{
    zmq::context_t ctx (1,1,0);
    zmq::socket_t p (ctx, ZMQ_PUB);
    p.bind ("inproc://x");
    zmq::socket_t s (ctx, ZMQ_SUB);
    s.connect ("inproc://x");

    zmq::message_t msg (10);
    p.send (msg);
    s.recv (&msg);

    return 0;
}

Seeing this in 2.0.7 when the peer being connected to crashes. Perhaps err == ECONNRESET in this case? According to the connect(2) manpage, there are a number of other errors that could plausibly result from a failed connection that isn't due to a ZMQ library bug, e.g. ENETDOWN, ENETUNREACH, EHOSTUNREACH

ZMQ devices with XREP as one socket are vulnerable to incoming messages on the other socket, if they are only one segment without ZMQ_SNDMORE ( because it is interpreted by the XREP as an IDENTITY prefix without content ).

Sample code here:
http://gist.github.com/517566

An example of this is that currently a PAIR socket can connect to other sockets types (other than the correct PAIR type). But, once send/recv starts to happen, things go bad. Right now, I see seg faults and assertion errors like:

Assertion failed: fetched (rep.cpp:258)

This brings up the bigger issue of what should happen when sockets try to connect to sockets of incompatible types.