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;
26class 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.
62class 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