Fix shutdown assert in new Mojo EDK.

mojo/edk/system/data_pipe_producer_dispatcher.cc

 // Copyright 2013 The Chromium 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 "mojo/edk/system/data_pipe_producer_dispatcher.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/edk/embedder/embedder_internal.h"
@@ skipping 68 matching lines @@
            serialized_write_buffer_size);
   return rv;
 }
 
 DataPipeProducerDispatcher::DataPipeProducerDispatcher(
     const MojoCreateDataPipeOptions& options)
     : options_(options), channel_(nullptr), error_(false), serialized_(false) {
 }
 
 DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
-  // |Close()|/|CloseImplNoLock()| should have taken care of the channel.
-  DCHECK(!channel_);
+  // See comment in ~MessagePipeDispatcher.
+  if (channel_ && internal::g_io_thread_task_runner->RunsTasksOnCurrentThread())
+    channel_->Shutdown();
+  else
+    DCHECK(!channel_);
 }
 
 void DataPipeProducerDispatcher::CancelAllAwakablesNoLock() {
   lock().AssertAcquired();
   awakable_list_.CancelAll();
 }
 
 void DataPipeProducerDispatcher::CloseImplNoLock() {
   lock().AssertAcquired();
   internal::g_io_thread_task_runner->PostTask(
@@ skipping 294 matching lines @@
     CHECK(fds.empty());
     if (write_error)
       serialized_platform_handle_.reset();
     channel_ = nullptr;
   }
   serialized_ = true;
 }
 
 }  // namespace edk
 }  // namespace mojo