1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. 2 // Use of this source code is governed by a BSD-style license that can be 3 // found in the LICENSE file. 4 5 #ifndef IPC_IPC_SYNC_CHANNEL_H_ 6 #define IPC_IPC_SYNC_CHANNEL_H_ 7 8 #include <string> 9 #include <deque> 10 11 #include "base/basictypes.h" 12 #include "base/memory/ref_counted.h" 13 #include "base/synchronization/lock.h" 14 #include "base/synchronization/waitable_event_watcher.h" 15 #include "ipc/ipc_channel_handle.h" 16 #include "ipc/ipc_channel_proxy.h" 17 #include "ipc/ipc_sync_message.h" 18 19 namespace base { 20 class WaitableEvent; 21 }; 22 23 namespace IPC { 24 25 class SyncMessage; 26 class ChannelFactory; 27 28 // This is similar to ChannelProxy, with the added feature of supporting sending 29 // synchronous messages. 30 // 31 // Overview of how the sync channel works 32 // -------------------------------------- 33 // When the sending thread sends a synchronous message, we create a bunch 34 // of tracking info (created in Send, stored in the PendingSyncMsg 35 // structure) associated with the message that we identify by the unique 36 // "MessageId" on the SyncMessage. Among the things we save is the 37 // "Deserializer" which is provided by the sync message. This object is in 38 // charge of reading the parameters from the reply message and putting them in 39 // the output variables provided by its caller. 40 // 41 // The info gets stashed in a queue since we could have a nested stack of sync 42 // messages (each side could send sync messages in response to sync messages, 43 // so it works like calling a function). The message is sent to the I/O thread 44 // for dispatch and the original thread blocks waiting for the reply. 45 // 46 // SyncContext maintains the queue in a threadsafe way and listens for replies 47 // on the I/O thread. When a reply comes in that matches one of the messages 48 // it's looking for (using the unique message ID), it will execute the 49 // deserializer stashed from before, and unblock the original thread. 50 // 51 // 52 // Significant complexity results from the fact that messages are still coming 53 // in while the original thread is blocked. Normal async messages are queued 54 // and dispatched after the blocking call is complete. Sync messages must 55 // be dispatched in a reentrant manner to avoid deadlock. 56 // 57 // 58 // Note that care must be taken that the lifetime of the ipc_thread argument 59 // is more than this object. If the message loop goes away while this object 60 // is running and it's used to send a message, then it will use the invalid 61 // message loop pointer to proxy it to the ipc thread. 62 class IPC_EXPORT SyncChannel : public ChannelProxy { 63 public: 64 enum RestrictDispatchGroup { 65 kRestrictDispatchGroup_None = 0, 66 }; 67 68 // Creates and initializes a sync channel. If create_pipe_now is specified, 69 // the channel will be initialized synchronously. 70 // The naming pattern follows IPC::Channel. 71 static scoped_ptr<SyncChannel> Create( 72 const IPC::ChannelHandle& channel_handle, 73 IPC::Channel::Mode mode, 74 Listener* listener, 75 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 76 bool create_pipe_now, 77 base::WaitableEvent* shutdown_event); 78 79 static scoped_ptr<SyncChannel> Create( 80 scoped_ptr<ChannelFactory> factory, 81 Listener* listener, 82 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 83 bool create_pipe_now, 84 base::WaitableEvent* shutdown_event); 85 86 // Creates an uninitialized sync channel. Call ChannelProxy::Init to 87 // initialize the channel. This two-step setup allows message filters to be 88 // added before any messages are sent or received. 89 static scoped_ptr<SyncChannel> Create( 90 Listener* listener, 91 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 92 base::WaitableEvent* shutdown_event); 93 94 virtual ~SyncChannel(); 95 96 virtual bool Send(Message* message) OVERRIDE; 97 98 // Sets the dispatch group for this channel, to only allow re-entrant dispatch 99 // of messages to other channels in the same group. 100 // 101 // Normally, any unblocking message coming from any channel can be dispatched 102 // when any (possibly other) channel is blocked on sending a message. This is 103 // needed in some cases to unblock certain loops (e.g. necessary when some 104 // processes share a window hierarchy), but may cause re-entrancy issues in 105 // some cases where such loops are not possible. This flags allows the tagging 106 // of some particular channels to only re-enter in known correct cases. 107 // 108 // Incoming messages on channels belonging to a group that is not 109 // kRestrictDispatchGroup_None will only be dispatched while a sync message is 110 // being sent on a channel of the *same* group. 111 // Incoming messages belonging to the kRestrictDispatchGroup_None group (the 112 // default) will be dispatched in any case. 113 void SetRestrictDispatchChannelGroup(int group); 114 115 protected: 116 class ReceivedSyncMsgQueue; 117 friend class ReceivedSyncMsgQueue; 118 119 // SyncContext holds the per object data for SyncChannel, so that SyncChannel 120 // can be deleted while it's being used in a different thread. See 121 // ChannelProxy::Context for more information. 122 class SyncContext : public Context { 123 public: 124 SyncContext( 125 Listener* listener, 126 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 127 base::WaitableEvent* shutdown_event); 128 129 // Adds information about an outgoing sync message to the context so that 130 // we know how to deserialize the reply. 131 void Push(SyncMessage* sync_msg); 132 133 // Cleanly remove the top deserializer (and throw it away). Returns the 134 // result of the Send call for that message. 135 bool Pop(); 136 137 // Returns an event that's set when the send is complete, timed out or the 138 // process shut down. 139 base::WaitableEvent* GetSendDoneEvent(); 140 141 // Returns an event that's set when an incoming message that's not the reply 142 // needs to get dispatched (by calling SyncContext::DispatchMessages). 143 base::WaitableEvent* GetDispatchEvent(); 144 145 void DispatchMessages(); 146 147 // Checks if the given message is blocking the listener thread because of a 148 // synchronous send. If it is, the thread is unblocked and true is 149 // returned. Otherwise the function returns false. 150 bool TryToUnblockListener(const Message* msg); 151 152 // Called on the IPC thread when a sync send that runs a nested message loop 153 // times out. 154 void OnSendTimeout(int message_id); 155 156 base::WaitableEvent* shutdown_event() { return shutdown_event_; } 157 158 ReceivedSyncMsgQueue* received_sync_msgs() { 159 return received_sync_msgs_.get(); 160 } 161 162 void set_restrict_dispatch_group(int group) { 163 restrict_dispatch_group_ = group; 164 } 165 166 int restrict_dispatch_group() const { 167 return restrict_dispatch_group_; 168 } 169 170 base::WaitableEventWatcher::EventCallback MakeWaitableEventCallback(); 171 172 private: 173 virtual ~SyncContext(); 174 // ChannelProxy methods that we override. 175 176 // Called on the listener thread. 177 virtual void Clear() OVERRIDE; 178 179 // Called on the IPC thread. 180 virtual bool OnMessageReceived(const Message& msg) OVERRIDE; 181 virtual void OnChannelError() OVERRIDE; 182 virtual void OnChannelOpened() OVERRIDE; 183 virtual void OnChannelClosed() OVERRIDE; 184 185 // Cancels all pending Send calls. 186 void CancelPendingSends(); 187 188 void OnWaitableEventSignaled(base::WaitableEvent* event); 189 190 typedef std::deque<PendingSyncMsg> PendingSyncMessageQueue; 191 PendingSyncMessageQueue deserializers_; 192 base::Lock deserializers_lock_; 193 194 scoped_refptr<ReceivedSyncMsgQueue> received_sync_msgs_; 195 196 base::WaitableEvent* shutdown_event_; 197 base::WaitableEventWatcher shutdown_watcher_; 198 base::WaitableEventWatcher::EventCallback shutdown_watcher_callback_; 199 int restrict_dispatch_group_; 200 }; 201 202 private: 203 SyncChannel( 204 Listener* listener, 205 const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner, 206 base::WaitableEvent* shutdown_event); 207 208 void OnWaitableEventSignaled(base::WaitableEvent* arg); 209 210 SyncContext* sync_context() { 211 return reinterpret_cast<SyncContext*>(context()); 212 } 213 214 // Both these functions wait for a reply, timeout or process shutdown. The 215 // latter one also runs a nested message loop in the meantime. 216 static void WaitForReply( 217 SyncContext* context, base::WaitableEvent* pump_messages_event); 218 219 // Runs a nested message loop until a reply arrives, times out, or the process 220 // shuts down. 221 static void WaitForReplyWithNestedMessageLoop(SyncContext* context); 222 223 // Starts the dispatch watcher. 224 void StartWatching(); 225 226 // Used to signal events between the IPC and listener threads. 227 base::WaitableEventWatcher dispatch_watcher_; 228 base::WaitableEventWatcher::EventCallback dispatch_watcher_callback_; 229 230 DISALLOW_COPY_AND_ASSIGN(SyncChannel); 231 }; 232 233 } // namespace IPC 234 235 #endif // IPC_IPC_SYNC_CHANNEL_H_ 236
