Make it possible to remove an Awakable with a specific "context" from a Dispatcher.

mojo/edk/system/data_pipe.cc

Index: mojo/edk/system/data_pipe.cc
diff --git a/mojo/edk/system/data_pipe.cc b/mojo/edk/system/data_pipe.cc
index b64eec402baedeb9717d0e6cab38e39e020b25a9..7167873266e740e66894892897d7a054f6f2e730 100644
--- a/mojo/edk/system/data_pipe.cc
+++ b/mojo/edk/system/data_pipe.cc
@@ -432,6 +432,17 @@ void DataPipe::ProducerRemoveAwakable(Awakable* awakable,
     *signals_state = impl_->ProducerGetHandleSignalsState();
 }
 
+void DataPipe::ProducerRemoveAwakableWithContext(
+    Awakable* awakable,
+    uint64_t context,
+    HandleSignalsState* signals_state) {
+  MutexLocker locker(&mutex_);
+  DCHECK(has_local_producer_no_lock());
+  producer_awakable_list_->RemoveWithContext(awakable, context);
+  if (signals_state)
+    *signals_state = impl_->ProducerGetHandleSignalsState();
+}
+
 void DataPipe::ProducerStartSerialize(Channel* channel,
                                       size_t* max_size,
                                       size_t* max_platform_handles) {
@@ -654,6 +665,17 @@ void DataPipe::ConsumerRemoveAwakable(Awakable* awakable,
     *signals_state = impl_->ConsumerGetHandleSignalsState();
 }
 
+void DataPipe::ConsumerRemoveAwakableWithContext(
+    Awakable* awakable,
+    uint64_t context,
+    HandleSignalsState* signals_state) {
+  MutexLocker locker(&mutex_);
+  DCHECK(has_local_consumer_no_lock());
+  consumer_awakable_list_->RemoveWithContext(awakable, context);
+  if (signals_state)
+    *signals_state = impl_->ConsumerGetHandleSignalsState();
+}
+
 void DataPipe::ConsumerStartSerialize(Channel* channel,
                                       size_t* max_size,
                                       size_t* max_platform_handles) {