From b7c0a59c0a86f289f55935b19efaf448e892eefb Mon Sep 17 00:00:00 2001 From: Josh Blum Date: Sat, 15 Dec 2012 09:53:25 -0800 Subject: work on the buffer queue api --- include/gras/CMakeLists.txt | 1 + include/gras/block.hpp | 9 +++-- include/gras/block.i | 1 + include/gras/buffer_queue.hpp | 43 ++++++++++++++++++++ lib/CMakeLists.txt | 2 + lib/block_allocator.cpp | 28 ++++++------- lib/block_task.cpp | 7 ++-- lib/buffer_queue_circ.cpp | 3 ++ lib/buffer_queue_pool.cpp | 54 +++++++++++++++++++++++++ lib/circular_buffer.cpp | 4 +- lib/gras_impl/block_actor.hpp | 3 +- lib/gras_impl/buffer_queue.hpp | 50 ----------------------- lib/gras_impl/input_buffer_queues.hpp | 6 +-- lib/gras_impl/messages.hpp | 3 +- lib/gras_impl/output_buffer_queues.hpp | 72 ++++++++++++++++++++++------------ lib/gras_impl/simple_buffer_queue.hpp | 50 +++++++++++++++++++++++ lib/input_handlers.cpp | 4 +- lib/output_handlers.cpp | 6 +-- 18 files changed, 233 insertions(+), 113 deletions(-) create mode 100644 include/gras/buffer_queue.hpp create mode 100644 lib/buffer_queue_circ.cpp create mode 100644 lib/buffer_queue_pool.cpp delete mode 100644 lib/gras_impl/buffer_queue.hpp create mode 100644 lib/gras_impl/simple_buffer_queue.hpp diff --git a/include/gras/CMakeLists.txt b/include/gras/CMakeLists.txt index e8fb85e..f0b0d3a 100644 --- a/include/gras/CMakeLists.txt +++ b/include/gras/CMakeLists.txt @@ -21,6 +21,7 @@ install(FILES thread_pool.hpp top_block.hpp work_buffer.hpp + buffer_queue.hpp DESTINATION include/gras COMPONENT ${GRAS_COMP_DEVEL} diff --git a/include/gras/block.hpp b/include/gras/block.hpp index 3899561..524ff00 100644 --- a/include/gras/block.hpp +++ b/include/gras/block.hpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include @@ -337,9 +338,9 @@ struct GRAS_API Block : Element * \param which_output the output 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) + * \return a shared ptr to a new buffer queue object */ - virtual SBufferToken output_buffer_allocator( + virtual BufferQueueSptr output_buffer_allocator( const size_t which_output, const SBufferToken &token, const size_t recommend_length @@ -356,9 +357,9 @@ struct GRAS_API Block : Element * \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) + * \return a shared ptr to a new buffer queue object */ - virtual SBufferToken input_buffer_allocator( + virtual BufferQueueSptr input_buffer_allocator( const size_t which_input, const SBufferToken &token, const size_t recommend_length diff --git a/include/gras/block.i b/include/gras/block.i index 80c3905..ef2738c 100644 --- a/include/gras/block.i +++ b/include/gras/block.i @@ -11,6 +11,7 @@ %import %include %import +%include %include #endif /*INCLUDED_GRAS_BLOCK_I*/ diff --git a/include/gras/buffer_queue.hpp b/include/gras/buffer_queue.hpp new file mode 100644 index 0000000..5125555 --- /dev/null +++ b/include/gras/buffer_queue.hpp @@ -0,0 +1,43 @@ +// Copyright (C) by Josh Blum. See LICENSE.txt for licensing information. + +#ifndef INCLUDED_GRAS_BUFFER_QUEUE_HPP +#define INCLUDED_GRAS_BUFFER_QUEUE_HPP + +#include +#include +#include + +namespace gras +{ + +struct BufferQueue; + +typedef boost::shared_ptr BufferQueueSptr; + +//! Buffer Queue is an interface enabling us to create custom buffer allocators. +struct BufferQueue +{ + + //! Create a buffer queue using the pool allocator + GRAS_API static BufferQueueSptr make_pool(const SBufferConfig &config, const size_t num_buffs); + + //! Create a buffer queue using the circular buffer allocator + GRAS_API static BufferQueueSptr make_circ(const SBufferConfig &config); + + //! Get a reference to the buffer at the front of the queue + virtual SBuffer &front(void) = 0; + + //! Pop off the used portion of the queue + virtual void pop(void) = 0; + + //! Push a used buffer back into the queue + virtual void push(const SBuffer &buff) = 0; + + //! Is the queue empty? + virtual bool empty(void) const = 0; + +}; + +} //namespace gras + +#endif /*INCLUDED_GRAS_BUFFER_QUEUE_HPP*/ diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt index 84c8f4e..bc38d06 100644 --- a/lib/CMakeLists.txt +++ b/lib/CMakeLists.txt @@ -41,6 +41,8 @@ list(APPEND GRAS_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/element.cpp ${CMAKE_CURRENT_SOURCE_DIR}/sbuffer.cpp ${CMAKE_CURRENT_SOURCE_DIR}/circular_buffer.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/buffer_queue_circ.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/buffer_queue_pool.cpp ${CMAKE_CURRENT_SOURCE_DIR}/tags.cpp ${CMAKE_CURRENT_SOURCE_DIR}/block.cpp ${CMAKE_CURRENT_SOURCE_DIR}/block_actor.cpp diff --git a/lib/block_allocator.cpp b/lib/block_allocator.cpp index 64b3aed..54654ba 100644 --- a/lib/block_allocator.cpp +++ b/lib/block_allocator.cpp @@ -58,7 +58,6 @@ void BlockActor::handle_top_alloc(const TopAllocMessage &, const Theron::Address //allocate output buffers which will also wake up the task const size_t num_outputs = this->get_num_outputs(); - this->output_buffer_tokens.resize(num_outputs); for (size_t i = 0; i < num_outputs; i++) { const size_t bytes = recommend_length( @@ -71,7 +70,8 @@ void BlockActor::handle_top_alloc(const TopAllocMessage &, const Theron::Address SBufferDeleter deleter = boost::bind(&BlockActor::buffer_returner, this, i, _1); SBufferToken token = SBufferToken(new SBufferDeleter(deleter)); - this->output_buffer_tokens[i] = block_ptr->output_buffer_allocator(i, token, bytes); + BufferQueueSptr queue = block_ptr->output_buffer_allocator(i, token, bytes); + this->output_queues.set_buffer_queue(i, queue); InputAllocMessage message; message.token = SBufferToken(new SBufferDeleter(deleter)); @@ -82,29 +82,23 @@ void BlockActor::handle_top_alloc(const TopAllocMessage &, const Theron::Address this->Send(0, from); //ACK } -SBufferToken Block::output_buffer_allocator( +BufferQueueSptr Block::output_buffer_allocator( const size_t, const SBufferToken &token, const size_t recommend_length ){ - for (size_t j = 0; j < THIS_MANY_BUFFERS; j++) - { - SBufferConfig config; - config.memory = NULL; - config.length = recommend_length; - config.affinity = (*this)->block->buffer_affinity; - config.token = token; - SBuffer buff(config); - std::memset(buff.get_actual_memory(), 0, buff.get_actual_length()); - //buffer derefs here and the token messages it back to the block - } - return token; + SBufferConfig config; + config.memory = NULL; + config.length = recommend_length; + config.affinity = (*this)->block->buffer_affinity; + config.token = token; + return BufferQueue::make_pool(config, THIS_MANY_BUFFERS); } -SBufferToken Block::input_buffer_allocator( +BufferQueueSptr Block::input_buffer_allocator( const size_t, const SBufferToken &, const size_t ){ - return SBufferToken(); //null + return BufferQueueSptr(); //null } diff --git a/lib/block_task.cpp b/lib/block_task.cpp index 9e5917d..6b18798 100644 --- a/lib/block_task.cpp +++ b/lib/block_task.cpp @@ -31,9 +31,6 @@ void BlockActor::mark_done(void) //release upstream, downstream, and executor tokens this->token_pool.clear(); - //release allocator tokens, buffers can now call deleters - this->output_buffer_tokens.clear(); - //release all buffers in queues this->input_queues.flush_all(); this->output_queues.flush_all(); @@ -156,7 +153,8 @@ void BlockActor::handle_task(void) this->input_items.max() = std::max(this->input_items.max(), items); //inline dealings, how and when input buffers can be inlined into output buffers - //continue; + continue; //FIXME to implement needs change + /* if ( buff.unique() and input_configs[i].inline_buffer and @@ -170,6 +168,7 @@ void BlockActor::handle_task(void) this->output_queues.push_front(output_inline_index, new_obuff); //you got inlined! output_inline_index++; //done do this output port again } + */ } //------------------------------------------------------------------ diff --git a/lib/buffer_queue_circ.cpp b/lib/buffer_queue_circ.cpp new file mode 100644 index 0000000..9d0dbcf --- /dev/null +++ b/lib/buffer_queue_circ.cpp @@ -0,0 +1,3 @@ +// Copyright (C) by Josh Blum. See LICENSE.txt for licensing information. + +#include diff --git a/lib/buffer_queue_pool.cpp b/lib/buffer_queue_pool.cpp new file mode 100644 index 0000000..a30f4a6 --- /dev/null +++ b/lib/buffer_queue_pool.cpp @@ -0,0 +1,54 @@ +// Copyright (C) by Josh Blum. See LICENSE.txt for licensing information. + +#include +#include +#include + +using namespace gras; + +struct BufferQueuePool : BufferQueue +{ + BufferQueuePool(const size_t num) + { + queue.resize(num); + } + + SBuffer &front(void) + { + return queue.front(); + } + + void pop(void) + { + ASSERT(not queue.empty()); + queue.front() = SBuffer(); //dont hold ref + queue.pop_front(); + } + + void push(const SBuffer &buff) + { + queue.push_back(buff); + } + + bool empty(void) const + { + return queue.empty(); + } + + boost::circular_buffer queue; + +}; + +BufferQueueSptr BufferQueue::make_pool( + const SBufferConfig &config, + const size_t num_buffs +){ + BufferQueueSptr bq(new BufferQueuePool(num_buffs)); + for (size_t i = 0; i < num_buffs; i++) + { + SBuffer buff(config); + std::memset(buff.get_actual_memory(), 0, buff.get_actual_length()); + bq->push(buff); + } + return bq; +} diff --git a/lib/circular_buffer.cpp b/lib/circular_buffer.cpp index 820129d..1aa745b 100644 --- a/lib/circular_buffer.cpp +++ b/lib/circular_buffer.cpp @@ -1,6 +1,6 @@ // Copyright (C) by Josh Blum. See LICENSE.txt for licensing information. -#include +#include #include #include #include @@ -137,6 +137,7 @@ static void circular_buffer_delete(SBuffer &buff, CircularBuffer *circ_buff) delete circ_buff; } +/* SBuffer EndlessBufferQueue::make_circular_buffer(const size_t num_bytes) { CircularBuffer *circ_buff = new CircularBuffer(num_bytes); @@ -147,3 +148,4 @@ SBuffer EndlessBufferQueue::make_circular_buffer(const size_t num_bytes) config.deleter = deleter; return SBuffer(config); } +*/ diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp index d204ec4..8717fa2 100644 --- a/lib/gras_impl/block_actor.hpp +++ b/lib/gras_impl/block_actor.hpp @@ -122,11 +122,10 @@ struct BlockActor : Apology::Worker BitSet inputs_done; BitSet outputs_done; std::set token_pool; - std::vector output_buffer_tokens; //buffer queues and ready conditions InputBufferQueues input_queues; - OutputBufferQueues output_queues; + OutputBufferQueues output_queues; BitSet inputs_available; //tag tracking diff --git a/lib/gras_impl/buffer_queue.hpp b/lib/gras_impl/buffer_queue.hpp deleted file mode 100644 index 0e6be9e..0000000 --- a/lib/gras_impl/buffer_queue.hpp +++ /dev/null @@ -1,50 +0,0 @@ -// Copyright (C) by Josh Blum. See LICENSE.txt for licensing information. - -#ifndef INCLUDED_LIBGRAS_IMPL_BUFFER_QUEUE_HPP -#define INCLUDED_LIBGRAS_IMPL_BUFFER_QUEUE_HPP - -#include -#include -#include - -namespace gras -{ - -struct BufferQueue : std::queue -{ - BufferQueue(void) - { - SBufferDeleter deleter = boost::bind(&BufferQueue::push_back, this, _1); - _token = SBufferToken(new SBufferDeleter(deleter)); - } - - ~BufferQueue(void) - { - _token.reset(); - while (not this->empty()) - { - this->pop(); - } - } - - void push_back(const SBuffer &buff) - { - this->push(buff); - } - - void allocate_one(const size_t num_bytes) - { - SBufferConfig config; - config.memory = NULL; - config.length = num_bytes; - config.token = _token; - SBuffer buff(config); - //buffer derefs here and the token messages it back to the queue - } - - SBufferToken _token; -}; - -} //namespace gras - -#endif /*INCLUDED_LIBGRAS_IMPL_BUFFER_QUEUE_HPP*/ diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp index 83ed44b..7f70d39 100644 --- a/lib/gras_impl/input_buffer_queues.hpp +++ b/lib/gras_impl/input_buffer_queues.hpp @@ -5,7 +5,7 @@ #include #include -#include +#include #include #include #include @@ -172,7 +172,7 @@ struct InputBufferQueues std::vector _maximum_bytes; std::vector > _queues; std::vector _preload_bytes; - std::vector > _aux_queues; + std::vector > _aux_queues; }; @@ -207,7 +207,7 @@ inline void InputBufferQueues::update_config( _aux_queues[i]->empty() or _aux_queues[i]->front().get_actual_length() != _maximum_bytes[i] ){ - _aux_queues[i] = boost::shared_ptr(new BufferQueue()); + _aux_queues[i].reset(new SimpleBufferQueue()); _aux_queues[i]->allocate_one(_maximum_bytes[i]); _aux_queues[i]->allocate_one(_maximum_bytes[i]); _aux_queues[i]->allocate_one(_maximum_bytes[i]); diff --git a/lib/gras_impl/messages.hpp b/lib/gras_impl/messages.hpp index eae3258..62d48e6 100644 --- a/lib/gras_impl/messages.hpp +++ b/lib/gras_impl/messages.hpp @@ -3,6 +3,7 @@ #ifndef INCLUDED_LIBGRAS_IMPL_MESSAGES_HPP #define INCLUDED_LIBGRAS_IMPL_MESSAGES_HPP +#include #include #include #include @@ -104,7 +105,7 @@ struct OutputHintMessage struct OutputAllocMessage { size_t index; - SBufferToken token; + BufferQueueSptr queue; }; //---------------------------------------------------------------------- diff --git a/lib/gras_impl/output_buffer_queues.hpp b/lib/gras_impl/output_buffer_queues.hpp index 722a4c2..5674c3d 100644 --- a/lib/gras_impl/output_buffer_queues.hpp +++ b/lib/gras_impl/output_buffer_queues.hpp @@ -3,64 +3,75 @@ #ifndef INCLUDED_LIBGRAS_IMPL_OUTPUT_BUFFER_QUEUES_HPP #define INCLUDED_LIBGRAS_IMPL_OUTPUT_BUFFER_QUEUES_HPP +#include #include #include -#include namespace gras { -template struct OutputBufferQueues { - enum {MAX_QUEUE_SIZE = 128}; - BitSet _bitset; - std::vector > _queues; + void set_buffer_queue(const size_t i, BufferQueueSptr queue) + { + _queues[i] = queue; + _update(i); + } + + void set_reserve_bytes(const size_t i, const size_t num_bytes) + { + _reserve_bytes[i] = num_bytes; + _update(i); + } GRAS_FORCE_INLINE void resize(const size_t size) { _bitset.resize(size); - _queues.resize(size, boost::circular_buffer(MAX_QUEUE_SIZE)); + _queues.resize(size); + _reserve_bytes.resize(size, 1); } - GRAS_FORCE_INLINE void push(const size_t i, const T &value) + GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &value) { - _queues[i].push_back(value); - _bitset.set(i); + _queues[i]->push(value); + _update(i); + } + + GRAS_FORCE_INLINE void flush_all(void) + { + const size_t old_size = this->size(); + this->resize(0); + this->resize(old_size); } //! used for input buffer inlining + /* GRAS_FORCE_INLINE void push_front(const size_t i, const T &value) { ASSERT(not _queues[i].full()); _queues[i].push_front(value); _bitset.set(i); } + */ - GRAS_FORCE_INLINE const T &front(const size_t i) const + GRAS_FORCE_INLINE SBuffer &front(const size_t i) { - return _queues[i].front(); - } - - GRAS_FORCE_INLINE T &front(const size_t i) - { - ASSERT(not _queues[i].empty()); - return _queues[i].front(); + ASSERT(not _queues[i]->empty()); + return _queues[i]->front(); } GRAS_FORCE_INLINE void pop(const size_t i) { - _queues[i].front() = T(); - _queues[i].pop_front(); - _bitset.set(i, not _queues[i].empty()); + _queues[i]->pop(); + _update(i); } GRAS_FORCE_INLINE void fail(const size_t i) { _bitset.reset(i); } - +/* GRAS_FORCE_INLINE void flush(const size_t i) { _queues[i].clear(); @@ -71,15 +82,18 @@ struct OutputBufferQueues { for (size_t i = 0; i < this->size(); i++) this->flush(i); } - +*/ GRAS_FORCE_INLINE bool ready(const size_t i) const { - return not _queues[i].empty(); + if (_queues[i]->empty()) return false; + const SBuffer &front = _queues[i]->front(); + const size_t avail = front.get_actual_length() - front.offset - front.length; + return avail >= _reserve_bytes[i]; } GRAS_FORCE_INLINE bool empty(const size_t i) const { - return _queues[i].empty(); + return _queues[i]->empty(); } GRAS_FORCE_INLINE bool all_ready(void) const @@ -91,6 +105,16 @@ struct OutputBufferQueues { return _queues.size(); } + + GRAS_FORCE_INLINE void _update(const size_t i) + { + _bitset.set(i, this->ready(i)); + } + + BitSet _bitset; + std::vector _queues; + std::vector _reserve_bytes; + }; } //namespace gras diff --git a/lib/gras_impl/simple_buffer_queue.hpp b/lib/gras_impl/simple_buffer_queue.hpp new file mode 100644 index 0000000..db6ecc0 --- /dev/null +++ b/lib/gras_impl/simple_buffer_queue.hpp @@ -0,0 +1,50 @@ +// Copyright (C) by Josh Blum. See LICENSE.txt for licensing information. + +#ifndef INCLUDED_LIBGRAS_IMPL_SIMPLE_BUFFER_QUEUE_HPP +#define INCLUDED_LIBGRAS_IMPL_SIMPLE_BUFFER_QUEUE_HPP + +#include +#include +#include + +namespace gras +{ + +struct SimpleBufferQueue : std::queue +{ + SimpleBufferQueue(void) + { + SBufferDeleter deleter = boost::bind(&SimpleBufferQueue::push_back, this, _1); + _token = SBufferToken(new SBufferDeleter(deleter)); + } + + ~SimpleBufferQueue(void) + { + _token.reset(); + while (not this->empty()) + { + this->pop(); + } + } + + void push_back(const SBuffer &buff) + { + this->push(buff); + } + + void allocate_one(const size_t num_bytes) + { + SBufferConfig config; + config.memory = NULL; + config.length = num_bytes; + config.token = _token; + SBuffer buff(config); + //buffer derefs here and the token messages it back to the queue + } + + SBufferToken _token; +}; + +} //namespace gras + +#endif /*INCLUDED_LIBGRAS_IMPL_SIMPLE_BUFFER_QUEUE_HPP*/ diff --git a/lib/input_handlers.cpp b/lib/input_handlers.cpp index 02f62f3..bef5234 100644 --- a/lib/input_handlers.cpp +++ b/lib/input_handlers.cpp @@ -66,8 +66,8 @@ void BlockActor::handle_input_alloc(const InputAllocMessage &message, const Ther //handle the upstream block allocation request OutputAllocMessage new_msg; - new_msg.token = block_ptr->input_buffer_allocator( + new_msg.queue = block_ptr->input_buffer_allocator( index, message.token, message.recommend_length ); - if (new_msg.token) this->post_upstream(index, new_msg); + if (new_msg.queue) this->post_upstream(index, new_msg); } diff --git a/lib/output_handlers.cpp b/lib/output_handlers.cpp index 5876f89..a0bad75 100644 --- a/lib/output_handlers.cpp +++ b/lib/output_handlers.cpp @@ -70,9 +70,5 @@ void BlockActor::handle_output_alloc(const OutputAllocMessage &message, const Th const size_t index = message.index; //return of a positive downstream allocation - //reset the token, and clear old output buffers - //the new token from the downstream is installed - this->output_buffer_tokens[index].reset(); - this->output_queues.flush(index); - this->output_buffer_tokens[index] = message.token; + this->output_queues.set_buffer_queue(index, message.queue); } -- cgit