qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cctriel...@apache.org
Subject svn commit: r818244 - in /qpid/trunk/qpid/cpp/src/qpid/broker: Broker.cpp Broker.h Queue.cpp QueueEvents.cpp QueueEvents.h
Date Wed, 23 Sep 2009 20:34:48 GMT
Author: cctrieloff
Date: Wed Sep 23 20:34:48 2009
New Revision: 818244

URL: http://svn.apache.org/viewvc?rev=818244&view=rev
Log:
This patch requires svn 817742, corrects the lock issue for ring queue in 817742, and protects
replication when used together with flow to disk

Modified:
    qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
    qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.cpp
    qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.h

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp?rev=818244&r1=818243&r2=818244&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.cpp Wed Sep 23 20:34:48 2009
@@ -91,7 +91,8 @@
     queueLimit(100*1048576/*100M default limit*/),
     tcpNoDelay(false),
     requireEncrypted(false),
-    maxSessionRate(0)
+    maxSessionRate(0),
+    asyncQueueEvents(true)
 {
     int c = sys::SystemInfo::concurrency();
     workerThreads=c+1;
@@ -121,7 +122,8 @@
         ("tcp-nodelay", optValue(tcpNoDelay), "Set TCP_NODELAY on TCP connections")
         ("require-encryption", optValue(requireEncrypted), "Only accept connections that
are encrypted")
         ("known-hosts-url", optValue(knownHosts, "URL or 'none'"), "URL to send as 'known-hosts'
to clients ('none' implies empty list)")
-        ("max-session-rate", optValue(maxSessionRate, "MESSAGES/S"), "Sets the maximum message
rate per session (0=unlimited)");
+        ("max-session-rate", optValue(maxSessionRate, "MESSAGES/S"), "Sets the maximum message
rate per session (0=unlimited)")
+        ("async-queue-events", optValue(asyncQueueEvents, "yes|no"), "Set Queue Events async,
used for services like replication");
 }
 
 const std::string empty;
@@ -150,7 +152,7 @@
         *this),
     managementAgent(conf.enableMgmt ? new ManagementAgent() : 0),
     queueCleaner(queues, timer),
-    queueEvents(poller),
+    queueEvents(poller,!conf.asyncQueueEvents), 
     recovery(true),
     expiryPolicy(new ExpiryPolicy),
     getKnownBrokers(boost::bind(&Broker::getKnownBrokersImpl, this))

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h?rev=818244&r1=818243&r2=818244&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Broker.h Wed Sep 23 20:34:48 2009
@@ -111,6 +111,7 @@
         bool requireEncrypted;
         std::string knownHosts;
         uint32_t maxSessionRate;
+        bool asyncQueueEvents;
 
       private:
         std::string getHome();

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=818244&r1=818243&r2=818244&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Wed Sep 23 20:34:48 2009
@@ -569,9 +569,6 @@
     {
         Mutex::ScopedLock locker(messageLock);   
         QueuedMessage qm(this, msg, ++sequence);
-        if (policy.get()) {
-            policy->enqueued(qm);
-        }
         if (insertSeqNo) msg->getOrInsertHeaders().setInt64(seqNoKey, sequence);
          
         LVQ::iterator i;
@@ -605,6 +602,10 @@
             if (eventMgr) eventMgr->enqueued(qm);
             else QPID_LOG(warning, "Enqueue manager not set, events not generated for " <<
getName());
         }
+        if (policy.get()) {
+            Mutex::ScopedUnlock locker(messageLock);   
+            policy->enqueued(qm);
+        }
     }
     copy.notify();
 }
@@ -792,9 +793,16 @@
 
 void Queue::configure(const FieldTable& _settings, bool recovering)
 {
+
+    eventMode = _settings.getAsInt(qpidQueueEventGeneration);
+
     if (QueuePolicy::getType(_settings) == QueuePolicy::FLOW_TO_DISK && 
-        (!store || NullMessageStore::isNullStore(store))) {
-        QPID_LOG(warning, "Flow to disk not valid for non-persisted queue");
+        (!store || NullMessageStore::isNullStore(store) || (eventMode && eventMgr
&& !eventMgr->isSync()) )) {
+        if ( NullMessageStore::isNullStore(store)) {
+            QPID_LOG(warning, "Flow to disk not valid for non-persisted queue:" <<
getName());
+        } else if (eventMgr && !eventMgr->isSync() ) {
+            QPID_LOG(warning, "Flow to disk not valid with async Queue Events:" <<
getName());
+        }
         FieldTable copy(_settings);
         copy.erase(QueuePolicy::typeKey);
         setPolicy(QueuePolicy::createQueuePolicy(getName(), copy));
@@ -803,19 +811,19 @@
     }
     //set this regardless of owner to allow use of no-local with exclusive consumers also
     noLocal = _settings.get(qpidNoLocal);
-    QPID_LOG(debug, "Configured queue with no-local=" << noLocal);
+    QPID_LOG(debug, "Configured queue " << getName() << " with no-local=" <<
noLocal);
 
     lastValueQueue= _settings.get(qpidLastValueQueue);
-    if (lastValueQueue) QPID_LOG(debug, "Configured queue as Last Value Queue");
+    if (lastValueQueue) QPID_LOG(debug, "Configured queue as Last Value Queue for: " <<
getName());
 
     lastValueQueueNoBrowse = _settings.get(qpidLastValueQueueNoBrowse);
     if (lastValueQueueNoBrowse){
-        QPID_LOG(debug, "Configured queue as Last Value Queue No Browse");
+        QPID_LOG(debug, "Configured queue as Last Value Queue No Browse for: " << getName());
         lastValueQueue = lastValueQueueNoBrowse;
     }
     
     persistLastNode= _settings.get(qpidPersistLastNode);
-    if (persistLastNode) QPID_LOG(debug, "Configured queue to Persist data if cluster fails
to one node");
+    if (persistLastNode) QPID_LOG(debug, "Configured queue to Persist data if cluster fails
to one node for: " << getName());
 
     traceId = _settings.getAsString(qpidTraceIdentity);
     std::string excludeList = _settings.getAsString(qpidTraceExclude);
@@ -825,8 +833,6 @@
     QPID_LOG(debug, "Configured queue " << getName() << " with qpid.trace.id='"
<< traceId 
              << "' and qpid.trace.exclude='"<< excludeList << "' i.e. "
<< traceExclude.size() << " elements");
 
-    eventMode = _settings.getAsInt(qpidQueueEventGeneration);
-
     FieldTable::ValuePtr p =_settings.get(qpidInsertSequenceNumbers);
     if (p && p->convertsTo<std::string>()) insertSequenceNumbers(p->get<std::string>());
 

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.cpp?rev=818244&r1=818243&r2=818244&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.cpp Wed Sep 23 20:34:48 2009
@@ -25,25 +25,41 @@
 namespace qpid {
 namespace broker {
 
-QueueEvents::QueueEvents(const boost::shared_ptr<sys::Poller>& poller) : 
-    eventQueue(boost::bind(&QueueEvents::handle, this, _1), poller), enabled(true) 
+QueueEvents::QueueEvents(const boost::shared_ptr<sys::Poller>& poller, bool isSync)
: 
+    eventQueue(boost::bind(&QueueEvents::handle, this, _1), poller), enabled(true), sync(isSync)

 {
-    eventQueue.start();
+    if (!sync) eventQueue.start();
 }
 
 QueueEvents::~QueueEvents() 
 {
-    eventQueue.stop();
+    if (!sync) eventQueue.stop();
 }
 
 void QueueEvents::enqueued(const QueuedMessage& m)
 {
-    if (enabled) eventQueue.push(Event(ENQUEUE, m));
+    if (enabled) {
+        Event enq(ENQUEUE, m);
+        if (sync) {
+            for (Listeners::iterator j = listeners.begin(); j != listeners.end(); j++) 
+                j->second(enq);
+        } else {
+            eventQueue.push(enq);
+        }
+    }
 }
 
 void QueueEvents::dequeued(const QueuedMessage& m)
 {
-    if (enabled) eventQueue.push(Event(DEQUEUE, m));
+    if (enabled) {
+        Event deq(DEQUEUE, m);
+        if (sync) {
+            for (Listeners::iterator j = listeners.begin(); j != listeners.end(); j++) 
+                j->second(deq);
+        } else {
+            eventQueue.push(Event(DEQUEUE, m));
+        }
+    }
 }
 
 void QueueEvents::registerListener(const std::string& id, const EventListener& listener)
@@ -70,15 +86,16 @@
 QueueEvents::handle(const EventQueue::Batch& events) {
     qpid::sys::Mutex::ScopedLock l(lock);
     for (EventQueue::Batch::const_iterator i = events.begin(); i != events.end(); ++i) {
-        for (Listeners::iterator j = listeners.begin(); j != listeners.end(); j++) 
-            j->second(*i);
+        for (Listeners::iterator j = listeners.begin(); j != listeners.end(); j++) {
+             j->second(*i);
+        }
     }
     return events.end();
 }
 
 void QueueEvents::shutdown()
 {
-    if (!eventQueue.empty() && !listeners.empty()) eventQueue.shutdown();
+    if (!sync && !eventQueue.empty() && !listeners.empty()) eventQueue.shutdown();
 }
 
 void QueueEvents::enable()
@@ -93,6 +110,12 @@
     QPID_LOG(debug, "Queue events disabled");
 }
 
+bool QueueEvents::isSync()
+{
+    return sync;
+}
+
+
 QueueEvents::Event::Event(EventType t, const QueuedMessage& m) : type(t), msg(m) {}
 
 

Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.h?rev=818244&r1=818243&r2=818244&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueEvents.h Wed Sep 23 20:34:48 2009
@@ -54,7 +54,7 @@
 
     typedef boost::function<void (Event)> EventListener;
 
-    QPID_BROKER_EXTERN QueueEvents(const boost::shared_ptr<sys::Poller>& poller);
+    QPID_BROKER_EXTERN QueueEvents(const boost::shared_ptr<sys::Poller>& poller,
bool isSync = false);
     QPID_BROKER_EXTERN ~QueueEvents();
     QPID_BROKER_EXTERN void enqueued(const QueuedMessage&);
     QPID_BROKER_EXTERN void dequeued(const QueuedMessage&);
@@ -65,6 +65,7 @@
     void disable();
     //process all outstanding events
     QPID_BROKER_EXTERN void shutdown();
+    QPID_BROKER_EXTERN bool isSync();
   private:
     typedef qpid::sys::PollableQueue<Event> EventQueue;
     typedef std::map<std::string, EventListener> Listeners;
@@ -73,6 +74,7 @@
     Listeners listeners;
     volatile bool enabled;
     qpid::sys::Mutex lock;//protect listeners from concurrent access
+    bool sync;
     
     EventQueue::Batch::const_iterator handle(const EventQueue::Batch& e);
 



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message