diff --git a/ci/licenses_golden/excluded_files b/ci/licenses_golden/excluded_files
index 8aa84c50351f1..41298f9154317 100644
--- a/ci/licenses_golden/excluded_files
+++ b/ci/licenses_golden/excluded_files
@@ -153,6 +153,7 @@
 ../../../flutter/impeller/playground
 ../../../flutter/impeller/renderer/backend/vulkan/blit_command_vk_unittests.cc
 ../../../flutter/impeller/renderer/backend/vulkan/command_encoder_vk_unittests.cc
+../../../flutter/impeller/renderer/backend/vulkan/command_pool_vk_unittests.cc
 ../../../flutter/impeller/renderer/backend/vulkan/context_vk_unittests.cc
 ../../../flutter/impeller/renderer/backend/vulkan/fence_waiter_vk_unittests.cc
 ../../../flutter/impeller/renderer/backend/vulkan/pass_bindings_cache_unittests.cc
diff --git a/impeller/renderer/backend/vulkan/BUILD.gn b/impeller/renderer/backend/vulkan/BUILD.gn
index 90781bdb0ad85..6b49311ed35a2 100644
--- a/impeller/renderer/backend/vulkan/BUILD.gn
+++ b/impeller/renderer/backend/vulkan/BUILD.gn
@@ -10,6 +10,7 @@ impeller_component("vulkan_unittests") {
   sources = [
     "blit_command_vk_unittests.cc",
     "command_encoder_vk_unittests.cc",
+    "command_pool_vk_unittests.cc",
     "context_vk_unittests.cc",
     "fence_waiter_vk_unittests.cc",
     "pass_bindings_cache_unittests.cc",
diff --git a/impeller/renderer/backend/vulkan/blit_command_vk_unittests.cc b/impeller/renderer/backend/vulkan/blit_command_vk_unittests.cc
index e1172651353b7..9b911c66cef17 100644
--- a/impeller/renderer/backend/vulkan/blit_command_vk_unittests.cc
+++ b/impeller/renderer/backend/vulkan/blit_command_vk_unittests.cc
@@ -2,7 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "flutter/testing/testing.h"
+#include "flutter/testing/testing.h"  // IWYU pragma: keep
 #include "impeller/renderer/backend/vulkan/blit_command_vk.h"
 #include "impeller/renderer/backend/vulkan/command_encoder_vk.h"
 #include "impeller/renderer/backend/vulkan/test/mock_vulkan.h"
@@ -12,7 +12,7 @@ namespace testing {
 
 TEST(BlitCommandVkTest, BlitCopyTextureToTextureCommandVK) {
   auto context = MockVulkanContextBuilder().Build();
-  auto pool = CommandPoolVK::GetThreadLocal(context.get());
+  auto pool = context->GetCommandPoolRecycler()->Get();
   auto encoder = std::make_unique<CommandEncoderFactoryVK>(context)->Create();
   BlitCopyTextureToTextureCommandVK cmd;
   cmd.source = context->GetResourceAllocator()->CreateTexture({
diff --git a/impeller/renderer/backend/vulkan/command_encoder_vk.cc b/impeller/renderer/backend/vulkan/command_encoder_vk.cc
index 9b187896a4f6d..8f973bd0fa4d1 100644
--- a/impeller/renderer/backend/vulkan/command_encoder_vk.cc
+++ b/impeller/renderer/backend/vulkan/command_encoder_vk.cc
@@ -5,7 +5,6 @@
 #include "impeller/renderer/backend/vulkan/command_encoder_vk.h"
 
 #include "flutter/fml/closure.h"
-#include "flutter/fml/trace_event.h"
 #include "impeller/renderer/backend/vulkan/context_vk.h"
 #include "impeller/renderer/backend/vulkan/fence_waiter_vk.h"
 #include "impeller/renderer/backend/vulkan/texture_vk.h"
@@ -21,7 +20,7 @@ class TrackedObjectsVK {
     if (!pool) {
       return;
     }
-    auto buffer = pool->CreateGraphicsCommandBuffer();
+    auto buffer = pool->CreateCommandBuffer();
     if (!buffer) {
       return;
     }
@@ -34,7 +33,7 @@ class TrackedObjectsVK {
     if (!buffer_) {
       return;
     }
-    pool_->CollectGraphicsCommandBuffer(std::move(buffer_));
+    pool_->CollectCommandBuffer(std::move(buffer_));
   }
 
   bool IsValid() const { return is_valid_; }
@@ -105,7 +104,11 @@ std::shared_ptr<CommandEncoderVK> CommandEncoderFactoryVK::Create() {
     return nullptr;
   }
   auto& context_vk = ContextVK::Cast(*context);
-  auto tls_pool = CommandPoolVK::GetThreadLocal(&context_vk);
+  auto recycler = context_vk.GetCommandPoolRecycler();
+  if (!recycler) {
+    return nullptr;
+  }
+  auto tls_pool = recycler->Get();
   if (!tls_pool) {
     return nullptr;
   }
diff --git a/impeller/renderer/backend/vulkan/command_encoder_vk_unittests.cc b/impeller/renderer/backend/vulkan/command_encoder_vk_unittests.cc
index 2314796f54415..3f17a6e99c1ac 100644
--- a/impeller/renderer/backend/vulkan/command_encoder_vk_unittests.cc
+++ b/impeller/renderer/backend/vulkan/command_encoder_vk_unittests.cc
@@ -4,10 +4,9 @@
 
 #include <thread>
 
-#include "flutter/fml/synchronization/count_down_latch.h"
-#include "flutter/testing/testing.h"
+#include "flutter/testing/testing.h"  // IWYU pragma: keep
+#include "fml/synchronization/waitable_event.h"
 #include "impeller/renderer/backend/vulkan/command_encoder_vk.h"
-#include "impeller/renderer/backend/vulkan/fence_waiter_vk.h"
 #include "impeller/renderer/backend/vulkan/test/mock_vulkan.h"
 
 namespace impeller {
@@ -26,6 +25,7 @@ TEST(CommandEncoderVKTest, DeleteEncoderAfterThreadDies) {
       encoder = factory.Create();
     });
     thread.join();
+    context->Shutdown();
   }
   auto destroy_pool =
       std::find(called_functions->begin(), called_functions->end(),
@@ -60,7 +60,9 @@ TEST(CommandEncoderVKTest, CleanupAfterSubmit) {
     wait_for_thread_join.Signal();
     wait_for_submit.Wait();
     called_functions = GetMockVulkanFunctions(context->GetDevice());
+    context->Shutdown();
   }
+
   auto destroy_pool =
       std::find(called_functions->begin(), called_functions->end(),
                 "vkDestroyCommandPool");
diff --git a/impeller/renderer/backend/vulkan/command_pool_vk.cc b/impeller/renderer/backend/vulkan/command_pool_vk.cc
index ad657e3dac60d..6b72f1a92361e 100644
--- a/impeller/renderer/backend/vulkan/command_pool_vk.cc
+++ b/impeller/renderer/backend/vulkan/command_pool_vk.cc
@@ -4,165 +4,201 @@
 
 #include "impeller/renderer/backend/vulkan/command_pool_vk.h"
 
-#include <map>
-#include <unordered_map>
-#include <vector>
+#include <memory>
+#include <optional>
+#include <utility>
 
-#include "flutter/fml/thread_local.h"
-#include "impeller/base/thread.h"
-#include "impeller/renderer/backend/vulkan/context_vk.h"
+#include "fml/macros.h"
+#include "fml/thread_local.h"
+#include "fml/trace_event.h"
+#include "impeller/renderer/backend/vulkan/resource_manager_vk.h"
+#include "impeller/renderer/backend/vulkan/vk.h"  // IWYU pragma: keep.
+#include "vulkan/vulkan_structs.hpp"
 
 namespace impeller {
 
-using CommandPoolMap = std::map<uint64_t, std::shared_ptr<CommandPoolVK>>;
-FML_THREAD_LOCAL fml::ThreadLocalUniquePtr<CommandPoolMap> tls_command_pool;
+// Holds the command pool in a background thread, recyling it when not in use.
+class BackgroundCommandPoolVK final {
+ public:
+  BackgroundCommandPoolVK(BackgroundCommandPoolVK&&) = default;
+
+  explicit BackgroundCommandPoolVK(
+      vk::UniqueCommandPool&& pool,
+      std::vector<vk::UniqueCommandBuffer>&& buffers,
+      std::weak_ptr<CommandPoolRecyclerVK> recycler)
+      : pool_(std::move(pool)),
+        buffers_(std::move(buffers)),
+        recycler_(std::move(recycler)) {}
+
+  ~BackgroundCommandPoolVK() {
+    auto const recycler = recycler_.lock();
+
+    // Not only does this prevent recycling when the context is being destroyed,
+    // but it also prevents the destructor from effectively being called twice;
+    // once for the original BackgroundCommandPoolVK() and once for the moved
+    // BackgroundCommandPoolVK().
+    if (!recycler) {
+      return;
+    }
+
+    recycler->Reclaim(std::move(pool_));
+  }
 
-static Mutex g_all_pools_mutex;
-static std::unordered_map<const ContextVK*,
-                          std::vector<std::weak_ptr<CommandPoolVK>>>
-    g_all_pools IPLR_GUARDED_BY(g_all_pools_mutex);
+ private:
+  FML_DISALLOW_COPY_AND_ASSIGN(BackgroundCommandPoolVK);
 
-std::shared_ptr<CommandPoolVK> CommandPoolVK::GetThreadLocal(
-    const ContextVK* context) {
+  vk::UniqueCommandPool pool_;
+
+  // These are retained because the destructor of the C++ UniqueCommandBuffer
+  // wrapper type will attempt to reset the cmd buffer, and doing so may be a
+  // thread safety violation as this may happen on the fence waiter thread.
+  std::vector<vk::UniqueCommandBuffer> buffers_;
+  std::weak_ptr<CommandPoolRecyclerVK> recycler_;
+};
+
+CommandPoolVK::~CommandPoolVK() {
+  auto const context = context_.lock();
   if (!context) {
-    return nullptr;
-  }
-  if (tls_command_pool.get() == nullptr) {
-    tls_command_pool.reset(new CommandPoolMap());
-  }
-  CommandPoolMap& pool_map = *tls_command_pool.get();
-  auto found = pool_map.find(context->GetHash());
-  if (found != pool_map.end() && found->second->IsValid()) {
-    return found->second;
-  }
-  auto pool = std::shared_ptr<CommandPoolVK>(new CommandPoolVK(context));
-  if (!pool->IsValid()) {
-    return nullptr;
+    return;
   }
-  pool_map[context->GetHash()] = pool;
-  {
-    Lock pool_lock(g_all_pools_mutex);
-    g_all_pools[context].push_back(pool);
+  auto const recycler = context->GetCommandPoolRecycler();
+  if (!recycler) {
+    return;
   }
-  return pool;
+
+  UniqueResourceVKT<BackgroundCommandPoolVK> pool(
+      context->GetResourceManager(),
+      BackgroundCommandPoolVK(std::move(pool_), std::move(collected_buffers_),
+                              recycler));
 }
 
-void CommandPoolVK::ClearAllPools(const ContextVK* context) {
-  if (tls_command_pool.get()) {
-    tls_command_pool.get()->erase(context->GetHash());
-  }
-  Lock pool_lock(g_all_pools_mutex);
-  if (auto found = g_all_pools.find(context); found != g_all_pools.end()) {
-    for (auto& weak_pool : found->second) {
-      auto pool = weak_pool.lock();
-      if (!pool) {
-        // The pool has already died because the thread died.
-        continue;
-      }
-      // The pool is reset but its reference in the TLS map remains till the
-      // thread dies.
-      pool->Reset();
-    }
-    g_all_pools.erase(found);
+// TODO(matanlurey): Return a status_or<> instead of {} when we have one.
+vk::UniqueCommandBuffer CommandPoolVK::CreateCommandBuffer() {
+  auto const context = context_.lock();
+  if (!context) {
+    return {};
   }
-}
 
-CommandPoolVK::CommandPoolVK(const ContextVK* context)
-    : owner_id_(std::this_thread::get_id()) {
-  vk::CommandPoolCreateInfo pool_info;
+  auto const device = context->GetDevice();
+  vk::CommandBufferAllocateInfo info;
+  info.setCommandPool(pool_.get());
+  info.setCommandBufferCount(1u);
+  info.setLevel(vk::CommandBufferLevel::ePrimary);
+  auto [result, buffers] = device.allocateCommandBuffersUnique(info);
+  if (result != vk::Result::eSuccess) {
+    return {};
+  }
+  return std::move(buffers[0]);
+}
 
-  pool_info.queueFamilyIndex = context->GetGraphicsQueue()->GetIndex().family;
-  pool_info.flags = vk::CommandPoolCreateFlagBits::eTransient |
-                    vk::CommandPoolCreateFlagBits::eResetCommandBuffer;
-  auto pool = context->GetDevice().createCommandPoolUnique(pool_info);
-  if (pool.result != vk::Result::eSuccess) {
+void CommandPoolVK::CollectCommandBuffer(vk::UniqueCommandBuffer&& buffer) {
+  if (!pool_) {
+    // If the command pool has already been destroyed, just free the buffer.
     return;
   }
-
-  device_holder_ = context->GetDeviceHolder();
-  graphics_pool_ = std::move(pool.value);
-  is_valid_ = true;
+  collected_buffers_.push_back(std::move(buffer));
 }
 
-CommandPoolVK::~CommandPoolVK() = default;
-
-bool CommandPoolVK::IsValid() const {
-  return is_valid_;
-}
+// Associates a resource with a thread and context.
+using CommandPoolMap =
+    std::unordered_map<uint64_t, std::shared_ptr<CommandPoolVK>>;
+FML_THREAD_LOCAL fml::ThreadLocalUniquePtr<CommandPoolMap> resources_;
 
-void CommandPoolVK::Reset() {
-  {
-    Lock lock(buffers_to_collect_mutex_);
-    graphics_pool_.reset();
+// TODO(matanlurey): Return a status_or<> instead of nullptr when we have one.
+std::shared_ptr<CommandPoolVK> CommandPoolRecyclerVK::Get() {
+  auto const strong_context = context_.lock();
+  if (!strong_context) {
+    return nullptr;
+  }
 
-    // When the command pool is destroyed, all of its command buffers are freed.
-    // Handles allocated from that pool are now invalid and must be discarded.
-    for (vk::UniqueCommandBuffer& buffer : buffers_to_collect_) {
-      buffer.release();
-    }
-    buffers_to_collect_.clear();
+  // If there is a resource in used for this thread and context, return it.
+  auto resources = resources_.get();
+  if (!resources) {
+    resources = new CommandPoolMap();
+    resources_.reset(resources);
+  }
+  auto const hash = strong_context->GetHash();
+  auto const it = resources->find(hash);
+  if (it != resources->end()) {
+    return it->second;
   }
 
-  for (vk::UniqueCommandBuffer& buffer : recycled_buffers_) {
-    buffer.release();
+  // Otherwise, create a new resource and return it.
+  auto pool = Create();
+  if (!pool) {
+    return nullptr;
   }
-  recycled_buffers_.clear();
 
-  is_valid_ = false;
+  auto const resource =
+      std::make_shared<CommandPoolVK>(std::move(*pool), context_);
+  resources->emplace(hash, resource);
+  return resource;
 }
 
-vk::UniqueCommandBuffer CommandPoolVK::CreateGraphicsCommandBuffer() {
-  std::shared_ptr<const DeviceHolder> strong_device = device_holder_.lock();
-  if (!strong_device) {
-    return {};
-  }
-  FML_DCHECK(std::this_thread::get_id() == owner_id_);
-  {
-    Lock lock(buffers_to_collect_mutex_);
-    GarbageCollectBuffersIfAble();
+// TODO(matanlurey): Return a status_or<> instead of nullopt when we have one.
+std::optional<vk::UniqueCommandPool> CommandPoolRecyclerVK::Create() {
+  // If we can reuse a command pool, do so.
+  if (auto pool = Reuse()) {
+    return pool;
   }
 
-  if (!recycled_buffers_.empty()) {
-    vk::UniqueCommandBuffer result = std::move(recycled_buffers_.back());
-    recycled_buffers_.pop_back();
-    return result;
+  // Otherwise, create a new one.
+  auto context = context_.lock();
+  if (!context) {
+    return std::nullopt;
   }
+  vk::CommandPoolCreateInfo info;
+  info.setQueueFamilyIndex(context->GetGraphicsQueue()->GetIndex().family);
+  info.setFlags(vk::CommandPoolCreateFlagBits::eTransient);
 
-  vk::CommandBufferAllocateInfo alloc_info;
-  alloc_info.commandPool = graphics_pool_.get();
-  alloc_info.commandBufferCount = 1u;
-  alloc_info.level = vk::CommandBufferLevel::ePrimary;
-  auto [result, buffers] =
-      strong_device->GetDevice().allocateCommandBuffersUnique(alloc_info);
+  auto device = context->GetDevice();
+  auto [result, pool] = device.createCommandPoolUnique(info);
   if (result != vk::Result::eSuccess) {
-    return {};
+    return std::nullopt;
   }
-  return std::move(buffers[0]);
+  return std::move(pool);
 }
 
-void CommandPoolVK::CollectGraphicsCommandBuffer(
-    vk::UniqueCommandBuffer buffer) {
-  Lock lock(buffers_to_collect_mutex_);
-  if (!graphics_pool_) {
-    // If the command pool has already been destroyed, then its command buffers
-    // have been freed and are now invalid.
-    buffer.release();
+std::optional<vk::UniqueCommandPool> CommandPoolRecyclerVK::Reuse() {
+  // If there are no recycled pools, return nullopt.
+  Lock recycled_lock(recycled_mutex_);
+  if (recycled_.empty()) {
+    return std::nullopt;
   }
-  buffers_to_collect_.emplace_back(std::move(buffer));
-  GarbageCollectBuffersIfAble();
+
+  // Otherwise, remove and return a recycled pool.
+  auto pool = std::move(recycled_.back());
+  recycled_.pop_back();
+  return std::move(pool);
 }
 
-void CommandPoolVK::GarbageCollectBuffersIfAble() {
-  if (std::this_thread::get_id() != owner_id_) {
+void CommandPoolRecyclerVK::Reclaim(vk::UniqueCommandPool&& pool) {
+  TRACE_EVENT0("impeller", "ReclaimCommandPool");
+
+  // Reset the pool on a background thread.
+  auto strong_context = context_.lock();
+  if (!strong_context) {
     return;
   }
+  auto device = strong_context->GetDevice();
+  device.resetCommandPool(pool.get());
 
-  for (auto& buffer : buffers_to_collect_) {
-    buffer->reset();
-    recycled_buffers_.emplace_back(std::move(buffer));
-  }
+  // Move the pool to the recycled list.
+  Lock recycled_lock(recycled_mutex_);
+  recycled_.push_back(std::move(pool));
+}
+
+CommandPoolRecyclerVK::~CommandPoolRecyclerVK() {
+  // Ensure all recycled pools are reclaimed before this is destroyed.
+  Dispose();
+}
 
-  buffers_to_collect_.clear();
+void CommandPoolRecyclerVK::Dispose() {
+  auto const resources = resources_.get();
+  if (!resources) {
+    return;
+  }
+  resources->clear();
 }
 
 }  // namespace impeller
diff --git a/impeller/renderer/backend/vulkan/command_pool_vk.h b/impeller/renderer/backend/vulkan/command_pool_vk.h
index 44722dba4b895..de7aa8b78ffcc 100644
--- a/impeller/renderer/backend/vulkan/command_pool_vk.h
+++ b/impeller/renderer/backend/vulkan/command_pool_vk.h
@@ -5,109 +5,130 @@
 #pragma once
 
 #include <memory>
-#include <thread>
-
-#include "flutter/fml/macros.h"
+#include <optional>
+#include <utility>
+#include "fml/macros.h"
 #include "impeller/base/thread.h"
 #include "impeller/renderer/backend/vulkan/context_vk.h"
-#include "impeller/renderer/backend/vulkan/device_holder.h"
+#include "impeller/renderer/backend/vulkan/vk.h"  // IWYU pragma: keep.
 
 namespace impeller {
 
+class ContextVK;
+class CommandPoolRecyclerVK;
+
 //------------------------------------------------------------------------------
-/// @brief      An opaque object that provides |vk::CommandBuffer| objects.
+/// @brief      Manages the lifecycle of a single |vk::CommandPool|.
 ///
-/// A best practice is to create a |CommandPoolVK| for each thread that will
-/// submit commands to the GPU, and to recycle (reuse) |vk::CommandBuffer|s by
-/// resetting them in a background thread.
+/// A |vk::CommandPool| is expensive to create and reset. This class manages
+/// the lifecycle of a single |vk::CommandPool| by returning to the origin
+/// (|CommandPoolRecyclerVK|) when it is destroyed to be reused.
 ///
-/// @see
-/// https://arm-software.github.io/vulkan_best_practice_for_mobile_developers/samples/performance/command_buffer_usage/command_buffer_usage_tutorial.html#resetting-the-command-pool
+/// @warning    This class is not thread-safe.
+///
+/// @see        |CommandPoolRecyclerVK|
 class CommandPoolVK final {
  public:
-  /// @brief      Gets the |CommandPoolVK| for the current thread.
-  ///
-  /// @param[in]  context  The |ContextVK| to use.
-  ///
-  /// If the current thread does not have a |CommandPoolVK|, one will be created
-  /// and returned. If the current thread already has a |CommandPoolVK|, it will
-  /// be returned.
-  ///
-  /// @return     The |CommandPoolVK| for the current thread, or |nullptr| if
-  ///             either the |ContextVK| is invalid, or a pool could not be
-  ///             created for any reason.
-  ///
-  ///             In other words an invalid command pool will never be returned.
-  static std::shared_ptr<CommandPoolVK> GetThreadLocal(
-      const ContextVK* context);
-
-  /// @brief      Clears all |CommandPoolVK|s for the given |ContextVK|.
-  ///
-  /// @param[in]  context  The |ContextVK| to clear.
-  ///
-  /// Every |CommandPoolVK| that was created for every thread that has ever
-  /// called |GetThreadLocal| with the given |ContextVK| will be cleared.
-  ///
-  /// @note       Should only be called when the |ContextVK| is being destroyed.
-  static void ClearAllPools(const ContextVK* context);
-
+  CommandPoolVK(CommandPoolVK&&) = default;
   ~CommandPoolVK();
 
-  /// @brief      Whether or not this |CommandPoolVK| is valid.
+  /// @brief      Creates a resource that manages the life of a command pool.
   ///
-  /// A command pool is no longer when valid once it's been |Reset|.
-  bool IsValid() const;
+  /// @param[in]  pool      The command pool to manage.
+  /// @param[in]  recycler  The context that will be notified on destruction.
+  explicit CommandPoolVK(vk::UniqueCommandPool pool,
+                         std::weak_ptr<ContextVK>& context)
+      : pool_(std::move(pool)), context_(context) {}
 
   /// @brief      Creates and returns a new |vk::CommandBuffer|.
   ///
-  /// An attempt is made to reuse existing buffers (instead of creating new
-  /// ones) by recycling buffers that have been collected by
-  /// |CollectGraphicsCommandBuffer|.
-  ///
   /// @return     Always returns a new |vk::CommandBuffer|, but if for any
   ///             reason a valid command buffer could not be created, it will be
   ///             a `{}` default instance (i.e. while being torn down).
-  vk::UniqueCommandBuffer CreateGraphicsCommandBuffer();
+  vk::UniqueCommandBuffer CreateCommandBuffer();
 
-  /// @brief      Collects the given |vk::CommandBuffer| for recycling.
-  ///
-  /// The given |vk::CommandBuffer| will be recycled (reused) in the future when
-  /// |CreateGraphicsCommandBuffer| is called.
+  /// @brief      Collects the given |vk::CommandBuffer| to be retained.
   ///
   /// @param[in]  buffer  The |vk::CommandBuffer| to collect.
   ///
-  /// @note       This method is a noop if a different thread created the pool.
-  ///
   /// @see        |GarbageCollectBuffersIfAble|
-  void CollectGraphicsCommandBuffer(vk::UniqueCommandBuffer buffer);
+  void CollectCommandBuffer(vk::UniqueCommandBuffer&& buffer);
 
  private:
-  const std::thread::id owner_id_;
-  std::weak_ptr<const DeviceHolder> device_holder_;
-  vk::UniqueCommandPool graphics_pool_;
-  Mutex buffers_to_collect_mutex_;
-  std::vector<vk::UniqueCommandBuffer> buffers_to_collect_
-      IPLR_GUARDED_BY(buffers_to_collect_mutex_);
-  std::vector<vk::UniqueCommandBuffer> recycled_buffers_;
-  bool is_valid_ = false;
-
-  /// @brief      Resets, releasing all |vk::CommandBuffer|s.
+  FML_DISALLOW_COPY_AND_ASSIGN(CommandPoolVK);
+
+  vk::UniqueCommandPool pool_;
+  std::weak_ptr<ContextVK>& context_;
+
+  // Used to retain a reference on these until the pool is reset.
+  std::vector<vk::UniqueCommandBuffer> collected_buffers_;
+};
+
+//------------------------------------------------------------------------------
+/// @brief      Creates and manages the lifecycle of |vk::CommandPool| objects.
+///
+/// A |vk::CommandPool| is expensive to create and reset. This class manages
+/// the lifecycle of |vk::CommandPool| objects by creating and recycling them;
+/// or in other words, a pool for command pools.
+///
+/// A single instance should be created per |ContextVK|.
+///
+/// Every "frame", a single |CommandPoolResourceVk| is made available for each
+/// thread that calls |Get|. After calling |Dispose|, the current thread's pool
+/// is moved to a background thread, reset, and made available for the next time
+/// |Get| is called and needs to create a command pool.
+///
+/// Commands in the command pool are not necessarily done executing when the
+/// pool is recycled, when all references are dropped to the pool, they are
+/// reset and returned to the pool of available pools.
+///
+/// @note       This class is thread-safe.
+///
+/// @see        |vk::CommandPoolResourceVk|
+/// @see        |ContextVK|
+/// @see
+/// https://arm-software.github.io/vulkan_best_practice_for_mobile_developers/samples/performance/command_buffer_usage/command_buffer_usage_tutorial.html
+class CommandPoolRecyclerVK final
+    : public std::enable_shared_from_this<CommandPoolRecyclerVK> {
+ public:
+  ~CommandPoolRecyclerVK();
+
+  /// @brief      Creates a recycler for the given |ContextVK|.
+  ///
+  /// @param[in]  context The context to create the recycler for.
+  explicit CommandPoolRecyclerVK(std::weak_ptr<ContextVK> context)
+      : context_(std::move(context)) {}
+
+  /// @brief      Gets a command pool for the current thread.
+  ///
+  /// @warning    Returns a |nullptr| if a pool could not be created.
+  std::shared_ptr<CommandPoolVK> Get();
+
+  /// @brief      Returns a command pool to be reset on a background thread.
   ///
-  /// @note       "All" includes active and recycled buffers.
-  void Reset();
+  /// @param[in]  pool The pool to recycler.
+  void Reclaim(vk::UniqueCommandPool&& pool);
 
-  explicit CommandPoolVK(const ContextVK* context);
+  /// @brief      Clears all recycled command pools to let them be reclaimed.
+  void Dispose();
 
-  /// @brief      Collects buffers for recycling if able.
+ private:
+  std::weak_ptr<ContextVK> context_;
+
+  Mutex recycled_mutex_;
+  std::vector<vk::UniqueCommandPool> recycled_ IPLR_GUARDED_BY(recycled_mutex_);
+
+  /// @brief      Creates a new |vk::CommandPool|.
   ///
-  /// If any buffers have been returned through |CollectGraphicsCommandBuffer|,
-  /// then they are reset and made available to future calls to
-  /// |CreateGraphicsCommandBuffer|.
+  /// @returns    Returns a |std::nullopt| if a pool could not be created.
+  std::optional<vk::UniqueCommandPool> Create();
+
+  /// @brief      Reuses a recycled |vk::CommandPool|, if available.
   ///
-  /// @note       This method is a noop if a different thread created the pool.
-  void GarbageCollectBuffersIfAble() IPLR_REQUIRES(buffers_to_collect_mutex_);
+  /// @returns    Returns a |std::nullopt| if a pool was not available.
+  std::optional<vk::UniqueCommandPool> Reuse();
 
-  FML_DISALLOW_COPY_AND_ASSIGN(CommandPoolVK);
+  FML_DISALLOW_COPY_AND_ASSIGN(CommandPoolRecyclerVK);
 };
 
 }  // namespace impeller
diff --git a/impeller/renderer/backend/vulkan/command_pool_vk_unittests.cc b/impeller/renderer/backend/vulkan/command_pool_vk_unittests.cc
new file mode 100644
index 0000000000000..66cde90c5cac2
--- /dev/null
+++ b/impeller/renderer/backend/vulkan/command_pool_vk_unittests.cc
@@ -0,0 +1,113 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "flutter/testing/testing.h"  // IWYU pragma: keep.
+#include "fml/synchronization/waitable_event.h"
+#include "impeller/renderer/backend/vulkan/resource_manager_vk.h"
+#include "impeller/renderer/backend/vulkan/test/mock_vulkan.h"
+
+namespace impeller {
+namespace testing {
+
+TEST(CommandPoolRecyclerVKTest, GetsACommandPoolPerThread) {
+  auto const context = MockVulkanContextBuilder().Build();
+
+  // Record the memory location of each pointer to a command pool.
+  int* pool1 = nullptr;
+  int* pool2 = nullptr;
+
+  // Create a command pool in two threads and record the memory location.
+  std::thread thread1([&]() {
+    auto const pool = context->GetCommandPoolRecycler()->Get();
+    pool1 = reinterpret_cast<int*>(pool.get());
+  });
+
+  std::thread thread2([&]() {
+    auto const pool = context->GetCommandPoolRecycler()->Get();
+    pool2 = reinterpret_cast<int*>(pool.get());
+  });
+
+  thread1.join();
+  thread2.join();
+
+  // The two command pools should be different.
+  EXPECT_NE(pool1, pool2);
+
+  context->Shutdown();
+}
+
+TEST(CommandPoolRecyclerVKTest, GetsTheSameCommandPoolOnSameThread) {
+  auto const context = MockVulkanContextBuilder().Build();
+
+  auto const pool1 = context->GetCommandPoolRecycler()->Get();
+  auto const pool2 = context->GetCommandPoolRecycler()->Get();
+
+  // The two command pools should be the same.
+  EXPECT_EQ(pool1.get(), pool2.get());
+
+  context->Shutdown();
+}
+
+namespace {
+
+// Invokes the provided callback when the destructor is called.
+//
+// Can be moved, but not copied.
+class DeathRattle final {
+ public:
+  explicit DeathRattle(std::function<void()> callback)
+      : callback_(std::move(callback)) {}
+
+  DeathRattle(DeathRattle&&) = default;
+  DeathRattle& operator=(DeathRattle&&) = default;
+
+  ~DeathRattle() { callback_(); }
+
+ private:
+  std::function<void()> callback_;
+};
+
+}  // namespace
+
+TEST(CommandPoolRecyclerVKTest, ReclaimMakesCommandPoolAvailable) {
+  auto const context = MockVulkanContextBuilder().Build();
+
+  {
+    // Fetch a pool (which will be created).
+    auto const recycler = context->GetCommandPoolRecycler();
+    auto const pool = recycler->Get();
+
+    // This normally is called at the end of a frame.
+    recycler->Dispose();
+  }
+
+  // Add something to the resource manager and have it notify us when it's
+  // destroyed. That should give us a non-flaky signal that the pool has been
+  // reclaimed as well.
+  auto waiter = fml::AutoResetWaitableEvent();
+  auto rattle = DeathRattle([&waiter]() { waiter.Signal(); });
+  {
+    UniqueResourceVKT<DeathRattle> resource(context->GetResourceManager(),
+                                            std::move(rattle));
+  }
+  waiter.Wait();
+
+  // On another thread explicitly, request a new pool.
+  std::thread thread([&]() {
+    auto const pool = context->GetCommandPoolRecycler()->Get();
+    EXPECT_NE(pool.get(), nullptr);
+  });
+
+  thread.join();
+
+  // Now check that we only ever created one pool.
+  auto const called = GetMockVulkanFunctions(context->GetDevice());
+  EXPECT_EQ(std::count(called->begin(), called->end(), "vkCreateCommandPool"),
+            1u);
+
+  context->Shutdown();
+}
+
+}  // namespace testing
+}  // namespace impeller
diff --git a/impeller/renderer/backend/vulkan/context_vk.cc b/impeller/renderer/backend/vulkan/context_vk.cc
index 88a86fa2c71a2..5453f060f7a44 100644
--- a/impeller/renderer/backend/vulkan/context_vk.cc
+++ b/impeller/renderer/backend/vulkan/context_vk.cc
@@ -13,11 +13,9 @@
 #include <map>
 #include <memory>
 #include <optional>
-#include <set>
 #include <string>
 #include <vector>
 
-#include "flutter/fml/build_config.h"
 #include "flutter/fml/trace_event.h"
 #include "impeller/base/validation.h"
 #include "impeller/renderer/backend/vulkan/allocator_vk.h"
@@ -114,7 +112,9 @@ ContextVK::~ContextVK() {
   if (device_holder_ && device_holder_->device) {
     [[maybe_unused]] auto result = device_holder_->device->waitIdle();
   }
-  CommandPoolVK::ClearAllPools(this);
+  if (command_pool_recycler_) {
+    command_pool_recycler_.get()->Dispose();
+  }
 }
 
 Context::BackendType ContextVK::GetBackendType() const {
@@ -379,7 +379,7 @@ void ContextVK::Setup(Settings settings) {
       std::shared_ptr<FenceWaiterVK>(new FenceWaiterVK(device_holder));
 
   //----------------------------------------------------------------------------
-  /// Create the resource manager.
+  /// Create the resource manager and command pool recycler.
   ///
   auto resource_manager = ResourceManagerVK::Create();
   if (!resource_manager) {
@@ -387,6 +387,13 @@ void ContextVK::Setup(Settings settings) {
     return;
   }
 
+  auto command_pool_recycler =
+      std::make_shared<CommandPoolRecyclerVK>(weak_from_this());
+  if (!command_pool_recycler) {
+    VALIDATION_LOG << "Could not create command pool recycler.";
+    return;
+  }
+
   //----------------------------------------------------------------------------
   /// Fetch the queues.
   ///
@@ -417,6 +424,7 @@ void ContextVK::Setup(Settings settings) {
   device_capabilities_ = std::move(caps);
   fence_waiter_ = std::move(fence_waiter);
   resource_manager_ = std::move(resource_manager);
+  command_pool_recycler_ = std::move(command_pool_recycler);
   device_name_ = std::string(physical_device_properties.deviceName);
   is_valid_ = true;
 
@@ -477,6 +485,14 @@ ContextVK::GetConcurrentWorkerTaskRunner() const {
 }
 
 void ContextVK::Shutdown() {
+  // There are multiple objects, for example |CommandPoolVK|, that in their
+  // destructors make a strong reference to |ContextVK|. Resetting these shared
+  // pointers ensures that cleanup happens in a correct order.
+  //
+  // tl;dr: Without it, we get thread::join failures on shutdown.
+  fence_waiter_.reset();
+  resource_manager_.reset();
+
   raster_message_loop_->Terminate();
 }
 
@@ -504,6 +520,11 @@ std::shared_ptr<ResourceManagerVK> ContextVK::GetResourceManager() const {
   return resource_manager_;
 }
 
+std::shared_ptr<CommandPoolRecyclerVK> ContextVK::GetCommandPoolRecycler()
+    const {
+  return command_pool_recycler_;
+}
+
 std::unique_ptr<CommandEncoderFactoryVK>
 ContextVK::CreateGraphicsCommandEncoderFactory() const {
   return std::make_unique<CommandEncoderFactoryVK>(weak_from_this());
diff --git a/impeller/renderer/backend/vulkan/context_vk.h b/impeller/renderer/backend/vulkan/context_vk.h
index 316cf1cbb2161..144de15b26913 100644
--- a/impeller/renderer/backend/vulkan/context_vk.h
+++ b/impeller/renderer/backend/vulkan/context_vk.h
@@ -12,6 +12,7 @@
 #include "flutter/fml/unique_fd.h"
 #include "impeller/base/backend_cast.h"
 #include "impeller/core/formats.h"
+#include "impeller/renderer/backend/vulkan/command_pool_vk.h"
 #include "impeller/renderer/backend/vulkan/device_holder.h"
 #include "impeller/renderer/backend/vulkan/pipeline_library_vk.h"
 #include "impeller/renderer/backend/vulkan/queue_vk.h"
@@ -26,6 +27,7 @@ bool HasValidationLayers();
 
 class CommandEncoderFactoryVK;
 class CommandEncoderVK;
+class CommandPoolRecyclerVK;
 class DebugReportVK;
 class FenceWaiterVK;
 class ResourceManagerVK;
@@ -140,6 +142,8 @@ class ContextVK final : public Context,
 
   std::shared_ptr<ResourceManagerVK> GetResourceManager() const;
 
+  std::shared_ptr<CommandPoolRecyclerVK> GetCommandPoolRecycler() const;
+
  private:
   struct DeviceHolderImpl : public DeviceHolder {
     // |DeviceHolder|
@@ -164,6 +168,7 @@ class ContextVK final : public Context,
   std::shared_ptr<const Capabilities> device_capabilities_;
   std::shared_ptr<FenceWaiterVK> fence_waiter_;
   std::shared_ptr<ResourceManagerVK> resource_manager_;
+  std::shared_ptr<CommandPoolRecyclerVK> command_pool_recycler_;
   std::string device_name_;
   std::shared_ptr<fml::ConcurrentMessageLoop> raster_message_loop_;
   bool sync_presentation_ = false;
diff --git a/impeller/renderer/backend/vulkan/context_vk_unittests.cc b/impeller/renderer/backend/vulkan/context_vk_unittests.cc
index 11f8b1599bc21..e658d94911865 100644
--- a/impeller/renderer/backend/vulkan/context_vk_unittests.cc
+++ b/impeller/renderer/backend/vulkan/context_vk_unittests.cc
@@ -15,8 +15,7 @@ TEST(ContextVKTest, DeletesCommandPools) {
   std::weak_ptr<CommandPoolVK> weak_pool;
   {
     std::shared_ptr<ContextVK> context = MockVulkanContextBuilder().Build();
-    std::shared_ptr<CommandPoolVK> pool =
-        CommandPoolVK::GetThreadLocal(context.get());
+    auto const pool = context->GetCommandPoolRecycler()->Get();
     weak_pool = pool;
     weak_context = context;
     ASSERT_TRUE(weak_pool.lock());
diff --git a/impeller/renderer/backend/vulkan/resource_manager_vk.cc b/impeller/renderer/backend/vulkan/resource_manager_vk.cc
index 382f7aa715e7c..f981321c7115e 100644
--- a/impeller/renderer/backend/vulkan/resource_manager_vk.cc
+++ b/impeller/renderer/backend/vulkan/resource_manager_vk.cc
@@ -22,6 +22,11 @@ std::shared_ptr<ResourceManagerVK> ResourceManagerVK::Create() {
 ResourceManagerVK::ResourceManagerVK() : waiter_([&]() { Start(); }) {}
 
 ResourceManagerVK::~ResourceManagerVK() {
+  FML_DCHECK(waiter_.get_id() != std::this_thread::get_id())
+      << "The ResourceManager being destructed on its own spawned thread is a "
+      << "sign that ContextVK was not properly destroyed. A usual fix for this "
+      << "is to ensure that ContextVK is shutdown (i.e. context->Shutdown()) "
+         "before the ResourceManager is destroyed (i.e. at the end of a test).";
   Terminate();
   waiter_.join();
 }
diff --git a/impeller/renderer/backend/vulkan/resource_manager_vk_unittests.cc b/impeller/renderer/backend/vulkan/resource_manager_vk_unittests.cc
index 30df654d3dac0..fd5f10d8e0f70 100644
--- a/impeller/renderer/backend/vulkan/resource_manager_vk_unittests.cc
+++ b/impeller/renderer/backend/vulkan/resource_manager_vk_unittests.cc
@@ -6,7 +6,6 @@
 #include <functional>
 #include <memory>
 #include <utility>
-#include "fml/logging.h"
 #include "fml/synchronization/waitable_event.h"
 #include "gtest/gtest.h"
 #include "impeller/renderer/backend/vulkan/resource_manager_vk.h"
@@ -75,5 +74,34 @@ TEST(ResourceManagerVKTest, TerminatesWhenOutOfScope) {
   EXPECT_EQ(manager.lock(), nullptr);
 }
 
+TEST(ResourceManagerVKTest, IsThreadSafe) {
+  // In a typical app, there is a single ResourceManagerVK per app, shared b/w
+  // threads.
+  //
+  // This test ensures that the ResourceManagerVK is thread-safe.
+  std::weak_ptr<ResourceManagerVK> manager;
+
+  {
+    auto const manager = ResourceManagerVK::Create();
+
+    // Spawn two threads, and have them both put resources into the manager.
+    struct MockResource {};
+
+    std::thread thread1([&manager]() {
+      UniqueResourceVKT<MockResource>(manager, MockResource{});
+    });
+
+    std::thread thread2([&manager]() {
+      UniqueResourceVKT<MockResource>(manager, MockResource{});
+    });
+
+    thread1.join();
+    thread2.join();
+  }
+
+  // The thread should have terminated.
+  EXPECT_EQ(manager.lock(), nullptr);
+}
+
 }  // namespace testing
 }  // namespace impeller
diff --git a/impeller/renderer/backend/vulkan/surface_context_vk.cc b/impeller/renderer/backend/vulkan/surface_context_vk.cc
index a1cddaef6e924..2e20d2af5cbe6 100644
--- a/impeller/renderer/backend/vulkan/surface_context_vk.cc
+++ b/impeller/renderer/backend/vulkan/surface_context_vk.cc
@@ -5,6 +5,7 @@
 #include "impeller/renderer/backend/vulkan/surface_context_vk.h"
 
 #include "flutter/fml/trace_event.h"
+#include "impeller/renderer/backend/vulkan/command_pool_vk.h"
 #include "impeller/renderer/backend/vulkan/context_vk.h"
 #include "impeller/renderer/backend/vulkan/swapchain_vk.h"
 
@@ -79,6 +80,7 @@ std::unique_ptr<Surface> SurfaceContextVK::AcquireNextSurface() {
   if (auto allocator = parent_->GetResourceAllocator()) {
     allocator->DidAcquireSurfaceFrame();
   }
+  parent_->GetCommandPoolRecycler()->Dispose();
   return surface;
 }
 
diff --git a/impeller/renderer/backend/vulkan/test/mock_vulkan.cc b/impeller/renderer/backend/vulkan/test/mock_vulkan.cc
index 393916b81023f..1140078386693 100644
--- a/impeller/renderer/backend/vulkan/test/mock_vulkan.cc
+++ b/impeller/renderer/backend/vulkan/test/mock_vulkan.cc
@@ -9,8 +9,7 @@
 #include "fml/macros.h"
 #include "fml/thread_local.h"
 #include "impeller/base/thread_safety.h"
-#include "vulkan/vulkan_core.h"
-#include "vulkan/vulkan_enums.hpp"
+#include "impeller/renderer/backend/vulkan/vk.h"  // IWYU pragma: keep.
 
 namespace impeller {
 namespace testing {
@@ -24,6 +23,8 @@ struct MockCommandBuffer {
   std::shared_ptr<std::vector<std::string>> called_functions_;
 };
 
+struct MockCommandPool {};
+
 class MockDevice final {
  public:
   explicit MockDevice() : called_functions_(new std::vector<std::string>()) {}
@@ -36,6 +37,25 @@ class MockDevice final {
     return result;
   }
 
+  MockCommandPool* NewCommandPool() {
+    auto pool = std::make_unique<MockCommandPool>();
+    MockCommandPool* result = pool.get();
+    Lock lock(commmand_pools_mutex_);
+    command_pools_.emplace_back(std::move(pool));
+    return result;
+  }
+
+  void DeleteCommandPool(MockCommandPool* pool) {
+    Lock lock(commmand_pools_mutex_);
+    auto it = std::find_if(command_pools_.begin(), command_pools_.end(),
+                           [pool](const std::unique_ptr<MockCommandPool>& p) {
+                             return p.get() == pool;
+                           });
+    if (it != command_pools_.end()) {
+      command_pools_.erase(it);
+    }
+  }
+
   const std::shared_ptr<std::vector<std::string>>& GetCalledFunctions() {
     return called_functions_;
   }
@@ -55,6 +75,10 @@ class MockDevice final {
   Mutex command_buffers_mutex_;
   std::vector<std::unique_ptr<MockCommandBuffer>> command_buffers_
       IPLR_GUARDED_BY(command_buffers_mutex_);
+
+  Mutex commmand_pools_mutex_;
+  std::vector<std::unique_ptr<MockCommandPool>> command_pools_
+      IPLR_GUARDED_BY(commmand_pools_mutex_);
 };
 
 void noop() {}
@@ -200,7 +224,16 @@ VkResult vkCreateCommandPool(VkDevice device,
                              const VkCommandPoolCreateInfo* pCreateInfo,
                              const VkAllocationCallbacks* pAllocator,
                              VkCommandPool* pCommandPool) {
-  *pCommandPool = reinterpret_cast<VkCommandPool>(0xc0de0001);
+  MockDevice* mock_device = reinterpret_cast<MockDevice*>(device);
+  mock_device->AddCalledFunction("vkCreateCommandPool");
+  *pCommandPool =
+      reinterpret_cast<VkCommandPool>(mock_device->NewCommandPool());
+  return VK_SUCCESS;
+}
+
+VkResult vkResetCommandPool(VkDevice device,
+                            VkCommandPool commandPool,
+                            VkCommandPoolResetFlags flags) {
   return VK_SUCCESS;
 }
 
@@ -402,6 +435,8 @@ void vkDestroyCommandPool(VkDevice device,
                           VkCommandPool commandPool,
                           const VkAllocationCallbacks* pAllocator) {
   MockDevice* mock_device = reinterpret_cast<MockDevice*>(device);
+  mock_device->DeleteCommandPool(
+      reinterpret_cast<MockCommandPool*>(commandPool));
   mock_device->AddCalledFunction("vkDestroyCommandPool");
 }
 
@@ -486,6 +521,8 @@ PFN_vkVoidFunction GetMockVulkanProcAddress(VkInstance instance,
     return (PFN_vkVoidFunction)vkCreatePipelineCache;
   } else if (strcmp("vkCreateCommandPool", pName) == 0) {
     return (PFN_vkVoidFunction)vkCreateCommandPool;
+  } else if (strcmp("vkResetCommandPool", pName) == 0) {
+    return (PFN_vkVoidFunction)vkResetCommandPool;
   } else if (strcmp("vkAllocateCommandBuffers", pName) == 0) {
     return (PFN_vkVoidFunction)vkAllocateCommandBuffers;
   } else if (strcmp("vkBeginCommandBuffer", pName) == 0) {
diff --git a/impeller/renderer/backend/vulkan/test/mock_vulkan_unittests.cc b/impeller/renderer/backend/vulkan/test/mock_vulkan_unittests.cc
index 1aa57b0345f21..2555994a8fd03 100644
--- a/impeller/renderer/backend/vulkan/test/mock_vulkan_unittests.cc
+++ b/impeller/renderer/backend/vulkan/test/mock_vulkan_unittests.cc
@@ -4,7 +4,6 @@
 
 #include "flutter/testing/testing.h"  // IWYU pragma: keep
 #include "gtest/gtest.h"
-#include "impeller/renderer/backend/vulkan/command_pool_vk.h"
 #include "impeller/renderer/backend/vulkan/test/mock_vulkan.h"
 #include "vulkan/vulkan_enums.hpp"
 
@@ -19,12 +18,12 @@ TEST(MockVulkanContextTest, IsThreadSafe) {
 
   // Spawn two threads, and have them create a CommandPoolVK each.
   std::thread thread1([&context]() {
-    auto const pool = CommandPoolVK::GetThreadLocal(context.get());
+    auto const pool = context->GetCommandPoolRecycler()->Get();
     EXPECT_TRUE(pool);
   });
 
   std::thread thread2([&context]() {
-    auto const pool = CommandPoolVK::GetThreadLocal(context.get());
+    auto const pool = context->GetCommandPoolRecycler()->Get();
     EXPECT_TRUE(pool);
   });