Fix race cond. Move get_awaiting_requests method to base class (openvinotoolkit#2174)

Fix race cond. Move get_awaiting_requests method to base class.
Fix Coverity issues

CVS-165981

Co-authored-by: Vladimir Zlobin <vladimir.zlobin@intel.com>

Author: olpipi

src/cpp/src/continuous_batching/pipeline_impl.cpp

@@ -398,7 +398,8 @@ ContinuousBatchingPipeline::ContinuousBatchingImpl::generate(const std::vector<o
         OPENVINO_ASSERT(1 == input_ids[request_id].get_shape().at(0), "Use multiple tensors to pass a batch.");
         generations.push_back(add_request(request_id, input_ids[request_id], sampling_params[request_id]));
     }
-    auto all_requests = m_awaiting_requests; // we need to store all requests to get results from them once generation has finished
+
+    auto all_requests = get_awaiting_requests(); // we need to store all requests to get results from them once generation has finished
 
     GenerationHandle& generation = generations.at(0);
 
@@ -720,4 +721,9 @@ void ContinuousBatchingPipeline::ContinuousBatchingImpl::_fill_prompt_log_probs(
         }
     }
 }
+
+std::vector<SequenceGroup::Ptr> ContinuousBatchingPipeline::ContinuousBatchingImpl::get_awaiting_requests() {
+    std::lock_guard<std::mutex> lock{m_awaiting_requests_mutex};
+    return m_awaiting_requests;
 }
+} // namespace ov::genai

src/cpp/src/continuous_batching/pipeline_impl.hpp

@@ -139,5 +139,7 @@ class ContinuousBatchingPipeline::ContinuousBatchingImpl : public ContinuousBatc
      * Updates LoRA adapters for current generation call
      */
     void set_adapters(const std::optional<AdapterConfig>& adapters);
+
+    std::vector<SequenceGroup::Ptr> get_awaiting_requests();
 };
 } // namespace ov::genai

src/cpp/src/prompt_lookup/continuous_batching_for_prompt_lookup.cpp

@@ -87,12 +87,7 @@ bool ContinuousBatchingPipeline::ContinuousBatchingForPromptLookupImpl::is_reque
     return m_requests.empty();
 }
 
-std::vector<SequenceGroup::Ptr> ContinuousBatchingPipeline::ContinuousBatchingForPromptLookupImpl::get_awaiting_requests() {
-    return m_awaiting_requests;
-}
-
 size_t ContinuousBatchingPipeline::ContinuousBatchingForPromptLookupImpl::get_processed_tokens_per_iteration() {
     return m_batch_size;
 }
-
-}
+}

src/cpp/src/prompt_lookup/continuous_batching_for_prompt_lookup.hpp

@@ -35,12 +35,11 @@ class ContinuousBatchingPipeline::ContinuousBatchingForPromptLookupImpl : public
     std::map<uint64_t, SequenceLen> get_generated_request_len();
 
     bool is_requests_empty();
-    std::vector<SequenceGroup::Ptr> get_awaiting_requests();
 
     size_t get_processed_tokens_per_iteration();
 
     using ContinuousBatchingPipeline::ContinuousBatchingImpl::drop_requests;
 protected:
     TokenIds generate_candidates(const TokenIds& input_ids, size_t num_pred_tokens, size_t max_ngram_size);
 };
-}
+}

src/cpp/src/speculative_decoding/continuous_batching_for_speculative_decoding_impl.cpp

@@ -286,10 +286,6 @@ bool ContinuousBatchingPipeline::ContinuousBatchingForSpeculativeDecodingImpl::i
     return m_requests.empty();
 }
 
-std::vector<SequenceGroup::Ptr> ContinuousBatchingPipeline::ContinuousBatchingForSpeculativeDecodingImpl::get_awaiting_requests() {
-    return m_awaiting_requests;
-}
-
 size_t ContinuousBatchingPipeline::ContinuousBatchingForSpeculativeDecodingImpl::get_processed_tokens_per_iteration() {
     return m_batch_size;
 }
@@ -339,4 +335,4 @@ void ContinuousBatchingPipeline::ContinuousBatchingForSpeculativeDecodingImpl::m
         }
     }
 }
-}
+}

src/cpp/src/speculative_decoding/continuous_batching_for_speculative_decoding_impl.hpp

@@ -28,7 +28,6 @@ class ContinuousBatchingPipeline::ContinuousBatchingForSpeculativeDecodingImpl :
     GeneratedRequests get_generated_requests();
     UpdateRequestResult update_request(uint64_t request_id, const GeneratedSequences& candidates, bool is_update_logit_processor);
     bool is_requests_empty();
-    std::vector<SequenceGroup::Ptr> get_awaiting_requests();
 
     size_t get_processed_tokens_per_iteration();
 
@@ -38,4 +37,4 @@ class ContinuousBatchingPipeline::ContinuousBatchingForSpeculativeDecodingImpl :
     void finish_request(SequenceGroup::Ptr request);
     void _pull_awaiting_requests() override {};
 };
-}
+}