Changeset 1b5188 for src

Timestamp:

Jul 30, 2015, 8:47:36 PM (10 years ago)

Author:

Frederik Heber <heber@…>

Children:

454bc54

Parents:

163eec

git-author:

Frederik Heber <heber@…> (06/18/15 00:02:38)

git-committer:

Frederik Heber <heber@…> (07/30/15 20:47:36)

Message:

Observables are now protected by mutexes, i.e. allow for concurrent use.

Location:

src

Files:

• CodePatterns/Observer/Observable.hpp (modified) (2 diffs)
• Observer/Observable.cpp (modified) (17 diffs)

src/CodePatterns/Observer/Observable.hpp

@@ -18 +18 @@
 #include <string>
 #include <boost/function.hpp>
+#include <boost/thread.hpp>
 
 #include "CodePatterns/Range.hpp"
@@ -135 +136 @@
   static std::set<Observable*> busyObservables;
 
+  static boost::recursive_mutex ObservablesMapLock; //!< a lock for the pointer of the instance
+
 private:
   friend class Zombie;

src/Observer/Observable.cpp

@@ -46 +46 @@
 std::set<Observable*> Observable::busyObservables; //!< Set of Observables that are currently busy notifying their sign-on'ed Observers
 Observable::ChannelMap Observable::NotificationChannels; //!< Map of Observables to their Channels.
+boost::recursive_mutex Observable::ObservablesMapLock;  //!< mutex for locking the above maps
 
 // ValidRange must be initialized before PriorityLevel.
@@ -77 +78 @@
  * \param *publisher reference of sub-observable
  */
-void Observable::start_observer_internal(Observable *publisher){
+void Observable::start_observer_internal(Observable *publisher)
+{
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   // increase the count for this observable by one
   // if no entry for this observable is found, an new one is created
@@ -97 +100 @@
  * \param *publisher reference of sub-observable
  */
-void Observable::finish_observer_internal(Observable *publisher){
+void Observable::finish_observer_internal(Observable *publisher)
+{
   // decrease the count for this observable
   // if zero is reached all observed blocks are done and we can
   // start to notify our observers
+  ObservablesMapLock.lock();
   --depth[publisher];
+  ObservablesMapLock.unlock();
 #ifdef LOG_OBSERVER
   observerLog().addMessage(depth[publisher]) << "<< Unlocking " << observerLog().getName(publisher);
 #endif
-  if(depth[publisher]){}
+  ObservablesMapLock.lock();
+  int depth_publisher = depth[publisher];
+  ObservablesMapLock.unlock();
+  if(depth_publisher){}
   else{
     publisher->notifyAll();
     // this item is done, so we don't have to keep the count with us
     // save some memory by erasing it
+    boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
     depth.erase(publisher);
   }
 }
 
-void Observable::enque_notification_internal(Observable *publisher, Notification_ptr notification){
+void Observable::enque_notification_internal(Observable *publisher, Notification_ptr notification)
+{
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   notifications[publisher].insert(notification);
 }
@@ -160 +172 @@
   // we are busy notifying others right now
   // add ourselves to the list of busy subjects to enable circle detection
+  ObservablesMapLock.lock();
   busyObservables.insert(this);
+  ObservablesMapLock.unlock();
   // see if anyone has signed up for observation
   // and call all observers
   try {
-    if(callTable.count(this)) {
+    ObservablesMapLock.lock();
+    const bool anybodyThere = callTable.count(this);
+    ObservablesMapLock.unlock();
+    if(anybodyThere) {
       // elements are stored sorted by keys in the multimap
       // so iterating over it gives us a the callees sorted by
       // the priorities
+      ObservablesMapLock.lock();
       callees_t callees = callTable[this];
+      ObservablesMapLock.unlock();
       callees_t::iterator iter;
       for(iter=callees.begin();iter!=callees.end();++iter){
@@ -190 +209 @@
   }
 
+  ObservablesMapLock.lock();
   notifications.erase(this);
 
    // done with notification, we can leave the set of busy subjects
   busyObservables.erase(this);
+  ObservablesMapLock.unlock();
 
 #ifdef LOG_OBSERVER
@@ -209 +230 @@
 void Observable::update(Observable *publisher) {
   // circle detection
-  if(busyObservables.find(this)!=busyObservables.end()) {
+  ObservablesMapLock.lock();
+  const bool presentCircle = busyObservables.find(this)!=busyObservables.end();
+  ObservablesMapLock.unlock();
+  if(presentCircle) {
     // somehow a circle was introduced... we were busy notifying our
     // observers, but still we are called by one of our sub-Observables
@@ -222 +246 @@
     // if we are changing ourselves at the same time our sub-observables change
     // we do not need to publish all the changes at each time we are called
-    if(depth.find(this)==depth.end()) {
+    ObservablesMapLock.lock();
+    const bool depth_this = depth.find(this)==depth.end();
+    ObservablesMapLock.unlock();
+    if(depth_this) {
 #ifdef LOG_OBSERVER
       observerLog().addMessage() << "-* Update from " << observerLog().getName(publisher)
@@ -249 +276 @@
 #endif
   bool res = false;
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   callees_t &callees = callTable[const_cast<Observable *>(this)];
 
@@ -265 +293 @@
 void Observable::signOff(Observer *target) const
 {
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   ASSERT(callTable.count(const_cast<Observable *>(this)),
       "SignOff called for an Observable without Observers.");
@@ -290 +319 @@
 void Observable::signOn(Observer *target, size_t channelno) const
 {
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   Notification_ptr notification = getChannel(channelno);
 #ifdef LOG_OBSERVER
@@ -301 +331 @@
 void Observable::signOff(Observer *target, size_t channelno) const
 {
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   Notification_ptr notification = getChannel(channelno);
 #ifdef LOG_OBSERVER
@@ -313 +344 @@
 bool Observable::isBlocked() const
 {
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   return depth.count(const_cast<Observable *>(this)) > 0;
 }
@@ -318 +350 @@
 Notification_ptr Observable::getChannel(size_t no) const
 {
+  ObservablesMapLock.lock();
   const ChannelMap::const_iterator iter = NotificationChannels.find(const_cast<Observable * const>(this));
-  ASSERT(iter != NotificationChannels.end(),
+  const bool status = iter != NotificationChannels.end();
+  Channels *OurChannel = NULL;
+  if (status)
+    OurChannel = iter->second;
+  ObservablesMapLock.unlock();
+  ASSERT(status,
       "Observable::getChannel() - we do not have a channel in NotificationChannels.");
-  const Channels *OurChannel = iter->second;
   ASSERT(OurChannel != NULL,
       "Observable::getChannel() - observable has no channels.");
@@ -329 +366 @@
 size_t Observable::getNumberOfObservers() const
 {
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   size_t ObserverCount = 0;
   {
@@ -354 +392 @@
  *  \param *publisher Sub-Observable.
  */
-void Observable::subjectKilled(Observable *publisher){
+void Observable::subjectKilled(Observable *publisher)
+{
 }
 
@@ -379 +418 @@
       << observerLog().getName(static_cast<Observable *>(this));
 #endif
+  boost::recursive_mutex::scoped_lock guard(ObservablesMapLock);
   if(callTable.count(this)) {
     // delete all entries for this observable
+    ObservablesMapLock.lock();
     callees_t callees = callTable[this];
+    ObservablesMapLock.unlock();
     callees_t::iterator iter;
     for(iter=callees.begin();iter!=callees.end();++iter)
       (*iter).second->subjectKilled(this);
+    ObservablesMapLock.lock();
     callTable.erase(this);
+    ObservablesMapLock.unlock();
   }
 
@@ -392 +436 @@
   if (iter != NotificationChannels.end()) {
     iter->second->subjectKilled(static_cast<Observable *>(this));
+    ObservablesMapLock.lock();
     delete iter->second;
     NotificationChannels.erase(iter);
-  }
-}
+    ObservablesMapLock.unlock();
+  }
+}