Add metrics for RWLockFIFO. [#CLICKHOUSE-3246]

dbms/src/Common/CurrentMetrics.cpp

@@ -35,6 +35,10 @@
     M(StorageBufferBytes) \
     M(DictCacheRequests) \
     M(Revision) \
+    M(RWLockWaitingReaders) \
+    M(RWLockWaitingWriters) \
+    M(RWLockActiveReaders) \
+    M(RWLockActiveWriters)
 
 
 namespace CurrentMetrics

dbms/src/Common/ProfileEvents.cpp

@@ -127,6 +127,10 @@
     M(DataAfterMergeDiffersFromReplica) \
     M(PolygonsAddedToPool) \
     M(PolygonsInPoolAllocatedBytes) \
+    M(RWLockAcquiredReadLocks) \
+    M(RWLockAcquiredWriteLocks) \
+    M(RWLockReadersWaitMilliseconds) \
+    M(RWLockWritersWaitMilliseconds)
 
 namespace ProfileEvents
 {

dbms/src/Common/RWLockFIFO.cpp

 #include "RWLockFIFO.h"
+#include <Common/Stopwatch.h>
 #include <Common/Exception.h>
-#include <iostream>
 #include <Poco/Ext/ThreadNumber.h>
+#include <Common/CurrentMetrics.h>
+#include <Common/ProfileEvents.h>
+
+
+namespace ProfileEvents
+{
+    extern const Event RWLockAcquiredReadLocks;
+    extern const Event RWLockAcquiredWriteLocks;
+    extern const Event RWLockReadersWaitMilliseconds;
+    extern const Event RWLockWritersWaitMilliseconds;
+}
+
+
+namespace CurrentMetrics
+{
+    extern const Metric RWLockWaitingReaders;
+    extern const Metric RWLockWaitingWriters;
+    extern const Metric RWLockActiveReaders;
+    extern const Metric RWLockActiveWriters;
+}
 
 
 namespace DB
@@ -13,12 +33,42 @@ namespace ErrorCodes
 }
 
 
-RWLockFIFO::LockHandler RWLockFIFO::getLock(RWLockFIFO::Type type, RWLockFIFO::Client client)
+class RWLockFIFO::LockHandlerImpl
 {
+    RWLockFIFOPtr parent;
     GroupsContainer::iterator it_group;
     ClientsContainer::iterator it_client;
+    ThreadToHandler::iterator it_handler;
+    CurrentMetrics::Increment active_client_increment;
+
+    LockHandlerImpl(RWLockFIFOPtr && parent, GroupsContainer::iterator it_group, ClientsContainer::iterator it_client);
+
+public:
+
+    LockHandlerImpl(const LockHandlerImpl & other) = delete;
+
+    ~LockHandlerImpl();
+
+    friend class RWLockFIFO;
+};
+
+
+RWLockFIFO::LockHandler RWLockFIFO::getLock(RWLockFIFO::Type type, RWLockFIFO::Client client)
+{
+    Stopwatch watch(CLOCK_MONOTONIC_COARSE);
+    CurrentMetrics::Increment waiting_client_increment((type == Read) ? CurrentMetrics::RWLockWaitingReaders
+                                                                      : CurrentMetrics::RWLockWaitingWriters);
+    auto finalize_metrics = [type, &watch] ()
+    {
+        ProfileEvents::increment((type == Read) ? ProfileEvents::RWLockAcquiredReadLocks
+                                                : ProfileEvents::RWLockAcquiredWriteLocks);
+        ProfileEvents::increment((type == Read) ? ProfileEvents::RWLockReadersWaitMilliseconds
+                                                : ProfileEvents::RWLockWritersWaitMilliseconds, watch.elapsedMilliseconds());
+    };
 
     auto this_thread_id = std::this_thread::get_id();
+    GroupsContainer::iterator it_group;
+    ClientsContainer::iterator it_client;
 
     std::unique_lock<std::mutex> lock(mutex);
 
@@ -79,6 +129,7 @@ RWLockFIFO::LockHandler RWLockFIFO::getLock(RWLockFIFO::Type type, RWLockFIFO::C
     if (it_group == queue.begin())
     {
         it_client->start_time = it_client->enqueue_time;
+        finalize_metrics();
         return res;
     }
 
@@ -86,6 +137,7 @@ RWLockFIFO::LockHandler RWLockFIFO::getLock(RWLockFIFO::Type type, RWLockFIFO::C
     it_group->cv.wait(lock, [&] () { return it_group == queue.begin(); } );
 
     it_client->start_time = time(nullptr);
+    finalize_metrics();
     return res;
 }
 
@@ -133,6 +185,9 @@ RWLockFIFO::LockHandlerImpl::~LockHandlerImpl()
 
 RWLockFIFO::LockHandlerImpl::LockHandlerImpl(RWLockFIFOPtr && parent, RWLockFIFO::GroupsContainer::iterator it_group,
                                              RWLockFIFO::ClientsContainer::iterator it_client)
-    : parent{std::move(parent)}, it_group{it_group}, it_client{it_client} {}
+    : parent{std::move(parent)}, it_group{it_group}, it_client{it_client},
+      active_client_increment{(it_client->type == RWLockFIFO::Read) ? CurrentMetrics::RWLockActiveReaders
+                                                                    : CurrentMetrics::RWLockActiveWriters}
+{}
 
 }

dbms/src/Common/RWLockFIFO.h

@@ -40,6 +40,7 @@ public:
 
         bool isStarted() { return start_time != 0; }
 
+        /// TODO: delete extra info below if there is no need fot it already.
         std::string info;
         int thread_number = 0;
         std::time_t enqueue_time = 0;
@@ -50,6 +51,7 @@ public:
 
     /// Just use LockHandler::reset() to release the lock
     class LockHandlerImpl;
+    friend class LockHandlerImpl;
     using LockHandler = std::shared_ptr<LockHandlerImpl>;
 
 
@@ -87,28 +89,6 @@ private:
         explicit Group(Type type) : type{type} {}
     };
 
-public:
-
-    class LockHandlerImpl
-    {
-        RWLockFIFOPtr parent;
-        GroupsContainer::iterator it_group;
-        ClientsContainer::iterator it_client;
-        ThreadToHandler::iterator it_handler;
-
-        LockHandlerImpl(RWLockFIFOPtr && parent, GroupsContainer::iterator it_group, ClientsContainer::iterator it_client);
-
-    public:
-
-        LockHandlerImpl(const LockHandlerImpl & other) = delete;
-
-        ~LockHandlerImpl();
-
-        friend class RWLockFIFO;
-    };
-
-private:
-
     mutable std::mutex mutex;
     GroupsContainer queue;
     ThreadToHandler thread_to_handler;

dbms/src/Common/Stopwatch.h

@@ -20,11 +20,12 @@ public:
       */
     Stopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { restart(); }
 
-    void start()                    { setStart(); is_running = true; }
-    void stop()                     { updateElapsed(); is_running = false; }
-    void restart()                  { elapsed_ns = 0; start(); }
-    UInt64 elapsed() const          { updateElapsed(); return elapsed_ns; }
-    double elapsedSeconds() const   { updateElapsed(); return static_cast<double>(elapsed_ns) / 1000000000ULL; }
+    void start()                        { setStart(); is_running = true; }
+    void stop()                         { updateElapsed(); is_running = false; }
+    void restart()                      { elapsed_ns = 0; start(); }
+    UInt64 elapsed() const              { updateElapsed(); return elapsed_ns; }
+    UInt64 elapsedMilliseconds() const  { updateElapsed(); return elapsed_ns / 1000000UL; }
+    double elapsedSeconds() const       { updateElapsed(); return static_cast<double>(elapsed_ns) / 1000000000ULL; }
 
 private:
     mutable UInt64 start_ns;