AthenaInterprocess Namespace Reference

Classes
struct	FdsRegistryEntry
class	IdentifiedSharedQueue
class	IMessageDecoder
class	IMPRunStop
class	Process
class	ProcessGroup
struct	ProcessResult
struct	ProcessStatus
struct	ScheduledWork
class	SharedQueue
class	UpdateAfterFork

Typedefs
typedef std::vector< FdsRegistryEntry >	FdsRegistry

Functions
std::string	randString ()
std::string	add_pid (const std::string &buf)
std::string	get_pid (const std::string &buf, pid_t &pid)
pid_t ProcessGroup::launchProcess	ATLAS_NOT_THREAD_SAFE ()
int ProcessGroup::map_async	ATLAS_NOT_THREAD_SAFE (const IMessageDecoder *func, const ScheduledWork *args, pid_t pid)
int ProcessGroup::wait	ATLAS_NOT_THREAD_SAFE (int options)
static bool	do_send (message_queue *mq, const std::string &buf, bool block)
static std::string	do_receive (message_queue *mq, bool block)

Variables
static const int	SHAREDQUEUE_MAX_MSG = 1000
static const std::size_t	MAX_MSG_SIZE = 256

Typedef Documentation

FdsRegistry

typedef std::vector<FdsRegistryEntry> AthenaInterprocess::FdsRegistry

Definition at line 22 of file FdsRegistry.h.

Function Documentation

add_pid()

std::string AthenaInterprocess::add_pid ( const std::string & buf )

inline

Definition at line 31 of file IdentifiedSharedQueue.cxx.

{
   std::ostringstream s;
   s << (long)getpid() << '\0' << buf;
   return s.str();
}

ATLAS_NOT_THREAD_SAFE() [1/3]

bool ProcessGroup::create AthenaInterprocess::ATLAS_NOT_THREAD_SAFE

(

void

)

Definition at line 49 of file ProcessGroup.cxx.

{
  if(m_pgid==-1)
    return -1;
 
  // Create new process
  SharedQueue queue = create_queue("child", m_processesCreated); 
  Process p = Process::launch();
  p.connectIn(queue);
  p.connectOut(m_inbox);
  pid_t newpid = p.getProcessID();
  if(newpid==0) {
    int status = p.mainloop();
    exit(status);
  }
 
  if(m_processes.empty()) 
    m_pgid = getpgid(newpid);
  else
    setpgid(newpid, m_pgid);
 
  m_processes.push_back(p);
  m_processesCreated++;
  return newpid;
}

ATLAS_NOT_THREAD_SAFE() [2/3]

int ProcessGroup::map_async AthenaInterprocess::ATLAS_NOT_THREAD_SAFE	(	const IMessageDecoder *	func,
		const ScheduledWork *	args,
		pid_t	pid )

Definition at line 75 of file ProcessGroup.cxx.

                                                                                                                    {
  // If pid=0, map the function-object 'func' onto all current child processes. Does
  // not wait for the results, but will return success only if the writing to
  // all the child queues succeeds.
  //
  // If pid!=0 then map only onto the process with given pid
  if(m_processes.empty()) {
    if(!create())
      return -1;
  }
 
  if(pid==0) { // Map on all processes in the group
    for(std::vector<Process>::iterator iproc = m_processes.begin();
    iproc!=m_processes.end(); ++iproc) {
 
      if(!iproc->schedule(func, args)) {
    // stopping the scheduling on all other processes is debatable ...
    return -1;
      }
    }
    return 0;
  }
  else { // Map only onto the given pid
    for(std::vector<Process>::iterator iproc=m_processes.begin();
    iproc!=m_processes.end(); ++iproc) {
      if(iproc->getProcessID()==pid)
    return (iproc->schedule(func,args) ? 0 : -1);
    }
    // pid not found in the group
    return -1;
  }
}

ATLAS_NOT_THREAD_SAFE() [3/3]

int ProcessGroup::wait AthenaInterprocess::ATLAS_NOT_THREAD_SAFE

(

int

options

)

Definition at line 108 of file ProcessGroup.cxx.

{
  // Wait for all child processes and store their status codes in m_statuses
  if(m_processes.empty())
      return 0;
 
  // Schedule an exit if we are to wait forever
  if(!(options & WNOHANG) && map_async(0,0))
    return -1;
 
   int result = 0;
   while(m_processes.size()) {
     int child_status = 0;
     pid_t child = waitpid(-m_pgid, &child_status, options);
     if(child < 0) {
       result = errno;
       break;
     }
     else if(child==0) {
       continue;
     }
 
     if(WIFSIGNALED(child_status))
       std::cout << "SIGNAL! " << child << " (" << child_status << "," << WTERMSIG(child_status) << ")" << std::endl;
 
      child_status = WIFSIGNALED(child_status) ? WTERMSIG(child_status) : WEXITSTATUS(child_status);
      ProcessStatus p = {child, child_status};
      m_statuses.push_back(p);
 
      // remove the process from m_processes
      for(std::vector<Process>::iterator iproc=m_processes.begin();
      iproc!=m_processes.end(); ++iproc) {
    if(iproc->getProcessID()==child) {
      m_processes.erase(iproc);
      break;
    }
      }
   }
   return result;
}

do_receive()

std::string AthenaInterprocess::do_receive ( message_queue * mq, bool block )

inlinestatic

Definition at line 112 of file SharedQueue.cxx.

{
   char buf[ MAX_MSG_SIZE ];
 
   try {
      std::size_t recvd_size = 0;
      unsigned int priority = 0;
 
      bool result = true;
      if ( block ) 
         mq->receive( buf, MAX_MSG_SIZE, recvd_size, priority );
      else
         result = mq->try_receive( buf, MAX_MSG_SIZE, recvd_size, priority );
 
      if ( result ) {
         if ( recvd_size > MAX_MSG_SIZE )
            recvd_size = MAX_MSG_SIZE;      // this is debatable, but send
                                            // should have failed already, so
                                            // "can not happen" applies
         return std::string( buf, recvd_size );
      }
   } catch ( interprocess_exception& e ) {
   /* silent for now: should return a callable error function */
   }
 
   return "";
}

do_send()

bool AthenaInterprocess::do_send ( message_queue * mq, const std::string & buf, bool block )

inlinestatic

Definition at line 86 of file SharedQueue.cxx.

{
   bool send_ok = true;
 
   try {
      if ( block )
         mq->send( buf.data(), buf.size(), 0 );
      else
         send_ok = mq->try_send( buf.data(), buf.size(), 0 );
   } catch ( interprocess_exception& e ) {
      send_ok = false;
   }
 
   return send_ok;
}

get_pid()

std::string AthenaInterprocess::get_pid ( const std::string & buf, pid_t & pid )

inline

Definition at line 48 of file IdentifiedSharedQueue.cxx.

{
   std::istringstream s( buf );
   long id = -1;
   s >> id;
   pid = (pid_t)id;
   std::string::size_type pos = s.tellg();
   if ( (pos+1) <= buf.size() )
      return buf.substr( pos+1 );
   return "";
}

randString()

std::string AthenaInterprocess::randString ( )

inline

Definition at line 13 of file Control/AthenaInterprocess/AthenaInterprocess/Utilities.h.

  {
    uuid_t uuid;
    char out[SSIZE];
    uuid_generate(uuid);
    uuid_unparse(uuid, out);
    return std::string(out);
  }

Variable Documentation

MAX_MSG_SIZE

const std::size_t AthenaInterprocess::MAX_MSG_SIZE = 256

static

Definition at line 19 of file SharedQueue.h.

SHAREDQUEUE_MAX_MSG

const int AthenaInterprocess::SHAREDQUEUE_MAX_MSG = 1000

static

Definition at line 18 of file SharedQueue.h.