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