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