[GitHub] [incubator-pegasus] Smityz commented on a change in pull request #631: feat(hotkey): capture data part3 - declare fine collector

[GitBox <gi...@apache.org>]

Smityz commented on a change in pull request #631:
URL: https://github.com/apache/incubator-pegasus/pull/631#discussion_r513903634



##########
File path: src/server/hotkey_collector.cpp
##########
@@ -251,5 +261,107 @@ void hotkey_coarse_data_collector::analyse_data(detect_hotkey_result &result)
     }
 }
 
+hotkey_fine_data_collector::hotkey_fine_data_collector(
+    replica_base *base,
+    dsn::replication::hotkey_type::type hotkey_type,
+    int target_bucket_index,
+    int max_queue_size)
+    : internal_collector_base(base),
+      _hotkey_type(hotkey_type),
+      _max_queue_size(max_queue_size),
+      _target_bucket_index(target_bucket_index)
+{
+    // Distinguish between single-threaded and multi-threaded environments
+    if (_hotkey_type == dsn::replication::hotkey_type::READ) {
+
+        auto threads = dsn::get_threadpool_threads_info(THREAD_POOL_LOCAL_APP);
+        int queue_num = threads.size();
+
+        _string_capture_queue_vec.reserve(queue_num);
+        for (int i = 0; i < queue_num; i++) {
+            _thread_queue_map.insert(std::make_pair(threads[i]->native_tid(), i));
+
+            // Create a vector of the ReaderWriterQueue whose size = _max_queue_size
+            _string_capture_queue_vec.emplace_back(_max_queue_size);
+        }
+
+    } else { // WRITE
+        _string_capture_queue_vec.emplace_back(_max_queue_size);
+    }
+}
+
+void hotkey_fine_data_collector::capture_data(const dsn::blob &hash_key, uint64_t weight)
+{
+    if (get_bucket_id(hash_key) != _target_bucket_index) {
+        return;
+    }
+    _string_capture_queue_vec[get_queue_index()].try_emplace(std::make_pair(hash_key, weight));
+}
+
+struct blob_hash
+{
+    std::size_t operator()(const dsn::blob &str) const
+    {
+        dsn::string_view cp(str);
+        return boost::hash_range(cp.begin(), cp.end());
+    }
+};
+struct blob_equal
+{
+    std::size_t operator()(const dsn::blob &lhs, const dsn::blob &rhs) const
+    {
+        return dsn::string_view(lhs) == dsn::string_view(rhs);
+    }
+};
+
+int hotkey_fine_data_collector::get_queue_index()
+{
+    if (_hotkey_type == dsn::replication::hotkey_type::WRITE) {
+        return 0;
+    }
+
+    int thread_native_tid = dsn::utils::get_current_tid();
+    auto result = _thread_queue_map.find(thread_native_tid);
+    dassert(result != _thread_queue_map.end(), "Can't find the queue corresponding to the thread");
+    return result->second;
+}
+
+void hotkey_fine_data_collector::analyse_data(detect_hotkey_result &result)
+{
+    std::unordered_map<dsn::blob, uint64_t, blob_hash, blob_equal> hash_key_accessed_cnt;
+    for (auto &rw_queue : _string_capture_queue_vec) {
+        std::pair<dsn::blob, int> hash_key_pair;
+        // prevent endless loop
+        int collect_sum = 0;
+        while (rw_queue.try_dequeue(hash_key_pair) && ++collect_sum <= _max_queue_size) {
+            hash_key_accessed_cnt[hash_key_pair.first] += hash_key_pair.second;
+        }
+    }

Review comment:
       https://github.com/apache/incubator-pegasus/pull/603#issue-487947949
   The data will be put into different queues according to the thread, and then they will be timely export to a hashmap to calculate the frequency of occurrence




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org