Merge pull request #168 from Mani-D/nativeTxn

Enhanced call site info to store probe name, file, function and line number of the probe location

.gitignore

@@ -23,3 +23,4 @@ ko/xpedite.ko.unsigned
 scripts/bin/xpeditec
 *.class
 .gradle
+**/.eggs

include/xpedite/framework/Persister.H

@@ -8,7 +8,7 @@
 
 #pragma once
 #include <xpedite/probes/Sample.H>
-#include <xpedite/framework/CallSiteInfo.H>
+#include <xpedite/framework/ProbeInfo.H>
 #include <vector>
 #include <cstring>
 
@@ -46,46 +46,88 @@ namespace xpedite { namespace framework {
     timeval  _time;
     uint64_t _tscHz;
     uint32_t _pmcCount;
-    uint32_t _callSiteCount;
-    CallSiteInfo _callSites[0];
+    uint32_t _probeCount;
+    uint32_t _probeInfoBufSize;
+    ProbeInfo _probes[0];
 
     public:
 
-    static constexpr uint64_t XPEDITE_VERSION {0x0200};
+    static constexpr uint64_t XPEDITE_MIN_COMPATIBLE_VERSION {0x0200};
+    static constexpr uint64_t XPEDITE_VERSION {0x0210};
     static constexpr uint64_t XPEDITE_FILE_HDR_SIG {0xC01DC01DC0FFEEEE};
 
-    static size_t callSiteSize(uint64_t callSiteCount_) {
-      return sizeof(CallSiteInfo) * callSiteCount_;
+    size_t size() const noexcept {
+      return sizeof(FileHeader) + _probeInfoBufSize;
     }
 
-    static size_t capacity(uint64_t callSiteCount_) {
-      return sizeof(FileHeader) + callSiteSize(callSiteCount_);
-    }
-
-    FileHeader(const std::vector<CallSiteInfo>& callSites_, timeval time_, uint64_t tscHz_, uint32_t pmcCount_)
+    FileHeader(timeval time_, uint64_t tscHz_, uint32_t pmcCount_)
       : _signature {XPEDITE_FILE_HDR_SIG}, _version {XPEDITE_VERSION}, _time (time_),
-        _tscHz {tscHz_}, _pmcCount {pmcCount_}, _callSiteCount {static_cast<uint32_t>(callSites_.size())} {
-      memcpy(reinterpret_cast<char*>(_callSites), callSites_.data(), callSiteSize(callSites_.size()));
+        _tscHz {tscHz_}, _pmcCount {pmcCount_}, _probeCount {}, _probeInfoBufSize {} {
     }
 
+    FileHeader(const FileHeader&)             = delete;
+    FileHeader& operator=(const FileHeader&)  = delete;
+    FileHeader(FileHeader&&)                  = delete;
+    FileHeader& operator=(FileHeader&&)       = delete;
+
     bool isValid() const noexcept {
-      return _signature == XPEDITE_FILE_HDR_SIG && _version == XPEDITE_VERSION;
+      return _signature == XPEDITE_FILE_HDR_SIG && _version >= XPEDITE_MIN_COMPATIBLE_VERSION && _version <= XPEDITE_VERSION;
     }
 
-    timeval time()      const noexcept { return _time;     }
-    uint64_t tscHz()    const noexcept { return _tscHz;    }
-    uint32_t pmcCount() const noexcept { return _pmcCount; }
+    timeval time()           const noexcept { return _time;       }
+    uint64_t tscHz()         const noexcept { return _tscHz;      }
+    uint32_t pmcCount()      const noexcept { return _pmcCount;   }
+    uint32_t probeCount()    const noexcept { return _probeCount; }
 
     const SegmentHeader* segmentHeader() const noexcept {
-      return reinterpret_cast<const SegmentHeader*>(reinterpret_cast<const char*>(this + 1) + callSiteSize(_callSiteCount));
+      if(_version < XPEDITE_VERSION) {
+        auto bufSize = sizeof(FileHeader) - 4 + _probeCount * 16;
+        return reinterpret_cast<const SegmentHeader*>(reinterpret_cast<const char*>(this) + bufSize);
+      }
+      return reinterpret_cast<const SegmentHeader*>(reinterpret_cast<const char*>(this + 1) + _probeInfoBufSize);
     }
 
-    std::tuple<const CallSiteInfo*, uint32_t> callSites() const noexcept {
-      return std::make_tuple(&_callSites[0], _callSiteCount);
+    void add(const void* callSite_, probes::CallSiteAttr attr_, uint32_t id_, Name probeName_,
+        Name fileName_, Name function_, uint32_t lineNo_) {
+      ++_probeCount;
+      auto* probeInfo = new (reinterpret_cast<char*>(this + 1) + _probeInfoBufSize) ProbeInfo {
+        callSite_, attr_, id_, probeName_, fileName_, function_, lineNo_
+      };
+      _probeInfoBufSize += probeInfo->size();
     }
+
+    template<typename T>
+    void forEachCallSiteInfo(T t) const {
+      if(_version < XPEDITE_VERSION) {
+        return;
+      }
+      const ProbeInfo* probeInfo {_probes};
+      for(uint32_t i=0; i<_probeCount; ++i) {
+        t(probeInfo);
+        probeInfo = reinterpret_cast<const ProbeInfo*>(reinterpret_cast<const char*>(probeInfo) + probeInfo->size());
+      }
+      assert(reinterpret_cast<const char*>(this)+size() == reinterpret_cast<const char*>(probeInfo));
+    }
+
   } __attribute__((packed));
 
-  void persistHeader(int fd_);
-  void persistData(int fd_, const probes::Sample* begin_, const probes::Sample* end_);
+  class Persister
+  {
+    FileHeader* _hdr;
+    std::vector<char> _buffer;
+
+    void resizeBuffer(size_t objSize_);
+
+    size_t freeSize() const noexcept {
+      return _buffer.size() - _hdr->size();
+    }
+
+    public:
+
+    Persister();
+
+    void persistHeader(int fd_);
+    void persistData(int fd_, const probes::Sample* begin_, const probes::Sample* end_);
+  };
 
 }}

include/xpedite/framework/ProbeInfo.H

@@ -0,0 +1,153 @@
+///////////////////////////////////////////////////////////////////////////////
+//
+// Provides implementation for storing and locating information about call sites
+// in a trager application.
+//
+// ProbeInfo - Stores the address, attribtues and id of a call site
+//
+// CallSiteMap - A collection of call sites in a target application
+//
+// Author: Manikandan Dhamodharan, Morgan Stanley
+//
+///////////////////////////////////////////////////////////////////////////////
+
+#pragma once
+#include <xpedite/probes/CallSite.H>
+#include <unordered_map>
+#include <sstream>
+#include <cstring>
+#include <cassert>
+
+namespace xpedite { namespace framework {
+
+  struct Name {
+    // null terminated c-string
+    const char* _data;
+
+    // size of the string including the null-char
+    uint32_t _size;
+  };
+
+  class FileHeader;
+
+  enum class ProbeType {
+    Invalid = 0,
+    TxnBeginProbe,
+    TxnSuspendProbe,
+    TxnResumeProbe,
+    TxnEndProbe
+  };
+
+  class ProbeInfo
+  {
+    friend class FileHeader;
+    const void* _callSite;
+    probes::CallSiteAttr _attr;
+    uint32_t _id;
+    uint32_t  _probeNameOffset;
+    uint32_t _fileNameOffset;
+    uint32_t _functionNameOffset;
+    uint32_t _lineNo;
+    uint32_t _size;
+    char _data[0];
+
+    ProbeInfo(const void* callSite_, probes::CallSiteAttr attr_, uint32_t id_, Name probeName_,
+        Name fileName_, Name functionName_, uint32_t lineNo_)
+      : _callSite {callSite_}, _attr {attr_}, _id {id_}, _lineNo {lineNo_}, _size {} {
+
+      _probeNameOffset = _size;
+      memcpy(_data+_size, probeName_._data, probeName_._size);
+      _size += probeName_._size;
+
+      _fileNameOffset  = _size;
+      memcpy(_data+_size, fileName_._data, fileName_._size);
+      _size += fileName_._size;
+
+      _functionNameOffset  = _size;
+      memcpy(_data+_size, functionName_._data, functionName_._size);
+      _size += functionName_._size;
+    }
+
+    ProbeInfo (const ProbeInfo&)            = delete;
+    ProbeInfo& operator=(const ProbeInfo&)  = delete;
+    ProbeInfo (ProbeInfo&&)                 = delete;
+    ProbeInfo& operator=(ProbeInfo&&)       = delete;
+
+    public:
+
+    const void* callSite()    const noexcept { 
+      return _callSite;
+    }
+
+    uint32_t id()              const noexcept { return _id;                          }
+    bool isActive()            const noexcept { return _attr.isActive();             }
+    bool canStoreData()        const noexcept { return _attr.canStoreData();         }
+    bool canBeginTxn()         const noexcept { return _attr.canBeginTxn();          }
+    bool canSuspendTxn()       const noexcept { return _attr.canSuspendTxn();        }
+    bool canResumeTxn()        const noexcept { return _attr.canResumeTxn();         }
+    bool canEndTxn()           const noexcept { return _attr.canEndTxn();            }
+    const char* probeName()    const noexcept { return _data + _probeNameOffset;     }
+    const char* fileName()     const noexcept { return _data + _fileNameOffset;      }
+    const char* functionName() const noexcept { return _data + _functionNameOffset;  }
+    uint32_t lineNo()          const noexcept { return _lineNo;                      }
+    size_t size()              const noexcept { return sizeof(ProbeInfo) + _size; }
+
+    ProbeType type() const noexcept {
+      if(canBeginTxn()) {
+        return ProbeType::TxnBeginProbe;
+      } else if(canSuspendTxn()) {
+        return ProbeType::TxnSuspendProbe;
+      } else if(canResumeTxn()) {
+        return ProbeType::TxnResumeProbe;
+      }
+      assert(canEndTxn());
+      return ProbeType::TxnEndProbe;
+    }
+
+    std::string toString() const {
+      std::ostringstream os;
+      os << "CallSite - " <<  callSite() << " | id - " << _id << " | " << _attr.toString()
+        << " | probe name - " << probeName() << " | file name - " << fileName()
+        << " | line no - " << _lineNo << " | function - " << functionName();
+      return os.str();
+    }
+
+  } __attribute__((packed));
+
+
+  class ProbeInfoMap
+  {
+    using Map = std::unordered_map<const void*, const ProbeInfo*>;
+    Map _map;
+
+    public:
+
+    using value_type = Map::value_type;
+
+    void add(const ProbeInfo* probeInfo_) {
+      _map.emplace(probeInfo_->callSite(), probeInfo_);
+    }
+
+    const ProbeInfo* locateInfo(const void* callSite_) const noexcept {
+      auto it = _map.find(callSite_);
+      if(it != _map.end()) {
+        return it->second;
+      }
+      return {};
+    }
+
+    const Map& data() const noexcept {
+      return _map;
+    }
+
+    std::string toString() const {
+      std::ostringstream os;
+      for(auto& kvp : _map) {
+        os << kvp.second->toString() << std::endl;
+      }
+      return os.str();
+    }
+
+  };
+
+}}

include/xpedite/framework/SamplesBuffer.H

@@ -58,11 +58,11 @@ namespace xpedite { namespace framework {
       return _head.load(std::memory_order_relaxed);
     }
 
-    static bool attachAll(const std::string& fileNamePattern_) noexcept {
+    static bool attachAll(Persister& persister_, const std::string& fileNamePattern_) noexcept {
       auto begin = SamplesBuffer::head();
       auto buffer = begin;
       while(buffer) {
-        if(!buffer->attachReader(fileNamePattern_)) {
+        if(!buffer->attachReader(persister_, fileNamePattern_)) {
           break;
         }
         buffer = buffer->next();
@@ -97,7 +97,7 @@ namespace xpedite { namespace framework {
       return _fd >= 0;
     }
 
-    bool attachReader(const std::string& fileNamePattern_) noexcept {
+    bool attachReader(Persister& persister_, const std::string& fileNamePattern_) noexcept {
       if(isReaderAttached()) {
         XpediteLogError << "xpedite - failed to attach reader to thread " << tid() 
           << " - reader already attached. attaching multiple readers not permitted" << XpediteLogEnd;
@@ -112,7 +112,7 @@ namespace xpedite { namespace framework {
         return false;
       }
 
-      persistHeader(_fd);
+      persister_.persistHeader(_fd);
       uint64_t rindex, windex;
       std::tie(rindex, windex) = _bufferPool.attachReader();
       XpediteLogInfo << "xpedite - attached reader to thread - " << tid() << " | buffer index state - [readIndex - "