messagequeue.cc

   1 /*
   2     Copyright 2004-2005 Chris Tallon
   3
   4     This file is part of VOMP.
   5
   6     VOMP is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     VOMP is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with VOMP.  If not, see <https://www.gnu.org/licenses/>.
  18 */
  19
  20 #include <algorithm>
  21
  22 #include "messagequeue.h"
  23
  24 #include "message.h"
  25 #include "log.h"
  26
  27 static const char* TAG = "MessageQueue";
  28
  29 MessageQueue* MessageQueue::instance{};
  30
  31 MessageQueue::MessageQueue()
  32 {
  33   instance = this;
  34   logger = LogNT::getInstance();
  35 }
  36
  37 MessageQueue::~MessageQueue() { instance = NULL; }
  38
  39 MessageQueue* MessageQueue::getInstance() { return instance; }
  40
  41 void MessageQueue::postMessage(Message* m)
  42 {
  43   LogNT::getInstance()->debug(TAG, "PostMessage");
  44   std::lock_guard<std::mutex> lg(messageQueueMutex); // Get the lock
  45   messages.push_back(m);
  46   LogNT::getInstance()->debug(TAG, "Pushed message. Notify...");
  47   messageQueueCond.notify_one();
  48 } // mutex unlocks
  49
  50
  51 void MessageQueue::flushMessageQueue()
  52 {
  53   std::lock_guard<std::mutex> lg(messageQueueMutex); // Get the lock
  54
  55   for (auto m : messages) delete m;
  56   messages.clear();
  57 }
  58
  59 bool MessageQueue::receiverExists(MessageReceiver* mr) // call with lock..
  60 {
  61   ReceiversI ri = std::find(receivers.begin(), receivers.end(), mr);
  62   return (ri != receivers.end());
  63 }
  64
  65 void MessageQueue::addReceiver(MessageReceiver* newMR)
  66 {
  67   std::lock_guard<std::mutex> lg(messageQueueMutex);
  68
  69   if (std::find(receivers.begin(), receivers.end(), newMR) == receivers.end())
  70   {
  71     logger->debug(TAG, "addReceiver: not found, adding {}", static_cast<void*>(newMR));
  72     receivers.push_back(newMR);
  73   }
  74 }
  75
  76 void MessageQueue::removeReceiver(MessageReceiver* toRemove)
  77 {
  78   std::unique_lock<std::mutex> ul(messageQueueMutex);
  79
  80   // Ignore the issue of the pre-defined targets
  81   // They outlive the message queue processing loop and therefore always exist
  82   // So just check message->to
  83
  84   while(1)
  85   {
  86     logger->debug(TAG, "Attempt remove receiver {}", static_cast<void*>(toRemove));
  87
  88     if (messageBeingProcessed && (messageBeingProcessed->to == toRemove))
  89     {
  90       // The message currently being processed by Control is with the receiver we're trying to remove
  91       // Release mutex, delay and retry
  92       logger->info(TAG, "Remove delay! Does this ever happen? {}", static_cast<void*>(toRemove));
  93
  94       ul.unlock();
  95       std::this_thread::sleep_for(std::chrono::milliseconds(100));
  96       continue;
  97     }
  98     else
  99     {
 100       ReceiversI toRemoveI = std::find(receivers.begin(), receivers.end(), toRemove);
 101       if (toRemoveI == receivers.end())
 102       {
 103         logger->error(TAG, "Remove error 1 {}", static_cast<void*>(toRemove));
 104         return;
 105       }
 106
 107       receivers.erase(toRemoveI);
 108       logger->debug(TAG, "Removed receiver {}", static_cast<void*>(toRemove));
 109       break;
 110     }
 111   }
 112 }
 113
 114 void MessageQueue::messageLoop()
 115 {
 116   std::unique_lock<std::mutex> lockWrapper(messageQueueMutex);  // locks. unlocks on out-of-scope
 117
 118   while(messageLoopRun)
 119   {
 120     messageQueueCond.wait(lockWrapper, [&] { return !messageLoopRun || !messages.empty(); });
 121     // locked
 122     logger->debug(TAG, "woke");
 123
 124     if (!messageLoopRun) break;
 125
 126     while(!messages.empty())
 127     {
 128       messageBeingProcessed = messages.front();
 129       messages.pop_front();
 130
 131       if (!messageBeingProcessed->p_to && !receiverExists(messageBeingProcessed->to))
 132       {
 133         // Receiver for this message has been deleted already
 134         logger->debug(TAG, "Dropping message {} for non-existent receiver {}", static_cast<void*>(messageBeingProcessed), static_cast<void*>(messageBeingProcessed->to));
 135         delete messageBeingProcessed;
 136         messageBeingProcessed = NULL;
 137         continue;
 138       }
 139
 140       logger->debug(TAG, "Dispatching message {} to {}", static_cast<void*>(messageBeingProcessed), static_cast<void*>(messageBeingProcessed->to));
 141
 142       lockWrapper.unlock();
 143       dispatchMessage(messageBeingProcessed);
 144       lockWrapper.lock();
 145
 146       delete messageBeingProcessed;
 147       messageBeingProcessed = NULL;
 148     }
 149   }
 150 }