Passing generic data between thread

[ittium]

I want to pass data between two threads. Type of data can vary.

data
{
list l;
map m;
}

data
{
int i;
long l;
}

I can think of following ways to do it

1. write serialise routines so that between the thread, encoded void*
buffer is passed. But since thread will run in same address space,
serialization seems to me as overkill.

2. Pass a token (specifying the type) and data typecasted to void*.
Receiving thread can typecast the data to actual type based on the
token

3. Use boost:any and then receiver can find out type using RTTI.

Since large amount of data is to be processed, CPU utilization is
important concern.

Can you comment, which of these approach will be good. I am open to
considering any approach other than above three.

I will appreciate your experiences of solving similar problem.

thanks
Ittium

 

[ittium]

Paavo thanks for the detailed reply. I might use boost::any, I am not
sure if RTTI is a safe and efficient method since I need to process
very large amount of data very fast.

 

[Joshua Maurice]

Paavo thanks for the detailed reply. I might use boost::any, I am not
sure if RTTI is a safe and efficient method since I need to process
very large amount of data very fast.

Are you using a discriminated unions elsewhere in the code?
http://en.wikipedia.org/wiki/Tagged_union
This is your #2 in the OP post.

That is, did you plan on keeping next to each data object a flag which
tells its data type, and then you do a switch on the type flag
whenever you need to do something with it?

My initial knee-jerk reaction is that this is exactly what virtual and
polymorphism was invented to solve, to avoid the switch on types.
However, you can maybe write it to run faster with a discriminated
union, but that comes with its own set of restrictions and problems.
One such onerous restriction is that it's not plug and play-able at
runtime - the code only knows how to deal with a specific finite
number of type flags. If written well, at least this could maybe be
changed by changing one or two places, and doing a full recompile (as
opposed to putting switches on the type flag everywhere in the code).
It's just like implementing virtual function calls yourself with a
switch statement instead of a function pointer.

If reasonable, I would implement it both with discriminated unions and
standard virtual polymorphism (such as Boost.variant and/or
Boost.any), and measure the difference on real data.

I don't know enough to comment further.

 

[Werner]

I want to pass data between two threads. Type of data can vary.

data
{
   list l;
   map m;

}

data
{
  int i;
  long l;

}

I can think of following ways to do it

1. write serialise routines so that between the thread, encoded void*
buffer is passed. But since thread will run in same address space,
serialization seems to me as overkill.

2. Pass a token (specifying the type) and data typecasted to void*.
Receiving thread can typecast the data to actual type based on the
token

3. Use boost:any and then receiver can find out type using RTTI.

It is not necessary to use boost any for this.

You can do this kind of thing by using commands
(GOF command pattern), a command having an
association with the queue in whose thread it will
be executed.

Continuing from the GOF pattern, "execute" is called
from the thread receiving the command. A clone of
the command is sent over the queue (pointer only), and
during the clone ownership of the argument of the command
is transferred from the envoked command to the cloned
command.

The envoked command sets the data using overloaded
operator()( argT ), then, depending on a storage policy,
on cloning the command, the ownership of the argument
is either transferred copied (for small objects).

Pre-allocated memory pools are used for memory of the
commands and their arguments (depending on storage pol).

Getting of memory in the pools are done by performing a binary
lookup to get the correct memory size, and returning of the
memory is done my storing a hidden pointer to the pool in
front of the memory.

Psuedo (very quick take):

class CmdExecutable
{
public:
//Called by executing thread...
virtual void execute() = 0;
};

class CmdProcessable : CmdExecutable
{
public:
virtual CmdProcessable* clone() const = 0;
};

class Cmd : CmdExecutable
{
public:
//<mngr> determines context of execution...
Cmd( CmdServer* server );

//If we have a queue, post myself over the queue
void process();
void operator()(); //Does the same as process

//I know clone has better/canonical implementations, but
// you get the idea
virtual Cmd* clone() const = 0;

private:
virtual void execute() = 0;
};


class CmdServer
{
public:
void process( const CmdProcessable& cmd )
{
mail( cmd.clone() );
}
// Implementation mails cmd over thread safe queue.
virtual void mail( CmdProcessable* cmd ) throw() = 0;
};

class CmdExecutor : public CmdServer
{
public:
void serviceCmdQueue() throw();

private:
//All v functions required by service...
};


void CmdExecutor::serviceCmdQueue() throw()
{
//Psuedo code...
CmdExecutable* cmd;
//This call should block until next cmd arrives on queue
unsigned size( retrieveNextCmdBlock( &cmd ) );

//Some error checking...

//Ensures deletion at end of this scope...
std::auto_ptr<CmdExecCmd> scopedCmd( cmd );

//Handles possible rendezvous with caller...
using namespace AviUtils;
ActionAtScopeEnd
handleRendez( mkAction( HandlePossibleRendezvous( *cmd ) ) );
valueUsed( handleRendez );

//Executes the command, passing associated arguments
// along with function (not shown for brevity)
cmd->execute();
}

Perhaps you get the idea, but I've left out lots
of gory detail.

Regards,

Werner

 

[ittium]

Perhaps you get the idea, but I've left out lots
of gory detail.

Regards,

Werner

In your example,my queue will be queue of cmd. I will derive classes
containing different type of data from the cmd class.

 

[Werner]

In your example,my queue will be queue of cmd. I will derive classes
containing different type of data from the cmd class.

Yes. Each of them will override execute and call their
respective callbacks (pointers to functions or pointers
to member functions, obviously then with the recipient...).

Example (for one argument case):

template <class Receiver, class RetVal, class Arg, class
StoragePolicy>
void CmdRetArg1<Receiver,RetVal,Arg,StoragePolicy>::execute()
{
using namespace AviUtils;
ActionAtScopeEnd a1(
mkAction( &StoragePolicyType::unprepare,
AviUtils::byRef( storage_ ) ) );

//Getting rid of some compiler warnings...
valueUsed( a1 );

//Invoke callback...
retVal_ = (receiver_->*action_)
( StoragePolicyType::retrieve( storage_ ) );
}

 

[Werner]

Example (for one argument case):

Needs some explaining...

template <class Receiver, class RetVal, class Arg, class
StoragePolicy>
void CmdRetArg1<Receiver,RetVal,Arg,StoragePolicy>::execute()
{
  using namespace AviUtils;
  ActionAtScopeEnd a1(
mkAction( &StoragePolicyType::unprepare,
AviUtils::byRef( storage_ ) ) );

reverse any preparation that might have happened
on the storage (some fancy name for the data), but
only do this once this function completes,
irrespective of whether it throws.

  //Getting rid of some compiler warnings...
  valueUsed( a1 );

  //Invoke callback...
  retVal_ = (receiver_->*action_)
( StoragePolicyType::retrieve( storage_ ) );

Calls the applicable member function action_, and
retrieves the argument using the storage policy.
Depending on the storage policy the argument may
be a pointer or a value... References are usually
converted to values as they can't be passed across
threads. It is assumed that if one passes pointers,
you know what you are doing...