[AWIBOF-5236] Reserve vector memory before latency logging

- Adjust dynamic rebalancing to latency token for memory optimization
- Reserve enough vector memory before latency logging so that memory allocation(in vector push_back) will not occur during logging

Signed-off-by: isaac.baek <isaac.baek@samsung.com>

Author: isaac-baek

src/collection/SwitchGear.cpp

@@ -68,7 +68,7 @@ collection::SwitchGear::_CheckDeadline(lib::Data* data)
         lat_data->start_deadline--;
         if (0 >= lat_data->start_deadline)
         {
-            lat_data->start_token = lib::LAT_TOKEN_SIZE;
+            lat_data->start_token = lat_data->start_token_size;
             lat_data->start_state = lib::TimeLogState::RUN;
         }
     }
@@ -78,7 +78,6 @@ collection::SwitchGear::_CheckDeadline(lib::Data* data)
         if (-9 > lat_data->start_deadline)
         {
             lat_data->start_state = lib::TimeLogState::STOP;
-            lat_data->access = false;
         }
     }
 
@@ -87,7 +86,7 @@ collection::SwitchGear::_CheckDeadline(lib::Data* data)
         lat_data->end_deadline--;
         if (0 >= lat_data->end_deadline)
         {
-            lat_data->end_token = lib::LAT_TOKEN_SIZE;
+            lat_data->end_token = lat_data->end_token_size;
             lat_data->end_state = lib::TimeLogState::RUN;
         }
     }
@@ -97,7 +96,6 @@ collection::SwitchGear::_CheckDeadline(lib::Data* data)
         if (-9 > lat_data->end_deadline)
         {
             lat_data->end_state = lib::TimeLogState::STOP;
-            lat_data->access = false;
         }
     }
 }

src/lib/Data.h

@@ -37,7 +37,6 @@ namespace lib
 const uint32_t IDLE_THRESHOLD {5};
 const uint32_t MAX_PACKET_CNT_SIZE {10};
 const uint32_t TIMELAG_SIZE {100};
-const int32_t LAT_TOKEN_SIZE {1000};
 
 struct Data
 {
@@ -91,11 +90,13 @@ struct LatencyData : public Data
     std::vector<struct TimeLog> start_v;
     TimeLogState start_state {TimeLogState::IDLE};
     int32_t start_token {0};
+    int32_t start_token_size {static_cast<int32_t>(TIMELAG_SIZE)};
     int32_t start_deadline {-45};
 
     std::vector<struct TimeLog> end_v;
     TimeLogState end_state {TimeLogState::IDLE};
     int32_t end_token {0};
+    int32_t end_token_size {static_cast<int32_t>(TIMELAG_SIZE)};
     int32_t end_deadline {-45};
 };
 

src/process/TimingDistributor.cpp

@@ -40,7 +40,7 @@ process::TimingDistributor::SetTiming(
             uint32_t index_size = node_meta_getter->IndexSize(nid);
             uint32_t filter_size = node_meta_getter->FilterSize(nid);
 
-            if (node_meta_getter->ProcessorType(nid) == air::ProcessorType::LATENCY)
+            if (air::ProcessorType::LATENCY == node_meta_getter->ProcessorType(nid))
             {
                 for (uint32_t hash_index = 0; hash_index < index_size; hash_index++)
                 {
@@ -53,6 +53,8 @@ process::TimingDistributor::SetTiming(
                         lib::LatencyData* from = static_cast<lib::LatencyData*>(
                             kv.second->node[nid]->GetUserDataByHashIndex(
                                 hash_index, filter_index));
+                        lib::AccLatencyData* acc = node_manager->GetAccLatData(
+                            nid, hash_index, filter_index);
 
                         uint64_t key {0};
                         int32_t value {0};
@@ -66,23 +68,72 @@ process::TimingDistributor::SetTiming(
                             timing_map.insert({key, value});
                         }
 
-                        _ResetTiming(from, to, timing_map[key]);
+                        _ResetTiming(from, to, timing_map[key], acc->sample_count);
                     }
                 }
             }
         }
     }
+
+    for (uint32_t nid = 0; nid < MAX_NID_SIZE; nid++)
+    {
+        if (node_meta_getter->ProcessorType(nid) == air::ProcessorType::LATENCY)
+        {
+            uint32_t index_size = node_meta_getter->IndexSize(nid);
+            for (uint32_t hash_index = 0; hash_index < index_size; hash_index++)
+            {
+                uint32_t filter_size = node_meta_getter->FilterSize(nid);
+                for (uint32_t filter_index = 0; filter_index < filter_size - 1;
+                     filter_index++)
+                {
+                    lib::AccLatencyData* acc =
+                        node_manager->GetAccLatData(nid, hash_index, filter_index);
+                    acc->sample_count = 0;
+                }
+            }
+        }
+    }
 }
 
 void
-process::TimingDistributor::_ResetTiming(
-    lib::LatencyData* curr_data, lib::LatencyData* next_data, int32_t time_value)
+process::TimingDistributor::_ResetTiming(lib::LatencyData* curr_data,
+    lib::LatencyData* next_data, int32_t time_value, uint32_t sampled_count)
 {
+    int32_t token {curr_data->start_token_size};
+    uint32_t minimum_sampled_count {lib::TIMELAG_SIZE / 4};         // 25%
+    uint32_t enough_sampled_count {(lib::TIMELAG_SIZE * 97) / 100}; // 97%
+    if (minimum_sampled_count > sampled_count)
+    {
+        if (MAX_TOKEN > token)
+        {
+            token += MIN_TOKEN;
+        }
+    }
+    else if (enough_sampled_count < sampled_count)
+    {
+        if (MIN_TOKEN < token)
+        {
+            token -= MIN_TOKEN;
+        }
+    }
+
     curr_data->start_deadline = time_value;
     curr_data->start_v.clear();
+    if (curr_data->access)
+    {
+        curr_data->start_v.reserve(token + 4);
+    }
+    curr_data->start_token_size = token;
     curr_data->start_state = lib::TimeLogState::IDLE;
+    curr_data->access = false;
 
     next_data->end_deadline = time_value;
     next_data->end_v.clear();
+    if (next_data->access)
+    {
+        next_data->end_v.reserve(token + 4);
+    }
+    next_data->end_token_size = token;
     next_data->end_state = lib::TimeLogState::IDLE;
+    next_data->access = false;
 }

src/process/TimingDistributor.h

@@ -47,12 +47,14 @@ class TimingDistributor
 
 private:
     void _ResetTiming(lib::LatencyData* curr_data, lib::LatencyData* next_data,
-        int32_t time_value);
+        int32_t time_value, uint32_t sampled_count);
     std::random_device rand_device;
     std::default_random_engine rand_engine {rand_device()};
     std::uniform_int_distribution<int32_t> dist {1, 10};
     const uint32_t MAX_NID_SIZE {
         cfg::GetSentenceCount(config::ParagraphType::NODE)};
+    const int32_t MIN_TOKEN {static_cast<int32_t>(lib::TIMELAG_SIZE)};
+    const int32_t MAX_TOKEN {10 * static_cast<int32_t>(lib::TIMELAG_SIZE)};
 };
 
 } // namespace process

src/process/processor/LatencyProcessor.cpp

@@ -157,7 +157,6 @@ process::LatencyProcessor::_InitData(lib::Data* air_data, lib::AccData* acc_data
     acc_lat_data->median = 0;
     acc_lat_data->lower_quartile = 0;
     acc_lat_data->upper_quartile = 0;
-    acc_lat_data->sample_count = 0;
     for (uint32_t i {0}; i < lib::TIMELAG_SIZE; i++)
     {
         acc_lat_data->timelag[i] = 0;