ipc/ipc_channel_proxy.cc

   1 // Copyright (c) 2011 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 #include "base/memory/ref_counted.h"
   6 #include "base/memory/scoped_ptr.h"
   7 #include "ipc/ipc_channel_proxy.h"
   8 #include "ipc/ipc_logging.h"
   9 #include "ipc/ipc_message_utils.h"
  10
  11 namespace IPC {
  12
  13 //------------------------------------------------------------------------------
  14
  15 // This task ensures the message is deleted if the task is deleted without
  16 // having been run.
  17 class SendTask : public Task {
  18  public:
  19   SendTask(ChannelProxy::Context* context, Message* message)
  20       : context_(context),
  21         message_(message) {
  22   }
  23
  24   virtual void Run() {
  25     context_->OnSendMessage(message_.release());
  26   }
  27
  28  private:
  29   scoped_refptr<ChannelProxy::Context> context_;
  30   scoped_ptr<Message> message_;
  31
  32   DISALLOW_COPY_AND_ASSIGN(SendTask);
  33 };
  34
  35 //------------------------------------------------------------------------------
  36
  37 ChannelProxy::MessageFilter::MessageFilter() {}
  38
  39 ChannelProxy::MessageFilter::~MessageFilter() {}
  40
  41 void ChannelProxy::MessageFilter::OnFilterAdded(Channel* channel) {}
  42
  43 void ChannelProxy::MessageFilter::OnFilterRemoved() {}
  44
  45 void ChannelProxy::MessageFilter::OnChannelConnected(int32 peer_pid) {}
  46
  47 void ChannelProxy::MessageFilter::OnChannelError() {}
  48
  49 void ChannelProxy::MessageFilter::OnChannelClosing() {}
  50
  51 bool ChannelProxy::MessageFilter::OnMessageReceived(const Message& message) {
  52   return false;
  53 }
  54
  55 void ChannelProxy::MessageFilter::OnDestruct() const {
  56   delete this;
  57 }
  58
  59 //------------------------------------------------------------------------------
  60
  61 ChannelProxy::Context::Context(Channel::Listener* listener,
  62                                base::MessageLoopProxy* ipc_message_loop)
  63     : listener_message_loop_(base::MessageLoopProxy::current()),
  64       listener_(listener),
  65       ipc_message_loop_(ipc_message_loop),
  66       peer_pid_(0),
  67       channel_connected_called_(false) {
  68 }
  69
  70 ChannelProxy::Context::~Context() {
  71 }
  72
  73 void ChannelProxy::Context::CreateChannel(const IPC::ChannelHandle& handle,
  74                                           const Channel::Mode& mode) {
  75   DCHECK(channel_.get() == NULL);
  76   channel_id_ = handle.name;
  77   channel_.reset(new Channel(handle, mode, this));
  78 }
  79
  80 bool ChannelProxy::Context::TryFilters(const Message& message) {
  81 #ifdef IPC_MESSAGE_LOG_ENABLED
  82   Logging* logger = Logging::GetInstance();
  83   if (logger->Enabled())
  84     logger->OnPreDispatchMessage(message);
  85 #endif
  86
  87   for (size_t i = 0; i < filters_.size(); ++i) {
  88     if (filters_[i]->OnMessageReceived(message)) {
  89 #ifdef IPC_MESSAGE_LOG_ENABLED
  90       if (logger->Enabled())
  91         logger->OnPostDispatchMessage(message, channel_id_);
  92 #endif
  93       return true;
  94     }
  95   }
  96   return false;
  97 }
  98
  99 // Called on the IPC::Channel thread
 100 bool ChannelProxy::Context::OnMessageReceived(const Message& message) {
 101   // First give a chance to the filters to process this message.
 102   if (!TryFilters(message))
 103     OnMessageReceivedNoFilter(message);
 104   return true;
 105 }
 106
 107 // Called on the IPC::Channel thread
 108 bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
 109   // NOTE: This code relies on the listener's message loop not going away while
 110   // this thread is active.  That should be a reasonable assumption, but it
 111   // feels risky.  We may want to invent some more indirect way of referring to
 112   // a MessageLoop if this becomes a problem.
 113   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
 114       this, &Context::OnDispatchMessage, message));
 115   return true;
 116 }
 117
 118 // Called on the IPC::Channel thread
 119 void ChannelProxy::Context::OnChannelConnected(int32 peer_pid) {
 120   // Add any pending filters.  This avoids a race condition where someone
 121   // creates a ChannelProxy, calls AddFilter, and then right after starts the
 122   // peer process.  The IO thread could receive a message before the task to add
 123   // the filter is run on the IO thread.
 124   OnAddFilter();
 125
 126   peer_pid_ = peer_pid;
 127   for (size_t i = 0; i < filters_.size(); ++i)
 128     filters_[i]->OnChannelConnected(peer_pid);
 129
 130   // See above comment about using listener_message_loop_ here.
 131   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
 132       this, &Context::OnDispatchConnected));
 133 }
 134
 135 // Called on the IPC::Channel thread
 136 void ChannelProxy::Context::OnChannelError() {
 137   for (size_t i = 0; i < filters_.size(); ++i)
 138     filters_[i]->OnChannelError();
 139
 140   // See above comment about using listener_message_loop_ here.
 141   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
 142       this, &Context::OnDispatchError));
 143 }
 144
 145 // Called on the IPC::Channel thread
 146 void ChannelProxy::Context::OnChannelOpened() {
 147   DCHECK(channel_ != NULL);
 148
 149   // Assume a reference to ourselves on behalf of this thread.  This reference
 150   // will be released when we are closed.
 151   AddRef();
 152
 153   if (!channel_->Connect()) {
 154     OnChannelError();
 155     return;
 156   }
 157
 158   for (size_t i = 0; i < filters_.size(); ++i)
 159     filters_[i]->OnFilterAdded(channel_.get());
 160 }
 161
 162 // Called on the IPC::Channel thread
 163 void ChannelProxy::Context::OnChannelClosed() {
 164   // It's okay for IPC::ChannelProxy::Close to be called more than once, which
 165   // would result in this branch being taken.
 166   if (!channel_.get())
 167     return;
 168
 169   for (size_t i = 0; i < filters_.size(); ++i) {
 170     filters_[i]->OnChannelClosing();
 171     filters_[i]->OnFilterRemoved();
 172   }
 173
 174   // We don't need the filters anymore.
 175   filters_.clear();
 176
 177   channel_.reset();
 178
 179   // Balance with the reference taken during startup.  This may result in
 180   // self-destruction.
 181   Release();
 182 }
 183
 184 // Called on the IPC::Channel thread
 185 void ChannelProxy::Context::OnSendMessage(Message* message) {
 186   if (!channel_.get()) {
 187     delete message;
 188     OnChannelClosed();
 189     return;
 190   }
 191   if (!channel_->Send(message))
 192     OnChannelError();
 193 }
 194
 195 // Called on the IPC::Channel thread
 196 void ChannelProxy::Context::OnAddFilter() {
 197   std::vector<scoped_refptr<MessageFilter> > new_filters;
 198   {
 199     base::AutoLock auto_lock(pending_filters_lock_);
 200     new_filters.swap(pending_filters_);
 201   }
 202
 203   for (size_t i = 0; i < new_filters.size(); ++i) {
 204     filters_.push_back(new_filters[i]);
 205
 206     // If the channel has already been created, then we need to send this
 207     // message so that the filter gets access to the Channel.
 208     if (channel_.get())
 209       new_filters[i]->OnFilterAdded(channel_.get());
 210     // Ditto for if the channel has been connected.
 211     if (peer_pid_)
 212       new_filters[i]->OnChannelConnected(peer_pid_);
 213   }
 214 }
 215
 216 // Called on the IPC::Channel thread
 217 void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
 218   for (size_t i = 0; i < filters_.size(); ++i) {
 219     if (filters_[i].get() == filter) {
 220       filter->OnFilterRemoved();
 221       filters_.erase(filters_.begin() + i);
 222       return;
 223     }
 224   }
 225
 226   NOTREACHED() << "filter to be removed not found";
 227 }
 228
 229 // Called on the listener's thread
 230 void ChannelProxy::Context::AddFilter(MessageFilter* filter) {
 231   base::AutoLock auto_lock(pending_filters_lock_);
 232   pending_filters_.push_back(make_scoped_refptr(filter));
 233   ipc_message_loop_->PostTask(
 234       FROM_HERE,
 235       NewRunnableMethod(this, &Context::OnAddFilter));
 236 }
 237
 238 // Called on the listener's thread
 239 void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
 240   if (!listener_)
 241     return;
 242
 243   OnDispatchConnected();
 244
 245 #ifdef IPC_MESSAGE_LOG_ENABLED
 246   Logging* logger = Logging::GetInstance();
 247   if (message.type() == IPC_LOGGING_ID) {
 248     logger->OnReceivedLoggingMessage(message);
 249     return;
 250   }
 251
 252   if (logger->Enabled())
 253     logger->OnPreDispatchMessage(message);
 254 #endif
 255
 256   listener_->OnMessageReceived(message);
 257
 258 #ifdef IPC_MESSAGE_LOG_ENABLED
 259   if (logger->Enabled())
 260     logger->OnPostDispatchMessage(message, channel_id_);
 261 #endif
 262 }
 263
 264 // Called on the listener's thread
 265 void ChannelProxy::Context::OnDispatchConnected() {
 266   if (channel_connected_called_)
 267     return;
 268
 269   channel_connected_called_ = true;
 270   if (listener_)
 271     listener_->OnChannelConnected(peer_pid_);
 272 }
 273
 274 // Called on the listener's thread
 275 void ChannelProxy::Context::OnDispatchError() {
 276   if (listener_)
 277     listener_->OnChannelError();
 278 }
 279
 280 //-----------------------------------------------------------------------------
 281
 282 ChannelProxy::ChannelProxy(const IPC::ChannelHandle& channel_handle,
 283                            Channel::Mode mode,
 284                            Channel::Listener* listener,
 285                            base::MessageLoopProxy* ipc_thread)
 286     : context_(new Context(listener, ipc_thread)),
 287       outgoing_message_filter_(NULL) {
 288   Init(channel_handle, mode, ipc_thread, true);
 289 }
 290
 291 ChannelProxy::ChannelProxy(const IPC::ChannelHandle& channel_handle,
 292                            Channel::Mode mode,
 293                            base::MessageLoopProxy* ipc_thread,
 294                            Context* context,
 295                            bool create_pipe_now)
 296     : context_(context),
 297       outgoing_message_filter_(NULL) {
 298   Init(channel_handle, mode, ipc_thread, create_pipe_now);
 299 }
 300
 301 ChannelProxy::~ChannelProxy() {
 302   Close();
 303 }
 304
 305 void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle,
 306                         Channel::Mode mode,
 307                         base::MessageLoopProxy* ipc_thread_loop,
 308                         bool create_pipe_now) {
 309 #if defined(OS_POSIX)
 310   // When we are creating a server on POSIX, we need its file descriptor
 311   // to be created immediately so that it can be accessed and passed
 312   // to other processes. Forcing it to be created immediately avoids
 313   // race conditions that may otherwise arise.
 314   if (mode & Channel::MODE_SERVER_FLAG) {
 315     create_pipe_now = true;
 316   }
 317 #endif  // defined(OS_POSIX)
 318
 319   if (create_pipe_now) {
 320     // Create the channel immediately.  This effectively sets up the
 321     // low-level pipe so that the client can connect.  Without creating
 322     // the pipe immediately, it is possible for a listener to attempt
 323     // to connect and get an error since the pipe doesn't exist yet.
 324     context_->CreateChannel(channel_handle, mode);
 325   } else {
 326     context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 327         context_.get(), &Context::CreateChannel, channel_handle, mode));
 328   }
 329
 330   // complete initialization on the background thread
 331   context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 332       context_.get(), &Context::OnChannelOpened));
 333 }
 334
 335 void ChannelProxy::Close() {
 336   // Clear the backpointer to the listener so that any pending calls to
 337   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
 338   // possible that the channel could be closed while it is receiving messages!
 339   context_->Clear();
 340
 341   if (context_->ipc_message_loop()) {
 342     context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 343         context_.get(), &Context::OnChannelClosed));
 344   }
 345 }
 346
 347 bool ChannelProxy::Send(Message* message) {
 348   if (outgoing_message_filter())
 349     message = outgoing_message_filter()->Rewrite(message);
 350
 351 #ifdef IPC_MESSAGE_LOG_ENABLED
 352   Logging::GetInstance()->OnSendMessage(message, context_->channel_id());
 353 #endif
 354
 355   context_->ipc_message_loop()->PostTask(FROM_HERE,
 356                                          new SendTask(context_.get(), message));
 357   return true;
 358 }
 359
 360 void ChannelProxy::AddFilter(MessageFilter* filter) {
 361   context_->AddFilter(filter);
 362 }
 363
 364 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
 365   context_->ipc_message_loop()->PostTask(
 366       FROM_HERE, NewRunnableMethod(
 367           context_.get(),
 368           &Context::OnRemoveFilter,
 369           make_scoped_refptr(filter)));
 370 }
 371
 372 void ChannelProxy::ClearIPCMessageLoop() {
 373   context()->ClearIPCMessageLoop();
 374 }
 375
 376 #if defined(OS_POSIX) && !defined(OS_NACL)
 377 // See the TODO regarding lazy initialization of the channel in
 378 // ChannelProxy::Init().
 379 // We assume that IPC::Channel::GetClientFileDescriptorMapping() is thread-safe.
 380 int ChannelProxy::GetClientFileDescriptor() const {
 381   Channel *channel = context_.get()->channel_.get();
 382   // Channel must have been created first.
 383   DCHECK(channel) << context_.get()->channel_id_;
 384   return channel->GetClientFileDescriptor();
 385 }
 386
 387 bool ChannelProxy::GetClientEuid(uid_t* client_euid) const {
 388   Channel *channel = context_.get()->channel_.get();
 389   // Channel must have been created first.
 390   DCHECK(channel) << context_.get()->channel_id_;
 391   return channel->GetClientEuid(client_euid);
 392 }
 393 #endif
 394
 395 //-----------------------------------------------------------------------------
 396
 397 }  // namespace IPC