---

IgNet Class Reference

#include <Iguana/Framework/interface/IgNet.h>

Inheritance diagram for IgNet:

List of all members.

Public Types
typedef std::vector< unsigned char >	DataBlob
typedef std::map< std::string, Object >	ObjectMap
typedef std::map< lat::Socket *, Peer >	PeerMap
typedef std::list< WaitObject >	WaitList
Public Member Functions
void	debug (bool doit)
	Enable or disable verbose debugging.
void	delay (int delay)
	Set the I/O dispatching delay.
	IgNet (const std::string &appname="")
void	listenToSource (const std::string &host, int port)
	Tell the network layer to connect to host and port and automatically receive updates from upstream source(s).
void	lock (void)
	Acquire a lock on the net layer.
virtual int	receive (void(*callback)(void *arg, uint32_t reason, Object &obj), void *arg)
virtual void	removeLocalObject (const std::string &name)
	Delete the local object.
void	run (void)
	Run the actual I/O processing loop.
void	sendLocalChanges (void)
void	shutdown (void)
	Stop the network layer and wait it to finish.
void	start (void)
	Start running the network layer in a new thread.
void	startLocalServer (int port)
	Start a server socket for accessing this node remotely.
void	unlock (void)
	Release the lock on the net layer.
virtual void	updateLocalObject (Object &o)
	Update the network cache for an object.
void	updateToCollector (const std::string &host, int port)
	Tell the network layer to connect to host and port and automatically send updates whenever local data changes.
virtual	~IgNet (void)
Static Public Attributes
static const uint32_t	MAX_PEER_WAITREQS = 128
static const uint32_t	VIS_FLAG_DEAD = 0x40000000
static const uint32_t	VIS_FLAG_NEW = 0x20000000
static const uint32_t	VIS_FLAG_RECEIVED = 0x10000000
static const uint32_t	VIS_FLAG_SCALAR = 0x1
static const uint32_t	VIS_FLAG_ZOMBIE = 0x80000000
static const uint32_t	VIS_MSG_GET_OBJECT = 3
static const uint32_t	VIS_MSG_HELLO = 0
static const uint32_t	VIS_MSG_LIST_OBJECTS = 2
static const uint32_t	VIS_MSG_UPDATE_ME = 1
static const uint32_t	VIS_REPLY_LIST_BEGIN = 101
static const uint32_t	VIS_REPLY_LIST_END = 102
static const uint32_t	VIS_REPLY_NONE = 103
static const uint32_t	VIS_REPLY_OBJECT = 104
Protected Member Functions
virtual Peer *	createPeer (lat::Socket *s)
virtual Object *	findObject (Peer *p, const std::string &name, Peer **owner=0)
virtual Peer *	getPeer (lat::Socket *s)
std::ostream &	logme (void)
virtual Object *	makeObject (Peer *p, const std::string &name)
virtual void	markObjectsDead (Peer *p)
virtual void	markObjectsZombies (Peer *p)
virtual bool	onMessage (Bucket *msg, Peer *p, unsigned char *data, size_t len)
virtual void	purgeDeadObjects (lat::Time oldobj, lat::Time deadobj)
virtual void	releaseFromWait (Bucket *msg, WaitObject &w, Object *o)
virtual void	removePeer (Peer *p, lat::Socket *s)
virtual void	sendObjectListToPeer (Bucket *msg, bool all, bool clear)
	Send all objects to a peer and optionally mark sent objects old.
virtual void	sendObjectListToPeers (bool all)
void	sendObjectToPeer (Bucket *msg, Object &o, bool data)
virtual bool	shouldStop (void)
void	updateMask (Peer *p)
	Update the selector mask for a peer based on data queues.
virtual void	updatePeerMasks (void)
void	waitForData (Peer *p, const std::string &name, const std::string &info, Peer *owner)
	Queue a request for an object and put a peer into the mode of waiting for object data to appear.
Static Protected Member Functions
static void	copydata (Bucket *b, const void *data, size_t len)
Protected Attributes
bool	debug_
Private Member Functions
	IgNet (const IgNet &)
bool	losePeer (const char *reason, Peer *peer, lat::IOSelectEvent *event, lat::Error *err=0)
	Handle errors with a peer socket.
bool	onLocalNotify (lat::IOSelectEvent *ev)
	React to notifications from the app thread.
bool	onPeerConnect (lat::IOSelectEvent *ev)
	Respond to new connections on the server socket.
bool	onPeerData (lat::IOSelectEvent *ev, Peer *p)
	Handle communication to a particular client.
IgNet &	operator= (const IgNet &)
void	releaseFromWait (WaitList::iterator i, Object *o)
void	releaseWaiters (Object *o)
void	requestObject (Peer *p, const char *name, size_t len)
	Queue an object request to the data server.
Static Private Member Functions
static void	discard (Bucket *&b)
Private Attributes
std::string	appname_
pthread_t	communicate_
int	delay_
AutoPeer	downstream_
bool	flush_
Peer *	local_
pthread_mutex_t	lock_
PeerMap	peers_
int	pid_
lat::IOSelector	sel_
lat::InetServerSocket *	server_
sig_atomic_t	shutdown_
AutoPeer	upstream_
lat::Time	version_
WaitList	waiting_
lat::Pipe	wakeup_
Classes
struct	AutoPeer
struct	Bucket
struct	Object
struct	Peer
struct	WaitObject

---

Detailed Description

Definition at line 18 of file IgNet.h.

---

Member Typedef Documentation

typedef std::vector<unsigned char> IgNet::DataBlob

Definition at line 41 of file IgNet.h.

typedef std::map<std::string, Object> IgNet::ObjectMap

Definition at line 45 of file IgNet.h.

typedef std::map<lat::Socket *, Peer> IgNet::PeerMap

Definition at line 46 of file IgNet.h.

typedef std::list<WaitObject> IgNet::WaitList

Definition at line 44 of file IgNet.h.

---

Constructor & Destructor Documentation

IgNet::IgNet

(

const std::string &

appname = ""

)

Definition at line 781 of file IgNet.cc.

References lat::IOSelector::attach(), lat::CreateHook(), createPeer(), downstream_, lat::IOChannel::fd(), IORead, local_, IgNet::AutoPeer::next, O_NONBLOCK, onLocalNotify(), IgNet::AutoPeer::peer, IgNet::AutoPeer::port, sel_, lat::Pipe::source(), IgNet::AutoPeer::update, upstream_, wakeup_, and IgNet::AutoPeer::warned.

  : debug_ (false),
    appname_ (appname.empty() ? "IgNet" : appname.c_str()),
    pid_ (getpid()),
    server_ (0),
    version_ (Time::current()),
    communicate_ ((pthread_t) -1),
    shutdown_ (0),
    delay_ (1000),
    flush_ (false)
{
  // Create a pipe for the local apps to tell the communicator
  // thread that local app data has changed and that the peers
  // should be notified.
  fcntl(wakeup_.source()->fd(), F_SETFL, O_RDONLY | O_NONBLOCK);
  sel_.attach(wakeup_.source(), IORead, CreateHook(this, &IgNet::onLocalNotify));

  // Initialise the upstream and downstream to empty.
  upstream_.peer   = downstream_.peer   = 0;
  upstream_.next   = downstream_.next   = 0;
  upstream_.port   = downstream_.port   = 0;
  upstream_.update = downstream_.update = false;
  upstream_.warned = downstream_.warned = false;

  local_ = createPeer((Socket *) -1);
}

IgNet::~IgNet

(

void

)

[virtual]

Definition at line 808 of file IgNet.cc.

{
  // FIXME
}

IgNet::IgNet

(

const IgNet &

)

[private]

---

Member Function Documentation

void IgNet::copydata	(	Bucket *	b,
		const void *	data,
		size_t	len
	)			[static, protected]

Definition at line 38 of file IgNet.cc.

References IgNet::Bucket::data.

Referenced by onMessage(), releaseFromWait(), requestObject(), run(), sendObjectListToPeer(), and sendObjectToPeer().

{
  b->data.insert(b->data.end(),
                 (const unsigned char *)data,
                 (const unsigned char *)data + len);
}

IgNet::Peer * IgNet::createPeer

(

lat::Socket *

s

)

[protected, virtual]

Definition at line 1225 of file IgNet.cc.

References IgNet::Peer::automatic, IgNet::Peer::mask, p, peers_, IgNet::Peer::sendpos, IgNet::Peer::sendq, IgNet::Peer::socket, IgNet::Peer::source, IgNet::Peer::update, IgNet::Peer::updated, IgNet::Peer::updates, and IgNet::Peer::waiting.

Referenced by IgNet(), onPeerConnect(), and run().

{
  Peer *p = &peers_[s];
  p->socket = 0;
  p->sendq = 0;
  p->sendpos = 0;
  p->mask = 0;
  p->source = false;
  p->update = false;
  p->updated = false;
  p->updates = 0;
  p->waiting = 0;
  p->automatic = 0;
  return p;
}

void IgNet::debug

(

bool

doit

)

Enable or disable verbose debugging.

Must be called before calling run() or start().

Definition at line 816 of file IgNet.cc.

References debug_.

Referenced by IguanaNetProducer::IguanaNetProducer().

{
  debug_ = doit;
}

void IgNet::delay

(

int

delay

)

Set the I/O dispatching delay.

Must be called before calling run() or start().

Definition at line 824 of file IgNet.cc.

References delay_.

{
  delay_ = delay;
}

void IgNet::discard

(

Bucket *&

b

)

[static, private]

Definition at line 47 of file IgNet.cc.

References IgNet::Bucket::next.

Referenced by losePeer(), and onPeerData().

{
  while (b)
  {
    Bucket *next = b->next;
    delete b;
    b = next;
  }
}

IgNet::Object * IgNet::findObject	(	Peer *	p,
		const std::string &	name,
		Peer **	owner = 0
	)			[protected, virtual]

Definition at line 1112 of file IgNet.cc.

References e, i, IgNet::Peer::objs, and peers_.

Referenced by onMessage(), and run().

{
  ObjectMap::iterator pos;
  PeerMap::iterator i, e;
  if (owner)
    *owner = 0;
  if (p)
  {
    pos = p->objs.find(name);
    if (pos == p->objs.end())
      return 0;
    else
    {
      if (owner) *owner = p;
      return &pos->second;
    }
  }
  else
  {
    for (i = peers_.begin(), e = peers_.end(); i != e; ++i)
    {
      pos = i->second.objs.find(name);
      if (pos != i->second.objs.end())
      {
        if (owner) *owner = &i->second;
        return &pos->second;
      }
    }
    return 0;
  }
}

IgNet::Peer * IgNet::getPeer

(

lat::Socket *

s

)

[protected, virtual]

Definition at line 1218 of file IgNet.cc.

References peers_.

Referenced by onPeerData().

{
  PeerMap::iterator pos = peers_.find(s);
  return pos == peers_.end() ? 0 : &pos->second;
}

void IgNet::listenToSource	(	const std::string &	host,
		int	port
	)

Tell the network layer to connect to host and port and automatically receive updates from upstream source(s).

Must be called before calling run() or start().

Definition at line 891 of file IgNet.cc.

References lat::endl(), IgNet::AutoPeer::host, logme(), IgNet::AutoPeer::port, IgNet::AutoPeer::update, and upstream_.

{
  if (! upstream_.host.empty())
  {
    logme()
      << "ERROR: Already receiving data from another collector at "
      << upstream_.host << ":" << upstream_.port << std::endl;
    return;
  }

  upstream_.update = false;
  upstream_.host = host;
  upstream_.port = port;
}

void IgNet::lock

(

void

)

Acquire a lock on the net layer.

Definition at line 931 of file IgNet.cc.

References communicate_, and lock_.

Referenced by onMessage(), edm::service::IguanaService::produceEvent(), receive(), and run().

{
  if (communicate_ != (pthread_t) -1)
    pthread_mutex_lock(&lock_);
}

std::ostream & IgNet::logme

(

void

)

[protected]

Definition at line 29 of file IgNet.cc.

References appname_, TestMuL1L2Filter_cff::cerr, and pid_.

Referenced by IguanaNetProducer::IguanaNetProducer(), listenToSource(), losePeer(), onLocalNotify(), onMessage(), onPeerConnect(), onPeerData(), purgeDeadObjects(), run(), sendObjectListToPeers(), start(), startLocalServer(), updateMask(), and updateToCollector().

{
  return std::cerr
    << Time::current().format(true, "%Y-%m-%d %H:%M:%S")
    << " " << appname_ << "[" << pid_ << "]: ";
}

bool IgNet::losePeer	(	const char *	reason,
		Peer *	peer,
		lat::IOSelectEvent *	event,
		lat::Error *	err = 0
	)			[private]

Handle errors with a peer socket.

Zaps the socket send queue, the socket itself, detaches the socket from the selector, and purges any pending wait requests linked to the socket.

Definition at line 62 of file IgNet.cc.

References IgNet::Peer::automatic, lat::Socket::close(), lat::IOSelector::detach(), discard(), e, lat::endl(), lat::Error::explain(), i, logme(), IgNet::AutoPeer::peer, IgNet::Peer::peeraddr, removePeer(), s, sel_, IgNet::Peer::sendq, IgNet::Peer::socket, lat::IOSelectEvent::source, and waiting_.

Referenced by onPeerData(), and updateMask().

{
  if (reason)
    logme ()
      << reason << peer->peeraddr
      << (err ? "; error was: " + err->explain() : std::string(""))
      << std::endl;

  Socket *s = peer->socket;

  for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    if (i->peer == peer)
      waiting_.erase(i++);
    else
      ++i;

  if (ev)
    ev->source = 0;

  discard(peer->sendq);
  if (peer->automatic)
    peer->automatic->peer = 0;

  sel_.detach(s);
  s->close();
  removePeer(peer, s);
  delete s;
  return true;
}

IgNet::Object * IgNet::makeObject	(	Peer *	p,
		const std::string &	name
	)			[protected, virtual]

Definition at line 1145 of file IgNet.cc.

References IgNet::Object::flags, IgNet::Object::lastreq, IgNet::Object::name, IgNet::Peer::objs, and IgNet::Object::version.

Referenced by onMessage().

{
  Object *o = &p->objs[name];
  o->version = 0;
  o->name = name;
  o->flags = 0;
  o->lastreq = 0;
  return o;
}

void IgNet::markObjectsDead

(

Peer *

p

)

[protected, virtual]

Definition at line 1171 of file IgNet.cc.

References e, i, IgNet::Peer::objs, VIS_FLAG_DEAD, and VIS_FLAG_ZOMBIE.

Referenced by onMessage().

{
  ObjectMap::iterator i, e;
  for (i = p->objs.begin(), e = p->objs.end(); i != e; ++i)
    if (i->second.flags & VIS_FLAG_ZOMBIE)
      i->second.flags = (i->second.flags & ~VIS_FLAG_ZOMBIE) | VIS_FLAG_DEAD;
}

void IgNet::markObjectsZombies

(

Peer *

p

)

[protected, virtual]

Definition at line 1162 of file IgNet.cc.

References e, i, IgNet::Peer::objs, and VIS_FLAG_ZOMBIE.

Referenced by onMessage().

{
  ObjectMap::iterator i, e;
  for (i = p->objs.begin(), e = p->objs.end(); i != e; ++i)
    i->second.flags |= VIS_FLAG_ZOMBIE;
}

bool IgNet::onLocalNotify

(

lat::IOSelectEvent *

ev

)

[private]

React to notifications from the app thread.

This is a simple message to tell this thread to wake up and send unsollicited updates to the peers when new app data appears. We don't send the updates here, but just set a flag to tell the main event pump to send a notification later. This avoids sending unnecessarily frequent object updates.

Definition at line 720 of file IgNet.cc.

References e, lat::endl(), lat::SysErr::ErrTryAgain, lat::Error::explain(), flush_, logme(), lat::Error::next(), lat::SystemError::portable(), lat::IOChannel::read(), and lat::IOSelectEvent::source.

Referenced by IgNet().

{
  // Discard the data in the pipe, we care only about the wakeup.
  try
  {
    IOSize sz;
    unsigned char buf [1024];
    while ((sz = ev->source->read(buf, sizeof(buf))))
      ;
  }
  catch (Error &e)
  {
    SystemError *next = dynamic_cast<SystemError *>(e.next());
    if (next && next->portable() == SysErr::ErrTryAgain)
      ; // Ignore it
    else
      logme()
        << "WARNING: error reading from notification pipe: "
        << e.explain() << std::endl;
  }

  // Tell the main event pump to send an update in a little while.
  flush_ = true;

  // We are never done, always keep going.
  return false;
}

bool IgNet::onMessage	(	Bucket *	msg,
		Peer *	p,
		unsigned char *	data,
		size_t	len
	)			[protected, virtual]

Definition at line 224 of file IgNet.cc.

References copydata(), IgNet::Bucket::data, debug_, lat::endl(), findObject(), IgNet::Object::flags, flags, flush_, IgNet::Object::lastreq, lock(), logme(), makeObject(), markObjectsDead(), markObjectsZombies(), name, IgNet::Peer::peeraddr, IgNet::Object::rawdata, releaseWaiters(), requestObject(), sendObjectListToPeer(), sendObjectToPeer(), IgNet::Peer::source, unlock(), IgNet::Peer::update, IgNet::Peer::updates, IgNet::Object::version, VIS_FLAG_DEAD, VIS_FLAG_RECEIVED, VIS_MSG_GET_OBJECT, VIS_MSG_LIST_OBJECTS, VIS_MSG_UPDATE_ME, VIS_REPLY_LIST_BEGIN, VIS_REPLY_LIST_END, VIS_REPLY_NONE, VIS_REPLY_OBJECT, and waitForData().

Referenced by onPeerData().

{
  // Decode and process this message.
  uint32_t type;
  memcpy (&type, data + sizeof(uint32_t), sizeof (type));
  switch (type)
  {
  case VIS_MSG_UPDATE_ME:
    {
      if (len != 2*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'UPDATE_ME' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      if (debug_)
        logme()
          << "DEBUG: received message 'UPDATE ME' from peer "
          << p->peeraddr << std::endl;

      p->update = true;
    }
    return true;

  case VIS_MSG_LIST_OBJECTS:
    {
      if (debug_)
        logme()
          << "DEBUG: received message 'LIST OBJECTS' from peer "
          << p->peeraddr << std::endl;

      // Send over current status: list of known objects.
      lock();
      sendObjectListToPeer(msg, true, false);
      unlock();
    }
    return true;

  case VIS_MSG_GET_OBJECT:
    {
      if (debug_)
        logme()
          << "DEBUG: received message 'GET OBJECT' from peer "
          << p->peeraddr << std::endl;

      if (len < 3*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'GET IMAGE' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      uint32_t namelen;
      memcpy(&namelen, data + 2*sizeof(uint32_t), sizeof(namelen));
      if (len != 3*sizeof(uint32_t) + namelen)
      {
        logme()
          << "ERROR: corrupt 'GET OBJECT' message of length " << len
          << " from peer " << p->peeraddr
          << ", expected length " << (3*sizeof(uint32_t))
          << " + " << namelen << std::endl;
        return false;
      }

      lock();
      std::string name((char *) data + 3*sizeof(uint32_t), namelen);
      Peer *owner = 0;
      Object *o = findObject(0, name, &owner);
      if (o)
      {
        o->lastreq = Time::current();
        if (o->rawdata.empty())
          waitForData(p, name, "", owner);
        else
          sendObjectToPeer(msg, *o, true);
      }
      else
      {
        uint32_t words[3];
        words[0] = sizeof(words) + name.size();
        words[1] = VIS_REPLY_NONE;
        words[2] = name.size();

        msg->data.reserve(msg->data.size() + words[0]);
        copydata(msg, &words[0], sizeof(words));
        copydata(msg, &name[0], name.size());
      }
      unlock();
    }
    return true;

  case VIS_REPLY_LIST_BEGIN:
    {
      if (len != 4*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'LIST BEGIN' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      if (debug_)
        logme()
          << "DEBUG: received message 'LIST BEGIN' from "
          << p->peeraddr << std::endl;

      // Get the update status: whether this is a full update.
      uint32_t flags;
      memcpy(&flags, data + 3*sizeof(uint32_t), sizeof(uint32_t));

      // If we are about to receive a full list of objects, flag all
      // objects dead.  Subsequent object notifications will undo this
      // for the live objects.  This tells us the object has been
      // removed, but we can keep making it available for a while if
      // there continues to be interest in it.
      if (flags)
      {
        lock();
        markObjectsZombies(p);
        unlock();
      }
    }
    return true;

  case VIS_REPLY_LIST_END:
    {
      if (len != 4*sizeof(uint32_t))
      {
        logme()
          << "ERROR: corrupt 'LIST END' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      // Get the update status: whether this is a full update.
      uint32_t flags;
      memcpy(&flags, data + 3*sizeof(uint32_t), sizeof(uint32_t));

      // If we received a full list of objects, flag all zombie objects
      // now dead. We need to do this in two stages in case we receive
      // updates in many parts, and end up sending updates to others in
      // between; this avoids us lying live objects are dead.
      if (flags)
      {
        lock();
        markObjectsDead(p);
        unlock();
      }

      if (debug_)
        logme()
          << "DEBUG: received message 'LIST END' from "
          << p->peeraddr << std::endl;

      // Indicate we have received another update from this peer.
      // Also indicate we should flush to our clients.
      flush_ = true;
      p->updates++;
    }
    return true;

  case VIS_REPLY_OBJECT:
    {
      uint32_t words[7];
      if (len < sizeof(words))
      {
        logme()
          << "ERROR: corrupt 'OBJECT' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      memcpy(&words[0], data, sizeof(words));
      uint32_t &namelen = words[5];
      uint32_t &datalen = words[6];

      if (len != sizeof(words) + namelen + datalen)
      {
        logme()
          << "ERROR: corrupt 'OBJECT' message of length " << len
          << " from peer " << p->peeraddr
          << ", expected length " << sizeof(words)
          << " + " << namelen
          << " + " << datalen
          << std::endl;
        return false;
      }

      unsigned char *namedata = data + sizeof(words);
      unsigned char *objdata = namedata + namelen;
      unsigned char *enddata = objdata + datalen;
      std::string name((char *) namedata, namelen);
      assert(enddata == data + len);

      if (debug_)
        logme()
          << "DEBUG: received message 'OBJECT " << name
          << "' from " << p->peeraddr << std::endl;

      // Mark the peer as a known object source.
      p->source = true;

      // Initialise or update an object entry.
      lock();
      Object *o = findObject(p, name);
      if (! o)
        o = makeObject(p, name);

      bool hadobject = ! o->rawdata.empty();
      o->flags = words[2] | VIS_FLAG_RECEIVED;
      o->version = ((uint64_t) words[4] << 32 | words[3]);
      o->rawdata.clear();
      o->rawdata.insert(o->rawdata.end(), objdata, enddata);

      // If we had an object for this one already and this is a list
      // update without data, issue an immediate data get request.
      if (hadobject && ! datalen)
        requestObject(p, (namelen ? &name[0] : 0), namelen);

      // If we have the object data, release from wait.
      if (datalen)
        releaseWaiters(o);
      unlock();
    }
    return true;

  case VIS_REPLY_NONE:
    {
      uint32_t words[3];
      if (len < sizeof(words))
      {
        logme()
          << "ERROR: corrupt 'NONE' message of length " << len
          << " from peer " << p->peeraddr << std::endl;
        return false;
      }

      memcpy(&words[0], data, sizeof(words));
      uint32_t &namelen = words[2];

      if (len != sizeof(words) + namelen)
      {
        logme()
          << "ERROR: corrupt 'NONE' message of length " << len
          << " from peer " << p->peeraddr
          << ", expected length " << sizeof(words)
          << " + " << namelen << std::endl;
        return false;
      }

      unsigned char *namedata = data + sizeof(words);
      unsigned char *enddata = namedata + namelen;
      std::string name((char *) namedata, namelen);
      assert(enddata == data + len);

      if (debug_)
        logme()
          << "DEBUG: received message 'NONE " << name
          << "' from " << p->peeraddr << std::endl;

      // Mark the peer as a known object source.
      p->source = true;

      // If this was a known object, update its entry.
      lock();
      Object *o = findObject(p, name);
      if (o)
        o->flags |= VIS_FLAG_DEAD;

      // If someone was waiting for this, let them go.
      releaseWaiters(o);
      unlock();
    }
    return true;

  default:
    logme()
      << "ERROR: unrecognised message of length " << len
      << " and type " << type << " from peer " << p->peeraddr
      << std::endl;
    return false;
  }
}

bool IgNet::onPeerConnect

(

lat::IOSelectEvent *

ev

)

[private]

Respond to new connections on the server socket.

Accepts the connection and creates a new socket for the peer, and sets it up for further communication. Returns false always to tell the IOSelector to keep processing events for the server socket.

Definition at line 680 of file IgNet.cc.

References lat::Socket::accept(), arg, lat::IOSelector::attach(), lat::CreateHook(), createPeer(), debug_, lat::endl(), lat::InetAddress::hostname(), IORead, IOUrgent, lat::IOChannel::isBlocking(), logme(), onPeerData(), p, lat::InetAddress::port(), s, sel_, server_, and lat::IOSelectEvent::source.

Referenced by startLocalServer().

{
  // Recover the server socket.
  assert(ev->source == server_);

  // Accept the connection.
  Socket *s = server_->accept();
  assert(s);
  assert(! s->isBlocking());

  // Record it to our list of peers.
  Peer *p = createPeer(s);
  InetAddress peeraddr = ((InetSocket *) s)->peername();
  InetAddress myaddr = ((InetSocket *) s)->sockname();
  p->peeraddr = StringFormat("%1:%2")
                .arg(peeraddr.hostname())
                .arg(peeraddr.port());
  p->mask = IORead|IOUrgent;
  p->socket = s;

  // Report the new connection.
  if (debug_)
    logme()
      << "INFO: new peer " << p->peeraddr << " is now connected to "
      << myaddr.hostname() << ":" << myaddr.port() << std::endl;

  // Attach it to the listener.
  sel_.attach(s, p->mask, CreateHook(this, &IgNet::onPeerData, p));

  // We are never done.
  return false;
}

bool IgNet::onPeerData	(	lat::IOSelectEvent *	ev,
		Peer *	p
	)			[private]

Handle communication to a particular client.

Definition at line 514 of file IgNet.cc.

References IgNet::Peer::automatic, b, data, IgNet::Bucket::data, debug_, discard(), e, lat::endl(), lat::SysErr::ErrTryAgain, lat::IOSelectEvent::events, getPeer(), IgNet::Peer::incoming, IORead, IOUrgent, IOWrite, len, logme(), losePeer(), IgNet::Peer::mask, MAX_PEER_WAITREQS, MESSAGE_SIZE_LIMIT, alivecheck_mergeAndRegister::msg, IgNet::Bucket::next, lat::Error::next(), old, onMessage(), IgNet::Peer::peeraddr, lat::SystemError::portable(), lat::IOChannel::read(), sel_, IgNet::Peer::sendpos, IgNet::Peer::sendq, lat::IOSelector::setMask(), IgNet::Peer::socket, SOCKET_READ_GROWTH, SOCKET_READ_SIZE, lat::IOSelectEvent::source, TrackValidation_HighPurity_cff::valid, IgNet::Peer::waiting, and lat::IOChannel::write().

Referenced by onPeerConnect(), and run().

{
  assert(getPeer(dynamic_cast<Socket *> (ev->source)) == p);

  // If there is a problem with the peer socket, discard the peer
  // and tell the selector to stop prcessing events for it.  If
  // this is a server connection, we will eventually recreate
  // everything if/when the data server comes back.
  if (ev->events & IOUrgent)
  {
    if (p->automatic)
    {
      logme()
        << "WARNING: connection to the server at " << p->peeraddr
        << " lost (will attempt to reconnect in 15 seconds)\n";
      return losePeer(0, p, ev);
    }
    else
      return losePeer("WARNING: lost peer connection ", p, ev);
  }

  // If we can write to the peer socket, pump whatever we can into it.
  if (ev->events & IOWrite)
  {
    while (Bucket *b = p->sendq)
    {
      IOSize len = b->data.size() - p->sendpos;
      const void *data = (len ? (const void *)&b->data[p->sendpos]
                          : (const void *)&data);
      IOSize done;

      try
      {
        done = (len ? ev->source->write (data, len) : 0);
        if (debug_ && len)
          logme()
            << "DEBUG: sent " << done << " bytes to peer "
            << p->peeraddr << std::endl;
      }
      catch (Error &e)
      {
        return losePeer("WARNING: unable to write to peer ",
                        p, ev, &e);
      }

      p->sendpos += done;
      if (p->sendpos == b->data.size())
      {
        Bucket *old = p->sendq;
        p->sendq = old->next;
        p->sendpos = 0;
        old->next = 0;
        discard(old);
      }

      if (! done && len)
        // Cannot write any more.
        break;
    }
  }

  // If there is data to be read from the peer, first receive what we
  // can get out the socket, the process all complete requests.
  if (ev->events & IORead)
  {
    // First build up the incoming buffer of data in the socket.
    // Remember the last size returned by the socket; we need
    // it to determine if the remote end closed the connection.
    IOSize sz;
    try
    {
      std::vector<unsigned char> buf(SOCKET_READ_SIZE);
      do
        if ((sz = ev->source->read(&buf[0], buf.size())))
        {
          if (debug_)
            logme()
              << "DEBUG: received " << sz << " bytes from peer "
              << p->peeraddr << std::endl;
          DataBlob &data = p->incoming;
          if (data.capacity () < data.size () + sz)
            data.reserve (data.size() + SOCKET_READ_GROWTH);
          data.insert (data.end(), &buf[0], &buf[0] + sz);
        }
      while (sz == sizeof (buf));
    }
    catch (Error &e)
    {
      SystemError *next = dynamic_cast<SystemError *>(e.next());
      if (next && next->portable() == SysErr::ErrTryAgain)
        sz = 1; // Ignore it, and fake no end of data.
      else
        // Houston we have a problem.
        return losePeer("WARNING: failed to read from peer ",
                        p, ev, &e);
    }

    // Process fully received messages as long as we can.
    size_t consumed = 0;
    DataBlob &data = p->incoming;
    while (data.size()-consumed >= sizeof(uint32_t)
           && p->waiting < MAX_PEER_WAITREQS)
    {
      uint32_t msglen;
      memcpy (&msglen, &data[0]+consumed, sizeof(msglen));

      if (msglen >= MESSAGE_SIZE_LIMIT)
        return losePeer("WARNING: excessively large message from ", p, ev);

      if (data.size()-consumed >= msglen)
      {
        bool valid = true;
        if (msglen < 2*sizeof(uint32_t))
        {
          logme()
            << "ERROR: corrupt peer message of length " << msglen
            << " from peer " << p->peeraddr << std::endl;
          valid = false;
        }
        else
        {
          // Decode and process this message.
          Bucket msg;
          msg.next = 0;
          valid = onMessage(&msg, p, &data[0]+consumed, msglen);

          // If we created a response, chain it to the write queue.
          if (! msg.data.empty())
          {
            Bucket **prev = &p->sendq;
            while (*prev)
               prev = &(*prev)->next;

            *prev = new Bucket;
            (*prev)->next = 0;
            (*prev)->data.swap(msg.data);
          }
        }

        if (! valid)
          return losePeer("WARNING: data stream error with ", p, ev);

        consumed += msglen;
      }
      else
        break;
    }

    data.erase(data.begin(), data.begin()+consumed);

    // If the client has closed the connection, shut down our end.  If
    // we have something to send back still, leave the write direction
    // open.  Otherwise close the shop for this client.
    if (sz == 0)
      sel_.setMask(p->socket, p->mask &= ~IORead);
  }

  // Yes, please keep processing events for this socket.
  return false;
}

IgNet& IgNet::operator=

(

const IgNet &

)

[private]

void IgNet::purgeDeadObjects	(	lat::Time	oldobj,
		lat::Time	deadobj
	)			[protected, virtual]

Definition at line 1181 of file IgNet.cc.

References debug_, lat::endl(), IgNet::Object::flags, IgNet::Object::lastreq, logme(), IgNet::Object::name, peers_, pi, IgNet::Object::rawdata, IgNet::Object::version, VIS_FLAG_DEAD, and VIS_FLAG_SCALAR.

Referenced by run().

{
  PeerMap::iterator pi, pe;
  ObjectMap::iterator oi, oe;
  for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
    for (oi = pi->second.objs.begin(), oe = pi->second.objs.end(); oi != oe; )
    {
      Object &o = oi->second;

      // Compact non-scalar objects that are unused.  We send scalar
      // objects to the web server so we keep them around.
      if (o.lastreq < oldobj && ! o.rawdata.empty() && ! (o.flags & VIS_FLAG_SCALAR))
      {
        if (debug_)
          logme()
            << "DEBUG: compacting idle '" << o.name
            << "' from " << pi->second.peeraddr << std::endl;
      }

      // Remove if dead, old and unused.
      if (o.lastreq < deadobj
          && o.version < deadobj
          && (o.flags & VIS_FLAG_DEAD))
      {
        if (debug_)
          logme()
            << "DEBUG: removing dead '" << o.name
            << "' from " << pi->second.peeraddr << std::endl;

        pi->second.objs.erase(oi++);
      }
      else
        ++oi;
    }
}

int IgNet::receive	(	void(*)(void *arg, uint32_t reason, Object &obj)	callback,
		void *	arg
	)			[virtual]

Definition at line 1335 of file IgNet.cc.

References IgNet::Object::flags, local_, lock(), IgNet::Peer::objs, p, peers_, pi, unlock(), IgNet::Peer::updates, VIS_FLAG_DEAD, and VIS_FLAG_RECEIVED.

{
  int updates = 0;

  lock();
  PeerMap::iterator pi, pe;
  ObjectMap::iterator oi, oe;
  for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
  {
    Peer &p = pi->second;
    if (&p == local_)
      continue;

    updates += p.updates;

    for (oi = p.objs.begin(), oe = p.objs.end(); oi != oe; )
    {
      Object &o = oi->second;
      if (o.flags & VIS_FLAG_DEAD)
      {
        callback(arg, VIS_FLAG_DEAD, o);
        p.objs.erase(oi++);
      }
      else if (o.flags & VIS_FLAG_RECEIVED)
      {
        callback(arg, VIS_FLAG_RECEIVED, o);
        o.flags &= ~VIS_FLAG_RECEIVED;
        ++oi;
      }
      else
        ++oi;
    }
  }
  unlock();

  return updates;
}

void IgNet::releaseFromWait	(	WaitList::iterator	i,
		Object *	o
	)			[private]

Definition at line 133 of file IgNet.cc.

References alivecheck_mergeAndRegister::msg, IgNet::Bucket::next, releaseFromWait(), and waiting_.

{
  Bucket **msg = &i->peer->sendq;
  while (*msg)
    msg = &(*msg)->next;
  *msg = new Bucket;
  (*msg)->next = 0;

  releaseFromWait(*msg, *i, o);

  assert(i->peer->waiting > 0);
  i->peer->waiting--;
  waiting_.erase(i);
}

void IgNet::releaseFromWait	(	Bucket *	msg,
		WaitObject &	w,
		Object *	o
	)			[protected, virtual]

Definition at line 170 of file IgNet.cc.

References copydata(), IgNet::Bucket::data, IgNet::WaitObject::name, sendObjectToPeer(), and VIS_REPLY_NONE.

Referenced by releaseFromWait(), releaseWaiters(), and run().

{
  if (o)
    sendObjectToPeer (msg, *o, true);
  else
  {
    uint32_t words[3];
    words[0] = sizeof(words) + w.name.size();
    words[1] = VIS_REPLY_NONE;
    words[2] = w.name.size();

    msg->data.reserve(msg->data.size() + words[0]);
    copydata(msg, &words[0], sizeof(words));
    copydata(msg, &w.name[0], w.name.size());
  }
}

void IgNet::releaseWaiters

(

Object *

o

)

[private]

Definition at line 150 of file IgNet.cc.

References e, i, IgNet::Object::name, releaseFromWait(), and waiting_.

Referenced by onMessage().

{
  for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    if (i->name == o->name)
      releaseFromWait(i++, o);
    else
      ++i;
}

void IgNet::removeLocalObject

(

const std::string &

path

)

[virtual]

Delete the local object.

The caller must call sendLocalChanges() later to push out the changes.

Definition at line 1393 of file IgNet.cc.

References local_, and IgNet::Peer::objs.

Referenced by edm::service::IguanaService::produceEvent().

{
  local_->objs.erase(path);
}

void IgNet::removePeer	(	Peer *	p,
		lat::Socket *	s
	)			[protected, virtual]

Definition at line 1242 of file IgNet.cc.

References IgNet::Peer::objs, peers_, and sendLocalChanges().

Referenced by losePeer().

{
  bool needflush = ! p->objs.empty();

  p->objs.clear();
  peers_.erase(s);

  // If we removed a peer with objects, our list of objects
  // has changed and we need to update downstream peers.
  if (needflush)
    sendLocalChanges();
}

void IgNet::requestObject	(	Peer *	p,
		const char *	name,
		size_t	len
	)			[private]

Queue an object request to the data server.

Definition at line 97 of file IgNet.cc.

References copydata(), alivecheck_mergeAndRegister::msg, IgNet::Bucket::next, IgNet::Peer::sendq, and VIS_MSG_GET_OBJECT.

Referenced by onMessage(), and waitForData().

{
  Bucket **msg = &p->sendq;
  while (*msg)
    msg = &(*msg)->next;
  *msg = new Bucket;
  (*msg)->next = 0;

  uint32_t words[3];
  words[0] = sizeof(words) + len;
  words[1] = VIS_MSG_GET_OBJECT;
  words[2] = len;
  (*msg)->data.reserve((*msg)->data.size() + words[0]);
  copydata(*msg, words, sizeof(words));
  copydata(*msg, name, len);
}

void IgNet::run

(

void

)

Run the actual I/O processing loop.

Definition at line 964 of file IgNet.cc.

References lat::Socket::abort(), arg, lat::IOSelector::attach(), IgNet::Peer::automatic, lat::InetSocket::connect(), copydata(), lat::CreateHook(), createPeer(), debug_, delay_, lat::IOSelector::dispatch(), downstream_, e, lat::endl(), lat::SysErr::ErrOperationInProgress, lat::Error::explain(), findObject(), flush_, IgNet::AutoPeer::host, lat::InetAddress::hostname(), i, IORead, IOUrgent, IOWrite, lock(), logme(), IgNet::Peer::mask, IgNet::Bucket::next, IgNet::AutoPeer::next, lat::Error::next(), onPeerData(), lat::SocketConst::OptSockReceiveBuffer, lat::SocketConst::OptSockSendBuffer, p, IgNet::AutoPeer::peer, IgNet::Peer::peeraddr, IgNet::AutoPeer::port, lat::InetAddress::port(), lat::SystemError::portable(), purgeDeadObjects(), releaseFromWait(), s, sel_, sendObjectListToPeers(), IgNet::Peer::sendq, lat::IOChannel::setBlocking(), lat::Socket::setopt(), shouldStop(), IgNet::Peer::socket, SOCKET_BUF_SIZE, lat::SocketConst::TypeStream, unlock(), IgNet::Peer::update, IgNet::AutoPeer::update, updatePeerMasks(), upstream_, VIS_MSG_LIST_OBJECTS, VIS_MSG_UPDATE_ME, waiting_, and IgNet::AutoPeer::warned.

{
  Time now;
  AutoPeer *automatic[2] = { &upstream_, &downstream_ };

  // Perform I/O.  Every once in a while flush updates to peers.
  while (! shouldStop())
  {
    for (int i = 0; i < 2; ++i)
    {
      AutoPeer *ap = automatic[i];

      // If we need a server connection and don't have one yet,
      // initiate asynchronous connection creation.  Swallow errors
      // in case the server won't talk to us.
      if (! ap->host.empty()
          && ! ap->peer
          && (now = Time::current()) > ap->next)
      {
        ap->next = now + TimeSpan(0, 0, 0, 15 /* seconds */, 0);
        InetSocket *s = 0;
        try
        {
          s = new InetSocket (SocketConst::TypeStream);
          s->setBlocking (false);
          s->connect(InetAddress (ap->host.c_str(), ap->port));
          s->setopt(lat::SocketConst::OptSockSendBuffer, SOCKET_BUF_SIZE);
          s->setopt(lat::SocketConst::OptSockReceiveBuffer, SOCKET_BUF_SIZE);
        }
        catch (Error &e)
        {
          SystemError *sys = dynamic_cast<SystemError *>(e.next());
          if (! sys || sys->portable() != SysErr::ErrOperationInProgress)
          {
            // "In progress" just means the connection is in progress.
            // The connection is ready when the socket is writeable.
            // Anything else is a real problem.
            if (! ap->warned)
            {
              logme()
                << "NOTE: server at " << ap->host << ":" << ap->port
                << " is unavailable.  Connection will be established"
                << " automatically on the background once the server"
                << " becomes available.  Error from the attempt was: "
                << e.explain() << '\n';
              ap->warned = true;
            }

            if (s)
              s->abort();
            delete s;
            s = 0;
          }
        }

        // Set up with the selector if we were successful.  If this is
        // the upstream collector, queue a request for updates.
        if (s)
        {
          lock();
          Peer *p = createPeer(s);
          ap->peer = p;
          ap->warned = false;
          unlock();

          InetAddress peeraddr = ((InetSocket *) s)->peername();
          InetAddress myaddr = ((InetSocket *) s)->sockname();
          p->peeraddr = StringFormat("%1:%2")
                        .arg(peeraddr.hostname())
                        .arg(peeraddr.port());
          p->mask = IORead|IOWrite|IOUrgent;
          p->update = ap->update;
          p->automatic = ap;
          p->socket = s;
          sel_.attach(s, p->mask, CreateHook(this, &IgNet::onPeerData, p));
          if (ap == &upstream_)
          {
            uint32_t words[4] = { 2*sizeof(uint32_t), VIS_MSG_LIST_OBJECTS,
                                  2*sizeof(uint32_t), VIS_MSG_UPDATE_ME };
            p->sendq = new Bucket;
            p->sendq->next = 0;
            copydata(p->sendq, words, sizeof(words));
          }

          // Report the new connection.
          if (debug_)
            logme()
              << "INFO: now connected to " << p->peeraddr << " from "
              << myaddr.hostname() << ":" << myaddr.port() << std::endl;
        }
      }
    }

    // Pump events for a while.
    sel_.dispatch(delay_);
    now = Time::current();

    // Check if flush is required.  Flush only if one is needed.
    // Always sends the full object list.  Compact objects no longer
    // in active use before sending out the update.
    if (flush_)
    {
      flush_ = false;

      lock();
      purgeDeadObjects(now - TimeSpan(0, 0, 2 /* minutes */, 0, 0),
                       now - TimeSpan(0, 0, 20 /* minutes */, 0, 0));
      sendObjectListToPeers(true);
      unlock();
    }

    // Update the data server and peer selection masks.  If we
    // have no more data to send and listening for writes, remove
    // the write mask.  If we have something to write and aren't
    // listening for writes, start listening so we can send off
    // the data.
    updatePeerMasks();

    // Release peers that have been waiting for data for too long.
    lock();
    Time waitold = now - TimeSpan(0, 0, 2 /* minutes */, 0, 0);
    for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    {
      // If the peer has waited for too long, send something.
      if (i->time < waitold)
        releaseFromWait(i++, findObject(0, i->name));

      // Keep it for now.
      else
        ++i;
    }
    unlock();
  }
}

void IgNet::sendLocalChanges

(

void

)

Definition at line 1102 of file IgNet.cc.

References lat::Pipe::sink(), wakeup_, and lat::IOChannel::write().

Referenced by edm::service::IguanaService::produceEvent(), and removePeer().

{
  char byte = 0;
  wakeup_.sink()->write(&byte, 1);
}

void IgNet::sendObjectListToPeer	(	Bucket *	msg,
		bool	all,
		bool	clear
	)			[protected, virtual]

Send all objects to a peer and optionally mark sent objects old.

Definition at line 1257 of file IgNet.cc.

References copydata(), IgNet::Bucket::data, peers_, pi, sendObjectToPeer(), VIS_FLAG_NEW, VIS_REPLY_LIST_BEGIN, and VIS_REPLY_LIST_END.

Referenced by onMessage(), and sendObjectListToPeers().

{
  PeerMap::iterator pi, pe;
  ObjectMap::iterator oi, oe;
  uint32_t numobjs = 0;
  for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
    numobjs += pi->second.objs.size();

  msg->data.reserve(msg->data.size() + 300*numobjs);

  uint32_t nupdates = 0;
  uint32_t words[4];
  words[0] = sizeof(words);
  words[1] = VIS_REPLY_LIST_BEGIN;
  words[2] = numobjs;
  words[3] = all;
  copydata(msg, &words[0], sizeof(words));

  for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
    for (oi = pi->second.objs.begin(), oe = pi->second.objs.end(); oi != oe; ++oi)
      if (all || (oi->second.flags & VIS_FLAG_NEW))
      {
        sendObjectToPeer(msg, oi->second, false);
        if (clear)
          oi->second.flags &= ~VIS_FLAG_NEW;
        ++nupdates;
      }

  words[1] = VIS_REPLY_LIST_END;
  words[2] = nupdates;
  copydata(msg, &words[0], sizeof(words));
}

void IgNet::sendObjectListToPeers

(

bool

all

)

[protected, virtual]

Definition at line 1291 of file IgNet.cc.

References IgNet::Bucket::data, debug_, e, lat::endl(), i, logme(), alivecheck_mergeAndRegister::msg, IgNet::Bucket::next, p, IgNet::Peer::peeraddr, peers_, sendObjectListToPeer(), IgNet::Peer::sendq, IgNet::Peer::update, and IgNet::Peer::updated.

Referenced by run().

{
  PeerMap::iterator i, e;
  for (i = peers_.begin(), e = peers_.end(); i != e; ++i)
  {
    Peer &p = i->second;
    if (! p.update)
      continue;

    if (debug_)
      logme()
        << "DEBUG: notifying " << p.peeraddr << std::endl;

    Bucket msg;
    msg.next = 0;
    sendObjectListToPeer(&msg, !p.updated || all, true);

    if (! msg.data.empty())
    {
      Bucket **prev = &p.sendq;
      while (*prev)
        prev = &(*prev)->next;

      *prev = new Bucket;
      (*prev)->next = 0;
      (*prev)->data.swap(msg.data);
    }
    p.updated = true;
  }
}

void IgNet::sendObjectToPeer	(	Bucket *	msg,
		Object &	o,
		bool	data
	)			[protected]

Definition at line 191 of file IgNet.cc.

References copydata(), IgNet::Bucket::data, IgNet::Object::flags, flags, IgNet::Object::name, IgNet::Object::rawdata, IgNet::Object::version, VIS_FLAG_SCALAR, VIS_FLAG_ZOMBIE, and VIS_REPLY_OBJECT.

Referenced by onMessage(), releaseFromWait(), and sendObjectListToPeer().

{
  uint32_t flags = o.flags & ~VIS_FLAG_ZOMBIE;
  DataBlob objdata;

  if (data || (flags & VIS_FLAG_SCALAR))
    objdata.insert(objdata.end(),
                   &o.rawdata[0],
                   &o.rawdata[0] + o.rawdata.size());

  uint32_t words[7];
  uint32_t namelen = o.name.size();
  uint32_t datalen = objdata.size();

  words[0] = sizeof(words) + namelen + datalen;
  words[1] = VIS_REPLY_OBJECT;
  words[2] = flags;
  words[3] = (o.version >> 0 ) & 0xffffffff;
  words[4] = (o.version >> 32) & 0xffffffff;
  words[5] = namelen;
  words[6] = datalen;

  msg->data.reserve(msg->data.size() + words[0]);
  copydata(msg, &words[0], sizeof(words));
  if (namelen)
    copydata(msg, &o.name[0], namelen);
  if (datalen)
    copydata(msg, &objdata[0], datalen);
}

bool IgNet::shouldStop

(

void

)

[protected, virtual]

Reimplemented in IguanaNetProducer.

Definition at line 162 of file IgNet.cc.

References shutdown_.

Referenced by run().

{
  return shutdown_;
}

void IgNet::shutdown

(

void

)

Stop the network layer and wait it to finish.

Definition at line 908 of file IgNet.cc.

References communicate_, and shutdown_.

{
  shutdown_ = 1;
  if (communicate_ != (pthread_t) -1)
    pthread_join(communicate_, 0);
}

void IgNet::start

(

void

)

Start running the network layer in a new thread.

This is an exclusive alternative to the run() method, which runs the network layer in the caller's thread.

Definition at line 949 of file IgNet.cc.

References communicate(), communicate_, lock_, logme(), and pthread_create.

Referenced by edm::service::IguanaService::init().

{
  if (communicate_ != (pthread_t) -1)
  {
    logme()
      << "ERROR: Shared memory networking thread has already been started\n";
    return;
  }

  pthread_mutex_init(&lock_, 0);
  pthread_create(&communicate_, 0, &communicate, this);
}

void IgNet::startLocalServer

(

int

port

)

Start a server socket for accessing this node remotely.

Must be called before calling run() or start(). May throw an Exception if the server socket cannot be initialised.

Definition at line 833 of file IgNet.cc.

References lat::IOSelector::attach(), lat::Error::clone(), lat::CreateHook(), e, lat::endl(), lat::Error::explain(), IOAccept, logme(), onPeerConnect(), lat::SocketConst::OptSockReceiveBuffer, lat::SocketConst::OptSockSendBuffer, sel_, server_, lat::IOChannel::setBlocking(), lat::Socket::setopt(), and SOCKET_BUF_SIZE.

Referenced by IguanaNetProducer::IguanaNetProducer().

{
  if (server_)
  {
    logme() << "ERROR: server was already started.\n";
    return;
  }

  try
  {
    server_ = new InetServerSocket(InetAddress (port), 10);
    server_->setopt(lat::SocketConst::OptSockSendBuffer, SOCKET_BUF_SIZE);
    server_->setopt(lat::SocketConst::OptSockReceiveBuffer, SOCKET_BUF_SIZE);
    server_->setBlocking(false);
    sel_.attach(server_, IOAccept, CreateHook(this, &IgNet::onPeerConnect));
  }
  catch (Error &e)
  {
    // FIXME: Do we need to do this when we throw an exception anyway?
    // FIXME: Abort instead?
    logme()
      << "ERROR: Failed to start server at port " << port << ": "
      << e.explain() << std::endl;

    throw IgNetError("Failed to start server at port ", e.clone());
    
    // FIXME: Throw something simpler that removes the dependency?
//     throw cms::Exception("IgNet::startLocalServer")
//       << "Failed to start server at port " << port << ": "
//       << e.explain();
  }
  
  logme() << "INFO: Shared memory server started at port " << port << std::endl;
}

void IgNet::unlock

(

void

)

Release the lock on the net layer.

Definition at line 939 of file IgNet.cc.

References communicate_, and lock_.

Referenced by onMessage(), edm::service::IguanaService::produceEvent(), receive(), and run().

{
  if (communicate_ != (pthread_t) -1)
    pthread_mutex_unlock(&lock_);
}

void IgNet::updateLocalObject

(

Object &

o

)

[virtual]

Update the network cache for an object.

The caller must call sendLocalChanges() later to push out the changes.

Definition at line 1376 of file IgNet.cc.

References IgNet::Object::flags, local_, IgNet::Object::name, IgNet::Peer::objs, IgNet::Object::rawdata, std::swap(), and IgNet::Object::version.

Referenced by edm::service::IguanaService::produceEvent().

{
  ObjectMap::iterator pos = local_->objs.find(o.name);
  if (pos == local_->objs.end())
    local_->objs.insert(ObjectMap::value_type(o.name, o));
  else
  {
    std::swap(pos->second.version,   o.version);
    std::swap(pos->second.flags,     o.flags);
    std::swap(pos->second.rawdata,   o.rawdata);
    pos->second.lastreq = 0;
  }
}

void IgNet::updateMask

(

Peer *

p

)

[protected]

Update the selector mask for a peer based on data queues.

Close the connection if there is no reason to maintain it open.

Definition at line 751 of file IgNet.cc.

References debug_, lat::endl(), IOUrgent, IOWrite, logme(), losePeer(), IgNet::Peer::mask, IgNet::Peer::peeraddr, sel_, IgNet::Peer::sendq, lat::IOSelector::setMask(), IgNet::Peer::socket, and IgNet::Peer::waiting.

Referenced by updatePeerMasks().

{
  if (! p->socket)
    return;

  // Listen to writes iff we have data to send.
  unsigned oldmask = p->mask;
  if (! p->sendq && (p->mask & IOWrite))
    sel_.setMask(p->socket, p->mask &= ~IOWrite);

  if (p->sendq && ! (p->mask & IOWrite))
    sel_.setMask(p->socket, p->mask |= IOWrite);

  if (debug_ && oldmask != p->mask)
    logme()
      << "DEBUG: updating mask for " << p->peeraddr << " to "
      << p->mask << " from " << oldmask << std::endl;

  // If we have nothing more to send and are no longer listening
  // for reads, close up the shop for this peer.
  if (p->mask == IOUrgent && ! p->waiting)
  {
    assert(! p->sendq);
    if (debug_)
      logme() << "INFO: connection closed to " << p->peeraddr << std::endl;
    losePeer(0, p, 0);
  }
}

void IgNet::updatePeerMasks

(

void

)

[protected, virtual]

Definition at line 1323 of file IgNet.cc.

References e, i, peers_, and updateMask().

Referenced by run().

{
  PeerMap::iterator i, e;
  for (i = peers_.begin(), e = peers_.end(); i != e; )
    updateMask(&(i++)->second);
}

void IgNet::updateToCollector	(	const std::string &	host,
		int	port
	)

Tell the network layer to connect to host and port and automatically send updates whenever local data changes.

Must be called before calling run() or start().

Definition at line 872 of file IgNet.cc.

References downstream_, lat::endl(), IgNet::AutoPeer::host, logme(), IgNet::AutoPeer::port, and IgNet::AutoPeer::update.

{
  if (! downstream_.host.empty())
  {
    logme()
      << "ERROR: Already updating another collector at "
      << downstream_.host << ":" << downstream_.port << std::endl;
    return;
  }

  downstream_.update = true;
  downstream_.host = host;
  downstream_.port = port;
}

void IgNet::waitForData	(	Peer *	p,
		const std::string &	name,
		const std::string &	info,
		Peer *	owner
	)			[protected]

Queue a request for an object and put a peer into the mode of waiting for object data to appear.

Definition at line 117 of file IgNet.cc.

References requestObject(), IgNet::Peer::waiting, and waiting_.

Referenced by onMessage().

{
  // FIXME: Should we automatically record which exact peer the waiter
  // is expecting to deliver data so we know to release the waiter if
  // the other peer vanishes?  The current implementation stands a
  // chance for the waiter to wait indefinitely -- although we do
  // force terminate the wait after a while.
  requestObject(owner, name.size() ? &name[0] : 0, name.size());
  WaitObject wo = { Time::current(), name, info, p };
  waiting_.push_back(wo);
  p->waiting++;
}

---

Member Data Documentation

std::string IgNet::appname_ [private]

Definition at line 162 of file IgNet.h.

Referenced by logme().

pthread_t IgNet::communicate_ [private]

Definition at line 177 of file IgNet.h.

Referenced by lock(), shutdown(), start(), and unlock().

bool IgNet::debug_ [protected]

Definition at line 146 of file IgNet.h.

Referenced by debug(), onMessage(), onPeerConnect(), onPeerData(), purgeDeadObjects(), run(), sendObjectListToPeers(), and updateMask().

int IgNet::delay_ [private]

Definition at line 180 of file IgNet.h.

Referenced by delay(), and run().

AutoPeer IgNet::downstream_ [private]

Definition at line 172 of file IgNet.h.

Referenced by IgNet(), run(), and updateToCollector().

bool IgNet::flush_ [private]

Definition at line 181 of file IgNet.h.

Referenced by onLocalNotify(), onMessage(), and run().

Peer* IgNet::local_ [private]

Definition at line 174 of file IgNet.h.

Referenced by IgNet(), receive(), removeLocalObject(), and updateLocalObject().

pthread_mutex_t IgNet::lock_ [private]

Definition at line 176 of file IgNet.h.

Referenced by lock(), start(), and unlock().

const uint32_t IgNet::MAX_PEER_WAITREQS = 128 [static]

Definition at line 37 of file IgNet.h.

Referenced by onPeerData().

PeerMap IgNet::peers_ [private]

Definition at line 170 of file IgNet.h.

Referenced by createPeer(), findObject(), getPeer(), purgeDeadObjects(), receive(), removePeer(), sendObjectListToPeer(), sendObjectListToPeers(), and updatePeerMasks().

int IgNet::pid_ [private]

Definition at line 163 of file IgNet.h.

Referenced by logme().

lat::IOSelector IgNet::sel_ [private]

Definition at line 165 of file IgNet.h.

Referenced by IgNet(), losePeer(), onPeerConnect(), onPeerData(), run(), startLocalServer(), and updateMask().

lat::InetServerSocket* IgNet::server_ [private]

Definition at line 166 of file IgNet.h.

Referenced by onPeerConnect(), and startLocalServer().

sig_atomic_t IgNet::shutdown_ [private]

Definition at line 178 of file IgNet.h.

Referenced by shouldStop(), and shutdown().

AutoPeer IgNet::upstream_ [private]

Definition at line 171 of file IgNet.h.

Referenced by IgNet(), listenToSource(), and run().

lat::Time IgNet::version_ [private]

Definition at line 168 of file IgNet.h.

const uint32_t IgNet::VIS_FLAG_DEAD = 0x40000000 [static]

Definition at line 34 of file IgNet.h.

Referenced by markObjectsDead(), onMessage(), purgeDeadObjects(), and receive().

const uint32_t IgNet::VIS_FLAG_NEW = 0x20000000 [static]

Definition at line 33 of file IgNet.h.

Referenced by edm::service::IguanaService::produceEvent(), and sendObjectListToPeer().

const uint32_t IgNet::VIS_FLAG_RECEIVED = 0x10000000 [static]

Definition at line 32 of file IgNet.h.

Referenced by onMessage(), and receive().

const uint32_t IgNet::VIS_FLAG_SCALAR = 0x1 [static]

Definition at line 31 of file IgNet.h.

Referenced by edm::service::IguanaService::produceEvent(), purgeDeadObjects(), and sendObjectToPeer().

const uint32_t IgNet::VIS_FLAG_ZOMBIE = 0x80000000 [static]

Definition at line 35 of file IgNet.h.

Referenced by markObjectsDead(), markObjectsZombies(), and sendObjectToPeer().

const uint32_t IgNet::VIS_MSG_GET_OBJECT = 3 [static]

Definition at line 24 of file IgNet.h.

Referenced by onMessage(), and requestObject().

const uint32_t IgNet::VIS_MSG_HELLO = 0 [static]

Definition at line 21 of file IgNet.h.

const uint32_t IgNet::VIS_MSG_LIST_OBJECTS = 2 [static]

Definition at line 23 of file IgNet.h.

Referenced by onMessage(), and run().

const uint32_t IgNet::VIS_MSG_UPDATE_ME = 1 [static]

Definition at line 22 of file IgNet.h.

Referenced by onMessage(), and run().

const uint32_t IgNet::VIS_REPLY_LIST_BEGIN = 101 [static]

Definition at line 26 of file IgNet.h.

Referenced by onMessage(), and sendObjectListToPeer().

const uint32_t IgNet::VIS_REPLY_LIST_END = 102 [static]

Definition at line 27 of file IgNet.h.

Referenced by onMessage(), and sendObjectListToPeer().

const uint32_t IgNet::VIS_REPLY_NONE = 103 [static]

Definition at line 28 of file IgNet.h.

Referenced by onMessage(), and releaseFromWait().

const uint32_t IgNet::VIS_REPLY_OBJECT = 104 [static]

Definition at line 29 of file IgNet.h.

Referenced by onMessage(), and sendObjectToPeer().

WaitList IgNet::waiting_ [private]

Definition at line 173 of file IgNet.h.

Referenced by losePeer(), releaseFromWait(), releaseWaiters(), run(), and waitForData().

lat::Pipe IgNet::wakeup_ [private]

Definition at line 167 of file IgNet.h.

Referenced by IgNet(), and sendLocalChanges().

---

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

• Iguana/Framework/interface/IgNet.h
• Iguana/Framework/src/IgNet.cc

---