[#260,!120] Removed Mutexing from IfaceMgr, PacketQueue must now be thread-safe

    Removed Mutexing from IfaceMgr, added it to default kea packet queue
    implementations.  PacketQueue derivations are required to be thread-safe,
    thus leaving upto implementers to choose how.

diff --git a/src/lib/dhcp/iface_mgr.cc b/src/lib/dhcp/iface_mgr.cc
index cdcbf5caccbc9e93665a030d03b99501d357c340..4a0dd9cfc27b26de0518acc858bc61ba2bcec3b1 100644 (file)
--- a/src/lib/dhcp/iface_mgr.cc
+++ b/src/lib/dhcp/iface_mgr.cc
@@ -227,11 +227,6 @@ Receiver::stop() {
     last_error_ = "thread stopped";
 }
 
-isc::util::thread::Mutex&
-Receiver::getLock() {
-    return(lock_);
-}
-
 void
 Receiver::setError(const std::string& error_msg) {
     last_error_ = error_msg;
@@ -1125,14 +1120,9 @@ Pkt4Ptr IfaceMgr::receive4Indirect(uint32_t timeout_sec, uint32_t timeout_usec /
     }
 
     // If we're here it should only be because there are DHCP packets waiting.
-    // Protected packet queue access.
-    Pkt4Ptr pkt;
-    {
-        Mutex::Locker lock(receiver_->getLock());
-        pkt = getPacketQueue4()->dequeuePacket();
-        if (!pkt) {
-            receiver_->clearReady(Receiver::RCV_READY);
-        }
+    Pkt4Ptr pkt = getPacketQueue4()->dequeuePacket();
+    if (!pkt) {
+        receiver_->clearReady(Receiver::RCV_READY);
     }
 
     return (pkt);
@@ -1456,14 +1446,9 @@ IfaceMgr::receive6Indirect(uint32_t timeout_sec, uint32_t timeout_usec /* = 0 */
     }
 
     // If we're here it should only be because there are DHCP packets waiting.
-    // Protected packet queue access.
-    Pkt6Ptr pkt;
-    {
-        Mutex::Locker lock(receiver_->getLock());
-        pkt = getPacketQueue6()->dequeuePacket();
-        if (!pkt) {
-            receiver_->clearReady(Receiver::RCV_READY);
-        }
+    Pkt6Ptr pkt = getPacketQueue6()->dequeuePacket();
+    if (!pkt) {
+        receiver_->clearReady(Receiver::RCV_READY);
     }
 
     return (pkt);
@@ -1643,7 +1628,6 @@ IfaceMgr::receiveDHCP4Packet(Iface& iface, const SocketInfo& socket_info) {
     }
 
     if (pkt) {
-        Mutex::Locker lock(receiver_->getLock());
         getPacketQueue4()->enqueuePacket(pkt, socket_info);
         receiver_->markReady(Receiver::RCV_READY);
     }
@@ -1675,7 +1659,6 @@ IfaceMgr::receiveDHCP6Packet(const SocketInfo& socket_info) {
     }
 
     if (pkt) {
-        Mutex::Locker lock(receiver_->getLock());
         getPacketQueue6()->enqueuePacket(pkt, socket_info);
         receiver_->markReady(Receiver::RCV_READY);
     }

diff --git a/src/lib/dhcp/iface_mgr.h b/src/lib/dhcp/iface_mgr.h
index dc950e9847771f3551e571c850b4787bc4724654..428523ea8265d5c3e4ed0d4cfa83557c6e6e76d5 100644 (file)
--- a/src/lib/dhcp/iface_mgr.h
+++ b/src/lib/dhcp/iface_mgr.h
@@ -19,7 +19,6 @@
 #include <util/optional_value.h>
 #include <util/watch_socket.h>
 #include <util/threads/thread.h>
-#include <util/threads/sync.h>
 
 #include <boost/function.hpp>
 #include <boost/noncopyable.hpp>
@@ -524,13 +523,6 @@ public:
     /// not done in the destructor to avoid race conditions.
     void stop();
 
-    /// @brief Fetches the receiver's thread mutex.
-    ///
-    /// This is used for instantation mutex locks to protect code blocks.
-    ///
-    /// @return a reference to the mutex
-    isc::util::thread::Mutex& getLock();
-
     /// @brief Sets the receiver error state
     ///
     /// This records the given error message and sets the error watch
@@ -557,9 +549,6 @@ public:
     /// Marked as ready when the DHCP packet receiver thread should terminate.
     isc::util::WatchSocket sockets_[RCV_TERMINATE + 1];
 
-    /// DHCP packet thread mutex.
-    isc::util::thread::Mutex lock_;
-
     /// DHCP packet receiver thread.
     isc::util::thread::ThreadPtr thread_ ;
 };

diff --git a/src/lib/dhcp/packet_queue.h b/src/lib/dhcp/packet_queue.h
index 0d33a4ef26c33f8aac6c9708729a2b574cdf290d..004bd36dffdb015b92cfb88eee248faad3103e32 100644 (file)
--- a/src/lib/dhcp/packet_queue.h
+++ b/src/lib/dhcp/packet_queue.h
@@ -11,6 +11,7 @@
 #include <dhcp/socket_info.h>
 #include <dhcp/pkt4.h>
 #include <dhcp/pkt6.h>
+#include <util/threads/sync.h>
 
 #include <boost/function.hpp>
 #include <boost/circular_buffer.hpp>
@@ -40,6 +41,8 @@ enum class QueueEnd {
 /// This class serves as the abstract interface for packet queue
 /// implementations which may be used by @c IfaceMgr to store
 /// inbound packets until they are a dequeued for processing.
+/// @note Derivations of this class MUST BE thread-safe.
+/// @endnote
 ///
 template<typename PacketTypePtr>
 class PacketQueue {
@@ -245,12 +248,14 @@ public:
 
     /// @brief Pushes a packet onto the queue
     ///
-    ///  Adds a packet onto the end of queue specified.
+    ///  Adds a packet onto the end of queue specified.  Note that this
+    ///  function is protected by a Mutex.
     ///
     /// @param packet packet to add to the queue
     /// @param to specifies the end of the queue to which the packet
     /// should be added.
     virtual void pushPacket(PacketTypePtr& packet, const QueueEnd& to=QueueEnd::BACK) {
+        isc::util::thread::Mutex::Locker lock(mutex_);
         if (to == QueueEnd::BACK) {
             queue_.push_back(packet);
         } else {
@@ -260,7 +265,8 @@ public:
 
     /// @brief Pops a packet from the queue
     ///
-    /// Removes a packet from the end of the queue specified and returns it.
+    /// Removes a packet from the end of the queue specified and returns it.  Note
+    /// that this function is protected by a Mutex.
     ///
     /// @param from specifies the end of the queue from which the packet
     /// should be taken.
@@ -268,6 +274,7 @@ public:
     /// @return A pointer to dequeued packet, or an empty pointer
     /// if the queue is empty.
     virtual PacketTypePtr popPacket(const QueueEnd& from = QueueEnd::FRONT) {
+        isc::util::thread::Mutex::Locker lock(mutex_);
         PacketTypePtr packet;
         if (queue_.empty()) {
             return (packet);
@@ -352,6 +359,9 @@ private:
 
     /// @brief Packet queue
     boost::circular_buffer<PacketTypePtr> queue_;
+
+    /// @brief Mutex for protecting queue accesses.
+    isc::util::thread::Mutex mutex_;
 };