使用可替换阶段在 C++ 中对管道进行建模

Question

我正在尝试构建一个 C++ 数据结构来建模一个简单的 N 阶段过程，其中每个阶段都可以用不同的函数替换。一种方法是使用 OO 方法，并为每个阶段创建一个带有虚拟方法的抽象基类；例如：

class Pipeline {
protected:
  virtual void processA(const In& in, BType& B) = 0;
  virtual void processB(const BType& B, BType& C) = 0;
  virtual void processC(const CType& C, BType& D) = 0;
  virtual void processD(const DType& D, Out& out) = 0;
public:
  void process(const In& in, Out& out) {
    Btype B;
    processA(in, B);
    Ctype C;
    processB(B, C);
    Btype D;
    processC(C, D);
    processD(D,out);
  }
};

这种方法的问题是，如果 N 个阶段中的每一个都可以与 M 个进程互换，那么你就有 N*M 个可能的子类。

另一种思路是存储函数对象：

class Pipeline {
public:
  std::function<void(const In& in, BType& B)> processA;
  std::function<void(const In& B, CType& C)> processB;
  std::function<void(const In& C, DType& D)> processC;
  std::function<void(const In& D, Out& out)> processD;
  void process(const In& in, Out& out) {
    Btype B;
    processA(in, B);
    Ctype C;
    processB(B, C);
    Btype D;
    processC(C, D);
    processD(D,out);
  }
};

我在使用这种方法时遇到的问题是这些阶段并不是真正独立的，在某些情况下，我希望一个对象来存储有关多个阶段的信息。

有没有人为带有可更换部件的管道找到一个好的数据结构？奖金将能够允许每个阶段同时运行。

Answer 1

指向 std 函数对象的指针是个坏主意。如果需要，它们已经可以存储指针。

我建议图表。

sink 是消费者：

template<class...Ts>
struct sink : std::function<void(Ts...)> {
  using std::function<void(Ts...)>::function;
};

来源是接受消费者并满足消费者的东西：

template<class...Ts>
using source = sink<sink<Ts...>>;

进程是连接生产者和消费者的东西，可能会改变类型：

template<class In, class Out>
using process = sink< source<In>, sink<Out> >;

然后我们可以定义一个流水线操作：

template<class In, class Out>
sink<In> operator|( process< In, Out > a, sink< Out > b ){
  return [a,b]( In in ){
    a( [&in]( sink<In> s )mutable{ s(std::forward<In>(in)); }, b );
  };
}
template<class In, class Out>
source<Out> operator|( source< In > a, process< In, Out > b ){
  return [a,b]( sink<Out> out ){
    b( a, out );
  };
}

template<class In, class Mid, class Out>
process<In, Out> operator|( process<In, Mid> a, process<Mid, Out> b ){
  return [a,b]( source<In> in, sink<Out> out ){
    a( in, b|out ); // or b( in|a, out )
  };
}
template<class...Ts>
sink<> operator|( source<Ts...> a, sink<Ts...> b ){
  return[a,b]{ a(b); };
}

应该这样做。

我假设组件管道元素的状态复制起来很便宜，所以共享 ptr 或原始指针或类似的东西。

如果您想要并发，只需启动提供值队列并通过管道传递期货的进程。但我认为通常最好将元素附加在一起并使管道异步，而不是阶段。

将管道元素设置为 gsl span 之类的东西也很有用，它允许阶段具有固定缓冲区并以块的形式传递计算结果而无需分配。

一个让你开始的玩具过程：

process<char, char> to_upper = []( source<char> in, sink<char> out ){
  in( [&out]( char c ) { out( std::toupper(c) ); } );
};

以及来源：

source<char> hello_world = [ptr="hello world"]( sink<char> s ){
  for (auto it = ptr; *it; ++it ){ s(*it); }
};
sink<char> print = [](char c){std::cout<<c;};

int main(){
  auto prog = hello_world|to_upper|print;
  prog();
}

输出"HELLO WORLD"。

现场演示：https://ideone.com/MC4fDV

请注意，这是一个基于推送的管道。基于拉的管道是一种替代方案。推送管道允许更轻松的作业批处理；拉管道可以避免制作没人想要的数据。 Push让数据传播自然； pull 让数据收集变得自然。

协程也可以使这更自然。从某种意义上说，源是一个协程，当它在推送管道中调用接收器时会挂起。反之亦然。协程可以让 push/pull 都在相同的处理代码中工作。

Answer 2

为了使您的第一种方法更具可互换性，您可以将抽象基类拆分为多个基类，每个进程一个。然后基类可以由一个或多个对象实现。管道将保存对每个基类的引用、指针或智能指针：

struct ProcessA {
  virtual void processA(const In& in, BType& B) = 0;
  virtual ~ProcessA() = default;
};
struct ProcessB {
  virtual void processB(const BType& B, CType& C) = 0;
  virtual ~ProcessB() = default;
};
// ...

struct Pipeline {
  ProcessA* processA;
  ProcessB* processB;
  ProcessC* processC;
  ProcessD* processD;

  void process(const In& in, Out& out) {
    BType B;
    processA->processA(in, B);
    CType C;
    processB->processB(B, C);
    DType D;
    processC->processC(C, D);
    processD->processD(D,out);
  }
};

struct SimpleProcessor : ProcessA, ProcessB, ProcessC, ProcessD {
  void processA(const In& in, BType& B) override;
  void processB(const BType& B, CType& C) override;
  void processC(const CType& C, DType& D) override;
  void processD(const DType& D, Out& out) override;
};

int main() {
  SimpleProcessor processor;
  Pipeline pipeline;
  pipeline.processA = &processor;
  pipeline.processB = &processor; 
  pipeline.processC = &processor; 
  pipeline.processD = &processor; 
  In in;
  Out out;
  pipeline.process(in, out);
}

Live demo.

您的第二种方法也可以。您可以使用 lambda 之类的东西来调整单个对象以适应每个 std::function：

struct Pipeline {
  std::function<void(const In& in, BType& B)>   processA;
  std::function<void(const BType& B, CType& C)> processB;
  std::function<void(const CType& C, DType& D)> processC;
  std::function<void(const DType& D, Out& out)> processD;

  void process(const In& in, Out& out) {
    BType B;
    processA(in, B);
    CType C;
    processB(B, C);
    DType D;
    processC(C, D);
    processD(D,out);
  }
};

int main() {
  SimpleProcessor proc;
  Pipeline pipeline;
  pipeline.processA = [&proc](const In& in, BType& B){ return proc.processA(in, B); };
  pipeline.processB = [&proc](const BType& B, CType& C){ return proc.processB(B, C); }; 
  pipeline.processC = [&proc](const CType& C, DType& D){ return proc.processC(C, D); }; 
  pipeline.processD = [&proc](const DType& D, Out& out){ return proc.processD(D, out); }; 
  In in;
  Out out;
  pipeline.process(in, out);
}

Live demo.

是的，任何一种方法都允许您同时运行每个进程，但您的 BType、CType 和 DType 必须支持并发访问，以便可以同时写入和读取它们。例如并发队列。