diff --git a/include/perfetto/ext/base/threading/future.h b/include/perfetto/ext/base/threading/future.h
index c509e9ba78..6ec6682e30 100644
--- a/include/perfetto/ext/base/threading/future.h
+++ b/include/perfetto/ext/base/threading/future.h
@@ -58,6 +58,10 @@ Future<T> MakeFuture(Args... args) {
 //    cancelled. Note, that the implementation of the source future still needs
 //    to propogate cancellation across thread/socket/pipe boundary.
 //
+// Note: Futures *must* be polled on the same thread on which they were created.
+// The |SpawnResultFuture| can be used to move the results of Futures between
+// threads in a safe manner.
+//
 // Implementation note:
 // An important point to note is that Future<T> is a final class. Implementation
 // of Future<T>::Poll happens through an indirection layer by implementing the
diff --git a/include/perfetto/ext/base/threading/spawn.h b/include/perfetto/ext/base/threading/spawn.h
index df2f8ec06f..8e848fbd9c 100644
--- a/include/perfetto/ext/base/threading/spawn.h
+++ b/include/perfetto/ext/base/threading/spawn.h
@@ -57,6 +57,9 @@ class SpawnHandle {
   SpawnHandle(TaskRunner* task_runner, std::function<Future<FVoid>()> fn);
   ~SpawnHandle();
 
+  SpawnHandle(SpawnHandle&&) = default;
+  SpawnHandle& operator=(SpawnHandle&&) = default;
+
  private:
   SpawnHandle(const SpawnHandle&) = delete;
   SpawnHandle& operator=(const SpawnHandle&) = delete;
@@ -65,25 +68,6 @@ class SpawnHandle {
   std::shared_ptr<std::unique_ptr<PolledFuture>> polled_future_;
 };
 
-// Specialization of SpawnHandle used by Futures/Streams which return T.
-//
-// Values of T are returned through a Channel<T> which allows reading these
-// values on a different thread to where the polling happens.
-template <typename T>
-class ResultSpawnHandle {
- public:
-  ResultSpawnHandle(TaskRunner* task_runner,
-                    std::shared_ptr<Channel<T>> channel,
-                    std::function<Future<FVoid>()> fn)
-      : handle_(task_runner, std::move(fn)), channel_(std::move(channel)) {}
-
-  Channel<T>* channel() const { return channel_.get(); }
-
- private:
-  SpawnHandle handle_;
-  std::shared_ptr<Channel<T>> channel_;
-};
-
 // "Spawns" a Future<FVoid> on the given TaskRunner and returns an RAII
 // SpawnHandle which can be used to cancel the spawn.
 //
@@ -103,10 +87,20 @@ PERFETTO_WARN_UNUSED_RESULT inline SpawnHandle SpawnFuture(
 
 // Variant of |SpawnFuture| for a Stream<T> allowing returning items of T.
 //
-// See ResultSpawnHandle for how elements from the stream can be consumed.
+// The Stream<T> returned by this function can be consumed on any thread, not
+// just the thread which ran this function.
+//
+// Dropping the returned stream does not affect the polling of the underlying
+// stream (i.e. the stream returned by |fn|); the polled values will simply be
+// dropped.
+//
+// Dropping the returned SpawnHandle causes the underlying stream to be
+// cancelled and dropped ASAP (this happens on the TaskRunner thread so there
+// can be some delay). The returned channel will return all the values that were
+// produced by the underlying stream before the cancellation.
 template <typename T>
-PERFETTO_WARN_UNUSED_RESULT inline ResultSpawnHandle<T> SpawnResultStream(
-    TaskRunner* task_runner,
+PERFETTO_WARN_UNUSED_RESULT std::pair<SpawnHandle, Stream<T>> SpawnResultStream(
+    TaskRunner* runner,
     std::function<Stream<T>()> fn) {
   class AllVoidCollector : public Collector<FVoid, FVoid> {
    public:
@@ -114,28 +108,38 @@ PERFETTO_WARN_UNUSED_RESULT inline ResultSpawnHandle<T> SpawnResultStream(
     FVoid OnDone() override { return FVoid(); }
   };
   auto channel = std::make_shared<Channel<T>>(4);
-  return ResultSpawnHandle<T>(
-      task_runner, channel, [c = channel, fn = std::move(fn)]() {
-        return fn()
-            .MapFuture([c](T value) {
-              return WriteChannelFuture(c.get(), std::move(value));
-            })
-            .Concat(OnDestroyStream<FVoid>([c]() { c->Close(); }))
-            .Collect(std::unique_ptr<Collector<FVoid, FVoid>>(
-                new AllVoidCollector()));
-      });
+  auto control = std::make_shared<Channel<FVoid>>(1);
+  SpawnHandle handle(runner, [channel, control, fn = std::move(fn)]() {
+    return fn()
+        .MapFuture([channel, control](T value) mutable {
+          if (control->ReadNonBlocking().is_closed) {
+            return base::Future<base::FVoid>(base::FVoid());
+          }
+          return WriteChannelFuture(channel.get(), std::move(value));
+        })
+        .Concat(OnDestroyStream<FVoid>([c = channel]() { c->Close(); }))
+        .template Collect<base::FVoid>(std::make_unique<AllVoidCollector>());
+  });
+  Stream<T> stream = ReadChannelStream(channel.get())
+                         .Concat(OnDestroyStream<T>([channel, control]() {
+                           // Close the control stream and drain an element from
+                           // the channel to unblock it in case it was blocked.
+                           // NOTE: the ordering here is important as we could
+                           // deadlock if it was the other way around!
+                           control->Close();
+                           base::ignore_result(channel->ReadNonBlocking());
+                         }));
+  return std::make_pair(std::move(handle), std::move(stream));
 }
 
-// Variant of |SpawnFuture| for a Future<T> allowing returning items of T.
-//
-// See ResultSpawnHandle for how elements from the future can be consumed.
+// Variant of |SpawnResultStream| but for Future<T>.
 template <typename T>
-PERFETTO_WARN_UNUSED_RESULT inline ResultSpawnHandle<T> SpawnResultFuture(
-    TaskRunner* task_runner,
-    std::function<Future<T>()> fn) {
-  return SpawnResultStream<T>(task_runner, [fn = std::move(fn)]() {
-    return StreamFromFuture(std::move(fn()));
-  });
+PERFETTO_WARN_UNUSED_RESULT inline std::pair<SpawnHandle, Future<T>>
+SpawnResultFuture(TaskRunner* task_runner, std::function<Future<T>()> fn) {
+  auto [handle, stream] = SpawnResultStream<T>(
+      task_runner, [fn = std::move(fn)]() { return StreamFromFuture(fn()); });
+  return std::make_pair(std::move(handle), std::move(stream).Collect(
+                                               ToFutureCheckedCollector<T>()));
 }
 
 }  // namespace base
diff --git a/include/perfetto/ext/base/threading/stream.h b/include/perfetto/ext/base/threading/stream.h
index 4d67eb4b46..d12f0b9af1 100644
--- a/include/perfetto/ext/base/threading/stream.h
+++ b/include/perfetto/ext/base/threading/stream.h
@@ -46,6 +46,10 @@ Stream<typename P::PollT> MakeStream(Args... args) {
 // version of Iterator<T>. Long-running compute/IO operations which return
 // multiple values can be represented with a Stream<T>.
 //
+// Note: Streams *must* be polled on the same thread on which they were
+// created. The |SpawnResultStreams| can be used to move of the results of
+// Streams between threads in a safe manner.
+//
 // Refer to the class documentation for Future<T> as most of the features and
 // implementation of Future<T> also apply to Stream<T>.
 template <typename T>
@@ -129,7 +133,9 @@ Stream<T> StreamOf(T first, Ts... rest) {
 // Creates a Stream<T> which returns the value of |future| before completing.
 template <typename T>
 Stream<T> StreamFromFuture(Future<T> future) {
-  return StreamOf(std::move(future)).MapFuture([](Future<T> value) { return value; });
+  return StreamOf(std::move(future)).MapFuture([](Future<T> value) {
+    return value;
+  });
 }
 
 // Creates a stream which returns no elements but calls |fn| in the destructor
@@ -173,8 +179,7 @@ inline std::unique_ptr<Collector<T, T>> ToFutureCheckedCollector() {
 // containing all the successful results from the stream. If any element is an
 // error, short-circuits the stream with the error.
 template <typename T>
-inline std::unique_ptr<
-    Collector<StatusOr<T>, StatusOr<std::vector<T>>>>
+inline std::unique_ptr<Collector<StatusOr<T>, StatusOr<std::vector<T>>>>
 StatusOrVectorCollector() {
   return std::make_unique<StatusOrVectorCollectorImpl<T>>();
 }
diff --git a/src/base/threading/spawn.cc b/src/base/threading/spawn.cc
index e727f84206..7ef183dd39 100644
--- a/src/base/threading/spawn.cc
+++ b/src/base/threading/spawn.cc
@@ -74,7 +74,7 @@ class PolledFuture {
     }
   }
 
-  void ClearFutureAndWatches(FlatSet<PlatformHandle> interested) {
+  void ClearFutureAndWatches(const FlatSet<PlatformHandle>& interested) {
     future_ = std::nullopt;
     for (PlatformHandle fd : interested) {
       task_runner_->RemoveFileDescriptorWatch(fd);
diff --git a/src/base/threading/spawn_unittest.cc b/src/base/threading/spawn_unittest.cc
index c1a2e8435e..b14397e296 100644
--- a/src/base/threading/spawn_unittest.cc
+++ b/src/base/threading/spawn_unittest.cc
@@ -16,12 +16,15 @@
 
 #include "perfetto/ext/base/threading/spawn.h"
 
+#include <memory>
 #include <optional>
 
+#include "perfetto/base/compiler.h"
 #include "perfetto/ext/base/event_fd.h"
 #include "perfetto/ext/base/thread_task_runner.h"
 #include "perfetto/ext/base/threading/future.h"
 #include "perfetto/ext/base/threading/poll.h"
+#include "perfetto/ext/base/threading/stream.h"
 #include "perfetto/ext/base/threading/util.h"
 #include "perfetto/ext/base/unix_task_runner.h"
 #include "src/base/test/test_task_runner.h"
@@ -46,87 +49,193 @@ class MockStreamPollable : public StreamPollable<T> {
   MOCK_METHOD(StreamPollResult<T>, PollNext, (PollContext*), (override));
 };
 
-TEST(SpawnUnittest, SpawnFuture) {
-  base::TestTaskRunner task_runner;
+class SpawnUnittest : public testing::Test {
+ protected:
+  void Drop(base::SpawnHandle) {}
+  void Drop(base::Stream<int>) {}
+
+  base::TestTaskRunner task_runner_;
+
+  base::FlatSet<base::PlatformHandle> interested_;
+  base::FlatSet<base::PlatformHandle> ready_;
+  PollContext ctx_{&interested_, &ready_};
+
+  base::EventFd fd_;
+  std::unique_ptr<MockFuturePollable<int>> future_pollable_ =
+      std::make_unique<MockFuturePollable<int>>();
+  std::unique_ptr<MockStreamPollable<int>> stream_pollable_ =
+      std::make_unique<MockStreamPollable<int>>();
+};
 
-  base::EventFd fd;
-  auto pollable = std::make_unique<MockFuturePollable<int>>();
-  EXPECT_CALL(*pollable, Poll(_))
-      .WillOnce([&fd](PollContext* ctx) {
-        fd.Clear();
-        ctx->RegisterInterested(fd.fd());
+TEST_F(SpawnUnittest, SpawnFuture) {
+  EXPECT_CALL(*future_pollable_, Poll(_))
+      .WillOnce([this](PollContext* ctx) {
+        fd_.Clear();
+        ctx->RegisterInterested(fd_.fd());
         return PendingPollResult();
       })
       .WillOnce(Return(FuturePollResult<int>(1024)));
-  auto res = SpawnResultFuture<int>(
-      &task_runner,
-      [pollable = std::make_shared<std::unique_ptr<MockFuturePollable<int>>>(
-           std::move(pollable))]() mutable {
-        return base::Future<int>(std::move(*pollable));
+  auto [handle, future] =
+      SpawnResultFuture<int>(&task_runner_, [this]() mutable {
+        return base::Future<int>(std::move(future_pollable_));
       });
+  base::ignore_result(handle);
 
-  task_runner.RunUntilIdle();
-  ASSERT_EQ(res.channel()->ReadNonBlocking().item, std::nullopt);
+  task_runner_.RunUntilIdle();
+  ASSERT_TRUE(future.Poll(&ctx_).IsPending());
 
-  task_runner.RunUntilIdle();
-  ASSERT_EQ(res.channel()->ReadNonBlocking().item, std::nullopt);
+  task_runner_.RunUntilIdle();
+  ASSERT_TRUE(future.Poll(&ctx_).IsPending());
 
-  fd.Notify();
-  task_runner.RunUntilIdle();
+  fd_.Notify();
+  task_runner_.RunUntilIdle();
 
-  auto read = res.channel()->ReadNonBlocking();
-  ASSERT_EQ(read.item, 1024);
-  ASSERT_TRUE(read.is_closed);
-
-  read = res.channel()->ReadNonBlocking();
-  ASSERT_TRUE(read.is_closed);
+  ASSERT_EQ(future.Poll(&ctx_).item(), 1024);
 }
 
-TEST(SpawnUnittest, SpawnStream) {
-  base::TestTaskRunner task_runner;
-
-  base::EventFd fd;
-  auto pollable = std::make_unique<MockStreamPollable<int>>();
-  EXPECT_CALL(*pollable, PollNext(_))
-      .WillOnce([&fd](PollContext* ctx) {
-        fd.Clear();
-        ctx->RegisterInterested(fd.fd());
+TEST_F(SpawnUnittest, SpawnStream) {
+  EXPECT_CALL(*stream_pollable_, PollNext(_))
+      .WillOnce([this](PollContext* ctx) {
+        fd_.Clear();
+        ctx->RegisterInterested(fd_.fd());
         return PendingPollResult();
       })
       .WillOnce(Return(StreamPollResult<int>(1024)))
-      .WillOnce([&fd](PollContext* ctx) {
-        fd.Clear();
-        ctx->RegisterInterested(fd.fd());
+      .WillOnce([this](PollContext* ctx) {
+        fd_.Clear();
+        ctx->RegisterInterested(fd_.fd());
         return PendingPollResult();
       })
       .WillOnce(Return(StreamPollResult<int>(2048)))
       .WillOnce(Return(DonePollResult()));
-  auto res = SpawnResultStream<int>(
-      &task_runner,
-      [pollable = std::make_shared<std::unique_ptr<MockStreamPollable<int>>>(
-           std::move(pollable))]() mutable {
-        return base::Stream<int>(std::move(*pollable));
+  auto [handle, stream] =
+      SpawnResultStream<int>(&task_runner_, [this]() mutable {
+        return base::Stream<int>(std::move(stream_pollable_));
+      });
+  base::ignore_result(handle);
+
+  task_runner_.RunUntilIdle();
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsPending());
+
+  fd_.Notify();
+  task_runner_.RunUntilIdle();
+
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 1024);
+
+  task_runner_.RunUntilIdle();
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsPending());
+
+  fd_.Notify();
+  task_runner_.RunUntilIdle();
+
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 2048);
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsDone());
+}
+
+TEST_F(SpawnUnittest, SpawnStreamDropStream) {
+  EXPECT_CALL(*stream_pollable_, PollNext(_))
+      .WillOnce([this](PollContext* ctx) {
+        fd_.Clear();
+        ctx->RegisterInterested(fd_.fd());
+        return PendingPollResult();
+      })
+      .WillOnce(Return(StreamPollResult<int>(1)))
+      .WillOnce(Return(StreamPollResult<int>(2)))
+      .WillOnce(Return(StreamPollResult<int>(4)))
+      .WillOnce(Return(StreamPollResult<int>(8)))
+      .WillOnce(Return(StreamPollResult<int>(16)))
+      .WillOnce(Return(StreamPollResult<int>(32)))
+      .WillOnce(Return(StreamPollResult<int>(64)))
+      .WillOnce(Return(StreamPollResult<int>(128)))
+      .WillOnce(Return(StreamPollResult<int>(256)))
+      .WillOnce(Return(StreamPollResult<int>(512)))
+      .WillOnce(Return(DonePollResult()));
+
+  auto [handle, stream] =
+      SpawnResultStream<int>(&task_runner_, [this]() mutable {
+        return base::Stream<int>(std::move(stream_pollable_));
+      });
+  base::ignore_result(handle);
+
+  task_runner_.RunUntilIdle();
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsPending());
+
+  fd_.Notify();
+  task_runner_.RunUntilIdle();
+
+  // We should get the first 4 elements and then nothing more: this corresponds
+  // to the internal channel buffer size being 4.
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 1);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 2);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 4);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 8);
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsPending());
+
+  // Should fill up a bunch more elements.
+  task_runner_.RunUntilIdle();
+
+  // Drop the stream.
+  Drop(std::move(stream));
+
+  // This should complete the stream.
+  task_runner_.RunUntilIdle();
+
+  // Drop the handle and ensure any resulting is completed.
+  Drop(std::move(handle));
+  task_runner_.RunUntilIdle();
+}
+
+TEST_F(SpawnUnittest, SpawnStreamDropHandle) {
+  EXPECT_CALL(*stream_pollable_, PollNext(_))
+      .WillOnce([this](PollContext* ctx) {
+        fd_.Clear();
+        ctx->RegisterInterested(fd_.fd());
+        return PendingPollResult();
+      })
+      .WillOnce(Return(StreamPollResult<int>(1)))
+      .WillOnce(Return(StreamPollResult<int>(2)))
+      .WillOnce(Return(StreamPollResult<int>(4)))
+      .WillOnce(Return(StreamPollResult<int>(8)))
+      .WillOnce(Return(StreamPollResult<int>(16)))
+      .WillOnce(Return(StreamPollResult<int>(32)))
+      .WillOnce(Return(StreamPollResult<int>(64)))
+      .WillOnce(Return(StreamPollResult<int>(128)))
+      .WillOnce(Return(DonePollResult()));
+
+  base::TestTaskRunner task_runner;
+  auto [handle, stream] =
+      SpawnResultStream<int>(&task_runner, [this]() mutable {
+        return base::Stream<int>(std::move(stream_pollable_));
       });
+  base::ignore_result(handle);
 
   task_runner.RunUntilIdle();
-  ASSERT_EQ(res.channel()->ReadNonBlocking().item, std::nullopt);
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsPending());
 
-  fd.Notify();
+  fd_.Notify();
   task_runner.RunUntilIdle();
 
-  auto read = res.channel()->ReadNonBlocking();
-  ASSERT_EQ(read.item, 1024);
-  ASSERT_FALSE(read.is_closed);
+  // We should get the first 4 elements and then nothing more: this corresponds
+  // to the internal channel buffer size being 4.
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 1);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 2);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 4);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 8);
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsPending());
 
+  // Should fill up a bunch more elements.
   task_runner.RunUntilIdle();
-  ASSERT_EQ(res.channel()->ReadNonBlocking().item, std::nullopt);
 
-  fd.Notify();
-  task_runner.RunUntilIdle();
+  // Drop the handle.
+  Drop(std::move(handle));
 
-  read = res.channel()->ReadNonBlocking();
-  ASSERT_EQ(read.item, 2048);
-  ASSERT_TRUE(read.is_closed);
+  // We should just get the next four buffered elements and the stream should
+  // complete.
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 16);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 32);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 64);
+  ASSERT_EQ(stream.PollNext(&ctx_).item(), 128);
+  ASSERT_TRUE(stream.PollNext(&ctx_).IsDone());
 }
 
 }  // namespace