[ET-VK] Fix pipeline indexing bug in batch create_pipelines (#17192)

Refactored Context::flush() by extracting two new public functions:
- wait_for_queue(): blocks until the GPU queue is idle
- clear_resources(): clears command/descriptor pools and cleanup lists

This allows callers to use these operations independently. ComputeGraph's
destructor now uses these functions directly and wraps them in try/catch
blocks to ensure it never throws exceptions, following C++ best practices
for destructors.

Authored with Claude.

Differential Revision: [D92171362](https://our.internmc.facebook.com/intern/diff/D92171362/)

[ghstack-poisoned]

Author: pytorchbot

backends/vulkan/runtime/vk_api/Pipeline.cpp

@@ -459,7 +459,22 @@ void ComputePipelineCache::create_pipelines(
     const std::unordered_set<Key, Hasher>& descriptors) {
   std::lock_guard<std::mutex> lock(cache_mutex_);
 
-  const auto num_pipelines = descriptors.size();
+  // Filter out descriptors already in cache to avoid creating duplicate
+  // pipelines and to ensure correct indexing between created pipelines and
+  // cache insertion.
+  std::vector<Key> keys_to_create;
+  keys_to_create.reserve(descriptors.size());
+  for (const auto& key : descriptors) {
+    if (cache_.find(key) == cache_.cend()) {
+      keys_to_create.push_back(key);
+    }
+  }
+
+  if (keys_to_create.empty()) {
+    return;
+  }
+
+  const auto num_pipelines = keys_to_create.size();
   std::vector<VkPipeline> pipelines(num_pipelines);
 
   std::vector<std::vector<VkSpecializationMapEntry>> map_entries;
@@ -474,7 +489,7 @@ void ComputePipelineCache::create_pipelines(
   std::vector<VkComputePipelineCreateInfo> create_infos;
   create_infos.reserve(num_pipelines);
 
-  for (auto& key : descriptors) {
+  for (const auto& key : keys_to_create) {
     map_entries.push_back(key.specialization_constants.generate_map_entries());
 
     specialization_infos.push_back(VkSpecializationInfo{
@@ -513,14 +528,10 @@ void ComputePipelineCache::create_pipelines(
       nullptr,
       pipelines.data()));
 
-  uint32_t i = 0;
-  for (auto& key : descriptors) {
-    auto it = cache_.find(key);
-    if (it != cache_.cend()) {
-      continue;
-    }
-    cache_.insert({key, ComputePipelineCache::Value(device_, pipelines[i])});
-    ++i;
+  for (size_t i = 0; i < keys_to_create.size(); ++i) {
+    cache_.insert(
+        {keys_to_create[i],
+         ComputePipelineCache::Value(device_, pipelines[i])});
   }
 }