thread_queue.h

Go to the documentation of this file.

/* Copyright (c) 2008-2022 the MRtrix3 contributors.
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Covered Software is provided under this License on an "as is"
 * basis, without warranty of any kind, either expressed, implied, or
 * statutory, including, without limitation, warranties that the
 * Covered Software is free of defects, merchantable, fit for a
 * particular purpose or non-infringing.
 * See the Mozilla Public License v. 2.0 for more details.
 *
 * For more details, see http://www.mrtrix.org/.
 */
 
#ifndef __mrtrix_thread_queue_h__
#define __mrtrix_thread_queue_h__
 
#include <stack>
#include <condition_variable>
 
#include "exception.h"
#include "memory.h"
#include "thread.h"
 
#define MRTRIX_QUEUE_DEFAULT_CAPACITY 128
#define MRTRIX_QUEUE_DEFAULT_BATCH_SIZE 128
 
namespace MR
{
  namespace Thread
  {
 
 
    //* \cond skip
    namespace {
 
      // to get multi/single job/functor seamlessly:
      template <class X>
        class __job { NOMEMALIGN
          public:
            using type = typename std::remove_reference<X>::type;
            using member_type = typename std::remove_reference<X>::type&;
            static X& functor (X& job) { return job; }
 
            template <class SingleFunctor>
              static SingleFunctor& get (X& /*f*/, SingleFunctor& functor) {
                return functor;
              }
        };
 
      template <class X>
        class __job <__Multi<X>> { NOMEMALIGN
          public:
            using type = typename std::remove_reference<X>::type;
            using member_type = typename std::remove_reference<X>::type;
            static X& functor (__Multi<X>& job) { return job.functor; }
 
            template <class SingleFunctor>
              static __Multi<SingleFunctor> get (__Multi<X>& f, SingleFunctor& functor) {
                return __Multi<SingleFunctor> (functor, f.num);
              }
        };
 
 
    }
 
 
 
 
 
 
     template <class T> class Queue { NOMEMALIGN
       public:
 
         Queue (const std::string& description = "unnamed", size_t buffer_size = MRTRIX_QUEUE_DEFAULT_CAPACITY) :
           buffer (new T* [buffer_size]),
           front (buffer),
           back (buffer),
           capacity (buffer_size),
           writer_count (0),
           reader_count (0),
           name (description) {
             assert (capacity > 0);
           }
 
         Queue (const Queue&) = delete;
         Queue (Queue&&) = default;
         Queue& operator= (const Queue&) = delete;
         Queue& operator= (Queue&&) = default;
 
 
         ~Queue () {
           delete [] buffer;
         }
 
 
         class Writer { NOMEMALIGN
           public:
 
             Writer (Queue<T>& queue) : Q (queue) {
               Q.register_writer();
             }
             Writer (const Writer& W) : Q (W.Q) {
               Q.register_writer();
             }
 
 
             class Item { NOMEMALIGN
               public:
 
                 Item (const Writer& writer) : Q (writer.Q), p (Q.get_item()) { }
                 ~Item () {
                   Q.unregister_writer();
                 }
 
                 using item_type = T;
 
                 FORCE_INLINE bool write () {
                   return Q.push (p);
                 }
                 FORCE_INLINE T& operator*() const throw ()   {
                   return *p;
                 }
                 FORCE_INLINE T* operator->() const throw ()  {
                   return p;
                 }
               private:
                 Queue<T>& Q;
                 T* p;
             };
 
             Item placeholder () const { return Item (*this); }
 
           private:
             Queue<T>& Q;
         };
 
 
 
         class Reader { NOMEMALIGN
           public:
 
             Reader (Queue<T>& queue) : Q (queue) {
               Q.register_reader();
             }
             Reader (const Reader& reader) : Q (reader.Q) {
               Q.register_reader();
             }
 
 
             class Item { NOMEMALIGN
               public:
 
                 Item (const Reader& reader) : Q (reader.Q), p (nullptr) { }
                 ~Item () {
                   Q.unregister_reader();
                 }
 
                 using item_type = T;
 
                 FORCE_INLINE bool read () {
                   return Q.pop (p);
                 }
                 FORCE_INLINE T* stash () throw () {
                   T* item = p;
                   p = nullptr;
                   return item;
                 }
                 FORCE_INLINE void recycle (T* item) const throw () {
                   Q.recycle (item);
                 }
                 FORCE_INLINE T& operator*() const throw ()   {
                   return *p;
                 }
                 FORCE_INLINE T* operator->() const throw ()  {
                   return p;
                 }
                 FORCE_INLINE bool operator! () const throw () {
                   return !p;
                 }
               private:
                 Queue<T>& Q;
                 T* p;
             };
 
             Item placeholder () const { return Item (*this); }
 
           private:
             Queue<T>& Q;
         };
 
         void status () {
           std::lock_guard<std::mutex> lock (mutex);
           std::cerr << "Thread::Queue \"" + name + "\": "
             << writer_count << " writer" << (writer_count > 1 ? "s" : "") << ", "
             << reader_count << " reader" << (reader_count > 1 ? "s" : "") << ", items waiting: " << size() << "\n";
         }
 
 
       private:
         std::mutex mutex;
         std::condition_variable more_data, more_space;
         T** buffer;
         T** front;
         T** back;
         size_t capacity;
         size_t writer_count, reader_count;
         std::stack<T*,vector<T*> > item_stack;
         vector<std::unique_ptr<T>> items;
         std::string name;
 
         void register_writer ()   {
           std::lock_guard<std::mutex> lock (mutex);
           ++writer_count;
         }
         void unregister_writer () {
           std::lock_guard<std::mutex> lock (mutex);
           assert (writer_count);
           --writer_count;
           if (!writer_count) {
             DEBUG ("no writers left on queue \"" + name + "\"");
             more_data.notify_all();
           }
         }
         void register_reader ()   {
           std::lock_guard<std::mutex> lock (mutex);
           ++reader_count;
         }
         void unregister_reader () {
           std::lock_guard<std::mutex> lock (mutex);
           assert (reader_count);
           --reader_count;
           if (!reader_count) {
             DEBUG ("no readers left on queue \"" + name + "\"");
             more_space.notify_all();
           }
         }
 
         FORCE_INLINE bool empty () const {
           return (front == back);
         }
         FORCE_INLINE bool full () const {
           return (inc (back) == front);
         }
         FORCE_INLINE size_t size () const {
           return ( (back < front ? back+capacity : back) - front);
         }
 
         FORCE_INLINE T* get_item () {
           std::lock_guard<std::mutex> lock (mutex);
           T* item (new T);
           items.push_back (std::unique_ptr<T> (item));
           return item;
         }
 
         FORCE_INLINE bool push (T*& item) {
           std::unique_lock<std::mutex> lock (mutex);
           more_space.wait (lock, [this]{ return !(full() && reader_count); });
           if (!reader_count) return false;
           *back = item;
           back = inc (back);
           if (item_stack.empty()) {
             item = new T;
             items.push_back (std::unique_ptr<T> (item));
           }
           else {
             item = item_stack.top();
             item_stack.pop();
           }
           more_data.notify_one();
           return true;
         }
 
         FORCE_INLINE bool pop (T*& item) {
           std::unique_lock<std::mutex> lock (mutex);
           if (item)
             item_stack.push (item);
           item = nullptr;
           more_data.wait (lock, [this]{ return !(empty() && writer_count); });
           if (empty() && !writer_count)
             return false;
           item = *front;
           front = inc (front);
           more_space.notify_one();
           return true;
         }
 
         FORCE_INLINE void recycle (T*& item) {
           std::unique_lock<std::mutex> lock (mutex);
           if (item)
             item_stack.push (item);
         }
 
         FORCE_INLINE T** inc (T** p) const {
           ++p;
           if (p >= buffer + capacity) p = buffer;
           return p;
         }
     };
 
 
 
 
 
 
     //* \cond skip
 
     namespace {
       /********************************************************************
        * convenience Functor classes for use in Thread::run_queue()
        ********************************************************************/
       template <class Item>
         struct __Batch { NOMEMALIGN
           __Batch (size_t number) : num (number) { }
           size_t num;
         };
 
 
 
       template <class Item> struct __batch_size { NOMEMALIGN
         __batch_size (const Item&) { }
         operator size_t () const { return 0; }
       };
       template <class Item> struct __batch_size <__Batch<Item>> { NOMEMALIGN
         __batch_size (const __Batch<Item>& item) : n (item.num) { }
         operator size_t () const { return n; }
         const size_t n;
       };
 
 
 
       template <class Item> struct Type { NOMEMALIGN
         using item = Item;
         using queue = Queue<Item>;
         using reader = typename queue::Reader;
         using writer = typename queue::Writer;
         using read_item = typename reader::Item;
         using write_item = typename writer::Item;
       };
 
       template <class Item> struct Type<__Batch<Item>> { NOMEMALIGN
         using item = Item;
         using queue = Queue<vector<Item>>;
         using reader = typename queue::Reader;
         using writer = typename queue::Writer;
         using read_item = typename reader::Item;
         using write_item = typename writer::Item;
       };
 
 
 
       template <class Item>
         struct FetchItem { NOMEMALIGN
           FetchItem (typename Type<Item>::reader& item) : in (item.placeholder()) { }
           bool read () { return in.read(); }
           Item& value () { return (*in); }
           typename Type<Item>::read_item in;
         };
 
       template <class Item>
         struct FetchItem<__Batch<Item>> { NOMEMALIGN
           FetchItem (typename Type<__Batch<Item>>::reader& in) : in (in.placeholder()), n (0) { }
           bool read () {
             if (!in)
               return in.read();
             ++n;
             if (n >= in->size()) {
               if (!in.read())
                 return false;
               n = 0;
             }
             return true;
           }
           Item& value () { return (*in)[n]; }
           typename Type<__Batch<Item>>::read_item in;
           size_t n;
         };
 
 
 
 
 
       template <class Item>
         struct StoreItem { NOMEMALIGN
           StoreItem (size_t, typename Type<Item>::writer& item) : out (item.placeholder()) { }
           bool write () { return out.write(); }
           Item& value () { return (*out); }
           bool flush () { return true; }
           typename Type<Item>::write_item out;
         };
 
       template <class Item>
         struct StoreItem<__Batch<Item>> { NOMEMALIGN
           StoreItem (size_t batch_size, typename Type<__Batch<Item>>::writer& item) :
             out (item.placeholder()), batch_size (batch_size), n(0) { out->resize (batch_size); }
           bool write () {
             ++n;
             if (n >= batch_size) {
               n = 0;
               if (!out.write())
                 return false;
               out->resize (batch_size);
             }
             return true;
           }
           Item& value () { return (*out)[n]; }
           void flush () { if (n) { out->resize (n); out.write(); } }
           typename Type<__Batch<Item>>::write_item out;
           const size_t batch_size;
           size_t n;
         };
 
 
 
 
       template <class Item, class Functor>
         struct __Source { MEMALIGN(__Source<Item,Functor>)
           using item_t = typename Type<Item>::item;
           using queue_t = typename Type<Item>::queue;
           using writer_t = typename Type<Item>::writer;
           using functor_t = typename __job<Functor>::member_type;
 
           writer_t writer;
           functor_t func;
           size_t batch_size;
 
           __Source (queue_t& queue, Functor& functor, const Item& item) :
             writer (queue),
             func (__job<Functor>::functor (functor)),
             batch_size (__batch_size<Item> (item)) { }
 
           void execute () {
             auto out = StoreItem<Item> (batch_size, writer);
             do {
               if (!func (out.value()))
                 break;
             } while (out.write());
             out.flush();
           }
         };
 
 
 
 
 
 
 
       template <class Item1, class Functor, class Item2>
         struct __Pipe { MEMALIGN(__Pipe<Item1,Functor,Item2>)
           using item1_t = typename Type<Item1>::item;
           using item2_t = typename Type<Item2>::item;
           using queue1_t = typename Type<Item1>::queue;
           using queue2_t = typename Type<Item2>::queue;
           using reader_t = typename Type<Item1>::reader;
           using writer_t = typename Type<Item2>::writer;
           using functor_t = typename __job<Functor>::member_type;
 
           reader_t reader;
           writer_t writer;
           functor_t func;
           const size_t batch_size;
 
           __Pipe (queue1_t& queue_in, Functor& functor, queue2_t& queue_out, const Item2& item2) :
             reader (queue_in),
             writer (queue_out),
             func (__job<Functor>::functor (functor)),
             batch_size (__batch_size<Item2> (item2)) { }
 
           void execute () {
             auto in = FetchItem<Item1> (reader);
             auto out = StoreItem<Item2> (batch_size, writer);
             while (in.read()) {
               if (func (in.value(), out.value())) {
                 if (!out.write())
                   break;
               }
             }
             out.flush();
           }
 
         };
 
 
 
 
 
 
       template <class Item, class Functor>
         struct __Sink { MEMALIGN(__Sink<Item,Functor>)
           using item_t = typename Type<Item>::item;
           using queue_t = typename Type<Item>::queue;
           using reader_t = typename Type<Item>::reader;
           using functor_t = typename __job<Functor>::member_type;
 
           reader_t reader;
           functor_t func;
 
           __Sink (queue_t& queue, Functor& functor) :
             reader (queue),
             func (__job<Functor>::functor (functor)) { }
 
           void execute () {
             auto in = FetchItem<Item> (reader);
             while (in.read()) {
               if (!func (in.value()))
                 return;
             }
           }
         };
 
 
 
 
 
     }
 
 
 
 
 
 
 
 
 
     template <class Item>
       inline __Batch<Item> batch (const Item&, size_t number = MRTRIX_QUEUE_DEFAULT_BATCH_SIZE)
       {
         return __Batch<Item> (number);
       }
 
 
 
 
 
 
 
       template <class Source, class Item, class Sink>
       inline void run_queue (
           Source&& source,
           const Item& item,
           Sink&& sink,
           size_t capacity = MRTRIX_QUEUE_DEFAULT_CAPACITY)
       {
         if (threads_to_execute() == 0) {
           typename Type<Item>::item item;
           while (__job<Source>::functor (source) (item))
             if (!__job<Sink>::functor (sink) (item))
               return;
           return;
         }
 
         typename Type<Item>::queue queue ("source->sink", capacity);
         __Source<Item,Source> source_functor (queue, source, item);
         __Sink<Item,Sink> sink_functor (queue, sink);
 
         auto t1 = run (__job<Source>::get (source, source_functor), "source");
         auto t2 = run (__job<Sink>::get (sink, sink_functor), "sink");
 
         t1.wait();
         t2.wait();
 
         check_app_exit_code();
       }
 
 
 
 
       template <class Source, class Item1, class Pipe, class Item2, class Sink>
         inline void run_queue (
             Source&& source,
             const Item1& item1,
             Pipe&& pipe,
             const Item2& item2,
             Sink&& sink,
             size_t capacity = MRTRIX_QUEUE_DEFAULT_CAPACITY)
         {
           if (threads_to_execute() == 0) {
             typename Type<Item1>::item item1;
             typename Type<Item2>::item item2;
             while (__job<Source>::functor (source) (item1)) {
               if (__job<Pipe>::functor (pipe) (item1, item2))
                 if (!__job<Sink>::functor (sink) (item2))
                   return;
             }
             return;
           }
 
 
           typename Type<Item1>::queue queue1 ("source->pipe", capacity);
           typename Type<Item2>::queue queue2 ("pipe->sink", capacity);
 
           __Source<Item1,Source> source_functor (queue1, source, item1);
           __Pipe<Item1,Pipe,Item2> pipe_functor (queue1, pipe, queue2, item2);
           __Sink<Item2,Sink> sink_functor (queue2, sink);
 
           auto t1 = run (__job<Source>::get (source, source_functor), "source");
           auto t2 = run (__job<Pipe>::get (pipe, pipe_functor), "pipe");
           auto t3 = run (__job<Sink>::get (sink, sink_functor), "sink");
 
           t1.wait();
           t2.wait();
           t3.wait();
 
           check_app_exit_code();
         }
 
 
 
 
       template <class Source, class Item1, class Pipe1, class Item2, class Pipe2, class Item3, class Sink>
         inline void run_queue (
             Source&& source,
             const Item1& item1,
             Pipe1&& pipe1,
             const Item2& item2,
             Pipe2&& pipe2,
             const Item3& item3,
             Sink&& sink,
             size_t capacity = MRTRIX_QUEUE_DEFAULT_CAPACITY)
         {
           if (threads_to_execute() == 0) {
             typename Type<Item1>::item item1;
             typename Type<Item2>::item item2;
             typename Type<Item3>::item item3;
             while (__job<Source>::functor (source) (item1)) {
               if (__job<Pipe1>::functor (pipe1) (item1, item2))
                 if (__job<Pipe2>::functor (pipe2) (item2, item3))
                   if (!__job<Sink>::functor (sink) (item3))
                     return;
             }
             return;
           }
 
 
           typename Type<Item1>::queue queue1 ("source->pipe", capacity);
           typename Type<Item2>::queue queue2 ("pipe->pipe", capacity);
           typename Type<Item3>::queue queue3 ("pipe->sink", capacity);
 
           __Source<Item1,Source> source_functor (queue1, source, item1);
           __Pipe<Item1,Pipe1,Item2> pipe1_functor (queue1, pipe1, queue2, item2);
           __Pipe<Item2,Pipe2,Item3> pipe2_functor (queue2, pipe2, queue3, item3);
           __Sink<Item3,Sink> sink_functor (queue3, sink);
 
           auto t1 = run (__job<Source>::get (source, source_functor), "source");
           auto t2 = run (__job<Pipe1>::get (pipe1, pipe1_functor), "pipe1");
           auto t3 = run (__job<Pipe2>::get (pipe2, pipe2_functor), "pipe2");
           auto t4 = run (__job<Sink>::get (sink, sink_functor), "sink");
 
           t1.wait();
           t2.wait();
           t3.wait();
           t4.wait();
 
           check_app_exit_code();
         }
 
 
  }
}
 
#endif