created hooks for input buffer allocator

5 files changed, 64 insertions, 6 deletions

diff --git a/include/gnuradio/block.hpp b/include/gnuradio/block.hpp
index ae09149..25cf77c 100644
--- a/include/gnuradio/block.hpp
+++ b/include/gnuradio/block.hpp
@@ -243,7 +243,24 @@ struct GRAS_API Block : Element
         const size_t recommend_length
     );
 
-    //TODO overload for allocate output buffer(index)
+    /*!
+     * The input buffer allocator method.
+     * This method is special and very different from allocate output buffers.
+     * Typically, blocks do not have control of their input buffers.
+     * When overloaded, an upstream block will ask this block
+     * to allocate its output buffers. This way, this block will get
+     * input buffers which were actually allocated by this method.
+     *
+     * \param which_input the input port index number
+     * \param token the token for the buffer's returner
+     * \param recommend_length the schedulers recommended length in bytes
+     * \return the token used for the buffer allocation (may be the same)
+     */
+    virtual SBufferToken input_buffer_allocator(
+        const size_t which_input,
+        const SBufferToken &token,
+        const size_t recommend_length
+    );
 
 };
 
diff --git a/lib/block_allocator.cpp b/lib/block_allocator.cpp
index 24369c7..ffd052b 100644
--- a/lib/block_allocator.cpp
+++ b/lib/block_allocator.cpp
@@ -86,15 +86,19 @@ void ElementImpl::handle_allocation(const tsbe::TaskInterface &task_iface)
         );
 
         SBufferDeleter deleter = boost::bind(&ElementImpl::buffer_returner, this, i, _1);
+        SBufferToken token = SBufferToken(new SBufferDeleter(deleter));
 
-        this->output_buffer_tokens[i] = block_ptr->output_buffer_allocator(
-            i, SBufferToken(new SBufferDeleter(deleter)), bytes
-        );
+        this->output_buffer_tokens[i] = block_ptr->output_buffer_allocator(i, token, bytes);
+
+        InputAllocatorMessage message;
+        message.token = token;
+        message.recommend_length = bytes;
+        task_iface.post_downstream(i, message);
     }
 }
 
 SBufferToken Block::output_buffer_allocator(
-    const size_t which_output,
+    const size_t,
     const SBufferToken &token,
     const size_t recommend_length
 ){
@@ -110,3 +114,11 @@ SBufferToken Block::output_buffer_allocator(
     }
     return token;
 }
+
+SBufferToken Block::input_buffer_allocator(
+    const size_t,
+    const SBufferToken &,
+    const size_t
+){
+    return SBufferToken(); //null
+}
diff --git a/lib/gras_impl/messages.hpp b/lib/gras_impl/messages.hpp
index 8c6664f..912e3d1 100644
--- a/lib/gras_impl/messages.hpp
+++ b/lib/gras_impl/messages.hpp
@@ -64,6 +64,12 @@ struct UpdateInputsMessage
     //empty
 };
 
+struct InputAllocatorMessage
+{
+    SBufferToken token;
+    size_t recommend_length;
+};
+
 } //namespace gnuradio
 
 #endif /*INCLUDED_LIBGRAS_IMPL_MESSAGES_HPP*/
diff --git a/lib/port_handlers.cpp b/lib/port_handlers.cpp
index d0d5896..848d9ef 100644
--- a/lib/port_handlers.cpp
+++ b/lib/port_handlers.cpp
@@ -60,6 +60,19 @@ void ElementImpl::handle_input_msg(
         return;
     }
 
+    //handle the upstream block allocation request
+    if (msg.type() == typeid(InputAllocatorMessage))
+    {
+        InputAllocatorMessage message;
+        message.token = block_ptr->input_buffer_allocator(
+            index,
+            msg.cast<InputAllocatorMessage>().token,
+            msg.cast<InputAllocatorMessage>().recommend_length
+        );
+        if (message.token) handle.post_upstream(index, message);
+        return;
+    }
+
     ASSERT(false);
 }
 
@@ -77,7 +90,6 @@ void ElementImpl::handle_output_msg(
         return;
     }
 
-
     //a downstream block has declared itself done, recheck the token
     if (msg.type() == typeid(CheckTokensMessage))
     {
@@ -111,5 +123,15 @@ void ElementImpl::handle_output_msg(
         return;
     }
 
+    //return of a positive downstream allocation
+    //reset the token, and clear old output buffers
+    //the new token from the downstream is installed
+    if (msg.type() == typeid(InputAllocatorMessage))
+    {
+        this->output_buffer_tokens[index].reset();
+        this->output_queues.flush(index);
+        this->output_buffer_tokens[index] = msg.cast<InputAllocatorMessage>().token;
+    }
+
     ASSERT(false);
 }
diff --git a/swig/gras.i b/swig/gras.i
index cc0f51a..b56318d 100644
--- a/swig/gras.i
+++ b/swig/gras.i
@@ -31,6 +31,7 @@
 %ignore gnuradio::IOSignature::operator->();
 %ignore gnuradio::IOSignature::operator->() const;
 
+%ignore gnuradio::Block::input_buffer_allocator;
 %ignore gnuradio::Block::output_buffer_allocator;
 
 %include <gnuradio/sbuffer.hpp>