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