<commit_before>#include <boost/thread/thread.hpp>

#include <util/ProgramOptions.h>
#include "InputSignals.h"
#include "ProcessNode.h"
#include "SimpleProcessNode.h"

logger::LogChannel simpleprocessnodelog("simpleprocessnodelog");

namespace pipeline {

util::ProgramOption optionNumThreads(
		util::_module           = "pipeline",
		util::_long_name        = "numThreads",
		util::_description_text = "Set the number of additional threads to parallelize independent processes.",
		util::_default_value    = 0);

template <typename LockingStrategy>
int SimpleProcessNode<LockingStrategy>::_numThreads = 0;

template <typename LockingStrategy>
boost::mutex SimpleProcessNode<LockingStrategy>::_threadCountMutex;

template <typename LockingStrategy>
SimpleProcessNode<LockingStrategy>::SimpleProcessNode() :
	_numInputs(0),
	_numMultiInputs(0),
	_numOutputs(0) {

	_numThreads = optionNumThreads;
}

template <typename LockingStrategy>
SimpleProcessNode<LockingStrategy>::~SimpleProcessNode() {

	foreach (signals::SlotsBase* slot, _multiInputUpdates)
		delete slot;
}

template <typename LockingStrategy>
void
SimpleProcessNode<LockingStrategy>::registerInput(InputBase& input, std::string name, InputType inputType) {

	LOG_ALL(simpleprocessnodelog) << "[" << typeName(this) << "] got a new input " << name << std::endl;

	ProcessNode::registerInput(input, name);

	int numInput = _numInputs;

	_inputDirty.push_back(true);

	_inputUpdate.addSlot();

	// create signal callbacks that store the number of the input with them
	boost::function<void(Modified&)> funOnModified = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputModified, this, _1, numInput);

	// register the callbacks and setup process node tracking
	input.registerBackwardCallback(funOnModified, this, signals::Transparent);

	if (inputType == Optional) {

		// Optional inputs are non-dirty by default (such that the output will
		// be computed, regardless of their presence).
		_inputDirty[numInput] = false;

		// optional inputs need not be present to update the output
		_inputRequired.push_back(false);

		// However, if an optional input is set, it has to be marked dirty,
		// except it was set to a shared pointer -- this is taken care of with
		// the following callbacks.
		boost::function<void(InputSetBase&)>                funOnInputSet                = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputSet,                this, _1, numInput);
		boost::function<void(InputSetToSharedPointerBase&)> funOnInputSetToSharedPointer = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputSetToSharedPointer, this, _1, numInput);

		input.registerBackwardCallback(funOnInputSet, this, signals::Transparent);
		input.registerBackwardCallback(funOnInputSetToSharedPointer, this, signals::Transparent);

	} else {

		// non-optional inputs have to be present before we can update the 
		// output
		_inputRequired.push_back(true);
	}

	// register the appropriate update signal for this input
	input.registerBackwardSlot(_inputUpdate[numInput]);

	_numInputs++;

	setOutputsDirty();
}

template <typename LockingStrategy>
void
SimpleProcessNode<LockingStrategy>::registerInputs(MultiInput& input, std::string name) {

	LOG_ALL(simpleprocessnodelog) << "[" << typeName(this) << "] got a new multi-input " << name << std::endl;

	ProcessNode::registerInputs(input, name);

	int numMultiInput = _numMultiInputs;

	_multiInputDirty.push_back(std::vector<bool>());
	_multiInputUpdates.push_back(new signals::Slots<Update>());

	// create signal callbacks that store the number of the multi-input with them
	boost::function<void(InputAddedBase&)>         funOnInputAdded    = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputAdded,         this, _1, numMultiInput);
	boost::function<void(InputsCleared&)>          funOnInputsCleared = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputsCleared,      this, _1, numMultiInput);
	boost::function<void(Modified&, unsigned int)> funOnModified      = boost::bind(&SimpleProcessNode<LockingStrategy>::onMultiInputModified, this, _1, _2, numMultiInput);

	// register the callbacks and setup process node tracking
	input.registerBackwardCallback(funOnInputAdded, this, signals::Transparent);
	input.registerBackwardCallback(funOnInputsCleared, this, signals::Transparent);
	input.registerBackwardCallbacks(funOnModified, this, signals::Transparent);<commit_msg>bugfix: SimpleProcessNode -- callback OnInputSetToSharedPointer was only registered for non-optional inputs, but is needed always<commit_after>#include <boost/thread/thread.hpp>

#include <util/ProgramOptions.h>
#include "InputSignals.h"
#include "ProcessNode.h"
#include "SimpleProcessNode.h"

logger::LogChannel simpleprocessnodelog("simpleprocessnodelog");

namespace pipeline {

util::ProgramOption optionNumThreads(
		util::_module           = "pipeline",
		util::_long_name        = "numThreads",
		util::_description_text = "Set the number of additional threads to parallelize independent processes.",
		util::_default_value    = 0);

template <typename LockingStrategy>
int SimpleProcessNode<LockingStrategy>::_numThreads = 0;

template <typename LockingStrategy>
boost::mutex SimpleProcessNode<LockingStrategy>::_threadCountMutex;

template <typename LockingStrategy>
SimpleProcessNode<LockingStrategy>::SimpleProcessNode() :
	_numInputs(0),
	_numMultiInputs(0),
	_numOutputs(0) {

	_numThreads = optionNumThreads;
}

template <typename LockingStrategy>
SimpleProcessNode<LockingStrategy>::~SimpleProcessNode() {

	foreach (signals::SlotsBase* slot, _multiInputUpdates)
		delete slot;
}

template <typename LockingStrategy>
void
SimpleProcessNode<LockingStrategy>::registerInput(InputBase& input, std::string name, InputType inputType) {

	LOG_ALL(simpleprocessnodelog) << "[" << typeName(this) << "] got a new input " << name << std::endl;

	ProcessNode::registerInput(input, name);

	int numInput = _numInputs;

	_inputDirty.push_back(true);

	_inputUpdate.addSlot();

	// create signal callbacks that store the number of the input with them
	boost::function<void(Modified&)> funOnModified = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputModified, this, _1, numInput);

	// register the callbacks and setup process node tracking
	input.registerBackwardCallback(funOnModified, this, signals::Transparent);

	if (inputType == Optional) {

		// Optional inputs are non-dirty by default (such that the output will
		// be computed, regardless of their presence).
		_inputDirty[numInput] = false;

		// optional inputs need not be present to update the output
		_inputRequired.push_back(false);

		// However, if an optional input is set, it has to be marked dirty,
		boost::function<void(InputSetBase&)> funOnInputSet = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputSet, this, _1, numInput);
		input.registerBackwardCallback(funOnInputSet, this, signals::Transparent);

	} else {

		// non-optional inputs have to be present before we can update the 
		// output
		_inputRequired.push_back(true);
	}

	// Regardless of the type of input -- if it was set to a shared pointer it 
	// has to be set dirty and Modified has to be sent.
	boost::function<void(InputSetToSharedPointerBase&)> funOnInputSetToSharedPointer = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputSetToSharedPointer, this, _1, numInput);
	input.registerBackwardCallback(funOnInputSetToSharedPointer, this, signals::Transparent);

	// register the appropriate update signal for this input
	input.registerBackwardSlot(_inputUpdate[numInput]);

	_numInputs++;

	setOutputsDirty();
}

template <typename LockingStrategy>
void
SimpleProcessNode<LockingStrategy>::registerInputs(MultiInput& input, std::string name) {

	LOG_ALL(simpleprocessnodelog) << "[" << typeName(this) << "] got a new multi-input " << name << std::endl;

	ProcessNode::registerInputs(input, name);

	int numMultiInput = _numMultiInputs;

	_multiInputDirty.push_back(std::vector<bool>());
	_multiInputUpdates.push_back(new signals::Slots<Update>());

	// create signal callbacks that store the number of the multi-input with them
	boost::function<void(InputAddedBase&)>         funOnInputAdded    = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputAdded,         this, _1, numMultiInput);
	boost::function<void(InputsCleared&)>          funOnInputsCleared = boost::bind(&SimpleProcessNode<LockingStrategy>::onInputsCleared,      this, _1, numMultiInput);
	boost::function<void(Modified&, unsigned int)> funOnModified      = boost::bind(&SimpleProcessNode<LockingStrategy>::onMultiInputModified, this, _1, _2, numMultiInput);

	// register the callbacks and setup process node tracking
	input.registerBackwardCallback(funOnInputAdded, this, signals::Transparent);
	input.registerBackwardCallback(funOnInputsCleared, this, signals::Transparent);
	input.registerBackwardCallbacks(funOnModified, this, signals::Transparent);
{'fraction_non_alphanumeric': 0.10919928025191183, 'fraction_numerical': 0.002699055330634278, 'mean_word_length': 6.247758761206194, 'pattern_counts': {'":': 0, '<': 99, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3178751', 'n_tokens_mistral': 2612, 'n_tokens_neox': 2539, 'n_words': 615}