Question: How to get dynamic parallelism to work

[civspan]

Hi!

I have been able to get many different static graphs to work with static split/joins, but cannot figure out how to create a dynamically parallelized graph.

In the simplest use case (see code example below) there are three kernels: producer, processor and consumer. The producer and consumer extend raft:parallel_k, and the processor uses the CLONE() macro and is a standard raft::kernel, with a single input and output port. The producer and consumer use addPortTo<>() to be able to dynamically create output/input ports respectively. I have also tried using the raft::order::out parameter. However, I only run with a single processor kernel with this approach, and can't find any examples where this approach is illustrated. Can someone please tell me what I'm missing?

Best,
Henrik Möller

#include <raft>
#include <raftio>
#include "matrix.h"

class producer : public raft::parallel_k
{
public:
    producer() : raft::parallel_k()
    {	
        addPortTo< Matrix >( output );
    }
    virtual ~producer() = default;
    virtual raft::kstatus run()
    {
        for( auto &port : output )
        {
            Matrix matrix = data_input_stream[input++]; //Read data from buffer
            auto out( port.template allocate_s< Matrix >() );
            (*out) = std::move( matrix );
        }            
        return( raft::proceed );
    }
};

class processor : public raft::kernel
{
public:
    processor() : raft::kernel()
    {
        input.addPort< Matrix >( "in" );
        output.addPort< Matrix >( "out" );
    }

    processor(const processor &proc)
    {
        input.addPort< Matrix >( "in" );
        output.addPort< Matrix >( "out" );
    }
    virtual ~processor() = default;
    virtual raft::kstatus run()
    {
        Matrix mat = input[ "in" ].peek< Matrix >();
        Matrix result;

        result.matrix_mul(mat, mat);	//Perform matrix multiplication defined in external Matrix class

        input[ "in" ].recycle();
        
        auto out( output["out"].template allocate_s< Matrix >() );
        (*out) = std::move( result );
        return( raft::proceed );
        
    }
    CLONE();
};

class consumer : public raft::parallel_k
{
public:
    consumer() : raft::parallel_k()
	{
	    addPortTo< Matrix>( input );
	}
    virtual ~consumer() = default;
    virtual raft::kstatus run()
    {
        for( auto &port : input )
        {
   	    Matrix mat = port.peek< Matrix >(); //Do something here..
            port.recycle();
        }
        return( raft::proceed );
    }
};

int  main()
{
    producer    a;
    processor   b;
    consumer    c;

    raft::map m;
    m += a >> raft::order::out >> b;
    m += b >> raft::order::out >> c;
    m.exe();
    return( EXIT_SUCCESS );
}

[vectronic]

Hello,

Did you get any further with this?

I'm fairly ignorant on usage of the framework at this stage, but looking at the documentation here:

https://github.com/RaftLib/RaftLib/wiki/Kernel-Linkage-Modifiers

instead of:

m += a >> raft::order::out >> b;
m += b >> raft::order::out >> c;

have you tried the following:

m += a >> raft::split >> raft::order::out >> b
m += b >> raft::order::out >> raft::reduce >> c;

---

When I look at the the docs in these two places:

https://github.com/RaftLib/RaftLib/wiki/Kernel-Linkage-Modifiers
https://github.com/RaftLib/RaftLib/wiki/Building-Kernels

I am still a bit confused as to how extending raft::parallel_k interacts with the raft::split, raft::reduce and raft::order::out stream modifiers.