embb/dataflow_cpp/include/embb/dataflow/network.h at master · bufferoverflow/embb

884 lines (769 loc) · 24.2 KB
 * Copyright (c) 2014, Siemens AG. All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 1. Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 * 2. 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.
#ifndef EMBB_DATAFLOW_NETWORK_H_
#define EMBB_DATAFLOW_NETWORK_H_
#if EMBB_DATAFLOW_TRACE_SIGNAL_HISTORY
#include <vector>
#include <embb/base/atomic.h>
#include <embb/base/thread.h>
#include <embb/dataflow/internal/select.h>
#include <embb/dataflow/internal/switch.h>
#include <embb/dataflow/internal/constant_source.h>
#include <embb/dataflow/internal/source.h>
#include <embb/dataflow/internal/process.h>
#include <embb/dataflow/internal/sink.h>
#include <embb/dataflow/internal/scheduler_sequential.h>
#include <embb/dataflow/internal/scheduler_mtapi.h>
namespace embb {
namespace dataflow {
#ifdef DOXYGEN
 * Represents a set of processes, that are connected by communication channels.
 * \tparam Slices Number of concurrently processed tokens.
 * \ingroup CPP_DATAFLOW
template <int Slices>
class Network {
   * Constructs an empty network.
  Network() {}
   * Input port class.
  template <typename Type>
  class In {
   * Output port class.
  template <typename Type>
  class Out {
     * Input port class that can be connected to this output port.
    typedef In<Type> InType;
     * Connects this output port to the input port \c input.
     * If the input port already was connected to a different
     * output an ErrorException is thrown.
     * \param input The input port to connect to.
    void Connect(InType & input);
     * Connects this output port to the input port \c input.
     * If the input port already was connected to a different
     * output an ErrorException is thrown.
     * \param input The input port to connect to.
    void operator >> (InType & input);
   * Provides the input port types for a process.
   * \tparam T1 Type of first port.
   * \tparam T2 Optional type of second port.
   * \tparam T3 Optional type of third port.
   * \tparam T4 Optional type of fourth port.
   * \tparam T5 Optional type of fifth port.
  template <typename T1, typename T2 = embb::base::internal::Nil,
    typename T3 = embb::base::internal::Nil,
    typename T4 = embb::base::internal::Nil,
    typename T5 = embb::base::internal::Nil>
  struct Inputs {
     * Type list used to derive input port types from Index.
     * \tparam Index The index of the input port type to query.
    template <int Index>
    struct Types {
       * Result of an input port type query.
       * T_Index is T1 if Index is 0, T2 if Index is 1 and so on.
      typedef In<T_Index> Result;
     * \returns Reference to input port at Index.
    template <int Index>
    typename Types<Index>::Result & Get();
   * Provides the output port types for a process.
   * \tparam T1 Type of first port.
   * \tparam T2 Optional type of second port.
   * \tparam T3 Optional type of third port.
   * \tparam T4 Optional type of fourth port.
   * \tparam T5 Optional type of fifth port.
  template <typename T1, typename T2 = embb::base::internal::Nil,
    typename T3 = embb::base::internal::Nil,
    typename T4 = embb::base::internal::Nil,
    typename T5 = embb::base::internal::Nil>
  struct Outputs {
     * Type list used to derive output port types from Index.
     * \tparam Index The index of the output port type to query.
    template <int Index>
    struct Types {
       * Result of an output port type query.
       * T_Index is T1 if Index is 0, T2 if Index is 1 and so on.
      typedef Out<T_Index> Result;
     * \returns Reference to output port at Index.
    template <int Index>
    typename Types<Index>::Result & Get();
   * Generic serial process template.
   * Implements a generic serial process in the network that may have one to
   * four input ports and one to four output ports but no more that five total
   * ports.
   * Tokens are processed in order.
   * \see Source, ParallelProcess, Sink, Switch, Select
   * \tparam Inputs Inputs of the process.
   * \tparam Outputs Outputs of the process.
  template <class Inputs, class Outputs>
  class SerialProcess {
     * Function type to use when processing tokens.
    typedef embb::base::Function<void, INPUT_TYPE_LIST, OUTPUT_TYPE_LIST>
      FunctionType;
     * Input port type list.
    typedef Inputs<INPUT_TYPE_LIST> InputsType;
     * Output port type list.
    typedef Outputs<OUTPUT_TYPE_LIST> OutputsType;
     * Constructs a SerialProcess with a user specified processing function.
     * \param function The Function to call to process a token.
    explicit SerialProcess(FunctionType function);
     * \returns \c true if the SerialProcess has any inputs, \c false
     *          otherwise.
    virtual bool HasInputs() const;
     * \returns Reference to a list of all input ports.
    InputsType & GetInputs();
     * \returns Input port at Index.
    template <int Index>
    typename InputsType::Types<Index>::Result & GetInput();
     * \returns \c true if the SerialProcess has any outputs, \c false
     *          otherwise.
    virtual bool HasOutputs() const;
     * \returns Reference to a list of all output ports.
    OutputsType & GetOutputs();
     * \returns Output port at Index.
    template <int Index>
    typename OutputsType::Types<Index>::Result & GetOutput();
   * Adds a new serial process to the network.
   * \param proc The process to add.
  template <class Inputs, class Outputs>
  void Add(SerialProcess<Inputs, Outputs> & proc);
  * Generic parallel process template.
  * Implements a generic parallel process in the network that may have one to
  * four input ports and one to four output ports but no more that five total
  * Tokens are processed as soon as all inputs for that token are complete.
  * \see Source, SerialProcess, Sink, Switch, Select
  * \tparam Inputs Inputs of the process.
  * \tparam Outputs Outputs of the process.
  template <class Inputs, class Outputs>
  class ParallelProcess {
    * Function type to use when processing tokens.
    typedef embb::base::Function<void, INPUT_TYPE_LIST, OUTPUT_TYPE_LIST>
      FunctionType;
    * Input port type list.
    typedef Inputs<INPUT_TYPE_LIST> InputsType;
    * Output port type list.
    typedef Outputs<OUTPUT_TYPE_LIST> OutputsType;
    * Constructs a ParallelProcess with a user specified processing function.
    * \param function The Function to call to process a token.
    explicit ParallelProcess(FunctionType function);
    * \returns \c true if the ParallelProcess has any inputs, \c false
    *          otherwise.
    virtual bool HasInputs() const;
    * \returns Reference to a list of all input ports.
    InputsType & GetInputs();
    * \returns Input port at Index.
    template <int Index>
    typename InputsType::Types<Index>::Result & GetInput();
    * \returns \c true if the ParallelProcess has any outputs, \c false
    *          otherwise.
    virtual bool HasOutputs() const;
    * \returns Reference to a list of all output ports.
    OutputsType & GetOutputs();
    * \returns Output port at Index.
    template <int Index>
    typename OutputsType::Types<Index>::Result & GetOutput();
  * Adds a new parallel process to the network.
  * \param proc The process to add.
  template <class Inputs, class Outputs>
  void Add(ParallelProcess<Inputs, Outputs> & proc);
   * Switch process template.
   * A switch has 2 inputs and 2 outputs. Input port 0 is of type boolean and
   * selects to which output port the value of input port 1 of type \c Type
   * is sent. If input port 0 is set to true the value goes to output port 0
   * and to output port 1 otherwise.
   * Tokens are processed as soon as all inputs for that token are complete.
   * \see Select
   * \tparam Type The type of input port 1 and output port 0 and 1.
  template<typename Type>
  class Switch {
     * Function type to use when processing tokens.
    typedef embb::base::Function<void, bool, Type, Type &> FunctionType;
     * Input port type list.
    typedef Inputs<bool, Type> InputsType;
     * Output port type list.
    typedef Outputs<Type> OutputsType;
     * \returns Always \c true.
    virtual bool HasInputs() const;
     * \returns Reference to a list of all input ports.
    InputsType & GetInputs();
     * \returns Input port at Index.
    template <int Index>
    typename InputsType::Types<Index>::Result & GetInput();
     * \returns Always \c true.
    virtual bool HasOutputs() const;
     * \returns Reference to a list of all output ports.
    OutputsType & GetOutputs();
     * \returns Output port at Index.
    template <int Index>
    typename OutputsType::Types<Index>::Result & GetOutput();
   * Adds a new switch process to the network.
   * \param sw The switch process to add.
  template <typename Type>
  void Add(Switch<Type> & sw);
   * Select process template.
   * A select has 3 inputs and 1 output. Input port 0 is of type boolean and
   * selects which of input port 1 or 2 (of type \c Type) is sent to output
   * port 0 (of type \c Type). If input port 0 is set to true the value of
   * input port 1 is selected, otherwise the value of input port 2 is taken.
   * Tokens are processed as soon as all inputs for that token are complete.
   * \see Switch
   * \tparam Type The type of input port 1 and 2 and output port 0.
  template<typename Type>
  class Select {
     * Function type to use when processing tokens.
    typedef embb::base::Function<void, bool, Type, Type, Type &> FunctionType;
     * Input port type list.
    typedef Inputs<bool, Type, Type> InputsType;
     * Output port type list.
    typedef Outputs<Type> OutputsType;
     * \returns Always \c true.
    virtual bool HasInputs() const;
     * \returns Reference to a list of all input ports.
    InputsType & GetInputs();
     * \returns Input port at Index.
    template <int Index>
    typename InputsType::Types<Index>::Result & GetInput();
     * \returns Always \c true.
    virtual bool HasOutputs() const;
     * \returns Reference to a list of all output ports.
    OutputsType & GetOutputs();
     * \returns Output port at Index.
    template <int Index>
    typename OutputsType::Types<Index>::Result & GetOutput();
   * Adds a new select process to the network.
   * \param sel The select process to add.
  template <typename Type>
  void Add(Select<Type> & sel);
   * Sink process template.
   * A sink marks the end of a particular processing chain. It can have one to
   * five input ports and no output ports.
   * Tokens are processed in order by the sink, regardless in which order they
   * arrive at the input ports.
   * \see Source, SerialProcess, ParallelProcess
   * \tparam I1 Type of first input port.
   * \tparam I2 Optional type of second input port.
   * \tparam I3 Optional type of third input port.
   * \tparam I4 Optional type of fourth input port.
   * \tparam I5 Optional type of fifth input port.
  template<typename I1, typename I2 = embb::base::internal::Nil,
    typename I3 = embb::base::internal::Nil,
    typename I4 = embb::base::internal::Nil,
    typename I5 = embb::base::internal::Nil>
  class Sink {
     * Function type to use when processing tokens.
    typedef embb::base::Function<void, INPUT_TYPE_LIST> FunctionType;
     * Input port type list.
    typedef Inputs<INPUT_TYPE_LIST> InputsType;
     * Constructs a Sink with a user specified processing function.
     * \param function The Function to call to process a token.
    explicit Sink(FunctionType function);
     * \returns Always \c true.
    virtual bool HasInputs() const;
     * \returns Reference to a list of all input ports.
    InputsType & GetInputs();
     * \returns Input port at Index.
    template <int Index>
    typename InputsType::Types<Index>::Result & GetInput();
     * \returns Always \c false.
    virtual bool HasOutputs() const;
   * Adds a new sink process to the network.
   * \param sink The sink process to add.
  template<typename I1, typename I2, typename I3, typename I4, typename I5>
  void Add(Sink<I1, I2, I3, I4, I5> & sink);
   * Source process template.
   * A source marks the start of a processing chain. It can have one to five
   * output ports and no input ports.
   * Tokens are emitted in order by the source.
   * \see SerialProcess, ParallelProcess, Sink
   * \tparam O1 Type of first output port.
   * \tparam O2 Optional type of second output port.
   * \tparam O3 Optional type of third output port.
   * \tparam O4 Optional type of fourth output port.
   * \tparam O5 Optional type of fifth output port.
  template<typename O1, typename O2 = embb::base::internal::Nil,
    typename O3 = embb::base::internal::Nil,
    typename O4 = embb::base::internal::Nil,
    typename O5 = embb::base::internal::Nil>
  class Source {
     * Function type to use when processing tokens.
    typedef embb::base::Function<void, OUTPUT_TYPE_LIST> FunctionType;
     * Output port type list.
    typedef Outputs<OUTPUT_TYPE_LIST> OutputsType;
     * Constructs a Source with a user specified processing function.
     * \param function The Function to call to emit a token.
    explicit Source(FunctionType function);
     * \returns Always \c false.
    virtual bool HasInputs() const;
     * \returns Always \c true.
    virtual bool HasOutputs() const;
     * \returns Reference to a list of all output ports.
    OutputsType & GetOutputs();
     * \returns Output port at INDEX.
    template <int Index>
    typename OutputsType::Types<Index>::Result & GetOutput();
   * Adds a new source process to the network.
   * \param source The source process to add.
  template<typename O1, typename O2, typename O3, typename O4, typename O5>
  void Add(Source<O1, O2, O3, O4, O5> & source);
   * Constant source process template.
   * A constant source has one output port and emits a constant value given
   * at construction time for each token.
   * \tparam Type The type of output port 0.
  template<typename Type>
  class ConstantSource {
     * Output port type list.
    typedef Outputs<OUTPUT_TYPE_LIST> OutputsType;
     * Constructs a ConstantSource with a value to emit on each token.
     * \param value The value to emit.
    explicit ConstantSource(Type value);
     * \returns Always \c false.
    virtual bool HasInputs() const;
     * \returns Always \c true.
    virtual bool HasOutputs() const;
     * \returns Reference to a list of all output ports.
    OutputsType & GetOutputs();
     * \returns Output port at Index.
    template <int Index>
    typename OutputsType::Types<Index>::Result & GetOutput();
   * Adds a new constant source process to the network.
   * \param source The constant source process to add.
  template<typename Type>
  void Add(ConstantSource<Type> & source);
   * Executes the network for at most \c elements tokens.
   * \param elements Maximum number of tokens to process.
  void operator () (int elements);
template <int Slices>
class Network : public internal::ClockListener {
  Network() {}
  template <typename T1, typename T2 = embb::base::internal::Nil,
    typename T3 = embb::base::internal::Nil,
    typename T4 = embb::base::internal::Nil,
    typename T5 = embb::base::internal::Nil>
  struct Inputs {
    typedef internal::Inputs<Slices, T1, T2, T3, T4, T5> Type;
  template <typename T1, typename T2 = embb::base::internal::Nil,
    typename T3 = embb::base::internal::Nil,
    typename T4 = embb::base::internal::Nil,
    typename T5 = embb::base::internal::Nil>
  struct Outputs {
    typedef internal::Outputs<Slices, T1, T2, T3, T4, T5> Type;
  template <class Inputs, class Outputs> class SerialProcess;
  template <
    typename I1, typename I2, typename I3, typename I4, typename I5,
    typename O1, typename O2, typename O3, typename O4, typename O5>
  class SerialProcess< internal::Inputs<Slices, I1, I2, I3, I4, I5>,
    internal::Outputs<Slices, O1, O2, O3, O4, O5> >
    : public internal::Process< Slices, true,
        internal::Inputs<Slices, I1, I2, I3, I4, I5>,
        internal::Outputs<Slices, O1, O2, O3, O4, O5> > {
    typedef typename internal::Process< Slices, true,
      internal::Inputs<Slices, I1, I2, I3, I4, I5>,
      internal::Outputs<Slices, O1, O2, O3, O4, O5> >::FunctionType
        FunctionType;
    explicit SerialProcess(FunctionType function)
      : internal::Process< Slices, true,
          internal::Inputs<Slices, I1, I2, I3, I4, I5>,
          internal::Outputs<Slices, O1, O2, O3, O4, O5> >(function) {
      //empty
  template <class Inputs, class Outputs>
  void Add(SerialProcess<Inputs, Outputs> & proc) {
    processes_.push_back(&proc);
  template <class Inputs, class Outputs> class ParallelProcess;
  template <
    typename I1, typename I2, typename I3, typename I4, typename I5,
    typename O1, typename O2, typename O3, typename O4, typename O5>
  class ParallelProcess< internal::Inputs<Slices, I1, I2, I3, I4, I5>,
    internal::Outputs<Slices, O1, O2, O3, O4, O5> >
    : public internal::Process< Slices, false,
        internal::Inputs<Slices, I1, I2, I3, I4, I5>,
        internal::Outputs<Slices, O1, O2, O3, O4, O5> >{
    typedef typename internal::Process< Slices, false,
      internal::Inputs<Slices, I1, I2, I3, I4, I5>,
      internal::Outputs<Slices, O1, O2, O3, O4, O5> >::FunctionType
        FunctionType;
    explicit ParallelProcess(FunctionType function)
      : internal::Process< Slices, false,
          internal::Inputs<Slices, I1, I2, I3, I4, I5>,
          internal::Outputs<Slices, O1, O2, O3, O4, O5> >(function) {
      //empty
  template <class Inputs, class Outputs>
  void Add(ParallelProcess<Inputs, Outputs> & proc) {
    processes_.push_back(&proc);
  template<typename Type>
  class Switch : public internal::Switch<Slices, Type> {
  template <typename Type>
  void Add(Switch<Type> & sw) {
    processes_.push_back(&sw);
  template<typename Type>
  class Select : public internal::Select<Slices, Type> {
  template <typename Type>
  void Add(Select<Type> & sel) {
    processes_.push_back(&sel);
  template<typename I1, typename I2 = embb::base::internal::Nil,
    typename I3 = embb::base::internal::Nil,
    typename I4 = embb::base::internal::Nil,
    typename I5 = embb::base::internal::Nil>
  class Sink : public internal::Sink<Slices,
    internal::Inputs<Slices, I1, I2, I3, I4, I5> > {
    typedef typename internal::Sink<Slices,
      internal::Inputs<Slices, I1, I2, I3, I4, I5> >::FunctionType FunctionType;
    explicit Sink(FunctionType function)
      : internal::Sink<Slices,
          internal::Inputs<Slices, I1, I2, I3, I4, I5> >(function) {
      //empty
  template<typename I1, typename I2, typename I3, typename I4, typename I5>
  void Add(Sink<I1, I2, I3, I4, I5> & sink) {
    sink.SetListener(this);
    sinks_.push_back(&sink);
  template<typename O1, typename O2 = embb::base::internal::Nil,
    typename O3 = embb::base::internal::Nil,
    typename O4 = embb::base::internal::Nil,
    typename O5 = embb::base::internal::Nil>
  class Source : public internal::Source<Slices,
    internal::Outputs<Slices, O1, O2, O3, O4, O5> > {
    typedef typename internal::Source<Slices,
      internal::Outputs<Slices, O1, O2, O3, O4, O5> >::FunctionType
        FunctionType;
    explicit Source(FunctionType function)
      : internal::Source<Slices,
          internal::Outputs<Slices, O1, O2, O3, O4, O5> >(function) {
      //empty
  template<typename O1, typename O2, typename O3, typename O4, typename O5>
  void Add(Source<O1, O2, O3, O4, O5> & source) {
    sources_.push_back(&source);
  template<typename Type>
  class ConstantSource : public internal::ConstantSource<Slices, Type> {
    explicit ConstantSource(Type value)
      : internal::ConstantSource<Slices, Type>(value) {
      //empty
  template<typename Type>
  void Add(ConstantSource<Type> & source) {
    sources_.push_back(&source);
  void operator () (int elements) {
    internal::SchedulerSequential sched_seq;
    internal::SchedulerMTAPI<Slices> sched_mtapi;
    internal::Scheduler * sched = &sched_mtapi;
    for (size_t it = 0; it < sources_.size(); it++)
      sources_[it]->SetScheduler(sched);
    for (size_t it = 0; it < processes_.size(); it++)
      processes_[it]->SetScheduler(sched);
    for (size_t it = 0; it < sinks_.size(); it++)
      sinks_[it]->SetScheduler(sched);
    for (int ii = 0; ii < Slices; ii++) sink_count_[ii] = 0;
    for (int clock = 0; clock < elements; clock++) {
      const int idx = clock % Slices;
      while (sink_count_[idx] > 0) embb::base::Thread::CurrentYield();
      sched->WaitForSlice(idx);
      SpawnClock(clock);
    for (int ii = 0; ii < Slices; ii++) {
      while (sink_count_[ii] > 0) embb::base::Thread::CurrentYield();
      sched->WaitForSlice(ii);
   * Internal.
   * \internal
   * Gets called when a token has reached all sinks and frees up the
   * corresponding slot, thus allowing a new token to be emitted.
  virtual void OnClock(int clock) {
    const int idx = clock % Slices;
    const int cnt = --sink_count_[idx];
    if (cnt < 0)
      EMBB_THROW(embb::base::ErrorException,
        "More sinks than expected signaled reception of given clock.")
  std::vector<internal::Node*> processes_;
  std::vector<internal::Node*> sources_;
  std::vector<internal::Node*> sinks_;
  embb::base::Atomic<int> sink_count_[Slices];
#if EMBB_DATAFLOW_TRACE_SIGNAL_HISTORY
  std::vector<int> spawn_history_[Slices];
  void SpawnClock(int clock) {
    const int idx = clock % Slices;
#if EMBB_DATAFLOW_TRACE_SIGNAL_HISTORY
    spawn_history_[idx].push_back(clock);
    sink_count_[idx] = static_cast<int>(sinks_.size());
    for (size_t kk = 0; kk < sources_.size(); kk++) {
      sources_[kk]->Start(clock);
#endif // DOXYGEN
} // namespace dataflow
} // namespace embb
#endif // EMBB_DATAFLOW_NETWORK_H_
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

network.h

Latest commit

History

network.h

File metadata and controls
