/afs/cern.ch/work/a/aaltunda/public/www/CMSSW_6_2_5/src/DQMServices/Core/src/DQMNet.cc

Go to the documentation of this file.

#include "DQMServices/Core/interface/DQMNet.h"
#include "DQMServices/Core/interface/DQMDefinitions.h"
#include "DQMServices/Core/src/DQMError.h"
#include "classlib/iobase/InetServerSocket.h"
#include "classlib/iobase/LocalServerSocket.h"
#include "classlib/iobase/Filename.h"
#include "classlib/sysapi/InetSocket.h" // for completing InetAddress
#include "classlib/utils/TimeInfo.h"
#include "classlib/utils/StringList.h"
#include "classlib/utils/StringFormat.h"
#include "classlib/utils/StringOps.h"
#include "classlib/utils/SystemError.h"
#include "classlib/utils/Regexp.h"
#include <unistd.h>
#include <fcntl.h>
#include <sys/wait.h>
#include <stdio.h>
#include <stdint.h>
#include <iostream>
#include <sstream>
#include <cassert>
#include <cfloat>
#include <inttypes.h>

#if __APPLE__
# define MESSAGE_SIZE_LIMIT (1*1024*1024)
# define SOCKET_BUF_SIZE    (1*1024*1024)
#else
# define MESSAGE_SIZE_LIMIT (8*1024*1024)
# define SOCKET_BUF_SIZE    (8*1024*1024)
#endif
#define SOCKET_READ_SIZE        (SOCKET_BUF_SIZE/8)
#define SOCKET_READ_GROWTH      (SOCKET_BUF_SIZE)

using namespace lat;

static const Regexp s_rxmeval("<(.*)>(i|f|s|qr)=(.*)</\\1>");

// Generate log prefix.
std::ostream &
DQMNet::logme (void)
{
  Time now = Time::current();
  return std::cout
    << now.format(true, "%Y-%m-%d %H:%M:%S.")
    << now.nanoformat(3, 3)
    << " " << appname_ << "[" << pid_ << "]: ";
}

// Append data into a bucket.
void
DQMNet::copydata(Bucket *b, const void *data, size_t len)
{
  b->data.insert(b->data.end(),
                 (const unsigned char *)data,
                 (const unsigned char *)data + len);
}

// Discard a bucket chain.
void
DQMNet::discard (Bucket *&b)
{
  while (b)
  {
    Bucket *next = b->next;
    delete b;
    b = next;
  }
}


void
DQMNet::losePeer(const char *reason,
                 Peer *peer,
                 IOSelectEvent *ev,
                 Error *err)
{
  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;
}

void
DQMNet::requestObjectData(Peer *p, const char *name, size_t len)
{
  // Issue request to peer.
  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] = DQM_MSG_GET_OBJECT;
  words[2] = len;
  copydata(*msg, words, sizeof(words));
  copydata(*msg, name, len);
}

void
DQMNet::waitForData(Peer *p, const std::string &name, const std::string &info, Peer *owner)
{
  // 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.
  requestObjectData(owner, name.size() ? &name[0] : 0, name.size());
  WaitObject wo = { Time::current(), name, info, p };
  waiting_.push_back(wo);
  p->waiting++;
}

// Once an object has been updated, this is invoked for all waiting
// peers.  Send the object back to the peer in suitable form.
void
DQMNet::releaseFromWait(WaitList::iterator i, Object *o)
{
  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);
}

// Release everyone waiting for the object @a o.
void
DQMNet::releaseWaiters(const std::string &name, Object *o)
{
  for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    if (i->name == name)
      releaseFromWait(i++, o);
    else
      ++i;
}

void
DQMNet::packQualityData(std::string &into, const QReports &qr)
{
  char buf[64];
  std::ostringstream qrs;
  QReports::const_iterator qi, qe;
  for (qi = qr.begin(), qe = qr.end(); qi != qe; ++qi)
  {
    int pos = 0;
    sprintf(buf, "%d%c%n%.*g", qi->code, 0, &pos, DBL_DIG+2, qi->qtresult);
    qrs << buf << '\0'
        << buf+pos << '\0'
        << qi->qtname << '\0'
        << qi->algorithm << '\0'
        << qi->message << '\0'
        << '\0';
  }
  into = qrs.str();
}

void
DQMNet::unpackQualityData(QReports &qr, uint32_t &flags, const char *from)
{
  const char *qdata = from;

  // Count how many qresults there are.
  size_t nqv = 0;
  while (*qdata)
  {
    ++nqv;
    while (*qdata) ++qdata; ++qdata;
    while (*qdata) ++qdata; ++qdata;
    while (*qdata) ++qdata; ++qdata;
    while (*qdata) ++qdata; ++qdata;
    while (*qdata) ++qdata; ++qdata;
  }

  // Now extract the qreports.
  qdata = from;
  qr.reserve(nqv);
  while (*qdata)
  {
    qr.push_back(DQMNet::QValue());
    DQMNet::QValue &qv = qr.back();

    qv.code = atoi(qdata);
    while (*qdata) ++qdata;
    switch (qv.code)
    {
    case dqm::qstatus::STATUS_OK:
      break;
    case dqm::qstatus::WARNING:
      flags |= DQMNet::DQM_PROP_REPORT_WARN;
      break;
    case dqm::qstatus::ERROR:
      flags |= DQMNet::DQM_PROP_REPORT_ERROR;
      break;
    default:
      flags |= DQMNet::DQM_PROP_REPORT_OTHER;
      break;
    }

    qv.qtresult = atof(++qdata);
    while (*qdata) ++qdata;

    qv.qtname = ++qdata;
    while (*qdata) ++qdata;

    qv.algorithm = ++qdata;
    while (*qdata) ++qdata;

    qv.message = ++qdata;
    while (*qdata) ++qdata;
    ++qdata;
  }
}

#if 0
// Deserialise a ROOT object from a buffer at the current position.
static TObject *
extractNextObject(TBufferFile &buf)
{
  if (buf.Length() == buf.BufferSize())
    return 0;

  buf.InitMap();
  Int_t pos = buf.Length();
  TClass *c = buf.ReadClass();
  buf.SetBufferOffset(pos);
  buf.ResetMap();
  return c ? buf.ReadObject(c) : 0;
}

// Reconstruct an object from the raw data.
bool
DQMNet::reconstructObject(Object &o)
{
  TBufferFile buf(TBufferFile::kRead, o.rawdata.size(), &o.rawdata[0], kFALSE);
  buf.Reset();

  // Extract the main object.
  if (! (o.object = extractNextObject(buf)))
    return false;
  
  // Extract the reference object.
  o.reference = extractNextObject(buf);

  // Extract quality reports.
  unpackQualityData(o.qreports, o.flags, o.qdata.c_str());
  return true;
}
#endif

#if 0
bool
DQMNet::reinstateObject(DQMStore *store, Object &o)
{
  if (! reconstructObject (o))
    return false;

  // Reconstruct the main object
  MonitorElement *obj = 0;
  store->setCurrentFolder(*o.dirname);
  switch (o.flags & DQM_PROP_TYPE_MASK)
  {
  case DQM_PROP_TYPE_INT:
    obj = store->bookInt(o.objname);
    obj->Fill(atoll(o.scalar.c_str()));
    break;

  case DQM_PROP_TYPE_REAL:
    obj = store->bookFloat(name);
    obj->Fill(atof(o.scalar.c_str()));
    break;

  case DQM_PROP_TYPE_STRING:
    obj = store->bookString(name, o.scalar);
    break;

  case DQM_PROP_TYPE_TH1F:
    obj = store->book1D(name, dynamic_cast<TH1F *>(o.object));
    break;

  case DQM_PROP_TYPE_TH1S:
    obj = store->book1S(name, dynamic_cast<TH1S *>(o.object));
    break;

  case DQM_PROP_TYPE_TH1D:
    obj = store->book1DD(name, dynamic_cast<TH1D *>(o.object));
    break;

  case DQM_PROP_TYPE_TH2F:
    obj = store->book2D(name, dynamic_cast<TH2F *>(o.object));
    break;

  case DQM_PROP_TYPE_TH2S:
    obj = store->book2S(name, dynamic_cast<TH2S *>(o.object));
    break;

  case DQM_PROP_TYPE_TH2D:
    obj = store->book2DD(name, dynamic_cast<TH2D *>(o.object));
    break;

  case DQM_PROP_TYPE_TH3F:
    obj = store->book3D(name, dynamic_cast<TH3F *>(o.object));
    break;

  case DQM_PROP_TYPE_TH3S:
    obj = store->book3S(name, dynamic_cast<TH3S *>(o.object));
    break;

  case DQM_PROP_TYPE_TH3D:
    obj = store->book3DD(name, dynamic_cast<TH3D *>(o.object));
    break;

  case DQM_PROP_TYPE_PROF:
    obj = store->bookProfile(name, dynamic_cast<TProfile *>(o.object));
    break;

  case DQM_PROP_TYPE_PROF2D:
    obj = store->bookProfile2D(name, dynamic_cast<TProfile2D *>(o.object));
    break;

  default:
    logme()
      << "ERROR: unexpected monitor element of type "
      << (o.flags & DQM_PROP_TYPE_MASK) << " called '"
      << *o.dirname << '/' << o.objname << "'\n";
    return false;
  }

  // Reconstruct tag and qreports.
  if (obj)
  {
    obj->data_.tag = o.tag;
    obj->data_.qreports = o.qreports;
  }

  // Inidicate success.
  return true;
}
#endif

// Check if the network layer should stop.
bool
DQMNet::shouldStop(void)
{
  return shutdown_;
}

// Once an object has been updated, this is invoked for all waiting
// peers.  Send the requested object to the waiting peer.
void
DQMNet::releaseFromWait(Bucket *msg, WaitObject &w, Object *o)
{
  if (o)
    sendObjectToPeer(msg, *o, true);
  else
  {
    uint32_t words [3];
    words[0] = sizeof(words) + w.name.size();
    words[1] = DQM_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());
  }
}

// Send an object to a peer.  If not @a data, only sends a summary
// without the object data, except the data is always sent for scalar
// objects.
void
DQMNet::sendObjectToPeer(Bucket *msg, Object &o, bool data)
{
  uint32_t flags = o.flags & ~DQM_PROP_DEAD;
  DataBlob objdata;

  if ((flags & DQM_PROP_TYPE_MASK) <= DQM_PROP_TYPE_SCALAR)
    objdata.insert(objdata.end(),
                   &o.scalar[0],
                   &o.scalar[0] + o.scalar.size());
  else if (data)
    objdata.insert(objdata.end(),
                   &o.rawdata[0],
                   &o.rawdata[0] + o.rawdata.size());

  uint32_t words [9];
  uint32_t namelen = o.dirname->size() + o.objname.size() + 1;
  uint32_t datalen = objdata.size();
  uint32_t qlen = o.qdata.size();

  if (o.dirname->empty())
    --namelen;

  words[0] = 9*sizeof(uint32_t) + namelen + datalen + qlen;
  words[1] = DQM_REPLY_OBJECT;
  words[2] = flags;
  words[3] = (o.version >> 0 ) & 0xffffffff;
  words[4] = (o.version >> 32) & 0xffffffff;
  words[5] = o.tag;
  words[6] = namelen;
  words[7] = datalen;
  words[8] = qlen;

  msg->data.reserve(msg->data.size() + words[0]);
  copydata(msg, &words[0], 9*sizeof(uint32_t));
  if (namelen)
  {
    copydata(msg, &(*o.dirname)[0], o.dirname->size());
    if (! o.dirname->empty())
      copydata(msg, "/", 1);
    copydata(msg, &o.objname[0], o.objname.size());
  }
  if (datalen)
    copydata(msg, &objdata[0], datalen);
  if (qlen)
    copydata(msg, &o.qdata[0], qlen);
}

// Handle peer messages.
bool
DQMNet::onMessage(Bucket *msg, Peer *p, unsigned char *data, size_t len)
{
  // Decode and process this message.
  uint32_t type;
  memcpy (&type, data + sizeof(uint32_t), sizeof (type));
  switch (type)
  {
  case DQM_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 << ", size " << len << std::endl;

      p->update = true;
    }
    return true;

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

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

  case DQM_MSG_GET_OBJECT:
    {
      if (debug_)
        logme()
          << "DEBUG: received message 'GET OBJECT' from peer "
          << p->peeraddr << ", size " << len << 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;
      }

      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().ns();
        if ((o->rawdata.empty() || (o->flags & DQM_PROP_STALE))
            && (o->flags & DQM_PROP_TYPE_MASK) > DQM_PROP_TYPE_SCALAR)
          waitForData(p, name, "", owner);
        else
          sendObjectToPeer(msg, *o, true);
      }
      else
      {
        uint32_t words [3];
        words[0] = sizeof(words) + name.size();
        words[1] = DQM_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());
      }
    }
    return true;

  case DQM_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;
      }

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

      if (debug_)
        logme()
          << "DEBUG: received message 'LIST BEGIN "
          << (flags ? "FULL" : "INCREMENTAL")
          << "' from " << p->peeraddr
          << ", size " << len << std::endl;

      // If we are about to receive a full list of objects, flag all
      // objects as possibly dead.  Subsequent object notifications
      // will undo this for the live objects.  We cannot delete
      // objects quite yet, as we may get inquiry from another client
      // while we are processing the incoming list, so we keep the
      // objects tentatively alive as long as we've not seen the end.
      if (flags)
        markObjectsDead(p);
    }
    return true;

  case DQM_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, now purge all dead
      // objects. 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)
        purgeDeadObjects(p);

      if (debug_)
        logme()
          << "DEBUG: received message 'LIST END "
          << (flags ? "FULL" : "INCREMENTAL")
          << "' from " << p->peeraddr
          << ", size " << len << 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 DQM_REPLY_OBJECT:
    {
      uint32_t words[9];
      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[6];
      uint32_t &datalen = words[7];
      uint32_t &qlen = words[8];

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

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

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

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

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

      o->flags = words[2] | DQM_PROP_NEW | DQM_PROP_RECEIVED;
      o->tag = words[5];
      o->version = ((uint64_t) words[4] << 32 | words[3]);
      o->scalar.clear();
      o->qdata.clear();
      if ((o->flags & DQM_PROP_TYPE_MASK) <= DQM_PROP_TYPE_SCALAR)
      {
        o->rawdata.clear();
        o->scalar.insert(o->scalar.end(), objdata, qdata);
      }
      else if (datalen)
      {
        o->rawdata.clear();
        o->rawdata.insert(o->rawdata.end(), objdata, qdata);
      }
      else if (! o->rawdata.empty())
        o->flags |= DQM_PROP_STALE;
      o->qdata.insert(o->qdata.end(), qdata, 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 (o->lastreq
          && ! datalen
          && (o->flags & DQM_PROP_TYPE_MASK) > DQM_PROP_TYPE_SCALAR)
        requestObjectData(p, (namelen ? &name[0] : 0), namelen);

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

  case DQM_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);
      std::string name((char *) namedata, namelen);

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

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

      // If this was a known object, kill it.
      if (Object *o = findObject(p, name))
      {
        o->flags |= DQM_PROP_DEAD;
        purgeDeadObjects(p);
      }

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

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

bool
DQMNet::onPeerData(IOSelectEvent *ev, Peer *p)
{
  lock();
  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 DQM server at " << p->peeraddr
        << " lost (will attempt to reconnect in 15 seconds)\n";
      losePeer(0, p, ev);
    }
    else
      losePeer("WARNING: lost peer connection ", p, ev);

    unlock();
    return true;
  }

  // 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)
      {
        losePeer("WARNING: unable to write to peer ", p, ev, &e);
        unlock();
        return true;
      }

      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.
        losePeer("WARNING: failed to read from peer ", p, ev, &e);
        unlock();
        return true;
      }
    }

    // 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)
      {
        losePeer("WARNING: excessively large message from ", p, ev);
        unlock();
        return true;
      }

      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)
        {
          losePeer("WARNING: data stream error with ", p, ev);
          unlock();
          return true;
        }

        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.
  unlock();
  return false;
}

bool
DQMNet::onPeerConnect(IOSelectEvent *ev)
{
  // 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.
  lock();
  Peer *p = createPeer(s);
  std::string localaddr;
  if (InetSocket *inet = dynamic_cast<InetSocket *>(s))
  {
    InetAddress peeraddr = inet->peername();
    InetAddress myaddr = inet->sockname();
    p->peeraddr = StringFormat("%1:%2")
                  .arg(peeraddr.hostname())
                  .arg(peeraddr.port());
    localaddr = StringFormat("%1:%2")
                .arg(myaddr.hostname())
                .arg(myaddr.port());
  }
  else if (LocalSocket *local = dynamic_cast<LocalSocket *>(s))
  {
    p->peeraddr = local->peername().path();
    localaddr = local->sockname().path();
  }
  else
    assert(false);

  p->mask = IORead|IOUrgent;
  p->socket = s;

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

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

  // We are never done.
  return false;
}

bool
DQMNet::onLocalNotify(IOSelectEvent *ev)
{
  // 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;
}

void
DQMNet::updateMask(Peer *p)
{
  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);
  }
}

DQMNet::DQMNet (const std::string &appname /* = "" */)
  : debug_ (false),
    appname_ (appname.empty() ? "DQMNet" : appname.c_str()),
    pid_ (getpid()),
    server_ (0),
    version_ (Time::current()),
    communicate_ ((pthread_t) -1),
    shutdown_ (0),
    delay_ (1000),
    waitStale_ (0, 0, 0, 0, 500000000 /* 500 ms */),
    waitMax_ (0, 0, 0, 5 /* seconds */, 0),
    flush_ (false)
{
  // Create a pipe for the local DQM to tell the communicator
  // thread that local DQM 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, &DQMNet::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;
}

DQMNet::~DQMNet(void)
{
  // FIXME
}

void
DQMNet::debug(bool doit)
{
  debug_ = doit;
}

void
DQMNet::delay(int delay)
{
  delay_ = delay;
}

void
DQMNet::staleObjectWaitLimit(lat::TimeSpan time)
{
  waitStale_ = time;
}

void
DQMNet::startLocalServer(int port)
{
  if (server_)
  {
    logme() << "ERROR: DQM server was already started.\n";
    return;
  }

  try
  {
    InetAddress addr("0.0.0.0", port);
    InetSocket *s = new InetSocket(SOCK_STREAM, 0, addr.family());
    s->bind(addr);
    s->listen(10);
    s->setopt(SO_SNDBUF, SOCKET_BUF_SIZE);
    s->setopt(SO_RCVBUF, SOCKET_BUF_SIZE);
    s->setBlocking(false);
    sel_.attach(server_ = s, IOAccept, CreateHook(this, &DQMNet::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;

    raiseDQMError("DQMNet::startLocalServer", "Failed to start server at port"
                  " %d: %s", port, e.explain().c_str());
  }
  
  logme() << "INFO: DQM server started at port " << port << std::endl;
}

void
DQMNet::startLocalServer(const char *path)
{
  if (server_)
  {
    logme() << "ERROR: DQM server was already started.\n";
    return;
  }

  try
  {
    server_ = new LocalServerSocket(path, 10);
    server_->setopt(SO_SNDBUF, SOCKET_BUF_SIZE);
    server_->setopt(SO_RCVBUF, SOCKET_BUF_SIZE);
    server_->setBlocking(false);
    sel_.attach(server_, IOAccept, CreateHook(this, &DQMNet::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 path " << path << ": "
      << e.explain() << std::endl;

    raiseDQMError("DQMNet::startLocalServer", "Failed to start server at path"
                  " %s: %s", path, e.explain().c_str());
  }
  
  logme() << "INFO: DQM server started at path " << path << std::endl;
}

void
DQMNet::updateToCollector(const std::string &host, int port)
{
  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
DQMNet::listenToCollector(const std::string &host, int port)
{
  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
DQMNet::shutdown(void)
{
  shutdown_ = 1;
  if (communicate_ != (pthread_t) -1)
    pthread_join(communicate_, 0);
}

static void *communicate(void *obj)
{
  sigset_t sigs;
  sigfillset (&sigs);
  pthread_sigmask (SIG_BLOCK, &sigs, 0);
  ((DQMNet *)obj)->run();
  return 0;
}

void
DQMNet::lock(void)
{
  if (communicate_ != (pthread_t) -1)
    pthread_mutex_lock(&lock_);
}

void
DQMNet::unlock(void)
{
  if (communicate_ != (pthread_t) -1)
    pthread_mutex_unlock(&lock_);
}

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

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

void
DQMNet::run(void)
{
  Time now;
  Time nextFlush = 0;
  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
        {
          InetAddress addr(ap->host.c_str(), ap->port);
          s = new InetSocket (SOCK_STREAM, 0, addr.family());
          s->setBlocking(false);
          s->connect(addr);
          s->setopt(SO_SNDBUF, SOCKET_BUF_SIZE);
          s->setopt(SO_RCVBUF, 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 (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)
        {
          Peer *p = createPeer(s);
          ap->peer = p;

          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, &DQMNet::onPeerData, p));
          if (ap == &upstream_)
          {
            uint32_t words[4] = { 2*sizeof(uint32_t), DQM_MSG_LIST_OBJECTS,
                                  2*sizeof(uint32_t), DQM_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();
    lock();

    // Check if flush is required.  Flush only if one is needed.
    // Always sends the full object list, but only rarely.
    if (flush_ && now > nextFlush)
    {
      flush_ = false;
      nextFlush = now + TimeSpan(0, 0, 0, 15 /* seconds */, 0);
      sendObjectListToPeers(true);
    }

    // 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.
    Time waitold = now - waitMax_;
    Time waitstale = now - waitStale_;
    for (WaitList::iterator i = waiting_.begin(), e = waiting_.end(); i != e; )
    {
      Object *o = findObject(0, i->name);

      // If we have (stale) object data, wait only up to stale limit.
      // Otherwise if we have no data at all, wait up to the max limit.
      if (i->time < waitold)
      {
        logme ()
          << "WARNING: source not responding in " << (waitMax_.ns() * 1e-9)
          << "s to retrieval, releasing '" << i->name << "' from wait, have "
          << (o ? o->rawdata.size() : 0) << " data available\n";
        releaseFromWait(i++, o);
      }
      else if (i->time < waitstale && o && (o->flags & DQM_PROP_STALE))
      {
        logme ()
          << "WARNING: source not responding in " << (waitStale_.ns() * 1e-9)
          << "s to update, releasing '" << i->name << "' from wait, have "
          << o->rawdata.size() << " data available\n";
        releaseFromWait(i++, o);
      }

      // Keep it for now.
      else
        ++i;
    }

    unlock();
  }
}

// Tell the network cache that there have been local changes that
// should be advertised to the downstream listeners.
void
DQMNet::sendLocalChanges(void)
{
  char byte = 0;
  wakeup_.sink()->write(&byte, 1);
}

DQMBasicNet::DQMBasicNet(const std::string &appname /* = "" */)
  : DQMImplNet<DQMNet::Object>(appname)
{
  local_ = static_cast<ImplPeer *>(createPeer((Socket *) -1));
}

void
DQMBasicNet::reserveLocalSpace(uint32_t size)
{
  local_->objs.resize(size);
}

void
DQMBasicNet::updateLocalObject(Object &o)
{
  o.dirname = &*local_->dirs.insert(*o.dirname).first;
  std::pair<ObjectMap::iterator, bool> info(local_->objs.insert(o));
  if (! info.second)
  {
    // Somewhat hackish. Sets are supposedly immutable, but we
    // need to change the non-key parts of the object. Erasing
    // and re-inserting would produce too much memory churn.
    Object &old = const_cast<Object &>(*info.first);
    std::swap(old.flags,     o.flags);
    std::swap(old.tag,       o.tag);
    std::swap(old.version,   o.version);
    std::swap(old.qreports,  o.qreports);
    std::swap(old.rawdata,   o.rawdata);
    std::swap(old.scalar,    o.scalar);
    std::swap(old.qdata,     o.qdata);
  }
}

bool
DQMBasicNet::removeLocalExcept(const std::set<std::string> &known)
{
  size_t removed = 0;
  std::string path;
  ObjectMap::iterator i, e;
  for (i = local_->objs.begin(), e = local_->objs.end(); i != e; )
  {
    path.clear();
    path.reserve(i->dirname->size() + i->objname.size() + 2);
    path += *i->dirname;
    if (! path.empty())
      path += '/';
    path += i->objname;

    if (! known.count(path))
      ++removed, local_->objs.erase(i++);
    else
      ++i;
  }

  return removed > 0;
}