ipc/ipc_sync_channel.h

   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   // TODO(erikchen): Remove default parameter for |broker|. It exists only to
  72   // make the upcoming refactor decomposable into smaller CLs.
  73   // http://crbug.com/493414.
  74   static scoped_ptr<SyncChannel> Create(
  75       const IPC::ChannelHandle& channel_handle,
  76       IPC::Channel::Mode mode,
  77       Listener* listener,
  78       const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  79       bool create_pipe_now,
  80       base::WaitableEvent* shutdown_event,
  81       AttachmentBroker* broker = nullptr);
  82
  83   static scoped_ptr<SyncChannel> Create(
  84       scoped_ptr<ChannelFactory> factory,
  85       Listener* listener,
  86       const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  87       bool create_pipe_now,
  88       base::WaitableEvent* shutdown_event);
  89
  90   // Creates an uninitialized sync channel. Call ChannelProxy::Init to
  91   // initialize the channel. This two-step setup allows message filters to be
  92   // added before any messages are sent or received.
  93   static scoped_ptr<SyncChannel> Create(
  94       Listener* listener,
  95       const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
  96       base::WaitableEvent* shutdown_event);
  97
  98   ~SyncChannel() override;
  99
 100   bool Send(Message* message) override;
 101
 102   // Sets the dispatch group for this channel, to only allow re-entrant dispatch
 103   // of messages to other channels in the same group.
 104   //
 105   // Normally, any unblocking message coming from any channel can be dispatched
 106   // when any (possibly other) channel is blocked on sending a message. This is
 107   // needed in some cases to unblock certain loops (e.g. necessary when some
 108   // processes share a window hierarchy), but may cause re-entrancy issues in
 109   // some cases where such loops are not possible. This flags allows the tagging
 110   // of some particular channels to only re-enter in known correct cases.
 111   //
 112   // Incoming messages on channels belonging to a group that is not
 113   // kRestrictDispatchGroup_None will only be dispatched while a sync message is
 114   // being sent on a channel of the *same* group.
 115   // Incoming messages belonging to the kRestrictDispatchGroup_None group (the
 116   // default) will be dispatched in any case.
 117   void SetRestrictDispatchChannelGroup(int group);
 118
 119  protected:
 120   class ReceivedSyncMsgQueue;
 121   friend class ReceivedSyncMsgQueue;
 122
 123   // SyncContext holds the per object data for SyncChannel, so that SyncChannel
 124   // can be deleted while it's being used in a different thread.  See
 125   // ChannelProxy::Context for more information.
 126   class SyncContext : public Context {
 127    public:
 128     SyncContext(
 129         Listener* listener,
 130         const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
 131         base::WaitableEvent* shutdown_event);
 132
 133     // Adds information about an outgoing sync message to the context so that
 134     // we know how to deserialize the reply.
 135     void Push(SyncMessage* sync_msg);
 136
 137     // Cleanly remove the top deserializer (and throw it away).  Returns the
 138     // result of the Send call for that message.
 139     bool Pop();
 140
 141     // Returns an event that's set when the send is complete, timed out or the
 142     // process shut down.
 143     base::WaitableEvent* GetSendDoneEvent();
 144
 145     // Returns an event that's set when an incoming message that's not the reply
 146     // needs to get dispatched (by calling SyncContext::DispatchMessages).
 147     base::WaitableEvent* GetDispatchEvent();
 148
 149     void DispatchMessages();
 150
 151     // Checks if the given message is blocking the listener thread because of a
 152     // synchronous send.  If it is, the thread is unblocked and true is
 153     // returned. Otherwise the function returns false.
 154     bool TryToUnblockListener(const Message* msg);
 155
 156     // Called on the IPC thread when a sync send that runs a nested message loop
 157     // times out.
 158     void OnSendTimeout(int message_id);
 159
 160     base::WaitableEvent* shutdown_event() { return shutdown_event_; }
 161
 162     ReceivedSyncMsgQueue* received_sync_msgs() {
 163       return received_sync_msgs_.get();
 164     }
 165
 166     void set_restrict_dispatch_group(int group) {
 167       restrict_dispatch_group_ = group;
 168     }
 169
 170     int restrict_dispatch_group() const {
 171       return restrict_dispatch_group_;
 172     }
 173
 174     base::WaitableEventWatcher::EventCallback MakeWaitableEventCallback();
 175
 176    private:
 177     ~SyncContext() override;
 178     // ChannelProxy methods that we override.
 179
 180     // Called on the listener thread.
 181     void Clear() override;
 182
 183     // Called on the IPC thread.
 184     bool OnMessageReceived(const Message& msg) override;
 185     void OnChannelError() override;
 186     void OnChannelOpened() override;
 187     void OnChannelClosed() override;
 188
 189     // Cancels all pending Send calls.
 190     void CancelPendingSends();
 191
 192     void OnWaitableEventSignaled(base::WaitableEvent* event);
 193
 194     typedef std::deque<PendingSyncMsg> PendingSyncMessageQueue;
 195     PendingSyncMessageQueue deserializers_;
 196     base::Lock deserializers_lock_;
 197
 198     scoped_refptr<ReceivedSyncMsgQueue> received_sync_msgs_;
 199
 200     base::WaitableEvent* shutdown_event_;
 201     base::WaitableEventWatcher shutdown_watcher_;
 202     base::WaitableEventWatcher::EventCallback shutdown_watcher_callback_;
 203     int restrict_dispatch_group_;
 204   };
 205
 206  private:
 207   SyncChannel(
 208       Listener* listener,
 209       const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
 210       base::WaitableEvent* shutdown_event);
 211
 212   void OnWaitableEventSignaled(base::WaitableEvent* arg);
 213
 214   SyncContext* sync_context() {
 215     return reinterpret_cast<SyncContext*>(context());
 216   }
 217
 218   // Both these functions wait for a reply, timeout or process shutdown.  The
 219   // latter one also runs a nested message loop in the meantime.
 220   static void WaitForReply(
 221       SyncContext* context, base::WaitableEvent* pump_messages_event);
 222
 223   // Runs a nested message loop until a reply arrives, times out, or the process
 224   // shuts down.
 225   static void WaitForReplyWithNestedMessageLoop(SyncContext* context);
 226
 227   // Starts the dispatch watcher.
 228   void StartWatching();
 229
 230   // Used to signal events between the IPC and listener threads.
 231   base::WaitableEventWatcher dispatch_watcher_;
 232   base::WaitableEventWatcher::EventCallback dispatch_watcher_callback_;
 233
 234   DISALLOW_COPY_AND_ASSIGN(SyncChannel);
 235 };
 236
 237 }  // namespace IPC
 238
 239 #endif  // IPC_IPC_SYNC_CHANNEL_H_