Make the embedder API use the ChannelManager. (This doesn't actually change any functionality.) R=yzshen@chromium.org Review URL: https://codereview.chromium.org/728613002
diff --git a/mojo/edk/embedder/channel_info_forward.h b/mojo/edk/embedder/channel_info_forward.h index 494d5ef..8d23836 100644 --- a/mojo/edk/embedder/channel_info_forward.h +++ b/mojo/edk/embedder/channel_info_forward.h
@@ -9,23 +9,14 @@ #define MOJO_EDK_EMBEDDER_CHANNEL_INFO_FORWARD_H_ namespace mojo { - -// Forward declare |system::ChannelInfo|, so that we can typedef it to -// |embedder::ChannelInfo|. Users of the embedder API shouldn't use this -// directly; instead they should use |embedder::ChannelInfo|. -namespace system { -struct ChannelInfo; -} - namespace embedder { // This is an opaque type. The embedder API uses (returns and takes as // arguments) pointers to this type. (We don't simply use |void*|, so that // custom deleters and such can be used without additional wrappers. -typedef system::ChannelInfo ChannelInfo; +struct ChannelInfo; } // namespace embedder - } // namespace mojo #endif // MOJO_EDK_EMBEDDER_CHANNEL_INFO_FORWARD_H_
diff --git a/mojo/edk/embedder/embedder.cc b/mojo/edk/embedder/embedder.cc index 1d8d89d..9c73eb0 100644 --- a/mojo/edk/embedder/embedder.cc +++ b/mojo/edk/embedder/embedder.cc
@@ -13,7 +13,7 @@ #include "mojo/edk/embedder/platform_support.h" #include "mojo/edk/system/channel.h" #include "mojo/edk/system/channel_endpoint.h" -#include "mojo/edk/system/channel_info.h" +#include "mojo/edk/system/channel_manager.h" #include "mojo/edk/system/configuration.h" #include "mojo/edk/system/core.h" #include "mojo/edk/system/message_pipe_dispatcher.h" @@ -25,13 +25,15 @@ namespace { -// Helper for |CreateChannel...()|. (Note: May return null for some failures.) -scoped_refptr<system::Channel> MakeChannel( +// Helper for |CreateChannel...()|. Returns 0 on failure. Called on the channel +// creation thread. +system::ChannelId MakeChannel( ScopedPlatformHandle platform_handle, scoped_refptr<system::ChannelEndpoint> channel_endpoint) { DCHECK(platform_handle.is_valid()); // Create and initialize a |system::Channel|. + DCHECK(internal::g_core); scoped_refptr<system::Channel> channel = new system::Channel(internal::g_core->platform_support()); if (!channel->Init(system::RawChannel::Create(platform_handle.Pass()))) { @@ -39,22 +41,25 @@ // reached some system resource limit). LOG(ERROR) << "Channel::Init() failed"; // Return null, since |Shutdown()| shouldn't be called in this case. - return scoped_refptr<system::Channel>(); + return 0; } - // Once |Init()| has succeeded, we have to return |channel| (since - // |Shutdown()| will have to be called on it). channel->AttachAndRunEndpoint(channel_endpoint, true); - return channel; + + DCHECK(internal::g_channel_manager); + return internal::g_channel_manager->AddChannel( + channel, base::MessageLoopProxy::current()); } +// Helper for |CreateChannel()|. Called on the channel creation thread. void CreateChannelHelper( ScopedPlatformHandle platform_handle, scoped_ptr<ChannelInfo> channel_info, scoped_refptr<system::ChannelEndpoint> channel_endpoint, DidCreateChannelCallback callback, scoped_refptr<base::TaskRunner> callback_thread_task_runner) { - channel_info->channel = MakeChannel(platform_handle.Pass(), channel_endpoint); + channel_info->channel_id = + MakeChannel(platform_handle.Pass(), channel_endpoint); // Hand the channel back to the embedder. if (callback_thread_task_runner.get()) { @@ -71,12 +76,15 @@ // Declared in embedder_internal.h. system::Core* g_core = nullptr; +system::ChannelManager* g_channel_manager = nullptr; } // namespace internal void Init(scoped_ptr<PlatformSupport> platform_support) { DCHECK(!internal::g_core); internal::g_core = new system::Core(platform_support.Pass()); + DCHECK(!internal::g_channel_manager); + internal::g_channel_manager = new system::ChannelManager(); } Configuration* GetConfiguration() { @@ -99,8 +107,7 @@ MessagePipeHandle(internal::g_core->AddDispatcher(dispatcher))); *channel_info = - new ChannelInfo(MakeChannel(platform_handle.Pass(), channel_endpoint), - base::MessageLoopProxy::current()); + new ChannelInfo(MakeChannel(platform_handle.Pass(), channel_endpoint)); return rv.Pass(); } @@ -122,9 +129,8 @@ ScopedMessagePipeHandle rv( MessagePipeHandle(internal::g_core->AddDispatcher(dispatcher))); + // We'll have to set |channel_info->channel_id| on the I/O thread. scoped_ptr<ChannelInfo> channel_info(new ChannelInfo()); - // We'll have to set |channel_info->channel| on the I/O thread. - channel_info->channel_thread_task_runner = io_thread_task_runner; if (rv.is_valid()) { io_thread_task_runner->PostTask( @@ -141,35 +147,25 @@ return rv.Pass(); } -void DestroyChannelOnIOThread(ChannelInfo* channel_info) { - DCHECK(channel_info); - if (!channel_info->channel.get()) { - // Presumably, |Init()| on the channel failed. - return; - } - - channel_info->channel->Shutdown(); - delete channel_info; -} - // TODO(vtl): Write tests for this. void DestroyChannel(ChannelInfo* channel_info) { DCHECK(channel_info); - DCHECK(channel_info->channel_thread_task_runner.get()); - - if (!channel_info->channel.get()) { + if (!channel_info->channel_id) { // Presumably, |Init()| on the channel failed. return; } - channel_info->channel->WillShutdownSoon(); - channel_info->channel_thread_task_runner->PostTask( - FROM_HERE, base::Bind(&DestroyChannelOnIOThread, channel_info)); + DCHECK(internal::g_channel_manager); + // This will destroy the channel synchronously if called from the channel + // thread. + internal::g_channel_manager->ShutdownChannel(channel_info->channel_id); + delete channel_info; } void WillDestroyChannelSoon(ChannelInfo* channel_info) { DCHECK(channel_info); - channel_info->channel->WillShutdownSoon(); + DCHECK(internal::g_channel_manager); + internal::g_channel_manager->WillShutdownChannel(channel_info->channel_id); } MojoResult CreatePlatformHandleWrapper(
diff --git a/mojo/edk/embedder/embedder.h b/mojo/edk/embedder/embedder.h index 11a352c..02b4ac3 100644 --- a/mojo/edk/embedder/embedder.h +++ b/mojo/edk/embedder/embedder.h
@@ -32,10 +32,12 @@ // A "channel" is a connection on top of an OS "pipe", on top of which Mojo // message pipes (etc.) can be multiplexed. It must "live" on some I/O thread. // -// There are two "channel" creation/destruction APIs: the synchronous -// |CreateChannelOnIOThread()|/|DestroyChannelOnIOThread()|, which must be -// called from the I/O thread, and the asynchronous -// |CreateChannel()|/|DestroyChannel()|, which may be called from any thread. +// There are two channel creation APIs: |CreateChannelOnIOThread()| creates a +// channel synchronously and must be called from the I/O thread, while +// |CreateChannel()| is asynchronous and may be called from any thread. +// |DestroyChannel()| is used to destroy the channel in either case and may be +// called from any thread, but completes synchronously when called from the I/O +// thread. // // Both creation functions have a |platform_handle| argument, which should be an // OS-dependent handle to one side of a suitable bidirectional OS "pipe" (e.g., @@ -67,9 +69,8 @@ // Creates a channel; must only be called from the I/O thread. |platform_handle| // should be a handle to a connected OS "pipe". Eventually (even on failure), -// the "out" value |*channel_info| should be passed to -// |DestroyChannelOnIOThread()| (or |DestoryChannel()|) to tear down the -// channel. Returns a handle to the bootstrap message pipe. +// the "out" value |*channel_info| should be passed to |DestoryChannel()| to +// tear down the channel. Returns a handle to the bootstrap message pipe. MOJO_SYSTEM_IMPL_EXPORT ScopedMessagePipeHandle CreateChannelOnIOThread(ScopedPlatformHandle platform_handle, ChannelInfo** channel_info); @@ -79,32 +80,26 @@ // |platform_handle| should be a handle to a connected OS "pipe". // |io_thread_task_runner| should be the |TaskRunner| for the I/O thread. // |callback| should be the callback to call with the |ChannelInfo*|, which -// should eventually be passed to |DestroyChannel()| (or -// |DestroyChannelOnIOThread()|) to tear down the channel; the callback will be -// called using |callback_thread_task_runner| if that is non-null, or otherwise -// it will be called using |io_thread_task_runner|. Returns a handle to the -// bootstrap message pipe. +// should eventually be passed to |DestroyChannel()| to tear down the channel; +// the callback will be called using |callback_thread_task_runner| if that is +// non-null, or otherwise it will be called using |io_thread_task_runner|. +// Returns a handle to the bootstrap message pipe. MOJO_SYSTEM_IMPL_EXPORT ScopedMessagePipeHandle CreateChannel(ScopedPlatformHandle platform_handle, scoped_refptr<base::TaskRunner> io_thread_task_runner, DidCreateChannelCallback callback, scoped_refptr<base::TaskRunner> callback_thread_task_runner); -// Destroys a channel that was created using either |CreateChannelOnIOThread()| -// or |CreateChannel()|; must only be called from the I/O thread. |channel_info| -// should be the "out" value from |CreateChannelOnIOThread()| or the value -// provided to the callback to |CreateChannel()|. -MOJO_SYSTEM_IMPL_EXPORT void DestroyChannelOnIOThread( - ChannelInfo* channel_info); - -// Destroys a channel (asynchronously) that was created using |CreateChannel()|; -// may be called from any thread. |channel_info| should be the value provided to -// the callback to |CreateChannel()|. +// Destroys a channel that was created using |CreateChannel()| (or +// |CreateChannelOnIOThread()|); may be called from any thread. |channel_info| +// should be the value provided to the callback to |CreateChannel()| (or +// returned by |CreateChannelOnIOThread()|). If called from the I/O thread, this +// will complete synchronously (in particular, it will post no tasks). MOJO_SYSTEM_IMPL_EXPORT void DestroyChannel(ChannelInfo* channel_info); // Inform the channel that it will soon be destroyed (doing so is optional). // This may be called from any thread, but the caller must ensure that this is -// called before |DestroyChannel()| or |DestroyChannelOnIOThread()|. +// called before |DestroyChannel()|. MOJO_SYSTEM_IMPL_EXPORT void WillDestroyChannelSoon(ChannelInfo* channel_info); // Creates a |MojoHandle| that wraps the given |PlatformHandle| (taking
diff --git a/mojo/edk/embedder/embedder_internal.h b/mojo/edk/embedder/embedder_internal.h index b8919b0..ab8388a 100644 --- a/mojo/edk/embedder/embedder_internal.h +++ b/mojo/edk/embedder/embedder_internal.h
@@ -2,22 +2,50 @@ // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. +// This header contains internal details for the *implementation* of the +// embedder API. It should not be included by any public header (nor by users of +// the embedder API). + #ifndef MOJO_EDK_EMBEDDER_EMBEDDER_INTERNAL_H_ #define MOJO_EDK_EMBEDDER_EMBEDDER_INTERNAL_H_ +#include <stdint.h> + namespace mojo { namespace system { + +class ChannelManager; class Core; + +// Repeat a typedef in mojo/edk/system/channel_manager.h, to avoid including it. +typedef uintptr_t ChannelId; + } // namespace system namespace embedder { + +// This is a type that's opaque to users of the embedder API (which only +// gives/takes |ChannelInfo*|s). We make it a struct to make it +// template-friendly. +struct ChannelInfo { + explicit ChannelInfo(system::ChannelId channel_id = 0) + : channel_id(channel_id) {} + + system::ChannelId channel_id; +}; + namespace internal { // Instance of |Core| used by the system functions (|Mojo...()|). extern system::Core* g_core; +// Instance of |ChannelManager| used by the channel management functions +// (|mojo::embedder::CreateChannel()|, etc.). +extern system::ChannelManager* g_channel_manager; + } // namespace internal + } // namepace embedder } // namespace mojo
diff --git a/mojo/edk/embedder/embedder_unittest.cc b/mojo/edk/embedder/embedder_unittest.cc index 78423d2..16a5ce2 100644 --- a/mojo/edk/embedder/embedder_unittest.cc +++ b/mojo/edk/embedder/embedder_unittest.cc
@@ -50,11 +50,7 @@ // Destructor: Shuts down the channel. (As noted above, for this to happen, // the I/O thread must be alive and pumping messages.) - ~ScopedTestChannel() { - system::test::PostTaskAndWait( - io_thread_task_runner_, FROM_HERE, - base::Bind(&ScopedTestChannel::DestroyChannel, base::Unretained(this))); - } + ~ScopedTestChannel() { DestroyChannel(channel_info_); } // Waits for channel creation to be completed. void WaitForChannelCreationCompletion() { did_create_channel_event_.Wait(); } @@ -73,12 +69,6 @@ did_create_channel_event_.Signal(); } - void DestroyChannel() { - CHECK(channel_info_); - DestroyChannelOnIOThread(channel_info_); - channel_info_ = nullptr; - } - scoped_refptr<base::TaskRunner> io_thread_task_runner_; // Valid from creation until whenever it gets closed (by the "owner" of this
diff --git a/mojo/edk/embedder/test_embedder.cc b/mojo/edk/embedder/test_embedder.cc index 1eca9c8..defab41 100644 --- a/mojo/edk/embedder/test_embedder.cc +++ b/mojo/edk/embedder/test_embedder.cc
@@ -10,6 +10,7 @@ #include "mojo/edk/embedder/embedder.h" #include "mojo/edk/embedder/embedder_internal.h" #include "mojo/edk/embedder/simple_platform_support.h" +#include "mojo/edk/system/channel_manager.h" #include "mojo/edk/system/core.h" #include "mojo/edk/system/handle_table.h" @@ -45,12 +46,14 @@ } bool Shutdown() { - system::Core* core = internal::g_core; - CHECK(core); - internal::g_core = nullptr; + CHECK(internal::g_channel_manager); + delete internal::g_channel_manager; + internal::g_channel_manager = nullptr; - bool rv = system::internal::ShutdownCheckNoLeaks(core); - delete core; + CHECK(internal::g_core); + bool rv = system::internal::ShutdownCheckNoLeaks(internal::g_core); + delete internal::g_core; + internal::g_core = nullptr; return rv; }
diff --git a/mojo/edk/embedder/test_embedder.h b/mojo/edk/embedder/test_embedder.h index b6203b4..76917ee 100644 --- a/mojo/edk/embedder/test_embedder.h +++ b/mojo/edk/embedder/test_embedder.h
@@ -20,6 +20,9 @@ // do more work to ensure that tests don't leak, etc.) Returns true if there // were no problems, false if there were leaks -- i.e., handles still open -- or // any other problems. +// +// Note: It is up to the caller to ensure that there are not outstanding +// callbacks from |CreateChannel()| before calling this. MOJO_SYSTEM_IMPL_EXPORT bool Shutdown(); } // namespace test
diff --git a/mojo/edk/system/channel_manager.cc b/mojo/edk/system/channel_manager.cc index e673299..89f5155 100644 --- a/mojo/edk/system/channel_manager.cc +++ b/mojo/edk/system/channel_manager.cc
@@ -6,6 +6,7 @@ #include "base/bind.h" #include "base/location.h" +#include "base/message_loop/message_loop_proxy.h" namespace mojo { namespace system { @@ -13,9 +14,14 @@ namespace { void ShutdownChannelHelper(const ChannelInfo& channel_info) { - channel_info.channel->WillShutdownSoon(); - channel_info.channel_thread_task_runner->PostTask( - FROM_HERE, base::Bind(&Channel::Shutdown, channel_info.channel)); + if (base::MessageLoopProxy::current() == + channel_info.channel_thread_task_runner) { + channel_info.channel->Shutdown(); + } else { + channel_info.channel->WillShutdownSoon(); + channel_info.channel_thread_task_runner->PostTask( + FROM_HERE, base::Bind(&Channel::Shutdown, channel_info.channel)); + } } } // namespace @@ -45,17 +51,22 @@ return channel_id; } -void ChannelManager::ShutdownChannel(ChannelId channel_id) { - ChannelInfo channel_info; - { - base::AutoLock locker(lock_); +void ChannelManager::WillShutdownChannel(ChannelId channel_id) { + GetChannelInfo(channel_id).channel->WillShutdownSoon(); +} - auto it = channel_infos_.find(channel_id); - DCHECK(it != channel_infos_.end()); - channel_info.Swap(&it->second); - channel_infos_.erase(it); - } - ShutdownChannelHelper(channel_info); +void ChannelManager::ShutdownChannel(ChannelId channel_id) { + ShutdownChannelHelper(GetChannelInfo(channel_id)); +} + +ChannelInfo ChannelManager::GetChannelInfo(ChannelId channel_id) { + ChannelInfo rv; + base::AutoLock locker(lock_); + auto it = channel_infos_.find(channel_id); + DCHECK(it != channel_infos_.end()); + rv.Swap(&it->second); + channel_infos_.erase(it); + return rv; } } // namespace system
diff --git a/mojo/edk/system/channel_manager.h b/mojo/edk/system/channel_manager.h index 244f8ad..60a3048 100644 --- a/mojo/edk/system/channel_manager.h +++ b/mojo/edk/system/channel_manager.h
@@ -43,23 +43,32 @@ // Adds |channel| to the set of |Channel|s managed by this |ChannelManager|; // |channel_thread_task_runner| should be the task runner for |channel|'s - // creation (a.k.a. I/O) thread. - // - // |channel| should either already be initialized or at least already have a - // task posted to |channel_thread_task_runner| to initialize it. It should not - // be managed by any |ChannelManager| yet. - // - // Returns the ID for the added channel. + // creation (a.k.a. I/O) thread. |channel| should either already be + // initialized. It should not be managed by any |ChannelManager| yet. Returns + // the ID for the added channel. ChannelId AddChannel( scoped_refptr<Channel> channel, scoped_refptr<base::TaskRunner> channel_thread_task_runner); + // Informs the channel manager (and thus channel) that it will be shutdown + // soon (by calling |ShutdownChannel()|). Calling this is optional (and may in + // fact be called multiple times) but it will suppress certain warnings (e.g., + // for the channel being broken) and enable others (if messages are written to + // the channel). + void WillShutdownChannel(ChannelId channel_id); + // Shuts down the channel specified by the given ID. It is up to the caller to // guarantee that this is only called once per channel (that was added using - // |AddChannel()|). + // |AddChannel()|). If called from the chanel's creation thread (i.e., + // |base::MessageLoopProxy::current()| is the channel thread's |TaskRunner|), + // this will complete synchronously. void ShutdownChannel(ChannelId channel_id); private: + // Gets the |ChannelInfo| for the channel specified by the given ID. (This + // should *not* be called under lock.) + ChannelInfo GetChannelInfo(ChannelId channel_id); + // Note: |Channel| methods should not be called under |lock_|. base::Lock lock_; // Protects the members below.