WorkflowTransformationExamples/src/11.elimination01/elimination.cpp

#include "elimination.h"

#include <algorithm>
#include <iostream>
#include <thread>
#include <chrono>
#include <boost/program_options.hpp>



WFR_INPUT_HOOK(toplevelInputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, toplevel_in);
	WFR_INPUT_HOOK_EPILOG;
}


WFR_OUTPUT_HOOK(toplevelOutputFunc) {
	WFR_OUTPUT_HOOK_PROLOG;
	WFR_SELF_OUTBUFFER_INIT(sout, toplevel_out);
	WFR_PORT_INIT_COLLECTION(sout->output1);
	WFR_PORT_INIT_COLLECTION(sout->output2);
	WFR_CHILD_OUTBUFFER_GET(a1out, for01_out, a1, 2, 0);
	if (a1out != nullptr) {
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sout->output1, a1out->output1));
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sout->output2, a1out->output2));
	}
	WFR_SELF_CHECK_AREALLOUTPUTSAVAILABLE(sout->output1.isReady() && sout->output2.isReady());
	WFR_OUTPUT_HOOK_EPILOG;
}


WFR_INPUT_HOOK(for01InputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, for01_in);
	WFR_PARENT_INBUFFER_GET(a1in, toplevel_in, a1);
	if (a1in != nullptr) {
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sin->loopCounter_to, a1in->iterCount));
	}
	WFR_INPUT_HOOK_EPILOG;
}


WFR_OUTPUT_HOOK(for01OutputFunc) {
	WFR_OUTPUT_HOOK_PROLOG;
	WFR_SELF_OUTBUFFER_INIT(sout, for01_out);
	WFR_PORT_INIT_COLLECTION(sout->output1);
	WFR_PORT_INIT_COLLECTION(sout->output2);

	WFR_COLLECT_OUTVALUES_FROM_CHILDREN(sout->output1, task03_out, output, 5);
	WFR_LOOP_FINALIZE_COLLECTION(sout, output1);
	WFR_COLLECT_OUTVALUES_FROM_CHILDREN(sout->output2, task04_out, output, 7);
	WFR_LOOP_FINALIZE_COLLECTION(sout, output2);

	WFR_SELF_CHECK_AREALLOUTPUTSAVAILABLE(sout->output1.isReady() && sout->output2.isReady());
	WFR_OUTPUT_HOOK_EPILOG;
}


WFR_FORCOUNTER_HOOK(for01CounterFunc) {
	WFR_FORCOUNTER_HOOK_PROLOG;
	WFR_SELF_INBUFFER_GET(sin, for01_in);
	if(sin != nullptr && sin->loopCounter_to.isReady()) {
		WFR_SELF_SETFORLOOPCOUNTERVALUES(1, 1, sin->loopCounter_to.getValue())
		WFR_LOOP_FINALIZE();
		WFR_SELF_CHECK_ISREADYFOREXECUTION(true);
	}
	WFR_FORCOUNTER_HOOK_EPILOG;
}


WFR_INPUT_HOOK(task01InputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, task01_in);
	WFR_PARENT_INBUFFER_GET(a1in, for01_in, a1);
	if (a1in != nullptr) {
		WFR_FORLOOP_GET_COUNTER_VALUE(a1, sin->input);
	}
	WFR_SELF_CHECK_ISREADYFOREXECUTION(sin->input.isReady());
	WFR_INPUT_HOOK_EPILOG;
}


WFR_EXEC_HOOK(task01ExecFunc_0) {
	WFR_EXEC_HOOK_PROLOG;
	WFR_SELF_INBUFFER_GET(sin, task01_in);
	WFR_SELF_OUTBUFFER_INIT(sout, task01_out);

	std::this_thread::sleep_for(std::chrono::seconds(1));
	sout->output.setValue(sin->input.getValue());
	
	WFR_EXEC_HOOK_EPILOG;
}


WFR_INPUT_HOOK(task02_1InputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, task02_1_in);
	WFR_PRED_OUTBUFFER_GET(a1in, task01_out, a1, 0);
	if (a1in != nullptr) {
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sin->input, a1in->output));
	}
	WFR_SELF_CHECK_ISREADYFOREXECUTION(sin->input.isReady());
	WFR_INPUT_HOOK_EPILOG;
}


WFR_EXEC_HOOK(task02_1ExecFunc_0) {
	WFR_EXEC_HOOK_PROLOG;
	WFR_SELF_INBUFFER_GET(sin, task02_1_in);
	WFR_SELF_OUTBUFFER_INIT(sout, task02_1_out);

	std::this_thread::sleep_for(std::chrono::seconds(1));
	sout->output.setValue(sin->input.getValue());
	
	WFR_EXEC_HOOK_EPILOG;
}


WFR_INPUT_HOOK(task03InputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, task03_in);
	WFR_PRED_OUTBUFFER_GET(a1in, task02_1_out, a1, 0);
	if (a1in != nullptr) {
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sin->input, a1in->output));
	}
	WFR_SELF_CHECK_ISREADYFOREXECUTION(sin->input.isReady());
	WFR_INPUT_HOOK_EPILOG;
}


WFR_EXEC_HOOK(task03ExecFunc_0) {
	WFR_EXEC_HOOK_PROLOG;
	WFR_SELF_INBUFFER_GET(sin, task03_in);
	WFR_SELF_OUTBUFFER_INIT(sout, task03_out);

	std::this_thread::sleep_for(std::chrono::seconds(1));
	sout->output.setValue(sin->input.getValue());
	
	WFR_EXEC_HOOK_EPILOG;
}


WFR_INPUT_HOOK(task02_2InputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, task02_2_in);
	WFR_PRED_OUTBUFFER_GET(a1in, task01_out, a1, 0);
	if (a1in != nullptr) {
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sin->input, a1in->output));
	}
	WFR_SELF_CHECK_ISREADYFOREXECUTION(sin->input.isReady());
	WFR_INPUT_HOOK_EPILOG;
}


WFR_EXEC_HOOK(task02_2ExecFunc_0) {
	WFR_EXEC_HOOK_PROLOG;
	WFR_SELF_INBUFFER_GET(sin, task02_2_in);
	WFR_SELF_OUTBUFFER_INIT(sout, task02_2_out);

	std::this_thread::sleep_for(std::chrono::seconds(1));
	sout->output.setValue(sin->input.getValue());
	
	WFR_EXEC_HOOK_EPILOG;
}


WFR_INPUT_HOOK(task04InputFunc) {
	WFR_INPUT_HOOK_PROLOG;
	WFR_SELF_INBUFFER_INIT(sin, task04_in);
	WFR_PRED_OUTBUFFER_GET(a1in, task02_2_out, a1, 0);
	if (a1in != nullptr) {
		WFR_HOOK_SET_RETVAL( portbuffer_copy(sin->input, a1in->output));
	}
	WFR_SELF_CHECK_ISREADYFOREXECUTION(sin->input.isReady());
	WFR_INPUT_HOOK_EPILOG;
}


WFR_EXEC_HOOK(task04ExecFunc_0) {
	WFR_EXEC_HOOK_PROLOG;
	WFR_SELF_INBUFFER_GET(sin, task04_in);
	WFR_SELF_OUTBUFFER_INIT(sout, task04_out);

	std::this_thread::sleep_for(std::chrono::seconds(1));
	sout->output.setValue(sin->input.getValue());
	
	WFR_EXEC_HOOK_EPILOG;
}

WFR_ACTIVITY_IMPLEMENTATIONS( task01Implementations, "task01Implementations", \
	wfr::wfs::AtomicActivityImplementation::PThread(task01ExecFunc_0, "task01ExecFunc_0" )
);

WFR_ACTIVITY_IMPLEMENTATIONS( task02_1Implementations, "task02_1Implementations", \
	wfr::wfs::AtomicActivityImplementation::PThread(task02_1ExecFunc_0, "task02_1ExecFunc_0" )
);

WFR_ACTIVITY_IMPLEMENTATIONS( task03Implementations, "task03Implementations", \
	wfr::wfs::AtomicActivityImplementation::PThread(task03ExecFunc_0, "task03ExecFunc_0" )
);

WFR_ACTIVITY_IMPLEMENTATIONS( task02_2Implementations, "task02_2Implementations", \
	wfr::wfs::AtomicActivityImplementation::PThread(task02_2ExecFunc_0, "task02_2ExecFunc_0" )
);

WFR_ACTIVITY_IMPLEMENTATIONS( task04Implementations, "task04Implementations", \
	wfr::wfs::AtomicActivityImplementation::PThread(task04ExecFunc_0, "task04ExecFunc_0" )
);


WFR_WORKFLOW(workflow, "generatedWorkflow", wfActivityTable, 1, \
	wfr::wfs::ActivityTemplate::container(1, "toplevel", 0, 0, {}, {}, nullptr, toplevelOutputFunc), \
	wfr::wfs::ActivityTemplate::forLoop(2, "for01", 1, 0, {}, {}, for01InputFunc, for01OutputFunc, for01CounterFunc, nullptr, true), \
	wfr::wfs::ActivityTemplate::atomic(3, "task01", 2, 0, {6, 4}, {}, task01InputFunc, task01Implementations), \
	wfr::wfs::ActivityTemplate::atomic(4, "task02_1", 2, 0, {5}, {3}, task02_1InputFunc, task02_1Implementations), \
	wfr::wfs::ActivityTemplate::atomic(5, "task03", 2, 0, {}, {4}, task03InputFunc, task03Implementations), \
	wfr::wfs::ActivityTemplate::atomic(6, "task02_2", 2, 0, {7}, {3}, task02_2InputFunc, task02_2Implementations), \
	wfr::wfs::ActivityTemplate::atomic(7, "task04", 2, 0, {}, {6}, task04InputFunc, task04Implementations), \
);


int wfstart(toplevel_in &wfInput, toplevel_out &wfOutput, int argc, char** argv) {
	wfr::WorkflowRuntime::init(argc, argv);
	workflow->execute(&wfInput, &wfOutput);
	wfr::WorkflowRuntime::finish();
	return 0;
}



int main(int argc, char** argv) {

	toplevel_in wfInput;
	// Fill workflow input struct
	long iterCount = 8;
	bool isHelp = false;
    boost::program_options::options_description opts("Raincloud Options");
    opts.add_options()
    	    ("help", "Print help message.")
    		("iterCount", boost::program_options::value<long>(), "NGroup parameter.");
    boost::program_options::variables_map vm;
    auto parsed = boost::program_options::command_line_parser(argc, argv).options(opts).allow_unregistered().run();
    boost::program_options::store(parsed, vm);
    boost::program_options::notify(vm);
    if (vm.count("help")) {
    	std::cerr << opts << std::endl;
    	isHelp = true;
    } else {
		if (vm.count("iterCount")) {
			iterCount = vm["iterCount"].as<long>();
		}
		std::cerr << "iterCount = " << iterCount << std::endl;
    }
	wfInput.iterCount.setValue(iterCount);

	toplevel_out wfOutput;

	wfstart(wfInput, wfOutput, argc, argv);
	// Do something with workflow output
	

	return 0;
}