Commit fc0cd893 authored by Sean Cannella's avatar Sean Cannella Committed by dcsommer

Support NotificationQueue on iOS/Android

Summary:
This is the last of the changes to make folly and thrift portable. The old
thrift SpinLock implementation is exactly we
want here so port this into folly. If someone later makes efficeint
versions of MicroSpinLock for platforms other than x64 they can easily
be plugged in without changing the calling code.

Test Plan: compiled on multiple platforms

Reviewed By: alandau@fb.com

Subscribers: folly-diffs@, ldbrandy, trunkagent, alandau, bmatheny, njormrod, mshneer, shikong, benyluo, kmdent, fma, ranjeeth, subodh

FB internal diff: D1621717

Tasks: 5183325
parent acc54589
/*
* Copyright 2014 Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once
#include <boost/noncopyable.hpp>
// This is a wrapper SpinLock implementation that works around the
// x64 limitation of the base folly MicroSpinLock. If that is available, this
// simply thinly wraps it. Otherwise, it uses the simplest analog available on
// iOS (or 32-bit Mac) or, failing that, POSIX (on Android et. al.)
#if __x86_64__
#include <folly/SmallLocks.h>
namespace folly { namespace io {
class PortableSpinLock {
public:
FOLLY_ALWAYS_INLINE PortableSpinLock() {
lock_.init();
}
FOLLY_ALWAYS_INLINE void lock() const {
lock_.lock();
}
FOLLY_ALWAYS_INLINE void unlock() const {
lock_.unlock();
}
FOLLY_ALWAYS_INLINE bool trylock() const {
return lock_.try_lock();
}
private:
mutable folly::MicroSpinLock lock_;
};
}}
#elif __APPLE__
#include <libkern/OSAtomic.h>
namespace folly { namespace io {
class PortableSpinLock {
public:
FOLLY_ALWAYS_INLINE PortableSpinLock() : lock_(0) {}
FOLLY_ALWAYS_INLINE void lock() const {
OSSpinLockLock(&lock_);
}
FOLLY_ALWAYS_INLINE void unlock() const {
OSSpinLockUnlock(&lock_);
}
FOLLY_ALWAYS_INLINE bool trylock() const {
return OSSpinLockTry(&lock_);
}
private:
mutable OSSpinLock lock_;
};
}}
#else
#include <pthread.h>
namespace folly { namespace io {
class PortableSpinLock {
public:
FOLLY_ALWAYS_INLINE PortableSpinLock() {
pthread_mutex_init(&lock_, nullptr);
}
void lock() const {
int rc = pthread_mutex_lock(&lock_);
CHECK_EQ(0, rc);
}
FOLLY_ALWAYS_INLINE void unlock() const {
int rc = pthread_mutex_unlock(&lock_);
CHECK_EQ(0, rc);
}
FOLLY_ALWAYS_INLINE bool trylock() const {
int rc = pthread_mutex_trylock(&lock_);
CHECK_GE(rc, 0);
return rc == 0;
}
private:
mutable pthread_mutex_t lock_;
};
}}
#endif
namespace folly { namespace io {
class PortableSpinLockGuard : private boost::noncopyable {
public:
FOLLY_ALWAYS_INLINE explicit PortableSpinLockGuard(PortableSpinLock& lock) :
lock_(lock) {
lock_.lock();
}
FOLLY_ALWAYS_INLINE ~PortableSpinLockGuard() {
lock_.unlock();
}
private:
PortableSpinLock& lock_;
};
}}
...@@ -19,17 +19,21 @@ ...@@ -19,17 +19,21 @@
#include <fcntl.h> #include <fcntl.h>
#include <unistd.h> #include <unistd.h>
#include <folly/io/PortableSpinLock.h>
#include <folly/io/async/EventBase.h> #include <folly/io/async/EventBase.h>
#include <folly/io/async/EventFDWrapper.h>
#include <folly/io/async/EventHandler.h> #include <folly/io/async/EventHandler.h>
#include <folly/io/async/Request.h> #include <folly/io/async/Request.h>
#include <folly/Likely.h> #include <folly/Likely.h>
#include <folly/SmallLocks.h>
#include <folly/ScopeGuard.h> #include <folly/ScopeGuard.h>
#include <glog/logging.h> #include <glog/logging.h>
#include <deque> #include <deque>
#if __linux__ && !__ANDROID__
#define FOLLY_HAVE_EVENTFD
#include <folly/io/async/EventFDWrapper.h>
#endif
namespace folly { namespace folly {
/** /**
...@@ -169,8 +173,10 @@ class NotificationQueue { ...@@ -169,8 +173,10 @@ class NotificationQueue {
}; };
enum class FdType { enum class FdType {
PIPE,
#ifdef FOLLY_HAVE_EVENTFD
EVENTFD, EVENTFD,
PIPE #endif
}; };
/** /**
...@@ -189,17 +195,20 @@ class NotificationQueue { ...@@ -189,17 +195,20 @@ class NotificationQueue {
* mostly for testing purposes. * mostly for testing purposes.
*/ */
explicit NotificationQueue(uint32_t maxSize = 0, explicit NotificationQueue(uint32_t maxSize = 0,
#ifdef FOLLY_HAVE_EVENTFD
FdType fdType = FdType::EVENTFD) FdType fdType = FdType::EVENTFD)
#else
FdType fdType = FdType::PIPE)
#endif
: eventfd_(-1), : eventfd_(-1),
pipeFds_{-1, -1}, pipeFds_{-1, -1},
advisoryMaxQueueSize_(maxSize), advisoryMaxQueueSize_(maxSize),
pid_(getpid()), pid_(getpid()),
queue_() { queue_() {
spinlock_.init();
RequestContext::getStaticContext(); RequestContext::getStaticContext();
#ifdef FOLLY_HAVE_EVENTFD
if (fdType == FdType::EVENTFD) { if (fdType == FdType::EVENTFD) {
eventfd_ = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK | EFD_SEMAPHORE); eventfd_ = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK | EFD_SEMAPHORE);
if (eventfd_ == -1) { if (eventfd_ == -1) {
...@@ -216,6 +225,7 @@ class NotificationQueue { ...@@ -216,6 +225,7 @@ class NotificationQueue {
} }
} }
} }
#endif
if (fdType == FdType::PIPE) { if (fdType == FdType::PIPE) {
if (pipe(pipeFds_)) { if (pipe(pipeFds_)) {
folly::throwSystemError("Failed to create pipe for NotificationQueue", folly::throwSystemError("Failed to create pipe for NotificationQueue",
...@@ -342,7 +352,7 @@ class NotificationQueue { ...@@ -342,7 +352,7 @@ class NotificationQueue {
try { try {
folly::MSLGuard g(spinlock_); folly::io::PortableSpinLockGuard g(spinlock_);
if (UNLIKELY(queue_.empty())) { if (UNLIKELY(queue_.empty())) {
return false; return false;
...@@ -366,7 +376,7 @@ class NotificationQueue { ...@@ -366,7 +376,7 @@ class NotificationQueue {
} }
int size() { int size() {
folly::MSLGuard g(spinlock_); folly::io::PortableSpinLockGuard g(spinlock_);
return queue_.size(); return queue_.size();
} }
...@@ -393,7 +403,7 @@ class NotificationQueue { ...@@ -393,7 +403,7 @@ class NotificationQueue {
NotificationQueue& operator=(NotificationQueue const &) = delete; NotificationQueue& operator=(NotificationQueue const &) = delete;
inline bool checkQueueSize(size_t maxSize, bool throws=true) const { inline bool checkQueueSize(size_t maxSize, bool throws=true) const {
DCHECK(0 == spinlock_.try_lock()); DCHECK(0 == spinlock_.trylock());
if (maxSize > 0 && queue_.size() >= maxSize) { if (maxSize > 0 && queue_.size() >= maxSize) {
if (throws) { if (throws) {
throw std::overflow_error("unable to add message to NotificationQueue: " throw std::overflow_error("unable to add message to NotificationQueue: "
...@@ -461,7 +471,7 @@ class NotificationQueue { ...@@ -461,7 +471,7 @@ class NotificationQueue {
checkPid(); checkPid();
bool signal = false; bool signal = false;
{ {
folly::MSLGuard g(spinlock_); folly::io::PortableSpinLockGuard g(spinlock_);
if (!checkQueueSize(maxSize, throws)) { if (!checkQueueSize(maxSize, throws)) {
return false; return false;
} }
...@@ -485,7 +495,7 @@ class NotificationQueue { ...@@ -485,7 +495,7 @@ class NotificationQueue {
checkPid(); checkPid();
bool signal = false; bool signal = false;
{ {
folly::MSLGuard g(spinlock_); folly::io::PortableSpinLockGuard g(spinlock_);
if (!checkQueueSize(maxSize, throws)) { if (!checkQueueSize(maxSize, throws)) {
return false; return false;
} }
...@@ -507,7 +517,7 @@ class NotificationQueue { ...@@ -507,7 +517,7 @@ class NotificationQueue {
bool signal = false; bool signal = false;
size_t numAdded = 0; size_t numAdded = 0;
{ {
folly::MSLGuard g(spinlock_); folly::io::PortableSpinLockGuard g(spinlock_);
while (first != last) { while (first != last) {
queue_.push_back(std::make_pair(*first, RequestContext::saveContext())); queue_.push_back(std::make_pair(*first, RequestContext::saveContext()));
++first; ++first;
...@@ -522,7 +532,7 @@ class NotificationQueue { ...@@ -522,7 +532,7 @@ class NotificationQueue {
} }
} }
mutable folly::MicroSpinLock spinlock_; mutable folly::io::PortableSpinLock spinlock_;
int eventfd_; int eventfd_;
int pipeFds_[2]; // to fallback to on older/non-linux systems int pipeFds_[2]; // to fallback to on older/non-linux systems
uint32_t advisoryMaxQueueSize_; uint32_t advisoryMaxQueueSize_;
...@@ -675,7 +685,7 @@ void NotificationQueue<MessageT>::Consumer::init( ...@@ -675,7 +685,7 @@ void NotificationQueue<MessageT>::Consumer::init(
queue_ = queue; queue_ = queue;
{ {
folly::MSLGuard g(queue_->spinlock_); folly::io::PortableSpinLockGuard g(queue_->spinlock_);
queue_->numConsumers_++; queue_->numConsumers_++;
} }
queue_->signalEvent(); queue_->signalEvent();
...@@ -695,7 +705,7 @@ void NotificationQueue<MessageT>::Consumer::stopConsuming() { ...@@ -695,7 +705,7 @@ void NotificationQueue<MessageT>::Consumer::stopConsuming() {
} }
{ {
folly::MSLGuard g(queue_->spinlock_); folly::io::PortableSpinLockGuard g(queue_->spinlock_);
queue_->numConsumers_--; queue_->numConsumers_--;
setActive(false); setActive(false);
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment