#include <NonRunnableDataSource.hpp>

Collaboration diagram for NES::Testing::NonRunnableDataSource:

Public Member Functions
	NonRunnableDataSource (const SchemaPtr &schema, const Runtime::BufferManagerPtr &bufferManager, const Runtime::QueryManagerPtr &queryManager, uint64_t numbersOfBufferToProduce, uint64_t gatheringInterval, OperatorId operatorId, OriginId originId, StatisticId statisticId, size_t numSourceLocalBuffers, const std::vector< Runtime::Execution::SuccessorExecutablePipeline > &successors, NES::Runtime::QueryTerminationType terminationType=NES::Runtime::QueryTerminationType::HardStop)

void	runningRoutine () override
	running routine while source is active More...

bool	stop (Runtime::QueryTerminationType termination) override
	method to stop the source. 1.) check if bool running is false, if false return, if not stop source 2.) stop thread by join More...

bool	performSoftStop () override
	This method will mark the running as false this will result in 1. the interruption of data gathering loop and 2. initiation of soft stop routine. More...

Runtime::MemoryLayouts::TestTupleBuffer	getBuffer ()

void	emitBuffer (Runtime::MemoryLayouts::TestTupleBuffer &buffer, bool addBufferMetaData=true)

void	emitBuffer (Runtime::TupleBuffer &buffer, bool addBufferMetaData=true)

Public Member Functions inherited from NES::DefaultSource
	DefaultSource (SchemaPtr schema, Runtime::BufferManagerPtr bufferManager, Runtime::QueryManagerPtr queryManager, uint64_t numberOfBufferToProduce, uint64_t gatheringInterval, OperatorId operatorId, OriginId originId, StatisticId statisticId, size_t numSourceLocalBuffers, std::vector< Runtime::Execution::SuccessorExecutablePipeline > successors={}, const std::string &physicalSourceName=std::string("defaultPhysicalSourceName"))

SourceType	getType () const override
	get source Type More...

std::optional< Runtime::TupleBuffer >	receiveData () override
	override function to create one buffer More...

std::vector< Schema::MemoryLayoutType >	getSupportedLayouts () override
	This method returns all supported layouts. More...

Public Member Functions inherited from NES::GeneratorSource
	GeneratorSource (SchemaPtr schema, Runtime::BufferManagerPtr bufferManager, Runtime::QueryManagerPtr queryManager, uint64_t numberOfBufferToProduce, OperatorId operatorId, OriginId originId, StatisticId statisticId, size_t numSourceLocalBuffers, GatheringMode gatheringMode, std::vector< Runtime::Execution::SuccessorExecutablePipeline > successors, const std::string &physicalSourceName=std::string("defaultPhysicalStreamName"))
	constructor to create a generator source More...

std::string	toString () const override
	override the toString method for the generator source More...

SourceType	getType () const override
	get source Type More...

Public Member Functions inherited from NES::DataSource
	DataSource (SchemaPtr schema, Runtime::BufferManagerPtr bufferManager, Runtime::QueryManagerPtr queryManager, OperatorId operatorId, OriginId originId, StatisticId statisticId, size_t numSourceLocalBuffers, GatheringMode gatheringMode, const std::string &physicalSourceName, bool persistentSource, std::vector< Runtime::Execution::SuccessorExecutablePipeline > executableSuccessors=std::vector< Runtime::Execution::SuccessorExecutablePipeline >(), uint64_t sourceAffinity=std::numeric_limits< uint64_t >::max(), uint64_t taskQueueId=0)
	public constructor for data source @Note the number of buffers to process is set to UINT64_MAX and the value is needed by some test to produce a deterministic behavior More...

	DataSource ()=delete

virtual void	open ()
	This methods initializes thread-local state. For instance, it creates the local buffer pool and is necessary because we cannot do it in the constructor. More...

virtual void	close ()
	This method cleans up thread-local state for the source. More...

virtual bool	start ()
	method to start the source. 1.) check if bool running is true, if true return if not start source 2.) start new thread with runningRoutine More...

virtual void	createOrLoadPersistedProperties ()
	This method defines the logic to load the properties persisted during the previous run of the data source. More...

virtual void	storePersistedProperties ()
	This method stores all properties that need to be persisted for future retrieval during createOrLoadPersistedProperties call. More...

virtual void	clearPersistedProperties ()
	This method clears the persisted properties of the data source on had stop. More...

SchemaPtr	getSchema () const
	method to return the current schema of the source More...

std::string	getSourceSchemaAsString ()
	method to return the current schema of the source as string More...

bool	isRunning () const noexcept
	debug function for testing to test if source is running More...

uint64_t	getNumberOfGeneratedTuples () const
	debug function for testing to get number of generated tuples More...

uint64_t	getNumberOfGeneratedBuffers () const
	debug function for testing to get number of generated buffer More...

void	setGatheringInterval (std::chrono::milliseconds interval)
	method to set the sampling interval More...

virtual	~DataSource () NES_NOEXCEPT(false) override
	Internal destructor to make sure that the data source is stopped before deconstrcuted @Note must be public because of boost serialize. More...

uint64_t	getNumBuffersToProcess () const
	Get number of buffers to be processed. More...

std::chrono::milliseconds	getGatheringInterval () const
	Get gathering interval. More...

uint64_t	getGatheringIntervalCount () const
	Get number representation of gathering interval. More...

OperatorId	getOperatorId () const
	Gets the operator id for the data source. More...

void	setOperatorId (OperatorId operatorId)
	Set the operator id for the data source. More...

std::vector< Runtime::Execution::SuccessorExecutablePipeline >	getExecutableSuccessors ()
	Returns the list of successor pipelines. More...

void	addExecutableSuccessors (std::vector< Runtime::Execution::SuccessorExecutablePipeline > newPipelines)
	Add a list of successor pipelines. More...

template<typename Derived >
std::shared_ptr< Derived >	shared_from_base ()
	This method is necessary to avoid problems with the shared_from_this machinery combined with multi-inheritance. More...

virtual void	onEvent (Runtime::BaseEvent &) override
	API method called upon receiving an event. More...

virtual void	onEvent (Runtime::BaseEvent &event, Runtime::WorkerContextRef workerContext)
	API method called upon receiving an event, whose handling requires the WorkerContext (e.g. its network channels). More...

virtual bool	fail ()

void	setSourceSharing (bool value)
	set source sharing value More...

void	incrementNumberOfConsumerQueries ()
	set the number of queries that use this source More...

virtual bool	handleReconfigurationMarker (ReconfigurationMarkerPtr marker)
	check if a reconfiguration marker contains an event for this source. If so, trigger the reconfiguration and propagate the marker downstream. More...

Public Member Functions inherited from NES::Runtime::Reconfigurable
	~Reconfigurable () NES_NOEXCEPT(false) override=default

virtual void	reconfigure (ReconfigurationMessage &, WorkerContext &)
	reconfigure callback that will be called per thread More...

virtual void	postReconfigurationCallback (ReconfigurationMessage &)
	callback that will be called on the last thread the executes the reconfiguration More...

Public Member Functions inherited from NES::detail::virtual_enable_shared_from_this< Reconfigurable, false >
	~virtual_enable_shared_from_this () NES_NOEXCEPT(isNoexceptDestructible) override=default

std::shared_ptr< T1 >	shared_from_this ()

std::weak_ptr< T1 >	weak_from_this ()

Public Member Functions inherited from NES::detail::virtual_enable_shared_from_this_base< isNoexceptDestructible >
virtual	~virtual_enable_shared_from_this_base () NES_NOEXCEPT(isNoexceptDestructible)=default

Public Member Functions inherited from NES::DataEmitter
virtual	~DataEmitter () NES_NOEXCEPT(false)=default

virtual void	onEndOfStream (Runtime::QueryTerminationType)

virtual DecomposedQueryPlanVersion	getVersion () const

virtual bool	startNewVersion ()
	start a previously scheduled new version for this data emitter More...

virtual bool	insertReconfigurationMarker (ReconfigurationMarkerPtr)
	check if a reconfiguration marker contains an event for this data emitter. If so, trigger the reconfiguration and propagate the marker downstream. More...

Public Member Functions inherited from NES::detail::virtual_enable_shared_from_this< RuntimeEventListener, false >
	~virtual_enable_shared_from_this () NES_NOEXCEPT(isNoexceptDestructible) override=default

std::shared_ptr< T1 >	shared_from_this ()

std::weak_ptr< T1 >	weak_from_this ()

Additional Inherited Members
Public Attributes inherited from NES::DataSource
const bool	persistentSource

const std::string	persistentSourceKey

Protected Member Functions inherited from NES::DataSource
void	emitWork (Runtime::TupleBuffer &buffer, bool addBufferMetaData=true) override
	create a task using the provided buffer and submit it to a task consumer, e.g., query manager More...

NES::Runtime::MemoryLayouts::TestTupleBuffer	allocateBuffer ()

Protected Attributes inherited from NES::DataSource
Runtime::QueryManagerPtr	queryManager

Runtime::BufferManagerPtr	localBufferManager

Runtime::FixedSizeBufferPoolPtr	bufferManager {nullptr}

std::vector< Runtime::Execution::SuccessorExecutablePipeline >	executableSuccessors

OperatorId	operatorId

OriginId	originId

StatisticId	statisticId

SchemaPtr	schema

uint64_t	generatedTuples {0}

uint64_t	generatedBuffers {0}

uint64_t	numberOfBuffersToProduce = std::numeric_limits<decltype(numberOfBuffersToProduce)>::max()

uint64_t	numSourceLocalBuffers

uint64_t	gatheringIngestionRate {}

std::chrono::milliseconds	gatheringInterval {0}

GatheringMode	gatheringMode

SourceType	type

Runtime::QueryTerminationType	wasGracefullyStopped {Runtime::QueryTerminationType::Graceful}

std::atomic_bool	wasStarted {false}

std::atomic_bool	futureRetrieved {false}

std::atomic_bool	running {false}

std::promise< bool >	completedPromise

uint64_t	sourceAffinity

uint64_t	taskQueueId

bool	sourceSharing = false

const std::string	physicalSourceName

std::atomic< uint64_t >	refCounter = 0

std::atomic< uint64_t >	numberOfConsumerQueries = 1

Runtime::MemoryLayouts::MemoryLayoutPtr	memoryLayout

Constructor & Destructor Documentation

◆ NonRunnableDataSource()

NES::Testing::NonRunnableDataSource::NonRunnableDataSource	(	const SchemaPtr &	schema,
		const Runtime::BufferManagerPtr &	bufferManager,
		const Runtime::QueryManagerPtr &	queryManager,
		uint64_t	numbersOfBufferToProduce,
		uint64_t	gatheringInterval,
		OperatorId	operatorId,
		OriginId	originId,
		StatisticId	statisticId,
		size_t	numSourceLocalBuffers,
		const std::vector< Runtime::Execution::SuccessorExecutablePipeline > &	successors,
		NES::Runtime::QueryTerminationType	terminationType = `NES::Runtime::QueryTerminationType::HardStop`
	)

explicit

References NES::DataSource::type, and NES::DataSource::wasGracefullyStopped.

Member Function Documentation

◆ emitBuffer() [1/2]

void NES::Testing::NonRunnableDataSource::emitBuffer	(	Runtime::MemoryLayouts::TestTupleBuffer &	buffer,
		bool	addBufferMetaData = `true`
	)

References NES::Runtime::MemoryLayouts::TestTupleBuffer::getBuffer().

Here is the call graph for this function:

◆ emitBuffer() [2/2]

void NES::Testing::NonRunnableDataSource::emitBuffer	(	Runtime::TupleBuffer &	buffer,
		bool	addBufferMetaData = `true`
	)

References NES::DataSource::emitWork().

Here is the call graph for this function:

◆ getBuffer()

Runtime::MemoryLayouts::TestTupleBuffer NES::Testing::NonRunnableDataSource::getBuffer ( )

References NES::DataSource::allocateBuffer().

Here is the call graph for this function:

◆ performSoftStop()

bool NES::Testing::NonRunnableDataSource::performSoftStop ( )

overridevirtual

This method will mark the running as false this will result in 1. the interruption of data gathering loop and 2. initiation of soft stop routine.

Returns: true if successfully interrupted else false

Reimplemented from NES::DataSource.

References NES::DataSource::performSoftStop().

Here is the call graph for this function:

◆ runningRoutine()

void NES::Testing::NonRunnableDataSource::runningRoutine ( )

overridevirtual

running routine while source is active

Reimplemented from NES::DataSource.

References NES::DataSource::close(), NES::DataSource::completedPromise, and NES::DataSource::open().

Here is the call graph for this function:

◆ stop()

bool NES::Testing::NonRunnableDataSource::stop ( Runtime::QueryTerminationType graceful )

overridevirtual

method to stop the source. 1.) check if bool running is false, if false return, if not stop source 2.) stop thread by join

Reimplemented from NES::DataSource.

References NES::DataSource::isRunning(), and NES::DataSource::stop().

Here is the call graph for this function:

The documentation for this class was generated from the following files:

/nebulastream/nes-coordinator/tests/include/Util/NonRunnableDataSource.hpp
/nebulastream/nes-coordinator/tests/src/Util/NonRunnableDataSource.cpp

Public Member Functions

Additional Inherited Members

Constructor & Destructor Documentation

◆ NonRunnableDataSource()

Member Function Documentation

◆ emitBuffer() [1/2]

◆ emitBuffer() [2/2]

◆ getBuffer()

◆ performSoftStop()

◆ runningRoutine()

◆ stop()