ipc/ipc_channel_proxy.cc

   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 #include "ipc/ipc_channel_proxy.h"
   6
   7 #include "base/bind.h"
   8 #include "base/compiler_specific.h"
   9 #include "base/location.h"
  10 #include "base/memory/ref_counted.h"
  11 #include "base/memory/scoped_ptr.h"
  12 #include "base/single_thread_task_runner.h"
  13 #include "base/thread_task_runner_handle.h"
  14 #include "ipc/ipc_channel_factory.h"
  15 #include "ipc/ipc_listener.h"
  16 #include "ipc/ipc_logging.h"
  17 #include "ipc/ipc_message_macros.h"
  18 #include "ipc/message_filter.h"
  19 #include "ipc/message_filter_router.h"
  20
  21 namespace IPC {
  22
  23 //------------------------------------------------------------------------------
  24
  25 ChannelProxy::Context::Context(
  26     Listener* listener,
  27     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
  28     : listener_task_runner_(base::ThreadTaskRunnerHandle::Get()),
  29       listener_(listener),
  30       ipc_task_runner_(ipc_task_runner),
  31       channel_connected_called_(false),
  32       message_filter_router_(new MessageFilterRouter()),
  33       peer_pid_(base::kNullProcessId) {
  34   DCHECK(ipc_task_runner_.get());
  35   // The Listener thread where Messages are handled must be a separate thread
  36   // to avoid oversubscribing the IO thread. If you trigger this error, you
  37   // need to either:
  38   // 1) Create the ChannelProxy on a different thread, or
  39   // 2) Just use Channel
  40   // Note, we currently make an exception for a NULL listener. That usage
  41   // basically works, but is outside the intent of ChannelProxy. This support
  42   // will disappear, so please don't rely on it. See crbug.com/364241
  43   DCHECK(!listener || (ipc_task_runner_.get() != listener_task_runner_.get()));
  44 }
  45
  46 ChannelProxy::Context::~Context() {
  47 }
  48
  49 void ChannelProxy::Context::ClearIPCTaskRunner() {
  50   ipc_task_runner_ = NULL;
  51 }
  52
  53 void ChannelProxy::Context::SetListenerTaskRunner(
  54     scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
  55   DCHECK(ipc_task_runner_.get() != task_runner.get());
  56   DCHECK(listener_task_runner_->BelongsToCurrentThread());
  57   DCHECK(task_runner->BelongsToCurrentThread());
  58   listener_task_runner_ = task_runner;
  59 }
  60
  61 void ChannelProxy::Context::CreateChannel(scoped_ptr<ChannelFactory> factory) {
  62   DCHECK(!channel_);
  63   channel_id_ = factory->GetName();
  64   channel_ = factory->BuildChannel(this);
  65 }
  66
  67 bool ChannelProxy::Context::TryFilters(const Message& message) {
  68   DCHECK(message_filter_router_);
  69 #ifdef IPC_MESSAGE_LOG_ENABLED
  70   Logging* logger = Logging::GetInstance();
  71   if (logger->Enabled())
  72     logger->OnPreDispatchMessage(message);
  73 #endif
  74
  75   if (message_filter_router_->TryFilters(message)) {
  76     if (message.dispatch_error()) {
  77       listener_task_runner_->PostTask(
  78           FROM_HERE, base::Bind(&Context::OnDispatchBadMessage, this, message));
  79     }
  80 #ifdef IPC_MESSAGE_LOG_ENABLED
  81     if (logger->Enabled())
  82       logger->OnPostDispatchMessage(message, channel_id_);
  83 #endif
  84     return true;
  85   }
  86   return false;
  87 }
  88
  89 // Called on the IPC::Channel thread
  90 bool ChannelProxy::Context::OnMessageReceived(const Message& message) {
  91   // First give a chance to the filters to process this message.
  92   if (!TryFilters(message))
  93     OnMessageReceivedNoFilter(message);
  94   return true;
  95 }
  96
  97 // Called on the IPC::Channel thread
  98 bool ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
  99   listener_task_runner_->PostTask(
 100       FROM_HERE, base::Bind(&Context::OnDispatchMessage, this, message));
 101   return true;
 102 }
 103
 104 // Called on the IPC::Channel thread
 105 void ChannelProxy::Context::OnChannelConnected(int32 peer_pid) {
 106   // We cache off the peer_pid so it can be safely accessed from both threads.
 107   peer_pid_ = channel_->GetPeerPID();
 108
 109   // Add any pending filters.  This avoids a race condition where someone
 110   // creates a ChannelProxy, calls AddFilter, and then right after starts the
 111   // peer process.  The IO thread could receive a message before the task to add
 112   // the filter is run on the IO thread.
 113   OnAddFilter();
 114
 115   // See above comment about using listener_task_runner_ here.
 116   listener_task_runner_->PostTask(
 117       FROM_HERE, base::Bind(&Context::OnDispatchConnected, this));
 118 }
 119
 120 // Called on the IPC::Channel thread
 121 void ChannelProxy::Context::OnChannelError() {
 122   for (size_t i = 0; i < filters_.size(); ++i)
 123     filters_[i]->OnChannelError();
 124
 125   // See above comment about using listener_task_runner_ here.
 126   listener_task_runner_->PostTask(
 127       FROM_HERE, base::Bind(&Context::OnDispatchError, this));
 128 }
 129
 130 // Called on the IPC::Channel thread
 131 void ChannelProxy::Context::OnChannelOpened() {
 132   DCHECK(channel_ != NULL);
 133
 134   // Assume a reference to ourselves on behalf of this thread.  This reference
 135   // will be released when we are closed.
 136   AddRef();
 137
 138   if (!channel_->Connect()) {
 139     OnChannelError();
 140     return;
 141   }
 142
 143   for (size_t i = 0; i < filters_.size(); ++i)
 144     filters_[i]->OnFilterAdded(channel_.get());
 145 }
 146
 147 // Called on the IPC::Channel thread
 148 void ChannelProxy::Context::OnChannelClosed() {
 149   // It's okay for IPC::ChannelProxy::Close to be called more than once, which
 150   // would result in this branch being taken.
 151   if (!channel_)
 152     return;
 153
 154   for (size_t i = 0; i < filters_.size(); ++i) {
 155     filters_[i]->OnChannelClosing();
 156     filters_[i]->OnFilterRemoved();
 157   }
 158
 159   // We don't need the filters anymore.
 160   message_filter_router_->Clear();
 161   filters_.clear();
 162   // We don't need the lock, because at this point, the listener thread can't
 163   // access it any more.
 164   pending_filters_.clear();
 165
 166   channel_.reset();
 167
 168   // Balance with the reference taken during startup.  This may result in
 169   // self-destruction.
 170   Release();
 171 }
 172
 173 void ChannelProxy::Context::Clear() {
 174   listener_ = NULL;
 175 }
 176
 177 // Called on the IPC::Channel thread
 178 void ChannelProxy::Context::OnSendMessage(scoped_ptr<Message> message) {
 179   if (!channel_) {
 180     OnChannelClosed();
 181     return;
 182   }
 183
 184   if (!channel_->Send(message.release()))
 185     OnChannelError();
 186 }
 187
 188 // Called on the IPC::Channel thread
 189 void ChannelProxy::Context::OnAddFilter() {
 190   // Our OnChannelConnected method has not yet been called, so we can't be
 191   // sure that channel_ is valid yet. When OnChannelConnected *is* called,
 192   // it invokes OnAddFilter, so any pending filter(s) will be added at that
 193   // time.
 194   if (peer_pid_ == base::kNullProcessId)
 195     return;
 196
 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     message_filter_router_->AddFilter(new_filters[i].get());
 207
 208     // The channel has already been created and connected, so we need to
 209     // inform the filters right now.
 210     new_filters[i]->OnFilterAdded(channel_.get());
 211     new_filters[i]->OnChannelConnected(peer_pid_);
 212   }
 213 }
 214
 215 // Called on the IPC::Channel thread
 216 void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
 217   if (peer_pid_ == base::kNullProcessId) {
 218     // The channel is not yet connected, so any filters are still pending.
 219     base::AutoLock auto_lock(pending_filters_lock_);
 220     for (size_t i = 0; i < pending_filters_.size(); ++i) {
 221       if (pending_filters_[i].get() == filter) {
 222         filter->OnFilterRemoved();
 223         pending_filters_.erase(pending_filters_.begin() + i);
 224         return;
 225       }
 226     }
 227     return;
 228   }
 229   if (!channel_)
 230     return;  // The filters have already been deleted.
 231
 232   message_filter_router_->RemoveFilter(filter);
 233
 234   for (size_t i = 0; i < filters_.size(); ++i) {
 235     if (filters_[i].get() == filter) {
 236       filter->OnFilterRemoved();
 237       filters_.erase(filters_.begin() + i);
 238       return;
 239     }
 240   }
 241
 242   NOTREACHED() << "filter to be removed not found";
 243 }
 244
 245 // Called on the listener's thread
 246 void ChannelProxy::Context::AddFilter(MessageFilter* filter) {
 247   base::AutoLock auto_lock(pending_filters_lock_);
 248   pending_filters_.push_back(make_scoped_refptr(filter));
 249   ipc_task_runner_->PostTask(
 250       FROM_HERE, base::Bind(&Context::OnAddFilter, this));
 251 }
 252
 253 // Called on the listener's thread
 254 void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
 255 #if defined(IPC_MESSAGE_LOG_ENABLED)
 256   Logging* logger = Logging::GetInstance();
 257   std::string name;
 258   logger->GetMessageText(message.type(), &name, &message, NULL);
 259   TRACE_EVENT1("ipc", "ChannelProxy::Context::OnDispatchMessage",
 260                "name", name);
 261 #else
 262   TRACE_EVENT2("ipc", "ChannelProxy::Context::OnDispatchMessage",
 263                "class", IPC_MESSAGE_ID_CLASS(message.type()),
 264                "line", IPC_MESSAGE_ID_LINE(message.type()));
 265 #endif
 266
 267   if (!listener_)
 268     return;
 269
 270   OnDispatchConnected();
 271
 272 #ifdef IPC_MESSAGE_LOG_ENABLED
 273   if (message.type() == IPC_LOGGING_ID) {
 274     logger->OnReceivedLoggingMessage(message);
 275     return;
 276   }
 277
 278   if (logger->Enabled())
 279     logger->OnPreDispatchMessage(message);
 280 #endif
 281
 282   listener_->OnMessageReceived(message);
 283   if (message.dispatch_error())
 284     listener_->OnBadMessageReceived(message);
 285
 286 #ifdef IPC_MESSAGE_LOG_ENABLED
 287   if (logger->Enabled())
 288     logger->OnPostDispatchMessage(message, channel_id_);
 289 #endif
 290 }
 291
 292 // Called on the listener's thread
 293 void ChannelProxy::Context::OnDispatchConnected() {
 294   if (channel_connected_called_)
 295     return;
 296
 297   channel_connected_called_ = true;
 298   if (listener_)
 299     listener_->OnChannelConnected(peer_pid_);
 300 }
 301
 302 // Called on the listener's thread
 303 void ChannelProxy::Context::OnDispatchError() {
 304   if (listener_)
 305     listener_->OnChannelError();
 306 }
 307
 308 // Called on the listener's thread
 309 void ChannelProxy::Context::OnDispatchBadMessage(const Message& message) {
 310   if (listener_)
 311     listener_->OnBadMessageReceived(message);
 312 }
 313
 314 //-----------------------------------------------------------------------------
 315
 316 // static
 317 scoped_ptr<ChannelProxy> ChannelProxy::Create(
 318     const IPC::ChannelHandle& channel_handle,
 319     Channel::Mode mode,
 320     Listener* listener,
 321     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
 322   scoped_ptr<ChannelProxy> channel(new ChannelProxy(listener, ipc_task_runner));
 323   channel->Init(channel_handle, mode, true);
 324   return channel.Pass();
 325 }
 326
 327 // static
 328 scoped_ptr<ChannelProxy> ChannelProxy::Create(
 329     scoped_ptr<ChannelFactory> factory,
 330     Listener* listener,
 331     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner) {
 332   scoped_ptr<ChannelProxy> channel(new ChannelProxy(listener, ipc_task_runner));
 333   channel->Init(factory.Pass(), true);
 334   return channel.Pass();
 335 }
 336
 337 ChannelProxy::ChannelProxy(Context* context)
 338     : context_(context),
 339       did_init_(false) {
 340 #if defined(ENABLE_IPC_FUZZER)
 341   outgoing_message_filter_ = NULL;
 342 #endif
 343 }
 344
 345 ChannelProxy::ChannelProxy(
 346     Listener* listener,
 347     const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner)
 348     : context_(new Context(listener, ipc_task_runner)), did_init_(false) {
 349 #if defined(ENABLE_IPC_FUZZER)
 350   outgoing_message_filter_ = NULL;
 351 #endif
 352 }
 353
 354 ChannelProxy::~ChannelProxy() {
 355   DCHECK(CalledOnValidThread());
 356
 357   Close();
 358 }
 359
 360 void ChannelProxy::Init(const IPC::ChannelHandle& channel_handle,
 361                         Channel::Mode mode,
 362                         bool create_pipe_now) {
 363 #if defined(OS_POSIX)
 364   // When we are creating a server on POSIX, we need its file descriptor
 365   // to be created immediately so that it can be accessed and passed
 366   // to other processes. Forcing it to be created immediately avoids
 367   // race conditions that may otherwise arise.
 368   if (mode & Channel::MODE_SERVER_FLAG) {
 369     create_pipe_now = true;
 370   }
 371 #endif  // defined(OS_POSIX)
 372   Init(ChannelFactory::Create(channel_handle, mode),
 373        create_pipe_now);
 374 }
 375
 376 void ChannelProxy::Init(scoped_ptr<ChannelFactory> factory,
 377                         bool create_pipe_now) {
 378   DCHECK(CalledOnValidThread());
 379   DCHECK(!did_init_);
 380
 381   if (create_pipe_now) {
 382     // Create the channel immediately.  This effectively sets up the
 383     // low-level pipe so that the client can connect.  Without creating
 384     // the pipe immediately, it is possible for a listener to attempt
 385     // to connect and get an error since the pipe doesn't exist yet.
 386     context_->CreateChannel(factory.Pass());
 387   } else {
 388     context_->ipc_task_runner()->PostTask(
 389         FROM_HERE, base::Bind(&Context::CreateChannel,
 390                               context_.get(), Passed(factory.Pass())));
 391   }
 392
 393   // complete initialization on the background thread
 394   context_->ipc_task_runner()->PostTask(
 395       FROM_HERE, base::Bind(&Context::OnChannelOpened, context_.get()));
 396
 397   did_init_ = true;
 398 }
 399
 400 void ChannelProxy::Close() {
 401   DCHECK(CalledOnValidThread());
 402
 403   // Clear the backpointer to the listener so that any pending calls to
 404   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
 405   // possible that the channel could be closed while it is receiving messages!
 406   context_->Clear();
 407
 408   if (context_->ipc_task_runner()) {
 409     context_->ipc_task_runner()->PostTask(
 410         FROM_HERE, base::Bind(&Context::OnChannelClosed, context_.get()));
 411   }
 412 }
 413
 414 bool ChannelProxy::Send(Message* message) {
 415   DCHECK(did_init_);
 416
 417   // TODO(alexeypa): add DCHECK(CalledOnValidThread()) here. Currently there are
 418   // tests that call Send() from a wrong thread. See http://crbug.com/163523.
 419
 420 #ifdef ENABLE_IPC_FUZZER
 421   // In IPC fuzzing builds, it is possible to define a filter to apply to
 422   // outgoing messages. It will either rewrite the message and return a new
 423   // one, freeing the original, or return the message unchanged.
 424   if (outgoing_message_filter())
 425     message = outgoing_message_filter()->Rewrite(message);
 426 #endif
 427
 428 #ifdef IPC_MESSAGE_LOG_ENABLED
 429   Logging::GetInstance()->OnSendMessage(message, context_->channel_id());
 430 #endif
 431
 432   context_->ipc_task_runner()->PostTask(
 433       FROM_HERE,
 434       base::Bind(&ChannelProxy::Context::OnSendMessage,
 435                  context_, base::Passed(scoped_ptr<Message>(message))));
 436   return true;
 437 }
 438
 439 void ChannelProxy::AddFilter(MessageFilter* filter) {
 440   DCHECK(CalledOnValidThread());
 441
 442   context_->AddFilter(filter);
 443 }
 444
 445 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
 446   DCHECK(CalledOnValidThread());
 447
 448   context_->ipc_task_runner()->PostTask(
 449       FROM_HERE, base::Bind(&Context::OnRemoveFilter, context_.get(),
 450                             make_scoped_refptr(filter)));
 451 }
 452
 453 void ChannelProxy::SetListenerTaskRunner(
 454     scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
 455   DCHECK(CalledOnValidThread());
 456
 457   context()->SetListenerTaskRunner(task_runner);
 458 }
 459
 460 void ChannelProxy::ClearIPCTaskRunner() {
 461   DCHECK(CalledOnValidThread());
 462
 463   context()->ClearIPCTaskRunner();
 464 }
 465
 466 #if defined(OS_POSIX) && !defined(OS_NACL_SFI)
 467 // See the TODO regarding lazy initialization of the channel in
 468 // ChannelProxy::Init().
 469 int ChannelProxy::GetClientFileDescriptor() {
 470   DCHECK(CalledOnValidThread());
 471
 472   Channel* channel = context_.get()->channel_.get();
 473   // Channel must have been created first.
 474   DCHECK(channel) << context_.get()->channel_id_;
 475   return channel->GetClientFileDescriptor();
 476 }
 477
 478 base::ScopedFD ChannelProxy::TakeClientFileDescriptor() {
 479   DCHECK(CalledOnValidThread());
 480
 481   Channel* channel = context_.get()->channel_.get();
 482   // Channel must have been created first.
 483   DCHECK(channel) << context_.get()->channel_id_;
 484   return channel->TakeClientFileDescriptor();
 485 }
 486 #endif
 487
 488 //-----------------------------------------------------------------------------
 489
 490 }  // namespace IPC