From ce4d32168440d5767805b7ee7cbfaac2f4c3f272 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lourens=20Naud=C3=A9?= Date: Sat, 17 Nov 2012 10:06:09 +0000 Subject: [PATCH 1/4] Revert "Merge pull request #473 from methodmissing/fix-engine-endpoint" This reverts commit 1a18c7b0a816e10f8a8a770b2141d47f661f0302, reversing changes made to bef9a41bdb991b867981317e0e20ee174ecd6fea. --- src/stream_engine.cpp | 6 ++---- src/stream_engine.hpp | 2 +- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/src/stream_engine.cpp b/src/stream_engine.cpp index 0c7f38fc..57381d5e 100644 --- a/src/stream_engine.cpp +++ b/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_, @@ -486,7 +484,7 @@ int zmq::stream_engine_t::push_msg (msg_t *msg_) void zmq::stream_engine_t::error () { zmq_assert (session); - socket->event_disconnected (endpoint, s); + socket->event_disconnected (endpoint.c_str(), s); session->detach (); unplug (); delete this; diff --git a/src/stream_engine.hpp b/src/stream_engine.hpp index 4a846c71..2d83995c 100644 --- a/src/stream_engine.hpp +++ b/src/stream_engine.hpp @@ -132,7 +132,7 @@ namespace zmq options_t options; // String representation of endpoint - char *endpoint; + std::string endpoint; bool plugged; From 6d245eb6bc614c050d7650c564fd1a986475cc67 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lourens=20Naud=C3=A9?= Date: Sat, 17 Nov 2012 11:29:47 +0000 Subject: [PATCH 2/4] Copy monitor specific event endpoints to event messages as the engine etc. can be released at anytime --- src/ipc_connecter.cpp | 8 +++--- src/ipc_listener.cpp | 10 +++---- src/socket_base.cpp | 61 +++++++++++++++++++++++++++---------------- src/socket_base.hpp | 25 ++++++++++-------- src/stream_engine.cpp | 2 +- src/tcp_connecter.cpp | 8 +++--- src/tcp_listener.cpp | 8 +++--- 7 files changed, 71 insertions(+), 51 deletions(-) diff --git a/src/ipc_connecter.cpp b/src/ipc_connecter.cpp index 37802cdb..0fd2e404 100644 --- a/src/ipc_connecter.cpp +++ b/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; } diff --git a/src/ipc_listener.cpp b/src/ipc_listener.cpp index 49f2eca1..28b01a6e 100644 --- a/src/ipc_listener.cpp +++ b/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; } diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 4715a6ba..d0ddae8f 100644 --- a/src/socket_base.cpp +++ b/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,116 +1047,133 @@ 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 (addr_); + event.data.connected.addr = new char[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 (addr_); + event.data.connect_delayed.addr = new char[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 (addr_); + event.data.connect_retried.addr = new char[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 (addr_); + event.data.listening.addr = new char[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 (addr_); + event.data.bind_failed.addr = new char[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 (addr_); + event.data.accepted.addr = new char[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 (addr_); + event.data.accept_failed.addr = new char[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 (addr_); + event.data.closed.addr = new char[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 (addr_); + event.data.close_failed.addr = new char[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 (addr_); + event.data.disconnected.addr = new char[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) { diff --git a/src/socket_base.hpp b/src/socket_base.hpp index 5237437d..4e70a018 100644 --- a/src/socket_base.hpp +++ b/src/socket_base.hpp @@ -102,18 +102,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 +151,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 (); diff --git a/src/stream_engine.cpp b/src/stream_engine.cpp index 57381d5e..883b0e8d 100644 --- a/src/stream_engine.cpp +++ b/src/stream_engine.cpp @@ -484,7 +484,7 @@ int zmq::stream_engine_t::push_msg (msg_t *msg_) void zmq::stream_engine_t::error () { zmq_assert (session); - socket->event_disconnected (endpoint.c_str(), s); + socket->event_disconnected (endpoint, s); session->detach (); unplug (); delete this; diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp index 90de92b3..6f62878e 100644 --- a/src/tcp_connecter.cpp +++ b/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; } diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp index e939c88a..4ee88d3d 100644 --- a/src/tcp_listener.cpp +++ b/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: From b1f4850153dbddb66c8e32fe3ce70361e954a38a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lourens=20Naud=C3=A9?= Date: Wed, 21 Nov 2012 18:27:53 +0000 Subject: [PATCH 3/4] Let socket event messages initialize with zmq_msg_init_data and let the framework handle cleanup --- src/socket_base.cpp | 6 ++++-- src/socket_base.hpp | 5 +++++ src/zmq.cpp | 41 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 2 deletions(-) diff --git a/src/socket_base.cpp b/src/socket_base.cpp index d0ddae8f..ace73f81 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -1178,8 +1178,10 @@ 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_)); + 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); } diff --git a/src/socket_base.hpp b/src/socket_base.hpp index 4e70a018..2d5f259d 100644 --- a/src/socket_base.hpp +++ b/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 { diff --git a/src/zmq.cpp b/src/zmq.cpp index a1428c09..ec8cc10e 100644 --- a/src/zmq.cpp +++ b/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 //////////////////////////////////////////////////////////////////////////////// From 359a50594930b3522d354d790fdbca73a92ef52b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lourens=20Naud=C3=A9?= Date: Thu, 22 Nov 2012 00:12:17 +0000 Subject: [PATCH 4/4] Prefer malloc to new for event address allocation as per Martin's recommendation --- src/socket_base.cpp | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/src/socket_base.cpp b/src/socket_base.cpp index ace73f81..2d853976 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -1052,7 +1052,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1064,7 +1064,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1076,7 +1076,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1088,7 +1088,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1100,7 +1100,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1112,7 +1112,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1124,7 +1124,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1136,7 +1136,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1148,7 +1148,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1160,7 +1160,7 @@ 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 = new char[addr_.size () + 1]; + 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); @@ -1179,7 +1179,7 @@ void zmq::socket_base_t::monitor_event (zmq_event_t event_) if (monitor_socket) { zmq_msg_t msg; void *event_data = malloc (sizeof (event_)); - assert (event_data); + 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);