t22_Pipeline.cpp File Reference

#include <pthread.h>
#include <iterator>
#include <vector>
#include <iostream>
#include <sstream>
#include <stdlib.h>
#include <algorithm>
#include <numeric>
#include <set>
#include <list>
#include <deque>
#include <rpa/architectures/posix.h>
#include <rpa/algorithms/accumulate.h>
#include <rpa/algorithms/transform.h>
#include <rpa/containers/list.h>
#include <rpa/containers/deque.h>
#include <rpa/containers/vector.h>
#include <rpa/containers/pipeline.h>

Classes
class	identity22< Type >
struct	t22adder< Type >
	This is used in the algorithm 'accumulate' for tracking how values. More...
class	Plus22< T >
	Because of the way this functor is used, some specific checks can be done. More...
Defines
#define	Test(a)   _Test22(#a,a)
Typedefs
typedef rpa::posix_condition	cond_main_t
	Used everywhere.
Functions
static void	_Test22 (const char *aMsg, bool aBool)
template<class SubThr, class Container, class PipTyp>
static void	tst22Sequential (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs)
template<class SubThr, class Container, class PipTyp>
static void	tst22OneSubThrCirc (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs, size_t szPipCirc)
template<class SubThr, class Container, class PipTyp>
static void	tst22OneSubThrArch (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs)
template<class Container, class PipTyp>
static void	tst22TwoSubThr (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs, size_t szPipCirc)
template<class SubThr, class Container, class PipTyp>
static void	tst22CpyOrdFinite (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs, size_t nbStps)
template<class SubThr, class Container, class PipTyp>
static void	tst22CpyOrdInfiniteEqual (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs)
template<class SubThr, class Container, class PipTyp>
static void	tst22Loop (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs, size_t szPipSub)
template<class SubThr, class Container, class PipTyp>
static void	tst22LoopTrans (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs)
template<class SubThr, class Container, class PipTyp>
static void	tst22MergeSequential (const Container &lstDat, typename Container::value_type sumDat, size_t nbDbls, size_t nbThrs)
template<class SubThr, class Cont, class PipTyp>
static void	tst22ContBuf (const Cont &lstDat, typename Cont::value_type sumDat, size_t nbDbls, size_t nbThrs)
template<class SubThr, class Cont>
static void	tst22Containers (size_t nbDbls, size_t nbThrs)
template<class SubThr>
void	tst22Thr (size_t nbE, size_t nbThrs)
int	main (int aArgC, const char **aArgV)
	The default number of samples is chosen for a Pentium, about 1GHz..

---

Define Documentation

#define Test

(

a

)

_Test22(#a,a)

---

Typedef Documentation

typedef rpa::posix_condition cond_main_t

Used everywhere.

---

Function Documentation

static void _Test22	(	const char *	aMsg,
		bool	aBool
	)			[static]

This is simply used to check whether a boolean condition is true or not, and fails if not with an error message.

int main	(	int	aArgC,
		const char **	aArgV
	)

The default number of samples is chosen for a Pentium, about 1GHz..

template<class SubThr, class Cont>

static void tst22Containers	(	size_t	nbDbls,
		size_t	nbThrs
	)			[static]

template<class SubThr, class Cont, class PipTyp>

static void tst22ContBuf	(	const Cont &	lstDat,
		typename Cont::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs
	)			[static]

template<class SubThr, class Container, class PipTyp>

static void tst22CpyOrdFinite	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs,
		size_t	nbStps
	)			[static]

This checks that the output must be the same as the input. Test with a pipeline of finite size. This proves that if the finite-size buffer is big enough, it will not block the reader and the writer.

The pipeline has a finite size which is automatically set at the size of the input data. Therefore, it is not needed to read it within a subthread because all the data is already there. When a pipeline is initialised with a pair of iterators, by default, the flag 'shut' is set to true, which means that there will be no more insertions.

Not very clean because we must create a buffer just to get the may size....

Now the container must be empty because all elements are read.

template<class SubThr, class Container, class PipTyp>

static void tst22CpyOrdInfiniteEqual	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs
	)			[static]

This checks that the output must be the same as the input. Test with a pipeline of infinite size. The elements of the pipe are read with begin/end.

When a pipeline is initialised with a pair of iterators, by default, the flag 'shut' is set to true, which means that there will be no more insertions.

Now the pipe must be empty.

template<class SubThr, class Container, class PipTyp>

static void tst22Loop	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs,
		size_t	szPipSub
	)			[static]

The point here is to test that it is possible to insert into a pipeline and read from it in the same loop. It is therefore possible to create an endless pipeline.

This pipeline has an infinite size : It stores everything. Therefore, we cannot use the 'PipTyp' which may have a max size, in these tests.

The flag shut=false is added because there will be other insertions.

This copies repetitively the elements of the input container, until there are 'nbCopies' elements.

These containers must be identical.

template<class SubThr, class Container, class PipTyp>

static void tst22LoopTrans	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs
	)			[static]

The point here is to test that it is possible to insert into a pipeline and read from it in the same loop. It is therefore possible to create an endless pipeline. The pipe has an finite size.

This pipeline has an infinite size : It stores everything. Therefore, we cannot use the 'PipTyp' which may have a max size, in these tests.

At the beginning, the pipe is filled with zeroes.

It is shut so we do not wait for new data input.

The read iterator does not move, because we read the underlying container.

It is cyclically filled. Instead of 'Plus22', it is possible to use std::plus.

The part which is inserted but not read.

Checks the content of the underlying container.

No more elements to read from the pipe. This test is not blocking because the pipe is shut.

template<class SubThr, class Container, class PipTyp>

static void tst22MergeSequential	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs
	)			[static]

This tests checks that elements can be read only once from a pipe line. The input elements must absolutely be sorted.

This pipeline has an infinite size : It stores everything. We must be sure that the underlying container has an infinite size too.

These containers must be identical.

template<class SubThr, class Container, class PipTyp>

static void tst22OneSubThrArch	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs
	)			[static]

Only one subthread for the input algorithm. The buffer is not big enough to store all the input data, but the main thread is running at the same time and reads from it. The difference with tst22OneSubThrCirc(), is that we use a pipe_archiver.

Finite-size pipeline. There may not be enough room to store the result of the 'transform' algorithm. The underlying parameter is not 'PipTyp' because it may be a fixed-size C-style array, and it would not work for a pipe_archiver which has an infinite size.

At the beginning, the pipeline is not shut. So, reading from it hangs, if there are no data inside.

This functor will be started in a subthread.

This is the thread tree containing just one top-level sub-thread.

This is the result of the detached execution of the functor in a sub-thread.

The output elements might not be in the right order. So we sort them before comparing. This happens if there are genuine subthreads.

template<class SubThr, class Container, class PipTyp>

static void tst22OneSubThrCirc	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs,
		size_t	szPipCirc
	)			[static]

Only one subthread for the input algorithm. The buffer is not big enough to store all the input data, but the main thread is running at the same time and reads from it.

Finite-size pipeline. There may not be enough room to store the result of the 'transform' algorithm.

This functor will be started in a subthread.

This is the thread tree containing just one top-level sub-thread.

This is the result of the detached execution of the functor in a sub-thread.

template<class SubThr, class Container, class PipTyp>

static void tst22Sequential	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs
	)			[static]

First fills the container, then uses it. Everything is in sequential mode, no threads nor mutex. If the pipe has a limited size (Static C-style array...), then the function loops until everything is processed.

This pipeline has an infinite size : It stores everything. Therefore, we cannot use the 'PipTyp' which may have a max size, in these tests.

Now the container must be empty because all elements are read.

template<class SubThr>

void tst22Thr	(	size_t	nbE,
		size_t	nbThrs
	)

There are two imbricated loop on the containers type. The external container type is always limit-less. The internal one may have a limit and will therefore never used for infinite pipelines.

template<class Container, class PipTyp>

static void tst22TwoSubThr	(	const Container &	lstDat,
		typename Container::value_type	sumDat,
		size_t	nbDbls,
		size_t	nbThrs,
		size_t	szPipCirc
	)			[static]

Two subthreads read from the input container. One write into a circular pipeline, the second into an archiver pipeline. Their output goes into a merge. The final result must be identical to the input. There are no sub-subthreads, because the sub-threads are sequential.

Finite-size pipeline. There may not be enough room to store the result of the 'transform' algorithm.

The kind of input range for this pipeline. It is intrinsincally synchronized.

The container underlying an archiver pipeline must have an infinite size.

The two sub-threads must shut automatically their output once they exit. Otherwise, the final merge would wait forever.

This is the thread tree containing just one top-level sub-thread.

This is the result of the detached execution of the functor in a sub-thread.

At the beginning, the pipeline is not shut. It means that reading from it will hang if it is empty.

This functor will be started in a subthread.

At the beginning, it is not shut. So, reading from it will hang if there are no data inside.

This functor will be started in a subthread.

The output must be automatically sorted because the elements will come from two different sources.

These containers must be identical.

---