---

VisNet Class Reference

#include <VisFramework/VisFrameworkBase/interface/VisNet.h>

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.
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.
	VisNet (const std::string &appname="")
virtual	~VisNet (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
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.
VisNet &	operator= (const VisNet &)
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.
	VisNet (const VisNet &)
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 VisNet.h.

---

Member Typedef Documentation

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

Definition at line 41 of file VisNet.h.

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

Definition at line 45 of file VisNet.h.

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

Definition at line 46 of file VisNet.h.

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

Definition at line 44 of file VisNet.h.

---

Constructor & Destructor Documentation

VisNet::VisNet

(

const std::string &

appname = ""

)

Definition at line 781 of file VisNet.cc.

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

  : debug_ (false),
    appname_ (appname.empty() ? "VisNet" : 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, &VisNet::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);
}

VisNet::~VisNet

(

void

)

[virtual]

Definition at line 808 of file VisNet.cc.

{
  // FIXME
}

VisNet::VisNet

(

const VisNet &

)

[private]

---

Member Function Documentation

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

Definition at line 38 of file VisNet.cc.

References VisNet::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);
}

VisNet::Peer * VisNet::createPeer

(

lat::Socket *

s

)

[protected, virtual]

Definition at line 1223 of file VisNet.cc.

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

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

{
  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 VisNet::debug

(

bool

doit

)

Enable or disable verbose debugging.

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

Definition at line 816 of file VisNet.cc.

References debug_.

{
  debug_ = doit;
}

void VisNet::delay

(

int

delay

)

Set the I/O dispatching delay.

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

Definition at line 824 of file VisNet.cc.

References delay_.

{
  delay_ = delay;
}

void VisNet::discard

(

Bucket *&

b

)

[static, private]

Definition at line 47 of file VisNet.cc.

References VisNet::Bucket::next.

Referenced by losePeer(), and onPeerData().

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

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

Definition at line 1110 of file VisNet.cc.

References e, i, VisNet::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;
  }
}

VisNet::Peer * VisNet::getPeer

(

lat::Socket *

s

)

[protected, virtual]

Definition at line 1216 of file VisNet.cc.

References peers_.

Referenced by onPeerData().

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

void VisNet::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 889 of file VisNet.cc.

References lat::endl(), VisNet::AutoPeer::host, logme(), VisNet::AutoPeer::port, VisNet::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 VisNet::lock

(

void

)

Acquire a lock on the net layer.

Definition at line 929 of file VisNet.cc.

References communicate_, and lock_.

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

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

std::ostream & VisNet::logme

(

void

)

[protected]

Definition at line 29 of file VisNet.cc.

References appname_, TestMuL1L2Filter_cff::cerr, and pid_.

Referenced by 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 VisNet::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 VisNet.cc.

References VisNet::Peer::automatic, lat::Socket::close(), lat::IOSelector::detach(), discard(), e, lat::endl(), lat::Error::explain(), i, logme(), VisNet::AutoPeer::peer, VisNet::Peer::peeraddr, removePeer(), s, sel_, VisNet::Peer::sendq, VisNet::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;
}

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

Definition at line 1143 of file VisNet.cc.

References VisNet::Object::flags, VisNet::Object::lastreq, VisNet::Object::name, VisNet::Peer::objs, and VisNet::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 VisNet::markObjectsDead

(

Peer *

p

)

[protected, virtual]

Definition at line 1169 of file VisNet.cc.

References e, i, VisNet::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 VisNet::markObjectsZombies

(

Peer *

p

)

[protected, virtual]

Definition at line 1160 of file VisNet.cc.

References e, i, VisNet::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 VisNet::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 VisNet.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 VisNet().

{
  // 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 VisNet::onMessage	(	Bucket *	msg,
		Peer *	p,
		unsigned char *	data,
		size_t	len
	)			[protected, virtual]

Definition at line 224 of file VisNet.cc.

References copydata(), VisNet::Bucket::data, debug_, lat::endl(), findObject(), VisNet::Object::flags, flags, flush_, VisNet::Object::lastreq, lock(), logme(), makeObject(), markObjectsDead(), markObjectsZombies(), name, VisNet::Peer::peeraddr, VisNet::Object::rawdata, releaseWaiters(), requestObject(), sendObjectListToPeer(), sendObjectToPeer(), VisNet::Peer::source, unlock(), VisNet::Peer::update, VisNet::Peer::updates, VisNet::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 VisNet::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 VisNet.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, &VisNet::onPeerData, p));

  // We are never done.
  return false;
}

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

Handle communication to a particular client.

Definition at line 514 of file VisNet.cc.

References VisNet::Peer::automatic, b, data, VisNet::Bucket::data, debug_, discard(), e, lat::endl(), lat::SysErr::ErrTryAgain, lat::IOSelectEvent::events, getPeer(), VisNet::Peer::incoming, IORead, IOUrgent, IOWrite, len, logme(), losePeer(), VisNet::Peer::mask, MAX_PEER_WAITREQS, MESSAGE_SIZE_LIMIT, alivecheck_mergeAndRegister::msg, VisNet::Bucket::next, lat::Error::next(), old, onMessage(), VisNet::Peer::peeraddr, lat::SystemError::portable(), lat::IOChannel::read(), sel_, VisNet::Peer::sendpos, VisNet::Peer::sendq, lat::IOSelector::setMask(), VisNet::Peer::socket, SOCKET_READ_GROWTH, SOCKET_READ_SIZE, lat::IOSelectEvent::source, TrackValidation_HighPurity_cff::valid, VisNet::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;
}

VisNet& VisNet::operator=

(

const VisNet &

)

[private]

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

Definition at line 1179 of file VisNet.cc.

References debug_, lat::endl(), VisNet::Object::flags, VisNet::Object::lastreq, logme(), VisNet::Object::name, peers_, pi, VisNet::Object::rawdata, VisNet::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 VisNet::receive	(	void(*)(void *arg, uint32_t reason, Object &obj)	callback,
		void *	arg
	)			[virtual]

Definition at line 1333 of file VisNet.cc.

References VisNet::Object::flags, local_, lock(), VisNet::Peer::objs, p, peers_, pi, unlock(), VisNet::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 VisNet::releaseFromWait	(	WaitList::iterator	i,
		Object *	o
	)			[private]

Definition at line 133 of file VisNet.cc.

References alivecheck_mergeAndRegister::msg, VisNet::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 VisNet::releaseFromWait	(	Bucket *	msg,
		WaitObject &	w,
		Object *	o
	)			[protected, virtual]

Definition at line 170 of file VisNet.cc.

References copydata(), VisNet::Bucket::data, VisNet::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 VisNet::releaseWaiters

(

Object *

o

)

[private]

Definition at line 150 of file VisNet.cc.

References e, i, VisNet::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 VisNet::removeLocalObject

(

const std::string &

path

)

[virtual]

Delete the local object.

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

Definition at line 1391 of file VisNet.cc.

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

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

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

Definition at line 1240 of file VisNet.cc.

References VisNet::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 VisNet::requestObject	(	Peer *	p,
		const char *	name,
		size_t	len
	)			[private]

Queue an object request to the data server.

Definition at line 97 of file VisNet.cc.

References copydata(), alivecheck_mergeAndRegister::msg, VisNet::Bucket::next, VisNet::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 VisNet::run

(

void

)

Run the actual I/O processing loop.

Definition at line 962 of file VisNet.cc.

References lat::Socket::abort(), arg, lat::IOSelector::attach(), VisNet::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_, VisNet::AutoPeer::host, lat::InetAddress::hostname(), i, IORead, IOUrgent, IOWrite, lock(), logme(), VisNet::Peer::mask, VisNet::Bucket::next, lat::Error::next(), VisNet::AutoPeer::next, onPeerData(), lat::SocketConst::OptSockReceiveBuffer, lat::SocketConst::OptSockSendBuffer, p, VisNet::AutoPeer::peer, VisNet::Peer::peeraddr, VisNet::AutoPeer::port, lat::InetAddress::port(), lat::SystemError::portable(), purgeDeadObjects(), releaseFromWait(), s, sel_, sendObjectListToPeers(), VisNet::Peer::sendq, lat::IOChannel::setBlocking(), lat::Socket::setopt(), shouldStop(), VisNet::Peer::socket, SOCKET_BUF_SIZE, lat::SocketConst::TypeStream, unlock(), VisNet::Peer::update, VisNet::AutoPeer::update, updatePeerMasks(), upstream_, VIS_MSG_LIST_OBJECTS, VIS_MSG_UPDATE_ME, waiting_, and VisNet::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, &VisNet::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 VisNet::sendLocalChanges

(

void

)

Definition at line 1100 of file VisNet.cc.

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

Referenced by removePeer().

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

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

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

Definition at line 1255 of file VisNet.cc.

References copydata(), VisNet::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 VisNet::sendObjectListToPeers

(

bool

all

)

[protected, virtual]

Definition at line 1289 of file VisNet.cc.

References VisNet::Bucket::data, debug_, e, lat::endl(), i, logme(), alivecheck_mergeAndRegister::msg, VisNet::Bucket::next, p, VisNet::Peer::peeraddr, peers_, sendObjectListToPeer(), VisNet::Peer::sendq, VisNet::Peer::update, and VisNet::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 VisNet::sendObjectToPeer	(	Bucket *	msg,
		Object &	o,
		bool	data
	)			[protected]

Definition at line 191 of file VisNet.cc.

References copydata(), VisNet::Bucket::data, VisNet::Object::flags, flags, VisNet::Object::name, VisNet::Object::rawdata, VisNet::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 VisNet::shouldStop

(

void

)

[protected, virtual]

Definition at line 162 of file VisNet.cc.

References shutdown_.

Referenced by run().

{
  return shutdown_;
}

void VisNet::shutdown

(

void

)

Stop the network layer and wait it to finish.

Definition at line 906 of file VisNet.cc.

References communicate_, and shutdown_.

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

void VisNet::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 947 of file VisNet.cc.

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

{
  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 VisNet::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 VisNet.cc.

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

{
  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, &VisNet::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;

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

void VisNet::unlock

(

void

)

Release the lock on the net layer.

Definition at line 937 of file VisNet.cc.

References communicate_, and lock_.

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

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

void VisNet::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 1374 of file VisNet.cc.

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

{
  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 VisNet::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 VisNet.cc.

References debug_, lat::endl(), IOUrgent, IOWrite, logme(), losePeer(), VisNet::Peer::mask, VisNet::Peer::peeraddr, sel_, VisNet::Peer::sendq, lat::IOSelector::setMask(), VisNet::Peer::socket, and VisNet::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 VisNet::updatePeerMasks

(

void

)

[protected, virtual]

Definition at line 1321 of file VisNet.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 VisNet::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 870 of file VisNet.cc.

References downstream_, lat::endl(), VisNet::AutoPeer::host, logme(), VisNet::AutoPeer::port, and VisNet::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 VisNet::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 VisNet.cc.

References requestObject(), VisNet::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 VisNet::appname_ [private]

Definition at line 162 of file VisNet.h.

Referenced by logme().

pthread_t VisNet::communicate_ [private]

Definition at line 177 of file VisNet.h.

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

bool VisNet::debug_ [protected]

Definition at line 146 of file VisNet.h.

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

int VisNet::delay_ [private]

Definition at line 180 of file VisNet.h.

Referenced by delay(), and run().

AutoPeer VisNet::downstream_ [private]

Definition at line 172 of file VisNet.h.

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

bool VisNet::flush_ [private]

Definition at line 181 of file VisNet.h.

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

Peer* VisNet::local_ [private]

Definition at line 174 of file VisNet.h.

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

pthread_mutex_t VisNet::lock_ [private]

Definition at line 176 of file VisNet.h.

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

const uint32_t VisNet::MAX_PEER_WAITREQS = 128 [static]

Definition at line 37 of file VisNet.h.

Referenced by onPeerData().

PeerMap VisNet::peers_ [private]

Definition at line 170 of file VisNet.h.

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

int VisNet::pid_ [private]

Definition at line 163 of file VisNet.h.

Referenced by logme().

lat::IOSelector VisNet::sel_ [private]

Definition at line 165 of file VisNet.h.

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

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

Definition at line 166 of file VisNet.h.

Referenced by onPeerConnect(), and startLocalServer().

sig_atomic_t VisNet::shutdown_ [private]

Definition at line 178 of file VisNet.h.

Referenced by shouldStop(), and shutdown().

AutoPeer VisNet::upstream_ [private]

Definition at line 171 of file VisNet.h.

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

lat::Time VisNet::version_ [private]

Definition at line 168 of file VisNet.h.

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

Definition at line 34 of file VisNet.h.

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

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

Definition at line 33 of file VisNet.h.

Referenced by sendObjectListToPeer().

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

Definition at line 32 of file VisNet.h.

Referenced by onMessage(), and receive().

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

Definition at line 31 of file VisNet.h.

Referenced by purgeDeadObjects(), and sendObjectToPeer().

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

Definition at line 35 of file VisNet.h.

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

const uint32_t VisNet::VIS_MSG_GET_OBJECT = 3 [static]

Definition at line 24 of file VisNet.h.

Referenced by onMessage(), and requestObject().

const uint32_t VisNet::VIS_MSG_HELLO = 0 [static]

Definition at line 21 of file VisNet.h.

const uint32_t VisNet::VIS_MSG_LIST_OBJECTS = 2 [static]

Definition at line 23 of file VisNet.h.

Referenced by onMessage(), and run().

const uint32_t VisNet::VIS_MSG_UPDATE_ME = 1 [static]

Definition at line 22 of file VisNet.h.

Referenced by onMessage(), and run().

const uint32_t VisNet::VIS_REPLY_LIST_BEGIN = 101 [static]

Definition at line 26 of file VisNet.h.

Referenced by onMessage(), and sendObjectListToPeer().

const uint32_t VisNet::VIS_REPLY_LIST_END = 102 [static]

Definition at line 27 of file VisNet.h.

Referenced by onMessage(), and sendObjectListToPeer().

const uint32_t VisNet::VIS_REPLY_NONE = 103 [static]

Definition at line 28 of file VisNet.h.

Referenced by onMessage(), and releaseFromWait().

const uint32_t VisNet::VIS_REPLY_OBJECT = 104 [static]

Definition at line 29 of file VisNet.h.

Referenced by onMessage(), and sendObjectToPeer().

WaitList VisNet::waiting_ [private]

Definition at line 173 of file VisNet.h.

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

lat::Pipe VisNet::wakeup_ [private]

Definition at line 167 of file VisNet.h.

Referenced by sendLocalChanges(), and VisNet().

---

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

• VisFramework/VisFrameworkBase/interface/VisNet.h
• VisFramework/VisFrameworkBase/src/VisNet.cc

---