Peter M. Groen / FlashMQ

Commit 295cf8eeabc557a317b1cb30f102f30a6d16182f

Authored by Wiebe Cazemier
Committed by Wiebe Cazemier
1 parent 6672b8c9

Lock (QoS data of) sessions during copying/saving 

The only mutable session data of a client is QoS related, so when we're
copying sessions (for saving them), we need to lock the QoS data,
because that gets modified from active client traffic in worker threads.

Note: not super well tested at this point, nor was I ever able to
trigger actual errors despite long stress testing, so it's a theoretical
fix.
Inline Side-by-side
Showing 2 changed files with 13 additions and 2 deletions
session.cpp
  View file @295cf8e
... ... @@ -67,6 +67,10 @@ void Session::setSessionTouch(int64_t ageInMs)
67 67 */
68 68 Session::Session(const Session &other)
69 69 {
  70 + // Only the QoS data is modified by worker threads (vs (locked) timed events), so it could change during copying, because
  71 + // it gets called from a separate thread.
  72 + std::unique_lock<std::mutex> locker(qosQueueMutex);
  73 +
70 74 this->username = other.username;
71 75 this->client_id = other.client_id;
72 76 this->incomingQoS2MessageIds = other.incomingQoS2MessageIds;
... ... @@ -236,17 +240,21 @@ bool Session::hasExpired(int expireAfterSeconds)
236 240  
237 241 void Session::addIncomingQoS2MessageId(uint16_t packet_id)
238 242 {
  243 + std::unique_lock<std::mutex> locker(qosQueueMutex);
239 244 incomingQoS2MessageIds.insert(packet_id);
240 245 }
241 246  
242   -bool Session::incomingQoS2MessageIdInTransit(uint16_t packet_id) const
  247 +bool Session::incomingQoS2MessageIdInTransit(uint16_t packet_id)
243 248 {
  249 + std::unique_lock<std::mutex> locker(qosQueueMutex);
244 250 const auto it = incomingQoS2MessageIds.find(packet_id);
245 251 return it != incomingQoS2MessageIds.end();
246 252 }
247 253  
248 254 void Session::removeIncomingQoS2MessageId(u_int16_t packet_id)
249 255 {
  256 + std::unique_lock<std::mutex> locker(qosQueueMutex);
  257 +
250 258 #ifndef NDEBUG
251 259 logger->logf(LOG_DEBUG, "As QoS 2 receiver: publish released (PUBREL) for '%s', packet id '%d'. Left in queue: %d", client_id.c_str(), packet_id, incomingQoS2MessageIds.size());
252 260 #endif
... ... @@ -258,11 +266,14 @@ void Session::removeIncomingQoS2MessageId(u_int16_t packet_id)
258 266  
259 267 void Session::addOutgoingQoS2MessageId(uint16_t packet_id)
260 268 {
  269 + std::unique_lock<std::mutex> locker(qosQueueMutex);
261 270 outgoingQoS2MessageIds.insert(packet_id);
262 271 }
263 272  
264 273 void Session::removeOutgoingQoS2MessageId(u_int16_t packet_id)
265 274 {
  275 + std::unique_lock<std::mutex> locker(qosQueueMutex);
  276 +
266 277 #ifndef NDEBUG
267 278 logger->logf(LOG_DEBUG, "As QoS 2 sender: publish complete (PUBCOMP) for '%s', packet id '%d'. Left in queue: %d", client_id.c_str(), packet_id, outgoingQoS2MessageIds.size());
268 279 #endif
... ...
session.h
  View file @295cf8e
... ... @@ -77,7 +77,7 @@ public:
77 77 bool hasExpired(int expireAfterSeconds);
78 78  
79 79 void addIncomingQoS2MessageId(uint16_t packet_id);
80   - bool incomingQoS2MessageIdInTransit(uint16_t packet_id) const;
  80 + bool incomingQoS2MessageIdInTransit(uint16_t packet_id);
81 81 void removeIncomingQoS2MessageId(u_int16_t packet_id);
82 82  
83 83 void addOutgoingQoS2MessageId(uint16_t packet_id);
... ...