QPID-1160 - Per-thread counters in management API to avoid locking

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@672864 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 42 insertions, 19 deletions

diff --git a/cpp/src/qpid/management/ManagementAgent.cpp b/cpp/src/qpid/management/ManagementAgent.cpp
deleted file mode 100644
index e69de29bb2..0000000000
--- a/cpp/src/qpid/management/ManagementAgent.cpp
+++ /dev/null
diff --git a/cpp/src/qpid/management/ManagementAgent.h b/cpp/src/qpid/management/ManagementAgent.h
index c38e273c49..c8a1b37823 100644
--- a/cpp/src/qpid/management/ManagementAgent.h
+++ b/cpp/src/qpid/management/ManagementAgent.h
@@ -37,6 +37,8 @@ class ManagementAgent
 
     static shared_ptr getAgent (void);
 
+    virtual int  getMaxThreads() = 0;
+
     virtual void RegisterClass (std::string packageName,
                                 std::string className,
                                 uint8_t*    md5Sum,
diff --git a/cpp/src/qpid/management/ManagementBroker.cpp b/cpp/src/qpid/management/ManagementBroker.cpp
index 24d18875b6..271a2ec73c 100644
--- a/cpp/src/qpid/management/ManagementBroker.cpp
+++ b/cpp/src/qpid/management/ManagementBroker.cpp
@@ -47,8 +47,8 @@ ManagementBroker::RemoteAgent::~RemoteAgent ()
         mgmtObject->resourceDestroy ();
 }
 
-ManagementBroker::ManagementBroker (string _dataDir, uint16_t _interval, Manageable* _broker) :
-    dataDir (_dataDir), interval (_interval), broker (_broker)
+ManagementBroker::ManagementBroker (string _dataDir, uint16_t _interval, Manageable* _broker, int _threads) :
+    threadPoolSize(_threads), dataDir(_dataDir), interval(_interval), broker(_broker)
 {
     timer.add (intrusive_ptr<TimerTask> (new Periodic(*this, interval)));
     localBank      = 5;
@@ -105,11 +105,11 @@ void ManagementBroker::writeData ()
     }
 }
 
-void ManagementBroker::enableManagement (string dataDir, uint16_t interval, Manageable* broker)
+void ManagementBroker::enableManagement (string dataDir, uint16_t interval, Manageable* broker, int threadPoolSize)
 {
     enabled = 1;
     if (agent.get () == 0)
-        agent = shared_ptr (new ManagementBroker (dataDir, interval, broker));
+        agent = shared_ptr (new ManagementBroker (dataDir, interval, broker, threadPoolSize));
 }
 
 ManagementAgent::shared_ptr ManagementAgent::getAgent (void)
@@ -634,7 +634,7 @@ void ManagementBroker::handleAttachRequestLH (Buffer& inBuffer, string replyToKe
     RemoteAgent* agent = new RemoteAgent;
     agent->objIdBank  = assignedBank;
     agent->mgmtObject = management::Agent::shared_ptr
-        (new management::Agent (agent));
+        (new management::Agent (this, agent));
     agent->mgmtObject->set_sessionId    (sessionId);
     agent->mgmtObject->set_label        (label);
     agent->mgmtObject->set_registeredTo (broker->GetManagementObject()->getObjectId());
diff --git a/cpp/src/qpid/management/ManagementBroker.h b/cpp/src/qpid/management/ManagementBroker.h
index 7548773960..18d30096e5 100644
--- a/cpp/src/qpid/management/ManagementBroker.h
+++ b/cpp/src/qpid/management/ManagementBroker.h
@@ -41,19 +41,21 @@ class ManagementBroker : public ManagementAgent
 {
   private:
 
-    ManagementBroker (std::string dataDir, uint16_t interval, Manageable* broker);
+    ManagementBroker (std::string dataDir, uint16_t interval, Manageable* broker, int threadPoolSize);
+    int threadPoolSize;
 
   public:
 
     virtual ~ManagementBroker ();
 
-    static void       enableManagement (std::string dataDir, uint16_t interval, Manageable* broker);
+    static void       enableManagement (std::string dataDir, uint16_t interval, Manageable* broker, int threadPoolSize);
     static shared_ptr getAgent (void);
     static void       shutdown (void);
 
     void setInterval     (uint16_t _interval) { interval = _interval; }
     void setExchange     (broker::Exchange::shared_ptr mgmtExchange,
                           broker::Exchange::shared_ptr directExchange);
+    int  getMaxThreads   () { return threadPoolSize; }
     void RegisterClass   (std::string packageName,
                           std::string className,
                           uint8_t*    md5Sum,
diff --git a/cpp/src/qpid/management/ManagementObject.cpp b/cpp/src/qpid/management/ManagementObject.cpp
index 68d7e5c886..2528ed4284 100644
--- a/cpp/src/qpid/management/ManagementObject.cpp
+++ b/cpp/src/qpid/management/ManagementObject.cpp
@@ -21,12 +21,15 @@
  
 #include "Manageable.h"
 #include "ManagementObject.h"
+#include "ManagementAgent.h"
 #include "qpid/framing/FieldTable.h"
 
 using namespace qpid::framing;
 using namespace qpid::management;
 using namespace qpid::sys;
 
+int ManagementObject::nextThreadIndex = 0;
+
 void ManagementObject::writeTimestamps (Buffer& buf)
 {
     buf.putShortString (getPackageName ());
@@ -40,3 +43,14 @@ void ManagementObject::writeTimestamps (Buffer& buf)
 
 void ManagementObject::setReference(uint64_t) {}
 
+int ManagementObject::getThreadIndex() {
+    static __thread int thisIndex = -1;
+    if (thisIndex == -1) {
+        sys::Mutex::ScopedLock mutex(accessLock);
+        thisIndex = nextThreadIndex;
+        if (nextThreadIndex < agent->getMaxThreads() - 1)
+            nextThreadIndex++;
+    }
+    return thisIndex;
+}
+
diff --git a/cpp/src/qpid/management/ManagementObject.h b/cpp/src/qpid/management/ManagementObject.h
index cf2da13b09..732dd14a24 100644
--- a/cpp/src/qpid/management/ManagementObject.h
+++ b/cpp/src/qpid/management/ManagementObject.h
@@ -32,19 +32,22 @@ namespace qpid {
 namespace management {
 
 class Manageable;
+class ManagementAgent;
 
 class ManagementObject
 {
   protected:
     
-    uint64_t    createTime;
-    uint64_t    destroyTime;
-    uint64_t    objectId;
-    bool        configChanged;
-    bool        instChanged;
-    bool        deleted;
-    Manageable* coreObject;
-    sys::Mutex  accessLock;
+    uint64_t         createTime;
+    uint64_t         destroyTime;
+    uint64_t         objectId;
+    bool             configChanged;
+    bool             instChanged;
+    bool             deleted;
+    Manageable*      coreObject;
+    sys::Mutex       accessLock;
+    ManagementAgent* agent;
+    int              maxThreads;
 
     static const uint8_t TYPE_U8        = 1;
     static const uint8_t TYPE_U16       = 2;
@@ -73,15 +76,18 @@ class ManagementObject
     static const uint8_t FLAG_INDEX  = 0x02;
     static const uint8_t FLAG_END    = 0x80;
 
+    static       int nextThreadIndex;
+        
+    int  getThreadIndex();
     void writeTimestamps (qpid::framing::Buffer& buf);
 
   public:
     typedef boost::shared_ptr<ManagementObject> shared_ptr;
     typedef void (*writeSchemaCall_t) (qpid::framing::Buffer&);
 
-    ManagementObject (Manageable* _core) :
+    ManagementObject (ManagementAgent* _agent, Manageable* _core) :
         destroyTime(0), objectId (0), configChanged(true),
-        instChanged(true), deleted(false), coreObject(_core)
+        instChanged(true), deleted(false), coreObject(_core), agent(_agent)
     { createTime = uint64_t (qpid::sys::Duration (qpid::sys::now ())); }
     virtual ~ManagementObject () {}
 
@@ -102,8 +108,7 @@ class ManagementObject
     uint64_t     getObjectId      (void) { return objectId; }
     inline  bool getConfigChanged (void) { return configChanged; }
     virtual bool getInstChanged   (void) { return instChanged; }
-    inline  void setAllChanged    (void)
-    {
+    inline  void setAllChanged    (void) {
         configChanged = true;
         instChanged   = true;
     }