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