Add multi-process notification class.

chrome/common/multi_process_notification_mac.mm

+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/common/multi_process_notification.h"
+
+#import <Foundation/Foundation.h>
+#include <notify.h>
+#include <sys/select.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <algorithm>
+
+#include "base/basictypes.h"
+#include "base/eintr_wrapper.h"
+#include "base/file_path.h"
+#include "base/logging.h"
+#include "base/mac/mac_util.h"
+#include "base/mac/scoped_nsautorelease_pool.h"
+#include "base/message_loop_proxy.h"
+#include "base/message_pump_libevent.h"
+#include "base/path_service.h"
+#include "base/ref_counted.h"
+#include "base/stringprintf.h"
+#include "base/synchronization/lock.h"
+#include "base/sys_string_conversions.h"
+#include "base/sys_info.h"
+#include "base/threading/simple_thread.h"
+#include "chrome/browser/browser_thread.h"
+#include "chrome/common/chrome_paths.h"
+
+// Enable this to build with leopard_switchboard_thread
+#define USE_LEOPARD_SWITCHBOARD_THREAD 1

[Mark Mentovai, 2011/01/13 23:10:34]

Turn this back off.

[dmac, 2011/01/13 23:17:43]

Done.

+
+namespace {
+
+std::string AddPrefixToNotification(const std::string& name,
+                                    multi_process_notification::Domain domain) {
+  // The ordering of the components in the string returned by this function
+  // is important. Read "NAMESPACE CONVENTIONS" in 'man 3 notify' for details.
+  base::mac::ScopedNSAutoreleasePool pool;
+  NSBundle* bundle = base::mac::MainAppBundle();
+  NSString* ns_bundle_id = [bundle bundleIdentifier];
+  std::string bundle_id = base::SysNSStringToUTF8(ns_bundle_id);
+  std::string domain_string;
+  switch (domain) {
+    case multi_process_notification::ProfileDomain: {
+      FilePath user_data_dir;
+      if (!PathService::Get(chrome::DIR_USER_DATA, &user_data_dir)) {
+        NOTREACHED();
+      }
+      domain_string = StringPrintf("user.uid.%u.%s.",
+                                   getuid(), user_data_dir.value().c_str());
+      break;
+    }
+
+    case multi_process_notification::UserDomain:
+      domain_string = StringPrintf("user.uid.%u.", getuid());
+      break;
+
+    case multi_process_notification::SystemDomain:
+      break;
+  }
+  return domain_string + bundle_id + "." + name;
+}
+
+bool UseLeopardSwitchboardThread() {
+#if USE_LEOPARD_SWITCHBOARD_THREAD
+  return true;
+#endif  // USE_LEOPARD_SWITCHBOARD_THREAD
+  int32 major_version, minor_version, bugfix_version;
+  base::SysInfo::OperatingSystemVersionNumbers(
+      &major_version, &minor_version, &bugfix_version);
+  return major_version < 10 || (major_version == 10 && minor_version <= 5);
+}
+
+}  // namespace
+
+namespace multi_process_notification {
+
+bool Post(const std::string& name, Domain domain) {
+  std::string notification = AddPrefixToNotification(name, domain);
+  uint32_t status = notify_post(notification.c_str());
+  DCHECK_EQ(status, static_cast<uint32_t>(NOTIFY_STATUS_OK));
+  return status == NOTIFY_STATUS_OK;
+}
+
+#if MAC_OS_X_VERSION_MIN_REQUIRED > MAC_OS_X_VERSION_10_5
+#error LeopardSwitchboardThread can be removed
+#endif  // MAC_OS_X_VERSION_MIN_REQUIRED > MAC_OS_X_VERSION_10_5
+
+// LeopardSwitchboardThread exists because the file descriptors returned by
+// notify_register_file_descriptor can't be monitored using kqueues on 10.5
+// ( http://openradar.appspot.com/8854692 ) and libevent uses kqueue to watch
+// file descriptors in IOMessageLoop.
+// This solution is to have a separate thread that monitors the file descriptor
+// returned by notify_register_file_descriptor using select, and then to
+// notify the MessageLoopForIO using a different file descriptor allocated by
+// socketpair that can be monitored using kqueues in libevent. This thread
+// only runs on 10.5, as 10.6 kqueues can monitor the notify file descriptors
+// without any problems.
+
+// LeopardSwitchboardThread creates three file descriptors:
+// internal_fd_: which communicates from the switchboard thread to other threads
+// external_fd_: which communicates from other threads to the switchboard thread
+// notify_fd_: which is the file descriptor returned from
+//             notify_register_file_descriptor
+//
+// The thread itself sits in a select loop waiting on internal_fd_, and
+// notify_fd_ for input. If it gets ANY input on internal_fd_ it exits.
+// If it gets input on notify_fd_ it sends the input through to external_fd_.
+// External_fd_ is monitored by MessageLoopForIO so that the lookup of any
+// matching listeners in entries_, and the triggering of those listeners,
+// occurs in the MessageLoopForIO thread.
+//
+// Lookups are linear right now, and could be optimized if they ever become
+// a performance issue.
+class LeopardSwitchboardThread
+    : public base::MessagePumpLibevent::Watcher,
+      public base::SimpleThread,
+      public MessageLoop::DestructionObserver {
+ public:
+  LeopardSwitchboardThread();
+  virtual ~LeopardSwitchboardThread();
+
+  bool Init();
+
+  bool AddListener(ListenerImpl* listener,
+                   const std::string& notification);
+  bool RemoveListener(ListenerImpl* listener, const std::string& notification);
+
+  bool finished() const { return finished_; }
+
+  // SimpleThread overrides
+  virtual void Run();
+
+  // Watcher overrides
+  virtual void OnFileCanReadWithoutBlocking(int fd);
+  virtual void OnFileCanWriteWithoutBlocking(int fd);
+
+  // DestructionObserver overrides
+  virtual void WillDestroyCurrentMessageLoop();
+
+ private:
+  // User to match tokens to notifications and vice-versa.

[Mark Mentovai, 2011/01/13 23:10:34]

“User” should be “Used”.

[dmac, 2011/01/13 23:17:43]

Done.

+  struct SwitchboardEntry {
+    int token_;
+    std::string notification_;
+    ListenerImpl* listener_;
+  };
+
+  enum {
+    kKillThreadMessage = 0xdecea5e
+  };
+
+  int internal_fd_;
+  int external_fd_;
+  int notify_fd_;
+  int notify_fd_token_;
+  mutable bool finished_;
+  fd_set fd_set_;
+
+  // all accesses to entries_ must be controlled by entries_lock_.
+  std::vector<SwitchboardEntry> entries_;
+  Lock entries_lock_;
+  base::MessagePumpLibevent::FileDescriptorWatcher watcher_;
+};
+
+class ListenerImpl : public base::MessagePumpLibevent::Watcher {
+ public:
+  ListenerImpl(const std::string& name,
+               Domain domain,
+               Listener::Delegate* delegate);
+  virtual ~ListenerImpl();
+
+  bool Start();
+  void OnListen();
+
+  // Watcher overrides
+  virtual void OnFileCanReadWithoutBlocking(int fd);
+  virtual void OnFileCanWriteWithoutBlocking(int fd);
+
+ private:
+  std::string name_;
+  Domain domain_;
+  Listener::Delegate* delegate_;
+  int fd_;
+  int token_;
+  Lock switchboard_lock_;
+  static LeopardSwitchboardThread* g_switchboard_thread_;
+  base::MessagePumpLibevent::FileDescriptorWatcher watcher_;
+  scoped_refptr<base::MessageLoopProxy> message_loop_proxy_;
+
+  void StartLeopard();
+  void StartSnowLeopard();
+  DISALLOW_COPY_AND_ASSIGN(ListenerImpl);
+};
+
+
+LeopardSwitchboardThread::LeopardSwitchboardThread()
+    : base::SimpleThread("LeopardSwitchboardThread"), internal_fd_(-1),
+      external_fd_(-1), notify_fd_(-1), notify_fd_token_(-1), finished_(false) {
+}
+
+LeopardSwitchboardThread::~LeopardSwitchboardThread() {
+  if (internal_fd_ != -1) {
+    close(internal_fd_);
+  }
+  if (external_fd_ != -1) {
+    close(external_fd_);
+  }
+  if (notify_fd_token_ != -1) {
+    // Cancelling this notification takes care of closing notify_fd_.
+    uint32_t status = notify_cancel(notify_fd_token_);
+    DCHECK_EQ(status, static_cast<uint32_t>(NOTIFY_STATUS_OK));
+  }
+}
+
+bool LeopardSwitchboardThread::Init() {
+  // Create a pair of sockets for communicating with the thread
+  // The file descriptors returned from socketpair can be kqueue'd on 10.5.
+  int sockets[2];
+  if (socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) < 0) {
+    PLOG(ERROR) << "socketpair";
+    return false;
+  }
+  internal_fd_ = sockets[0];
+  external_fd_ = sockets[1];
+
+  // Register a bogus notification so that there is single notify_fd_ to
+  // monitor. This runs a small risk of overflowing the notification buffer
+  // if notifications are used heavily (see man 3 notify), however it greatly
+  // simplifies the select loop code as there are only 2 file descriptors
+  // that need to be monitored, and there is no need to add/remove file
+  // descriptors from fd_set_ as listeners are added and removed.
+  // This also keep the total fd usage on 10.5 to three for all
+  // notifications. The 10.6 implementation will use one fd per notification,
+  // but doesn't run the risk of notification buffer overflow. If fds ever
+  // become tight, the 10.6 code could be changed to use only one fd for
+  // all notifications.
+  std::string notification = StringPrintf("LeopardSwitchboardThread.%d",
+                                          getpid());
+  notification = AddPrefixToNotification(notification, ProfileDomain);
+  uint32_t status = notify_register_file_descriptor(
+      notification.c_str(), &notify_fd_, 0, &notify_fd_token_);
+  if (status != NOTIFY_STATUS_OK) {
+    return false;
+  }
+
+  FD_ZERO(&fd_set_);
+  FD_SET(internal_fd_, &fd_set_);
+  FD_SET(notify_fd_, &fd_set_);
+
+  MessageLoopForIO* io_loop = MessageLoopForIO::current();
+  // Watch for destruction of the BrowserThread::IO message loop so that
+  // the thread can be exited cleanly.

[Mark Mentovai, 2011/01/13 23:10:34]

the thread can be stopped cleanly. “Exit” sounds to me more like something that
an object would do itself, in which case you’d say “so that the thread can exit
cleanly,” but that’s not what happens. The observer will ask the thread to stop.

[dmac, 2011/01/13 23:17:43]

Done.

+  io_loop->AddDestructionObserver(this);
+  return io_loop->WatchFileDescriptor(
+      external_fd_, true, MessageLoopForIO::WATCH_READ, &watcher_, this);
+}
+
+void LeopardSwitchboardThread::WillDestroyCurrentMessageLoop() {
+  DCHECK_EQ(MessageLoop::current(), MessageLoopForIO::current());
+  watcher_.StopWatchingFileDescriptor();
+
+  // Send the appropriate message to end our thread, and then wait for it
+  // to finish before continuing.
+  int message = kKillThreadMessage;
+  write(external_fd_, &message, sizeof(message));
+  Join();
+}
+
+void LeopardSwitchboardThread::Run() {
+  DCHECK(!finished_);
+  int nfds = std::max(internal_fd_, notify_fd_) + 1;
+  while (1) {
+    fd_set working_set;
+    FD_COPY(&fd_set_, &working_set);
+    int count = HANDLE_EINTR(select(nfds, &working_set, NULL, NULL, NULL));
+    if (count < 0) {
+      PLOG(ERROR) << "select";
+      break;
+    } else if (count == 0) {
+      DLOG(INFO) << "select timed out";
+      continue;
+    }
+    if (FD_ISSET(notify_fd_, &working_set)) {
+      int token;
+      int status = HANDLE_EINTR(read(notify_fd_, &token, sizeof(token)));
+      if (status < 0) {
+        PLOG(ERROR) << "read";
+        break;
+      } else if (status == 0) {
+        LOG(ERROR) << "notify fd closed";
+        break;
+      } else if (status != sizeof(token)) {
+        LOG(ERROR) << "read wrong size: " << status;
+        break;
+      } else if (token == notify_fd_token_) {
+        LOG(ERROR) << "invalid token: " << token;
+      }
+      status = HANDLE_EINTR(write(internal_fd_, &token, sizeof(token)));
+      if (status < 0) {
+        PLOG(ERROR) << "write";
+        break;
+      } else if (status == 0) {
+        LOG(ERROR) << "external_fd_ closed";
+        break;
+      } else if (status != sizeof(token)) {
+        LOG(ERROR) << "write wrong size: " << status;
+        break;
+      }
+    }
+    if (FD_ISSET(internal_fd_, &working_set)) {
+      int value;
+      int status = HANDLE_EINTR(read(internal_fd_, &value, sizeof(value)));
+      if (status < 0) {
+        PLOG(ERROR) << "read";
+      } else if (status == 0) {
+        LOG(ERROR) << "internal_fd_ closed";
+      } else if (value != kKillThreadMessage) {
+        LOG(ERROR) << "unknown message sent: " << value;
+      }
+      break;
+    }
+  }
+  finished_ = true;
+}
+
+bool LeopardSwitchboardThread::AddListener(ListenerImpl* listener,
+                                           const std::string& notification) {
+  DCHECK(!finished());
+  base::AutoLock autolock(entries_lock_);
+  for (std::vector<SwitchboardEntry>::iterator i = entries_.begin();
+       i < entries_.end(); ++i) {
+    if (i->listener_ == listener && i->notification_ == notification) {
+      LOG(ERROR) << "listener " << listener
+                 << " already registered for '" << notification << "'";
+      return false;
+    }
+  }
+  int token;
+  uint32_t status = notify_register_file_descriptor(
+     notification.c_str(), &notify_fd_, NOTIFY_REUSE, &token);
+  if (status != NOTIFY_STATUS_OK) {
+    LOG(ERROR) << "unable to notify_register_file_descriptor for '"
+               << notification << "' status: " << status;
+    return false;
+  }
+  SwitchboardEntry entry;
+  entry.token_ = token;
+  entry.notification_ = notification;
+  entry.listener_ = listener;
+  entries_.push_back(entry);
+  return true;
+}
+
+bool LeopardSwitchboardThread::RemoveListener(ListenerImpl* listener,
+                                              const std::string& notification) {
+  DCHECK(!finished());
+  base::AutoLock autolock(entries_lock_);
+  for (std::vector<SwitchboardEntry>::iterator i = entries_.begin();
+       i < entries_.end(); ++i) {
+    if (i->listener_ == listener && i->notification_ == notification) {
+      uint32_t status = notify_cancel(i->token_);
+      DCHECK_EQ(status, static_cast<uint32_t>(NOTIFY_STATUS_OK));
+      entries_.erase(i);
+      return true;
+    }
+  }
+  LOG(ERROR) << "unable to remove listener '" << listener
+             << "' for '" << notification << "'.";
+  return false;
+}
+
+void LeopardSwitchboardThread::OnFileCanReadWithoutBlocking(int fd) {
+  DCHECK_EQ(MessageLoop::current(), MessageLoopForIO::current());
+  DCHECK_EQ(fd, external_fd_);
+  int token;
+  int status = HANDLE_EINTR(read(fd, &token, sizeof(token)));
+  if (status < 0) {
+    PLOG(ERROR) << "read";
+  } else if (status == 0) {
+    LOG(ERROR) << "external_fd_ closed";
+  } else if (status != sizeof(token)) {
+    LOG(ERROR) << "unexpected read size " << status;
+  } else {
+    // Have to swap to native endianness <http://openradar.appspot.com/8821081>.
+    token = static_cast<int>(ntohl(token));
+    base::AutoLock autolock(entries_lock_);
+    bool found_token = false;
+    for (std::vector<SwitchboardEntry>::iterator i = entries_.begin();
+         i < entries_.end(); ++i) {
+      if (i->token_ == token) {
+        found_token = true;
+        i->listener_->OnListen();
+      }
+    }
+    if (!found_token) {
+      LOG(ERROR) << "read unknown token " << token;
+    }
+  }
+}
+
+void LeopardSwitchboardThread::OnFileCanWriteWithoutBlocking(int fd) {
+  NOTREACHED();
+}
+
+LeopardSwitchboardThread* ListenerImpl::g_switchboard_thread_ = NULL;
+
+ListenerImpl::ListenerImpl(
+    const std::string& name, Domain domain, Listener::Delegate* delegate)
+    : name_(name), domain_(domain), delegate_(delegate), fd_(-1), token_(-1) {
+}
+
+ListenerImpl::~ListenerImpl() {
+  if (!UseLeopardSwitchboardThread()) {
+    if (token_ != -1) {
+      uint32_t status = notify_cancel(token_);
+      DCHECK_EQ(status, static_cast<uint32_t>(NOTIFY_STATUS_OK));
+      token_ = -1;
+    }
+  } else {
+    base::AutoLock autolock(switchboard_lock_);
+    if (g_switchboard_thread_) {
+      std::string notification = AddPrefixToNotification(name_, domain_);
+      CHECK(g_switchboard_thread_->RemoveListener(this, notification));
+    }
+  }
+}
+
+bool ListenerImpl::Start() {
+  DCHECK_EQ(fd_, -1);
+  DCHECK_EQ(token_, -1);
+  message_loop_proxy_ = base::MessageLoopProxy::CreateForCurrentThread();
+  Task* task;
+  if(UseLeopardSwitchboardThread()) {
+    task = NewRunnableMethod(this, &ListenerImpl::StartLeopard);
+  } else {
+    task = NewRunnableMethod(this, &ListenerImpl::StartSnowLeopard);
+  }
+  return BrowserThread::PostTask(BrowserThread::IO, FROM_HERE, task);
+}
+
+void ListenerImpl::StartLeopard() {
+  DCHECK(UseLeopardSwitchboardThread());
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  bool success = true;
+  {
+    base::AutoLock autolock(switchboard_lock_);
+    if (g_switchboard_thread_ && g_switchboard_thread_->HasBeenJoined()) {
+      delete g_switchboard_thread_;
+      g_switchboard_thread_ = NULL;
+    }
+    DCHECK(!(g_switchboard_thread_ && g_switchboard_thread_->finished()));
+    if (!g_switchboard_thread_) {
+      g_switchboard_thread_ = new LeopardSwitchboardThread();
+      success = g_switchboard_thread_->Init();
+      if (success) {
+        g_switchboard_thread_->Start();
+      }
+    }
+    if (success) {
+      std::string notification = AddPrefixToNotification(name_, domain_);
+      success = g_switchboard_thread_->AddListener(this, notification);
+    }
+  }
+  Task* task =
+      new Listener::ListenerStartedTask(name_, domain_, delegate_, success);
+  CHECK(message_loop_proxy_->PostTask(FROM_HERE, task));
+}
+
+void ListenerImpl::StartSnowLeopard() {
+  DCHECK(!UseLeopardSwitchboardThread());
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  bool success = true;
+  std::string notification = AddPrefixToNotification(name_, domain_);
+  uint32_t status = notify_register_file_descriptor(
+      notification.c_str(), &fd_, 0, &token_);
+  if (status != NOTIFY_STATUS_OK) {
+    LOG(ERROR) << "unable to notify_register_file_descriptor for '"
+               << notification << "' Status: " << status;
+    success = false;
+  }
+  if (success) {
+    MessageLoopForIO* io_loop = MessageLoopForIO::current();
+    success = io_loop->WatchFileDescriptor(
+        fd_, true, MessageLoopForIO::WATCH_READ, &watcher_, this);
+  }
+  Task* task =
+      new Listener::ListenerStartedTask(name_, domain_, delegate_, success);
+  CHECK(message_loop_proxy_->PostTask(FROM_HERE, task));
+}
+
+void ListenerImpl::OnFileCanReadWithoutBlocking(int fd) {
+  DCHECK(!UseLeopardSwitchboardThread());
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK_EQ(fd, fd_);
+  int token;
+  int status = HANDLE_EINTR(read(fd, &token, sizeof(token)));
+  if (status < 0) {
+    PLOG(ERROR) << "read";
+  } else if (status == 0) {
+    LOG(ERROR) << "external_fd_ closed";
+  } else if (status != sizeof(token)) {
+    LOG(ERROR) << "unexpected read size " << status;
+  } else {
+    // Have to swap to native endianness <http://openradar.appspot.com/8821081>.
+    token = static_cast<int>(ntohl(token));
+    if (token == token_) {
+      OnListen();
+    } else {
+      LOG(ERROR) << "unexpected value " << token;
+    }
+  }
+}
+
+void ListenerImpl::OnListen() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  Task* task =
+      new Listener::NotificationReceivedTask(name_, domain_, delegate_);
+  CHECK(message_loop_proxy_->PostTask(FROM_HERE, task));
+}
+
+void ListenerImpl::OnFileCanWriteWithoutBlocking(int fd) {
+  NOTREACHED();
+}
+
+Listener::Listener(
+    const std::string& name, Domain domain, Listener::Delegate* delegate)
+    : impl_(new ListenerImpl(name, domain, delegate)) {
+}
+
+Listener::~Listener() {
+}
+
+bool Listener::Start() {
+  return impl_->Start();
+}
+
+}  // namespace multi_process_notification
+
+DISABLE_RUNNABLE_METHOD_REFCOUNT(multi_process_notification::ListenerImpl);
+

[Mark Mentovai, 2011/01/13 23:10:34]

Dump the blank line at EOF.

[dmac, 2011/01/13 23:17:43]

Done.