DQMNet.h

Go to the documentation of this file.

#ifndef DQMSERVICES_CORE_DQM_NET_H
# define DQMSERVICES_CORE_DQM_NET_H

# include "classlib/iobase/Socket.h"
# include "classlib/iobase/IOSelector.h"
# include "classlib/iobase/Pipe.h"
# include "classlib/utils/Signal.h"
# include "classlib/utils/Error.h"
# include "classlib/utils/Time.h"
# include <pthread.h>
# include <stdint.h>
# include <iostream>
# include <vector>
# include <string>
# include <list>
# include <map>
# include <set>
# include <ext/hash_set>

//class DQMStore;

class DQMNet
{
public:
  static const uint32_t DQM_PROP_TYPE_MASK  = 0x000000ff;
  static const uint32_t DQM_PROP_TYPE_SCALAR    = 0x0000000f;
  static const uint32_t DQM_PROP_TYPE_INVALID   = 0x00000000;
  static const uint32_t DQM_PROP_TYPE_INT   = 0x00000001;
  static const uint32_t DQM_PROP_TYPE_REAL  = 0x00000002;
  static const uint32_t DQM_PROP_TYPE_STRING    = 0x00000003;
  static const uint32_t DQM_PROP_TYPE_TH1F  = 0x00000010;
  static const uint32_t DQM_PROP_TYPE_TH1S  = 0x00000011;
  static const uint32_t DQM_PROP_TYPE_TH1D  = 0x00000012;
  static const uint32_t DQM_PROP_TYPE_TH2F  = 0x00000020;
  static const uint32_t DQM_PROP_TYPE_TH2S  = 0x00000021;
  static const uint32_t DQM_PROP_TYPE_TH2D  = 0x00000022;
  static const uint32_t DQM_PROP_TYPE_TH3F  = 0x00000030;
  static const uint32_t DQM_PROP_TYPE_TH3S  = 0x00000031;
  static const uint32_t DQM_PROP_TYPE_TH3D  = 0x00000032;
  static const uint32_t DQM_PROP_TYPE_TPROF = 0x00000040;
  static const uint32_t DQM_PROP_TYPE_TPROF2D   = 0x00000041;
  static const uint32_t DQM_PROP_TYPE_DATABLOB  = 0x00000050;
  
  static const uint32_t DQM_PROP_REPORT_MASK    = 0x00000f00;
  static const uint32_t DQM_PROP_REPORT_CLEAR   = 0x00000000;
  static const uint32_t DQM_PROP_REPORT_ERROR   = 0x00000100;
  static const uint32_t DQM_PROP_REPORT_WARN    = 0x00000200;
  static const uint32_t DQM_PROP_REPORT_OTHER   = 0x00000400;
  static const uint32_t DQM_PROP_REPORT_ALARM   = (DQM_PROP_REPORT_ERROR
                           | DQM_PROP_REPORT_WARN
                           | DQM_PROP_REPORT_OTHER);

  static const uint32_t DQM_PROP_HAS_REFERENCE  = 0x00001000;
  static const uint32_t DQM_PROP_TAGGED     = 0x00002000;
  static const uint32_t DQM_PROP_ACCUMULATE = 0x00004000;
  static const uint32_t DQM_PROP_RESET      = 0x00008000;

  static const uint32_t DQM_PROP_NEW        = 0x00010000;
  static const uint32_t DQM_PROP_RECEIVED   = 0x00020000;
  static const uint32_t DQM_PROP_LUMI       = 0x00040000;
  static const uint32_t DQM_PROP_DEAD       = 0x00080000;
  static const uint32_t DQM_PROP_STALE      = 0x00100000;

  static const uint32_t DQM_MSG_HELLO       = 0;
  static const uint32_t DQM_MSG_UPDATE_ME   = 1;
  static const uint32_t DQM_MSG_LIST_OBJECTS    = 2;
  static const uint32_t DQM_MSG_GET_OBJECT  = 3;

  static const uint32_t DQM_REPLY_LIST_BEGIN    = 101;
  static const uint32_t DQM_REPLY_LIST_END  = 102;
  static const uint32_t DQM_REPLY_NONE      = 103;
  static const uint32_t DQM_REPLY_OBJECT    = 104;

  static const uint32_t MAX_PEER_WAITREQS   = 128;

  struct Peer;
  struct QValue;
  struct WaitObject;

  typedef std::vector<unsigned char>    DataBlob;
  typedef std::vector<QValue>       QReports;
  typedef std::vector<uint32_t>     TagList; // DEPRECATED
  typedef std::list<WaitObject>     WaitList;

  struct QValue
  {
    int         code;
    float       qtresult;
    std::string     message;
    std::string     qtname;
    std::string     algorithm;
  };

  struct CoreObject
  {
    uint32_t        flags;
    uint32_t        tag;
    uint64_t        version;
    const std::string   *dirname;
    std::string     objname;
    QReports        qreports;
  };
  
  struct Object : CoreObject
  {
    uint64_t        hash;
    uint64_t        lastreq;
    DataBlob        rawdata;
    std::string     scalar;
    std::string     qdata;
  };

  struct Bucket
  {
    Bucket      *next;
    DataBlob        data;
  };

  struct WaitObject
  {
    lat::Time       time;
    std::string     name;
    std::string     info;
    Peer        *peer;
  };

  struct AutoPeer;
  struct Peer
  {
    std::string     peeraddr;
    lat::Socket     *socket;
    DataBlob        incoming;
    Bucket      *sendq;
    size_t      sendpos;

    unsigned        mask;
    bool        source;
    bool        update;
    bool        updated;
    size_t      updates;
    size_t      waiting;
    AutoPeer        *automatic;
  };

  struct AutoPeer
  {
    Peer        *peer;
    lat::Time       next;
    std::string     host;
    int         port;
    bool        update;
  };

  DQMNet(const std::string &appname = "");
  virtual ~DQMNet(void);

  void          debug(bool doit);
  void          delay(int delay);
  void          startLocalServer(int port);
  void          startLocalServer(const char *path);
  void          staleObjectWaitLimit(lat::TimeSpan time);
  void          updateToCollector(const std::string &host, int port);
  void          listenToCollector(const std::string &host, int port);
  void          shutdown(void);
  void          lock(void);
  void          unlock(void);

  void          start(void);
  void          run(void);

  void          sendLocalChanges(void);

  static bool setOrder(const CoreObject &a, const CoreObject &b)
    {
      int diff = a.dirname->compare(*b.dirname);
      return (diff < 0 ? true
          : diff == 0 ? a.objname < b.objname
          : false);
    }

  struct HashOp
  {
    uint32_t operator()(const Object &a) const
      {
        return a.hash;
      }
  };

  struct HashEqual
  {
    bool operator()(const Object &a, const Object &b) const
      {
        return a.hash == b.hash && *a.dirname == *b.dirname && a.objname == b.objname;
      }
  };

  static size_t
  dqmhash(const void *key, size_t keylen)
    {
      // Reduced version of Bob Jenkins' hash function at:
      //   http://www.burtleburtle.net/bob/c/lookup3.c
#     define dqmhashrot(x,k) (((x)<<(k)) | ((x)>>(32-(k))))
#     define dqmhashmix(a,b,c) { \
        a -= c; a ^= dqmhashrot(c, 4); c += b; \
        b -= a; b ^= dqmhashrot(a, 6); a += c; \
        c -= b; c ^= dqmhashrot(b, 8); b += a; \
        a -= c; a ^= dqmhashrot(c,16); c += b; \
        b -= a; b ^= dqmhashrot(a,19); a += c; \
        c -= b; c ^= dqmhashrot(b, 4); b += a; }
#     define dqmhashfinal(a,b,c) { \
        c ^= b; c -= dqmhashrot(b,14); \
        a ^= c; a -= dqmhashrot(c,11); \
        b ^= a; b -= dqmhashrot(a,25); \
        c ^= b; c -= dqmhashrot(b,16); \
        a ^= c; a -= dqmhashrot(c,4);  \
        b ^= a; b -= dqmhashrot(a,14); \
        c ^= b; c -= dqmhashrot(b,24); }
 
      uint32_t a, b, c;
      a = b = c = 0xdeadbeef + (uint32_t) keylen;
      const unsigned char *k = (const unsigned char *) key;

      // all but the last block: affect some bits of (a, b, c)
      while (keylen > 12)
      {
        a += k[0];
        a += ((uint32_t)k[1]) << 8;
        a += ((uint32_t)k[2]) << 16;
        a += ((uint32_t)k[3]) << 24;
        b += k[4];
        b += ((uint32_t)k[5]) << 8;
        b += ((uint32_t)k[6]) << 16;
        b += ((uint32_t)k[7]) << 24;
        c += k[8];
        c += ((uint32_t)k[9]) << 8;
        c += ((uint32_t)k[10]) << 16;
        c += ((uint32_t)k[11]) << 24;
        dqmhashmix(a,b,c);
        keylen -= 12;
        k += 12;
      }

      // last block: affect all 32 bits of (c); all case statements fall through
      switch (keylen)
      {
      case 12: c += ((uint32_t)k[11]) << 24;
      case 11: c += ((uint32_t)k[10]) << 16;
      case 10: c += ((uint32_t)k[9]) << 8;
      case 9 : c += k[8];
      case 8 : b += ((uint32_t)k[7]) << 24;
      case 7 : b += ((uint32_t)k[6]) << 16;
      case 6 : b += ((uint32_t)k[5]) << 8;
      case 5 : b += k[4];
      case 4 : a += ((uint32_t)k[3]) << 24;
      case 3 : a += ((uint32_t)k[2]) << 16;
      case 2 : a += ((uint32_t)k[1]) << 8;
      case 1 : a += k[0];
               break;
      case 0 : return c;
      }

      dqmhashfinal(a, b, c);
      return c;
#     undef dqmhashrot
#     undef dqmhashmix
#     undef dqmhashfinal
    }

  static void       packQualityData(std::string &into, const QReports &qr);
  static void       unpackQualityData(QReports &qr, uint32_t &flags, const char *from);

protected:
  std::ostream &    logme(void);
  static void       copydata(Bucket *b, const void *data, size_t len);
  virtual void      sendObjectToPeer(Bucket *msg, Object &o, bool data);

  virtual bool      shouldStop(void);
  void          waitForData(Peer *p, const std::string &name, const std::string &info, Peer *owner);
  virtual void      releaseFromWait(Bucket *msg, WaitObject &w, Object *o);
  virtual bool      onMessage(Bucket *msg, Peer *p, unsigned char *data, size_t len);

  // bool           reconstructObject(Object &o);
  // bool           reinstateObject(DQMStore *store, Object &o);
  virtual Object *  findObject(Peer *p, const std::string &name, Peer **owner = 0) = 0;
  virtual Object *  makeObject(Peer *p, const std::string &name) = 0;
  virtual void      markObjectsDead(Peer *p) = 0;
  virtual void      purgeDeadObjects(Peer *p) = 0;

  virtual Peer *    getPeer(lat::Socket *s) = 0;
  virtual Peer *    createPeer(lat::Socket *s) = 0;
  virtual void      removePeer(Peer *p, lat::Socket *s) = 0;
  virtual void      sendObjectListToPeer(Bucket *msg, bool all, bool clear) = 0;
  virtual void      sendObjectListToPeers(bool all) = 0;

  void          updateMask(Peer *p);
  virtual void      updatePeerMasks(void) = 0;
  static void       discard(Bucket *&b);

  bool          debug_;
  pthread_mutex_t   lock_;

private:
  void          losePeer(const char *reason,
                 Peer *peer,
                 lat::IOSelectEvent *event,
                 lat::Error *err = 0);
  void          requestObjectData(Peer *p, const char *name, size_t len);
  void          releaseFromWait(WaitList::iterator i, Object *o);
  void          releaseWaiters(const std::string &name, Object *o);

  bool          onPeerData(lat::IOSelectEvent *ev, Peer *p);
  bool          onPeerConnect(lat::IOSelectEvent *ev);
  bool          onLocalNotify(lat::IOSelectEvent *ev);

  std::string       appname_;
  int           pid_;

  lat::IOSelector   sel_;
  lat::Socket       *server_;
  lat::Pipe     wakeup_;
  lat::Time     version_;

  AutoPeer      upstream_;
  AutoPeer      downstream_;
  WaitList      waiting_;

  pthread_t     communicate_;
  sig_atomic_t      shutdown_;

  int           delay_;
  lat::TimeSpan     waitStale_;
  lat::TimeSpan     waitMax_;
  bool          flush_;

  // copying is not available
  DQMNet(const DQMNet &);
  DQMNet &operator=(const DQMNet &);
};

template <class ObjType>
class DQMImplNet : public DQMNet
{
public:
  struct ImplPeer;

  typedef std::set<std::string> DirMap;
  typedef __gnu_cxx::hash_set<ObjType, HashOp, HashEqual> ObjectMap;
  typedef std::map<lat::Socket *, ImplPeer> PeerMap;
  struct ImplPeer : Peer
  {
    ImplPeer(void) {}
    ObjectMap objs;
    DirMap dirs;
  };

  DQMImplNet(const std::string &appname = "")
    : DQMNet(appname)
    {}
  
  ~DQMImplNet(void)
    {}

protected:
  virtual Object *
  findObject(Peer *p, const std::string &name, Peer **owner = 0)
    {
      size_t slash = name.rfind('/');
      size_t dirpos = (slash == std::string::npos ? 0 : slash);
      size_t namepos = (slash == std::string::npos ? 0 : slash+1);
      std::string path(name, 0, dirpos);
      ObjType proto;
      proto.hash = dqmhash(name.c_str(), name.size());
      proto.dirname = &path;
      proto.objname.append(name, namepos, std::string::npos);

      typename ObjectMap::iterator pos;
      typename PeerMap::iterator i, e;
      if (owner)
    *owner = 0;
      if (p)
      {
    ImplPeer *ip = static_cast<ImplPeer *>(p);
    pos = ip->objs.find(proto);
    if (pos == ip->objs.end())
      return 0;
    else
    {
      if (owner) *owner = ip;
      return const_cast<ObjType *>(&*pos);
    }
      }
      else
      {
    for (i = peers_.begin(), e = peers_.end(); i != e; ++i)
    {
      pos = i->second.objs.find(proto);
      if (pos != i->second.objs.end())
      {
        if (owner) *owner = &i->second;
        return const_cast<ObjType *>(&*pos);
      }
    }
    return 0;
      }
    }

  virtual Object *
  makeObject(Peer *p, const std::string &name)
    {
      ImplPeer *ip = static_cast<ImplPeer *>(p);
      size_t slash = name.rfind('/');
      size_t dirpos = (slash == std::string::npos ? 0 : slash);
      size_t namepos = (slash == std::string::npos ? 0 : slash+1);
      ObjType o;
      o.flags = 0;
      o.tag = 0;
      o.version = 0;
      o.lastreq = 0;
      o.dirname = &*ip->dirs.insert(name.substr(0, dirpos)).first;
      o.objname.append(name, namepos, std::string::npos);
      o.hash = dqmhash(name.c_str(), name.size());
      return const_cast<ObjType *>(&*ip->objs.insert(o).first);
    }

  // Mark all the objects dead.  This is intended to be used when
  // starting to process a complete list of objects, in order to
  // flag the objects that need to be killed at the end.  After
  // call to this method, revive all live objects by removing the
  // DQM_PROP_DEAD flag, then call purgeDeadObjects() at the end
  // to remove the dead ones.  This also turns off object request
  // for objects we've lost interest in.
  virtual void
  markObjectsDead(Peer *p)
    {
      uint64_t minreq
    = (lat::Time::current()
      - lat::TimeSpan(0, 0, 5 /* minutes */, 0, 0)).ns();
      ImplPeer *ip = static_cast<ImplPeer *>(p);
      typename ObjectMap::iterator i, e;
      for (i = ip->objs.begin(), e = ip->objs.end(); i != e; ++i)
      {
    if (i->lastreq && i->lastreq < minreq)
      const_cast<ObjType &>(*i).lastreq = 0;
    const_cast<ObjType &>(*i).flags |= DQM_PROP_DEAD;
      }
    }

  // Mark remaining zombie objects as dead.  See markObjectsDead().
  virtual void
  purgeDeadObjects(Peer *p)
    {
      ImplPeer *ip = static_cast<ImplPeer *>(p);
      typename ObjectMap::iterator i, e;
      for (i = ip->objs.begin(), e = ip->objs.end(); i != e; )
      {
    if (i->flags & DQM_PROP_DEAD)
      ip->objs.erase(i++);
    else
      ++i;
      }
    }

  virtual Peer *
  getPeer(lat::Socket *s)
    {
      typename PeerMap::iterator pos = peers_.find(s);
      typename PeerMap::iterator end = peers_.end();
      return pos == end ? 0 : &pos->second;
    }

  virtual Peer *
  createPeer(lat::Socket *s)
    {
      ImplPeer *ip = &peers_[s];
      ip->socket = 0;
      ip->sendq = 0;
      ip->sendpos = 0;
      ip->mask = 0;
      ip->source = false;
      ip->update = false;
      ip->updated = false;
      ip->updates = 0;
      ip->waiting = 0;
      ip->automatic = 0;
      return ip;
    }

  virtual void
  removePeer(Peer *p, lat::Socket *s)
    {
      ImplPeer *ip = static_cast<ImplPeer *>(p);
      bool needflush = ! ip->objs.empty();

      typename ObjectMap::iterator i, e;
      for (i = ip->objs.begin(), e = ip->objs.end(); i != e; )
    ip->objs.erase(i++);
    
      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();
    }

  virtual void
  sendObjectListToPeer(Bucket *msg, bool all, bool clear)
    {
      typename PeerMap::iterator pi, pe;
      typename ObjectMap::iterator oi, oe;
      size_t size = 0;
      size_t numobjs = 0;
      for (pi = peers_.begin(), pe = peers_.end(); pi != pe; ++pi)
    for (oi = pi->second.objs.begin(), oe = pi->second.objs.end(); oi != oe; ++oi, ++numobjs)
      if (all || (oi->flags & DQM_PROP_NEW))
        size += 9*sizeof(uint32_t) + oi->dirname->size()
            + oi->objname.size() + 1 + oi->scalar.size() + oi->qdata.size()
            + (oi->lastreq > 0 ? oi->rawdata.size() : 0);

      msg->data.reserve(msg->data.size() + size + 8 * sizeof(uint32_t));

      uint32_t nupdates = 0;
      uint32_t words [4];
      words[0] = sizeof(words);
      words[1] = DQM_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->flags & DQM_PROP_NEW))
      {
        sendObjectToPeer(msg, const_cast<ObjType &>(*oi), oi->lastreq > 0);
        if (clear)
          const_cast<ObjType &>(*oi).flags &= ~DQM_PROP_NEW;
        ++nupdates;
      }

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

  virtual void
  sendObjectListToPeers(bool all)
    {
      typename PeerMap::iterator i, e;
      typename ObjectMap::iterator oi, oe;
      for (i = peers_.begin(), e = peers_.end(); i != e; ++i)
      {
    ImplPeer &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;
      }
    }

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

protected:
  PeerMap       peers_;
};
  

class DQMBasicNet : public DQMImplNet<DQMNet::Object>
{
public:
  DQMBasicNet(const std::string &appname = "");

  void          reserveLocalSpace(uint32_t size);
  void          updateLocalObject(Object &o);
  bool          removeLocalExcept(const std::set<std::string> &known);

private:
  ImplPeer      *local_;
};


#endif // DQMSERVICES_CORE_DQM_NET_H