[WIP, RFC] Problem: lot of use of dynamic memory allocation, leading to performance degradation and non-determism

CMakeLists.txt

@@ -615,6 +615,8 @@ if(MSVC)
 
     zmq_check_cxx_flag_prepend("/analyze")
 
+    zmq_check_cxx_flag_prepend("/Zc:__cplusplus") # enables right reporting of __cplusplus, ref. https://devblogs.microsoft.com/cppblog/msvc-now-correctly-reports-__cplusplus/
+
     # C++11/14/17-specific, but maybe possible via conditional defines
     zmq_check_cxx_flag_prepend("/wd26440") # Function '...' can be declared 'noexcept'
     zmq_check_cxx_flag_prepend("/wd26432") # If you define or delete any default operation in the type '...', define or
@@ -893,6 +895,8 @@ endif()
 set(cxx-sources
     precompiled.cpp
     address.cpp
+    allocator_default.cpp
+    allocator_global_pool.cpp
     channel.cpp
     client.cpp
     clock.cpp
@@ -993,9 +997,12 @@ set(cxx-sources
     zmtp_engine.cpp
     # at least for VS, the header files must also be listed
     address.hpp
+    allocator_default.hpp
+    allocator_global_pool.cpp
     array.hpp
     atomic_counter.hpp
     atomic_ptr.hpp
+    basic_concurrent_queue.hpp
     blob.hpp
     channel.hpp
     client.hpp
@@ -1136,6 +1143,8 @@ set(cxx-sources
     zap_client.hpp
     zmtp_engine.hpp)
 
+list(APPEND sources ${CMAKE_CURRENT_SOURCE_DIR}/external/mpmcqueue/concurrentqueue.h)
+
 if(MINGW)
   # Generate the right type when using -m32 or -m64
   macro(set_rc_arch rc_target)

Makefile.am

@@ -20,11 +20,17 @@ include_HEADERS = \
 	include/zmq_utils.h
 
 src_libzmq_la_SOURCES = \
+	external/mpmcqueue/concurrentqueue.h \
 	src/address.cpp \
 	src/address.hpp \
+	src/allocator_default.cpp \
+	src/allocator_default.hpp \
+	src/allocator_global_pool.cpp \
+	src/allocator_global_pool.hpp \
 	src/array.hpp \
 	src/atomic_counter.hpp \
 	src/atomic_ptr.hpp \
+	src/basic_concurrent_queue.hpp \
 	src/blob.hpp \
 	src/channel.cpp \
 	src/channel.hpp \

external/mpmcqueue/license.txt

@@ -0,0 +1,61 @@
+This license file applies to everything in this repository except that which
+is explicitly annotated as being written by other authors, i.e. the Boost
+queue (included in the benchmarks for comparison), Intel's TBB library (ditto),
+the CDSChecker tool (used for verification), the Relacy model checker (ditto),
+and Jeff Preshing's semaphore implementation (used in the blocking queue) which
+has a zlib license (embedded in lightweightsempahore.h).
+
+---
+
+Simplified BSD License:
+
+Copyright (c) 2013-2016, Cameron Desrochers.
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+- Redistributions of source code must retain the above copyright notice, this list of
+conditions and the following disclaimer.
+- Redistributions in binary form must reproduce the above copyright notice, this list of
+conditions and the following disclaimer in the documentation and/or other materials
+provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
+OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
+TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
+EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+---
+
+I have also chosen to dual-license under the Boost Software License as an alternative to
+the Simplified BSD license above:
+
+Boost Software License - Version 1.0 - August 17th, 2003
+
+Permission is hereby granted, free of charge, to any person or organization
+obtaining a copy of the software and accompanying documentation covered by
+this license (the "Software") to use, reproduce, display, distribute,
+execute, and transmit the Software, and to prepare derivative works of the
+Software, and to permit third-parties to whom the Software is furnished to
+do so, all subject to the following:
+
+The copyright notices in the Software and this entire statement, including
+the above license grant, this restriction and the following disclaimer,
+must be included in all copies of the Software, in whole or in part, and
+all derivative works of the Software, unless such copies or derivative
+works are solely in the form of machine-executable object code generated by
+a source language processor.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
+SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
+FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.

external/mpmcqueue/version.txt

@@ -0,0 +1 @@
+https://github.com/cameron314/concurrentqueue/commit/38e6a6f0185a98c3aaf2a95aa109ba041221d527

include/zmq.h

@@ -699,6 +699,35 @@ ZMQ_EXPORT int zmq_ctx_get_ext (void *context_,
                                 void *optval_,
                                 size_t *optvallen_);
 
+// ZMQ-provided message-pool implementations.                                 */
+// default allocator using malloc/free
+#define ZMQ_MSG_ALLOCATOR_DEFAULT 0
+// using internally a SPSC queue (cannot be used with inproc maybe?) or perhaps an MPMC queue anyway
+#define ZMQ_MSG_ALLOCATOR_PER_THREAD_POOL 1
+// using internally a MPMC queue, C++11 required
+#define ZMQ_MSG_ALLOCATOR_GLOBAL_POOL 2
+
+ZMQ_EXPORT void *zmq_msg_allocator_new (int type_);
+ZMQ_EXPORT int zmq_msg_allocator_destroy (void **allocator_);
+ZMQ_EXPORT int
+zmq_msg_init_allocator (zmq_msg_t *msg_, size_t size_, void *allocator_);
+
+struct zmq_allocator_t
+{
+    // Allocate a chunk of memory of size len and return the pointer
+    void*(*allocate_fn) (void *allocator, size_t len);
+
+    // Deallocate the memory chunk pointed to by data_
+    void(*deallocate_fn) (void *allocator, void *data_);
+
+    // Return true if this is an allocator and alive, otherwise false
+    bool(*check_tag_fn) (void *allocator);
+
+    void(*destroy_fn)( void *allocator);
+
+    void *allocator;
+};
+
 /*  DRAFT Socket methods.                                                     */
 ZMQ_EXPORT int zmq_join (void *s, const char *group);
 ZMQ_EXPORT int zmq_leave (void *s, const char *group);

perf/remote_thr.cpp

@@ -27,6 +27,7 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
+#include "platform.hpp"
 #include "../include/zmq.h"
 #include <stdio.h>
 #include <stdlib.h>
@@ -67,6 +68,11 @@ int main (int argc, char *argv[])
         return -1;
     }
 
+#if (defined ZMQ_BUILD_DRAFT_API && defined ZMQ_MSG_ALLOCATOR_GLOBAL_POOL)
+    // EXPERIMENTAL ALLOCATOR FOR MSG_T
+    void *allocator = zmq_msg_allocator_new (ZMQ_MSG_ALLOCATOR_GLOBAL_POOL);
+#endif
+
     s = zmq_socket (ctx, ZMQ_PUSH);
     if (!s) {
         printf ("error in zmq_socket: %s\n", zmq_strerror (errno));
@@ -105,7 +111,11 @@ int main (int argc, char *argv[])
     }
 
     for (i = 0; i != message_count; i++) {
+#if (defined ZMQ_BUILD_DRAFT_API && defined ZMQ_MSG_ALLOCATOR_GLOBAL_POOL)
+        rc = zmq_msg_init_allocator (&msg, message_size, allocator);
+#else
         rc = zmq_msg_init_size (&msg, message_size);
+#endif
         if (rc != 0) {
             printf ("error in zmq_msg_init_size: %s\n", zmq_strerror (errno));
             return -1;
@@ -134,5 +144,11 @@ int main (int argc, char *argv[])
         return -1;
     }
 
+#if (defined ZMQ_BUILD_DRAFT_API && defined ZMQ_MSG_ALLOCATOR_GLOBAL_POOL)
+    // IMPORTANT: destroy the allocator only after zmq_ctx_term() since otherwise
+    // some zmq_msg_t may still be "in fly"
+    zmq_msg_allocator_destroy (&allocator);
+#endif
+
     return 0;
 }

src/allocator_default.cpp

@@ -0,0 +1,60 @@
+/*
+    Copyright (c) 2019-2020 Contributors as noted in the AUTHORS file
+
+    This file is part of libzmq, the ZeroMQ core engine in C++.
+
+    libzmq is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License (LGPL) as published
+    by the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    As a special exception, the Contributors give you permission to link
+    this library with independent modules to produce an executable,
+    regardless of the license terms of these independent modules, and to
+    copy and distribute the resulting executable under terms of your choice,
+    provided that you also meet, for each linked independent module, the
+    terms and conditions of the license of that module. An independent
+    module is a module which is not derived from or based on this library.
+    If you modify this library, you must extend this exception to your
+    version of the library.
+
+    libzmq 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 GNU Lesser General Public
+    License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "precompiled.hpp"
+#include "allocator_default.hpp"
+
+#include <new>
+
+zmq::allocator_default_t::allocator_default_t ()
+{
+    _tag = 0xCAFEEBEB;
+}
+
+
+zmq::allocator_default_t::~allocator_default_t ()
+{
+    //  Mark this instance as dead
+    _tag = 0xdeadbeef;
+}
+
+void *zmq::allocator_default_t::allocate (size_t len_)
+{
+    return operator new (len_, std::nothrow);
+}
+
+void zmq::allocator_default_t::deallocate (void *data_)
+{
+    operator delete (data_);
+}
+
+bool zmq::allocator_default_t::check_tag () const
+{
+    return _tag == 0xCAFEEBEB;
+}

src/allocator_default.hpp

@@ -0,0 +1,71 @@
+/*
+    Copyright (c) 2019-2020 Contributors as noted in the AUTHORS file
+
+    This file is part of libzmq, the ZeroMQ core engine in C++.
+
+    libzmq is free software; you can redistribute it and/or modify it under
+    the terms of the GNU Lesser General Public License (LGPL) as published
+    by the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    As a special exception, the Contributors give you permission to link
+    this library with independent modules to produce an executable,
+    regardless of the license terms of these independent modules, and to
+    copy and distribute the resulting executable under terms of your choice,
+    provided that you also meet, for each linked independent module, the
+    terms and conditions of the license of that module. An independent
+    module is a module which is not derived from or based on this library.
+    If you modify this library, you must extend this exception to your
+    version of the library.
+
+    libzmq 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 GNU Lesser General Public
+    License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_I_ALLOCATOR_HPP_INCLUDED__
+#define __ZMQ_I_ALLOCATOR_HPP_INCLUDED__
+
+namespace zmq
+{
+class allocator_default_t
+{
+  public:
+    allocator_default_t ();
+
+    ~allocator_default_t ();
+
+    static void *allocate_fn (void *allocator_, size_t len_)
+    {
+        return static_cast<allocator_default_t *> (allocator_)->allocate (len_);
+    }
+
+    static void deallocate_fn (void *allocator_, void *data_)
+    {
+        return static_cast<allocator_default_t *> (allocator_)
+          ->deallocate (data_);
+    }
+
+    static bool check_tag_fn (void *allocator_)
+    {
+        return static_cast<allocator_default_t *> (allocator_)->check_tag ();
+    }
+
+    // allocate() typically gets called by the consumer thread: the user app thread(s)
+    void *allocate (size_t len_);
+
+    void deallocate (void *data_);
+
+    bool check_tag () const;
+
+  private:
+    //  Used to check whether the object is a socket.
+    uint32_t _tag;
+};
+}
+
+#endif