register messages, and custom allocator for performance

3 files changed, 138 insertions, 0 deletions

diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt
index 1eeb832..956e5e1 100644
--- a/lib/CMakeLists.txt
+++ b/lib/CMakeLists.txt
@@ -11,6 +11,10 @@ if(CMAKE_BUILD_TYPE STREQUAL "Debug")
     add_definitions(-DGRAS_DEBUG)
 endif()
 
+list(APPEND gnuradio_core_sources
+    ${CMAKE_CURRENT_SOURCE_DIR}/register_messages.cpp #exports messages, must be first
+)
+
 ########################################################################
 # Setup Theron Deps
 ########################################################################
@@ -54,6 +58,7 @@ list(APPEND gnuradio_core_sources
     ${CMAKE_CURRENT_SOURCE_DIR}/block.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block_actor.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block_task.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/theron_allocator.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block_allocator.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block_handlers.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/topology_handler.cpp
diff --git a/lib/register_messages.cpp b/lib/register_messages.cpp
new file mode 100644
index 0000000..18a3cd8
--- /dev/null
+++ b/lib/register_messages.cpp
@@ -0,0 +1,43 @@
+//
+// Copyright 2012 Josh Blum
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program 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 io_sig program.  If not, see <http://www.gnu.org/licenses/>.
+
+#include <Theron/Register.h>
+#include <gras_impl/messages.hpp>
+#include <gras_impl/interruptible_thread.hpp>
+#include <Apology/Worker.hpp>
+
+THERON_REGISTER_MESSAGE(Apology::WorkerTopologyMessage);
+THERON_REGISTER_MESSAGE(gnuradio::TopAllocMessage);
+THERON_REGISTER_MESSAGE(gnuradio::TopActiveMessage);
+THERON_REGISTER_MESSAGE(gnuradio::TopInertMessage);
+THERON_REGISTER_MESSAGE(gnuradio::TopTokenMessage);
+THERON_REGISTER_MESSAGE(gnuradio::TopHintMessage);
+THERON_REGISTER_MESSAGE(gnuradio::SharedThreadGroup);
+
+THERON_REGISTER_MESSAGE(gnuradio::InputTagMessage);
+THERON_REGISTER_MESSAGE(gnuradio::InputBufferMessage);
+THERON_REGISTER_MESSAGE(gnuradio::InputTokenMessage);
+THERON_REGISTER_MESSAGE(gnuradio::InputCheckMessage);
+THERON_REGISTER_MESSAGE(gnuradio::InputAllocMessage);
+
+THERON_REGISTER_MESSAGE(gnuradio::OutputBufferMessage);
+THERON_REGISTER_MESSAGE(gnuradio::OutputTokenMessage);
+THERON_REGISTER_MESSAGE(gnuradio::OutputCheckMessage);
+THERON_REGISTER_MESSAGE(gnuradio::OutputHintMessage);
+THERON_REGISTER_MESSAGE(gnuradio::OutputAllocMessage);
+
+THERON_REGISTER_MESSAGE(gnuradio::SelfKickMessage);
+THERON_REGISTER_MESSAGE(gnuradio::UpdateInputsMessage);
diff --git a/lib/theron_allocator.cpp b/lib/theron_allocator.cpp
new file mode 100644
index 0000000..b8f875f
--- /dev/null
+++ b/lib/theron_allocator.cpp
@@ -0,0 +1,90 @@
+//
+// Copyright 2012 Josh Blum
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program 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 io_sig program.  If not, see <http://www.gnu.org/licenses/>.
+
+/***********************************************************************
+ * There is allocation overhead for sending messages.
+ * Want per worker-allocator for message allocation...
+ * But until thats possible, install a new global allocator.
+ * This allocator uses a fixed pool for small sized buffers,
+ * and otherwise the regular malloc/free for larger buffers.
+ **********************************************************************/
+
+#include <gras_impl/debug.hpp>
+#include <Theron/Detail/Threading/SpinLock.h>
+#include <Theron/IAllocator.h>
+#include <Theron/AllocatorManager.h>
+#include <boost/circular_buffer.hpp>
+
+#define MY_ALLOCATOR_CHUNK_SIZE 88 //Theron asks for a lot of 88-byte buffers
+#define MY_ALLOCATOR_POOL_SIZE (MY_ALLOCATOR_CHUNK_SIZE * 4096)
+
+static struct WorkerAllocator : Theron::IAllocator
+{
+    WorkerAllocator(void)
+    {
+        const size_t N = MY_ALLOCATOR_POOL_SIZE/MY_ALLOCATOR_CHUNK_SIZE;
+        queue.resize(N);
+        for (size_t i = 0; i < N; i++)
+        {
+            const ptrdiff_t pool_ptr = ptrdiff_t(pool) + i*MY_ALLOCATOR_CHUNK_SIZE;
+            queue.push_back((void *)pool_ptr);
+        }
+        Theron::AllocatorManager::Instance().SetAllocator(this);
+    }
+
+    ~WorkerAllocator(void)
+    {
+        //NOP
+    }
+
+    void *Allocate(const SizeType size)
+    {
+        if (size <= MY_ALLOCATOR_CHUNK_SIZE)
+        {
+            mSpinLock.Lock();
+            ASSERT(not queue.empty());
+            void *memory = queue.front();
+            queue.pop_front();
+            mSpinLock.Unlock();
+            return memory;
+        }
+        else
+        {
+            //std::cout << "malloc size " << size << std::endl;
+            return std::malloc(size);
+        }
+    }
+
+    void Free(void *const memory)
+    {
+        const bool in_pool = ptrdiff_t(memory) >= ptrdiff_t(pool) and ptrdiff_t(memory) < (ptrdiff_t(pool) + MY_ALLOCATOR_POOL_SIZE);
+        if (in_pool)
+        {
+            mSpinLock.Lock();
+            queue.push_front(memory);
+            mSpinLock.Unlock();
+        }
+        else
+        {
+            std::free(memory);
+        }
+    }
+
+    boost::circular_buffer<void *> queue;
+    long pool[MY_ALLOCATOR_POOL_SIZE/sizeof(long)];
+    Theron::Detail::SpinLock mSpinLock;
+
+} my_alloc;