issue86.c

/*
    Reproduces segmentation fault in 0MQ
    1. Start in two windows
        myname d1 d2
        myname d2 d1
    2. Kill either instance
    3. Other instance crashes

#0  0x04f6e824 in ?? ()
#1  0x00152ac9 in zmq::session_t::activated (this=0x8054088, pipe_=0x8056598) at session.cpp:152
#2  0x0014c98c in zmq::reader_t::process_activate_reader (this=0x8056598) at pipe.cpp:140
#3  0x0014ac70 in zmq::object_t::process_command (this=0x8056598, cmd_=...) at object.cpp:61
#4  0x00147c2f in zmq::io_thread_t::in_event (this=0x804c8d0) at io_thread.cpp:82
#5  0x00146479 in zmq::epoll_t::loop (this=0x804c8f8) at epoll.cpp:197
#6  0x0014657d in zmq::epoll_t::worker_routine (arg_=0x804c8f8) at epoll.cpp:210
#7  0x0015a557 in zmq::thread_t::thread_routine (arg_=0x804c918) at thread.cpp:79
#8  0x001ad96e in start_thread () from /lib/tls/i686/cmov/libpthread.so.0
#9  0x002b4a4e in clone () from /lib/tls/i686/cmov/libc.so.6

    Trigger of crash is ZMQ_IDENTITY on sub socket; without this the code
    runs without crashing.

*/
#include "zhelpers.h"

int main (int argc, char *argv[])
{
    //  Router master endpoint is first argument, remaining arguments
    //  are other master routers to slave to.
    //
    if (argc < 3) {
        printf ("syntax: dacenter this other ...\n");
        printf ("e.g. dacenter name1 name2...\n");
        exit (EXIT_FAILURE);
    }
    char *self = argv [1];
    printf ("I: preparing broker at %s...\n", self);

    //  Prepare our context and sockets
    void *context = zmq_init (1);
    char endpoint [256];

    //  Bind tweeter socket which we use to broadcast our
    //  The meta endpoint is for worker availability
    void *tweeter = zmq_socket (context, ZMQ_PUB);
    uint64_t hwm = 1;
    zmq_setsockopt (tweeter, ZMQ_HWM, &hwm, sizeof (hwm));
    snprintf (endpoint, 255, "ipc://%s-meta.ipc", self);
    assert (zmq_bind (tweeter, endpoint) == 0);

    //  Peer to each other brokers, by connecting slave and sensor
    int argn;
    void *sensor = zmq_socket (context, ZMQ_SUB);
    zmq_setsockopt (sensor, ZMQ_SUBSCRIBE, "STATUS", 6);
    //
    //  ***********************************************************
    //                     THIS CAUSES THE CRASH
    zmq_setsockopt (sensor, ZMQ_IDENTITY, self, strlen (self));
    //  ***********************************************************
    for (argn = 2; argn < argc; argn++) {
        char *peer = argv [argn];
        printf ("I: peering with broker at '%s'\n", peer);
        snprintf (endpoint, 255, "ipc://%s-meta.ipc", peer);
        assert (zmq_connect (sensor, endpoint) == 0);
    }

    //  Send out status messages to peers, and collect from peers
    //  The zmq_poll timeout defines our own heartbeating
    while (1) {
        //  Initialize poll set
        zmq_pollitem_t items [1] = {
            { sensor, 0, ZMQ_POLLIN, 0 }
        };
        assert (zmq_poll (items, 1, 1000000) >= 0);

        //  Handle incoming status message
        if (items [0].revents & ZMQ_POLLIN) {
            //  Get message key and discard it
            char *message_key = s_recv (sensor);
            //  Get address of peer broker
            char *broker_addr = s_recv (sensor);
            //  Get number of available workers
            char *broker_status = s_recv (sensor);

            printf ("%s - %s workers free\n", broker_addr, broker_status);
            free (message_key);
            free (broker_addr);
            free (broker_status);
        }
        else {
            //  Send message envelope and body
            //  Here, just some random value for worker availability
            char status [3];
            snprintf (status, 2, "%d", within (10));
            s_sendmore (tweeter, "STATUS");
            s_sendmore (tweeter, self);
            s_send     (tweeter, status);
        }
    }
    zmq_term (context);
    return 0;
}

zhelpers.h

/*  =========================================================================
    zhelpers.h - ZeroMQ helpers for example applications

    Copyright (c) 1991-2010 iMatix Corporation and contributors

    This is free software; you can redistribute it and/or modify it under
    the terms of the Lesser GNU General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.

    This software 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
    Lesser GNU General Public License for more details.

    You should have received a copy of the Lesser GNU General Public License
    along with this program. If not, see <http://www.gnu.org/licenses/>.
    =========================================================================
*/

#ifndef __ZHELPERS_H_INCLUDED__
#define __ZHELPERS_H_INCLUDED__

//  Include a bunch of headers that we will need in the examples

#include <zmq.h>

#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <assert.h>

#define within(num) (int) ((float) num * random () / (RAND_MAX + 1.0))

//  Receive 0MQ string from socket and convert into C string
static char *
s_recv (void *socket) {
    zmq_msg_t message;
    zmq_msg_init (&message);
    if (zmq_recv (socket, &message, 0)) {
        printf ("E: %s\n", zmq_strerror (errno));
        exit (1);
    }
    int size = zmq_msg_size (&message);
    char *string = malloc (size + 1);
    memcpy (string, zmq_msg_data (&message), size);
    zmq_msg_close (&message);
    string [size] = 0;
    return (string);
}

//  Convert C string to 0MQ string and send to socket
static int
s_send (void *socket, char *string) {
    int rc;
    zmq_msg_t message;
    zmq_msg_init_size (&message, strlen (string));
    memcpy (zmq_msg_data (&message), string, strlen (string));
    rc = zmq_send (socket, &message, 0);
    assert (!rc);
    zmq_msg_close (&message);
    return (rc);
}

//  Sends string as 0MQ string, as multipart non-terminal
static int
s_sendmore (void *socket, char *string) {
    int rc;
    zmq_msg_t message;
    zmq_msg_init_size (&message, strlen (string));
    memcpy (zmq_msg_data (&message), string, strlen (string));
    rc = zmq_send (socket, &message, ZMQ_SNDMORE);
    zmq_msg_close (&message);
    assert (!rc);
    return (rc);
}

//  Receives all message parts from socket, prints neatly
//
static void
s_dump (void *socket)
{
    puts ("----------------------------------------");
    while (1) {
        //  Process all parts of the message
        zmq_msg_t message;
        zmq_msg_init (&message);
        zmq_recv (socket, &message, 0);

        //  Dump the message as text or binary
        char *data = zmq_msg_data (&message);
        int size = zmq_msg_size (&message);
        int is_text = 1;
        int char_nbr;
        for (char_nbr = 0; char_nbr < size; char_nbr++)
            if (data [char_nbr] < 32 || data [char_nbr] > 127)
                is_text = 0;

        printf ("[%03d] ", size);
        for (char_nbr = 0; char_nbr < size; char_nbr++) {
            if (is_text)
                printf ("%c", data [char_nbr]);
            else
                printf ("%02X", (unsigned char) data [char_nbr]);
        }
        printf ("\n");

        int64_t more;           //  Multipart detection
        size_t more_size = sizeof (more);
        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);
        zmq_msg_close (&message);
        if (!more)
            break;      //  Last message part
    }
}

//  Set simple random printable identity on socket
//
static void
s_set_id (void *socket)
{
    char identity [10];
    sprintf (identity, "%04X-%04X", within (0x10000), within (0x10000));
    zmq_setsockopt (socket, ZMQ_IDENTITY, identity, strlen (identity));
}

#endif