Merge pull request #478 from methodmissing/event-messages-3.2

Event message memory corruption fixes

src/ipc_connecter.cpp

@@ -122,7 +122,7 @@ void zmq::ipc_connecter_t::out_event ()
     //  Shut the connecter down.
     terminate ();
 
-    socket->event_connected (endpoint.c_str(), fd);
+    socket->event_connected (endpoint, fd);
 }
 
 void zmq::ipc_connecter_t::timer_event (int id_)
@@ -150,7 +150,7 @@ void zmq::ipc_connecter_t::start_connecting ()
         handle = add_fd (s);
         handle_valid = true;
         set_pollout (handle);
-        socket->event_connect_delayed (endpoint.c_str(), zmq_errno());
+        socket->event_connect_delayed (endpoint, zmq_errno());
     }
 
     //  Handle any other error condition by eventual reconnect.
@@ -165,7 +165,7 @@ void zmq::ipc_connecter_t::add_reconnect_timer()
 {
     int rc_ivl = get_new_reconnect_ivl();
     add_timer (rc_ivl, reconnect_timer_id);
-    socket->event_connect_retried (endpoint.c_str(), rc_ivl);
+    socket->event_connect_retried (endpoint, rc_ivl);
     timer_started = true;
 }
 
@@ -226,7 +226,7 @@ int zmq::ipc_connecter_t::close ()
     zmq_assert (s != retired_fd);
     int rc = ::close (s);
     errno_assert (rc == 0);
-    socket->event_closed (endpoint.c_str(), s);
+    socket->event_closed (endpoint, s);
     s = retired_fd;
     return 0;
 }

src/ipc_listener.cpp

@@ -76,7 +76,7 @@ void zmq::ipc_listener_t::in_event ()
     //  If connection was reset by the peer in the meantime, just ignore it.
     //  TODO: Handle specific errors like ENFILE/EMFILE etc.
     if (fd == retired_fd) {
-        socket->event_accept_failed (endpoint.c_str(), zmq_errno());
+        socket->event_accept_failed (endpoint, zmq_errno());
         return;
     }
 
@@ -96,7 +96,7 @@ void zmq::ipc_listener_t::in_event ()
     session->inc_seqnum ();
     launch_child (session);
     send_attach (session, engine, false);
-    socket->event_accepted (endpoint.c_str(), fd);
+    socket->event_accepted (endpoint, fd);
 }
 
 int zmq::ipc_listener_t::get_address (std::string &addr_)
@@ -155,7 +155,7 @@ int zmq::ipc_listener_t::set_address (const char *addr_)
     if (rc != 0)
         goto error;
 
-    socket->event_listening (endpoint.c_str(), s);
+    socket->event_listening (endpoint, s);
     return 0;
 
 error:
@@ -178,12 +178,12 @@ int zmq::ipc_listener_t::close ()
     if (has_file && !filename.empty ()) {
         rc = ::unlink(filename.c_str ());
         if (rc != 0) {
-            socket->event_close_failed (endpoint.c_str(), zmq_errno());
+            socket->event_close_failed (endpoint, zmq_errno());
             return -1;
         }
     }
 
-    socket->event_closed (endpoint.c_str(), s);
+    socket->event_closed (endpoint, s);
     return 0;
 }
 

src/socket_base.cpp

@@ -357,7 +357,7 @@ int zmq::socket_base_t::bind (const char *addr_)
         int rc = listener->set_address (address.c_str ());
         if (rc != 0) {
             delete listener;
-            event_bind_failed (addr_, zmq_errno());
+            event_bind_failed (address, zmq_errno());
             return -1;
         }
 
@@ -376,7 +376,7 @@ int zmq::socket_base_t::bind (const char *addr_)
         int rc = listener->set_address (address.c_str ());
         if (rc != 0) {
             delete listener;
-            event_bind_failed (addr_, zmq_errno());
+            event_bind_failed (address, zmq_errno());
             return -1;
         }
 
@@ -1047,122 +1047,141 @@ int zmq::socket_base_t::monitor (const char *addr_, int events_)
     return rc;
 }
 
-void zmq::socket_base_t::event_connected (const char *addr_, int fd_)
+void zmq::socket_base_t::event_connected (std::string &addr_, int fd_)
 {
     if (monitor_events & ZMQ_EVENT_CONNECTED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_CONNECTED;
-        event.data.connected.addr = const_cast <char *> (addr_);
+        event.data.connected.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.connected.addr, addr_);
         event.data.connected.fd = fd_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_connect_delayed (const char *addr_, int err_)
+void zmq::socket_base_t::event_connect_delayed (std::string &addr_, int err_)
 {
     if (monitor_events & ZMQ_EVENT_CONNECT_DELAYED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_CONNECT_DELAYED;
-        event.data.connect_delayed.addr = const_cast <char *> (addr_);
+        event.data.connect_delayed.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.connect_delayed.addr, addr_);
         event.data.connect_delayed.err = err_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_connect_retried (const char *addr_, int interval_)
+void zmq::socket_base_t::event_connect_retried (std::string &addr_, int interval_)
 {
     if (monitor_events & ZMQ_EVENT_CONNECT_RETRIED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_CONNECT_RETRIED;
-        event.data.connect_retried.addr = const_cast <char *> (addr_);
+        event.data.connect_retried.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.connect_retried.addr, addr_);
         event.data.connect_retried.interval = interval_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_listening (const char *addr_, int fd_)
+void zmq::socket_base_t::event_listening (std::string &addr_, int fd_)
 {
     if (monitor_events & ZMQ_EVENT_LISTENING) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_LISTENING;
-        event.data.listening.addr = const_cast <char *> (addr_);
+        event.data.listening.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.listening.addr, addr_);
         event.data.listening.fd = fd_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_bind_failed (const char *addr_, int err_)
+void zmq::socket_base_t::event_bind_failed (std::string &addr_, int err_)
 {
     if (monitor_events & ZMQ_EVENT_BIND_FAILED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_BIND_FAILED;
-        event.data.bind_failed.addr = const_cast <char *> (addr_);
+        event.data.bind_failed.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.bind_failed.addr, addr_);
         event.data.bind_failed.err = err_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_accepted (const char *addr_, int fd_)
+void zmq::socket_base_t::event_accepted (std::string &addr_, int fd_)
 {
     if (monitor_events & ZMQ_EVENT_ACCEPTED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_ACCEPTED;
-        event.data.accepted.addr = const_cast <char *> (addr_);
+        event.data.accepted.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.accepted.addr, addr_);
         event.data.accepted.fd = fd_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_accept_failed (const char *addr_, int err_)
+void zmq::socket_base_t::event_accept_failed (std::string &addr_, int err_)
 {
     if (monitor_events & ZMQ_EVENT_ACCEPT_FAILED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_ACCEPT_FAILED;
-        event.data.accept_failed.addr = const_cast <char *> (addr_);
+        event.data.accept_failed.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.accept_failed.addr, addr_);
         event.data.accept_failed.err= err_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_closed (const char *addr_, int fd_)
+void zmq::socket_base_t::event_closed (std::string &addr_, int fd_)
 {
     if (monitor_events & ZMQ_EVENT_CLOSED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_CLOSED;
-        event.data.closed.addr = const_cast <char *> (addr_);
+        event.data.closed.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.closed.addr, addr_);
         event.data.closed.fd = fd_;
         monitor_event (event);
     }
 }
         
-void zmq::socket_base_t::event_close_failed (const char *addr_, int err_)
+void zmq::socket_base_t::event_close_failed (std::string &addr_, int err_)
 {
     if (monitor_events & ZMQ_EVENT_CLOSE_FAILED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_CLOSE_FAILED;
-        event.data.close_failed.addr = const_cast <char *> (addr_);
+        event.data.close_failed.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.close_failed.addr, addr_);
         event.data.close_failed.err = err_;
         monitor_event (event);
     }
 }
 
-void zmq::socket_base_t::event_disconnected (const char *addr_, int fd_)
+void zmq::socket_base_t::event_disconnected (std::string &addr_, int fd_)
 {
     if (monitor_events & ZMQ_EVENT_DISCONNECTED) {
         zmq_event_t event;
         event.event = ZMQ_EVENT_DISCONNECTED;
-        event.data.disconnected.addr = const_cast <char *> (addr_);
+        event.data.disconnected.addr = (char *) malloc (addr_.size () + 1);
+        copy_monitor_address (event.data.disconnected.addr, addr_);
         event.data.disconnected.fd = fd_;
         monitor_event (event);
     }
 }
 
+void zmq::socket_base_t::copy_monitor_address (char *dest_, std::string &src_)
+{
+    alloc_assert (dest_);
+    dest_[src_.size ()] = 0;
+    memcpy (dest_, src_.c_str (), src_.size ());
+}
+
 void zmq::socket_base_t::monitor_event (zmq_event_t event_)
 {
     if (monitor_socket) {
         zmq_msg_t msg;
-        zmq_msg_init_size (&msg, sizeof (event_));
-        memcpy (zmq_msg_data (&msg), &event_, sizeof (event_));
+        void *event_data = malloc (sizeof (event_));
+        alloc_assert (event_data);
+        memcpy (event_data, &event_, sizeof (event_));
+        zmq_msg_init_data (&msg, event_data, sizeof (event_), zmq_free_event, NULL);
         zmq_sendmsg (monitor_socket, &msg, 0);
         zmq_msg_close (&msg);
     }

src/socket_base.hpp

@@ -38,6 +38,11 @@
 #include "clock.hpp"
 #include "pipe.hpp"
 
+extern "C"
+{
+    void zmq_free_event (void *data, void *hint);
+}
+
 namespace zmq
 {
 
@@ -102,18 +107,18 @@ namespace zmq
         void lock();
         void unlock();
 
-        int monitor(const char *endpoint_, int events_);
+        int monitor (const char *endpoint_, int events_);
 
-        void event_connected(const char *addr_, int fd_);
-        void event_connect_delayed(const char *addr_, int err_);
-        void event_connect_retried(const char *addr_, int interval_);
-        void event_listening(const char *addr_, int fd_);
-        void event_bind_failed(const char *addr_, int err_);
-        void event_accepted(const char *addr_, int fd_);
-        void event_accept_failed(const char *addr_, int err_);
-        void event_closed(const char *addr_, int fd_);        
-        void event_close_failed(const char *addr_, int fd_);  
-        void event_disconnected(const char *addr_, int fd_); 
+        void event_connected (std::string &addr_, int fd_);
+        void event_connect_delayed (std::string &addr_, int err_);
+        void event_connect_retried (std::string &addr_, int interval_);
+        void event_listening (std::string &addr_, int fd_);
+        void event_bind_failed (std::string &addr_, int err_);
+        void event_accepted (std::string &addr_, int fd_);
+        void event_accept_failed (std::string &addr_, int err_);
+        void event_closed (std::string &addr_, int fd_);        
+        void event_close_failed (std::string &addr_, int fd_);  
+        void event_disconnected (std::string &addr_, int fd_); 
 
     protected:
 
@@ -151,6 +156,9 @@ namespace zmq
         // Socket event data dispath
         void monitor_event (zmq_event_t data_);
 
+        // Copy monitor specific event endpoints to event messages
+        void copy_monitor_address (char *dest_, std::string &src_);
+
         // Monitor socket cleanup
         void stop_monitor ();
 

src/stream_engine.cpp

@@ -63,6 +63,7 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_, cons
     greeting_bytes_read (0),
     session (NULL),
     options (options_),
+    endpoint (endpoint_),
     plugged (false),
     socket (NULL)
 {
@@ -95,8 +96,6 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_, cons
     int rc = setsockopt (s, SOL_SOCKET, SO_NOSIGPIPE, &set, sizeof (int));
     errno_assert (rc == 0);
 #endif
-    endpoint = new char[endpoint_.length() + 1];
-    strcpy (endpoint, endpoint_.c_str());
 }
 
 zmq::stream_engine_t::~stream_engine_t ()
@@ -118,7 +117,6 @@ zmq::stream_engine_t::~stream_engine_t ()
         delete encoder;
     if (decoder != NULL)
         delete decoder;
-    delete [] endpoint;
 }
 
 void zmq::stream_engine_t::plug (io_thread_t *io_thread_,

src/stream_engine.hpp

@@ -132,7 +132,7 @@ namespace zmq
         options_t options;
 
         // String representation of endpoint
-        char *endpoint;
+        std::string endpoint;
 
         bool plugged;
 

src/tcp_connecter.cpp

@@ -136,7 +136,7 @@ void zmq::tcp_connecter_t::out_event ()
     //  Shut the connecter down.
     terminate ();
 
-    socket->event_connected (endpoint.c_str(), fd);
+    socket->event_connected (endpoint, fd);
 }
 
 void zmq::tcp_connecter_t::timer_event (int id_)
@@ -164,7 +164,7 @@ void zmq::tcp_connecter_t::start_connecting ()
         handle = add_fd (s);
         handle_valid = true;
         set_pollout (handle);
-        socket->event_connect_delayed (endpoint.c_str(), zmq_errno());
+        socket->event_connect_delayed (endpoint, zmq_errno());
     }
 
     //  Handle any other error condition by eventual reconnect.
@@ -179,7 +179,7 @@ void zmq::tcp_connecter_t::add_reconnect_timer()
 {
     int rc_ivl = get_new_reconnect_ivl();
     add_timer (rc_ivl, reconnect_timer_id);
-    socket->event_connect_retried (endpoint.c_str(), rc_ivl);
+    socket->event_connect_retried (endpoint, rc_ivl);
     timer_started = true;
 }
 
@@ -305,6 +305,6 @@ void zmq::tcp_connecter_t::close ()
     int rc = ::close (s);
     errno_assert (rc == 0);
 #endif
-    socket->event_closed (endpoint.c_str(), s);
+    socket->event_closed (endpoint, s);
     s = retired_fd;
 }

src/tcp_listener.cpp

@@ -85,7 +85,7 @@ void zmq::tcp_listener_t::in_event ()
     //  If connection was reset by the peer in the meantime, just ignore it.
     //  TODO: Handle specific errors like ENFILE/EMFILE etc.
     if (fd == retired_fd) {
-        socket->event_accept_failed (endpoint.c_str(), zmq_errno());
+        socket->event_accept_failed (endpoint, zmq_errno());
         return;
     }
 
@@ -108,7 +108,7 @@ void zmq::tcp_listener_t::in_event ()
     session->inc_seqnum ();
     launch_child (session);
     send_attach (session, engine, false);
-    socket->event_accepted (endpoint.c_str(), fd);
+    socket->event_accepted (endpoint, fd);
 }
 
 void zmq::tcp_listener_t::close ()
@@ -121,7 +121,7 @@ void zmq::tcp_listener_t::close ()
     int rc = ::close (s);
     errno_assert (rc == 0);
 #endif
-    socket->event_closed (endpoint.c_str(), s);
+    socket->event_closed (endpoint, s);
     s = retired_fd;
 }
 
@@ -223,7 +223,7 @@ int zmq::tcp_listener_t::set_address (const char *addr_)
         goto error;
 #endif
 
-    socket->event_listening (endpoint.c_str(), s);
+    socket->event_listening (endpoint, s);
     return 0;
 
 error:

src/zmq.cpp

@@ -987,6 +987,47 @@ int zmq_device (int type, void *frontend_, void *backend_)
         (zmq::socket_base_t*) backend_, NULL);
 }
 
+//  Callback to free socket event data
+
+void zmq_free_event (void *event_data, void *hint)
+{
+    zmq_event_t *event = (zmq_event_t *) event_data;
+
+    switch (event->event) {
+    case ZMQ_EVENT_CONNECTED:
+        free (event->data.connected.addr);
+        break;
+    case ZMQ_EVENT_CONNECT_DELAYED:
+        free (event->data.connect_delayed.addr);
+        break;
+    case ZMQ_EVENT_CONNECT_RETRIED:
+        free (event->data.connect_retried.addr);
+        break;
+    case ZMQ_EVENT_LISTENING:
+        free (event->data.listening.addr);
+        break;
+    case ZMQ_EVENT_BIND_FAILED:
+        free (event->data.bind_failed.addr);
+        break;
+    case ZMQ_EVENT_ACCEPTED:
+        free (event->data.accepted.addr);
+        break;
+    case ZMQ_EVENT_ACCEPT_FAILED:
+        free (event->data.accept_failed.addr);
+        break;
+    case ZMQ_EVENT_CLOSED:
+        free (event->data.closed.addr);
+        break;
+    case ZMQ_EVENT_CLOSE_FAILED:
+        free (event->data.close_failed.addr);
+        break;
+    case ZMQ_EVENT_DISCONNECTED:
+        free (event->data.disconnected.addr);
+        break;
+    }
+    free (event_data);
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 //  0MQ utils - to be used by perf tests
 ////////////////////////////////////////////////////////////////////////////////