libudx: Move ServiceDispatcher from libpdx_uds to libpdx
The two separate implementations of ServiceDispatcher for UDS and ServiceFS
are practically identical. Moved the implementation into the base libpdx lib,
and removed the transport-specific versions.
Bug: None
Test: `m -j32` succeeds; pdx_tests, pdx_servicefs_tests, pdx_uds_tests pass
Change-Id: I2344f4f23bc3da27eb7b192344844b87660e8610
diff --git a/libs/vr/libpdx/Android.bp b/libs/vr/libpdx/Android.bp
index f55e994..8fce140 100644
--- a/libs/vr/libpdx/Android.bp
+++ b/libs/vr/libpdx/Android.bp
@@ -5,12 +5,15 @@
"-Wall",
"-Wextra",
"-Werror",
+ "-DLOG_TAG=\"libpdx\"",
+ "-DTRACE=0",
],
export_include_dirs: ["private"],
local_include_dirs: ["private"],
srcs: [
"client.cpp",
"service.cpp",
+ "service_dispatcher.cpp",
"status.cpp",
],
}
diff --git a/libs/vr/libpdx/client.cpp b/libs/vr/libpdx/client.cpp
index bfa2d87..a01c4d6 100644
--- a/libs/vr/libpdx/client.cpp
+++ b/libs/vr/libpdx/client.cpp
@@ -1,6 +1,5 @@
#include "pdx/client.h"
-#define LOG_TAG "ServiceFramework"
#include <log/log.h>
#include <pdx/trace.h>
diff --git a/libs/vr/libpdx/mock_tests.cpp b/libs/vr/libpdx/mock_tests.cpp
index 76fd154..4143837 100644
--- a/libs/vr/libpdx/mock_tests.cpp
+++ b/libs/vr/libpdx/mock_tests.cpp
@@ -3,7 +3,6 @@
#include <pdx/mock_client_channel_factory.h>
#include <pdx/mock_message_reader.h>
#include <pdx/mock_message_writer.h>
-#include <pdx/mock_service_dispatcher.h>
#include <pdx/mock_service_endpoint.h>
TEST(MockTypes, Instantiation) {
@@ -15,6 +14,5 @@
android::pdx::MockMessageReader message_reader;
android::pdx::MockOutputResourceMapper output_resource_mapper;
android::pdx::MockMessageWriter message_writer;
- android::pdx::MockServiceDispatcher service_dispatcher;
android::pdx::MockEndpoint endpoint;
}
diff --git a/libs/vr/libpdx/private/pdx/mock_service_dispatcher.h b/libs/vr/libpdx/private/pdx/mock_service_dispatcher.h
deleted file mode 100644
index 9b51d30..0000000
--- a/libs/vr/libpdx/private/pdx/mock_service_dispatcher.h
+++ /dev/null
@@ -1,24 +0,0 @@
-#ifndef ANDROID_PDX_MOCK_SERVICE_DISPATCHER_H_
-#define ANDROID_PDX_MOCK_SERVICE_DISPATCHER_H_
-
-#include <gmock/gmock.h>
-#include <pdx/service_dispatcher.h>
-
-namespace android {
-namespace pdx {
-
-class MockServiceDispatcher : public ServiceDispatcher {
- public:
- MOCK_METHOD1(AddService, int(const std::shared_ptr<Service>& service));
- MOCK_METHOD1(RemoveService, int(const std::shared_ptr<Service>& service));
- MOCK_METHOD0(ReceiveAndDispatch, int());
- MOCK_METHOD1(ReceiveAndDispatch, int(int timeout));
- MOCK_METHOD0(EnterDispatchLoop, int());
- MOCK_METHOD1(SetCanceled, void(bool cancel));
- MOCK_CONST_METHOD0(IsCanceled, bool());
-};
-
-} // namespace pdx
-} // namespace android
-
-#endif // ANDROID_PDX_MOCK_SERVICE_DISPATCHER_H_
diff --git a/libs/vr/libpdx/private/pdx/mock_service_endpoint.h b/libs/vr/libpdx/private/pdx/mock_service_endpoint.h
index e741d4a..7f829e7 100644
--- a/libs/vr/libpdx/private/pdx/mock_service_endpoint.h
+++ b/libs/vr/libpdx/private/pdx/mock_service_endpoint.h
@@ -66,6 +66,7 @@
MOCK_METHOD0(AllocateMessageState, void*());
MOCK_METHOD1(FreeMessageState, void(void* state));
MOCK_METHOD0(Cancel, Status<void>());
+ MOCK_CONST_METHOD0(epoll_fd, int());
};
} // namespace pdx
diff --git a/libs/vr/libpdx/private/pdx/service_dispatcher.h b/libs/vr/libpdx/private/pdx/service_dispatcher.h
index c5e342a..bd27000 100644
--- a/libs/vr/libpdx/private/pdx/service_dispatcher.h
+++ b/libs/vr/libpdx/private/pdx/service_dispatcher.h
@@ -2,6 +2,11 @@
#define ANDROID_PDX_SERVICE_DISPATCHER_H_
#include <memory>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+
+#include <pdx/file_handle.h>
namespace android {
namespace pdx {
@@ -15,7 +20,10 @@
*/
class ServiceDispatcher {
public:
- virtual ~ServiceDispatcher() = default;
+ // Get a new instance of ServiceDispatcher, or return nullptr if init failed.
+ static std::unique_ptr<ServiceDispatcher> Create();
+
+ ~ServiceDispatcher();
/*
* Adds a service to the list of services handled by this dispatcher. This
@@ -24,7 +32,7 @@
*
* Returns 0 on success; -EEXIST if the service was already added.
*/
- virtual int AddService(const std::shared_ptr<Service>& service) = 0;
+ int AddService(const std::shared_ptr<Service>& service);
/*
* Removes a service from this dispatcher. This will fail if any threads are
@@ -33,7 +41,7 @@
* Returns 0 on success; -ENOENT if the service was not previously added;
* -EBUSY if there are threads in the dispatcher.
*/
- virtual int RemoveService(const std::shared_ptr<Service>& service) = 0;
+ int RemoveService(const std::shared_ptr<Service>& service);
/*
* Receive and dispatch one set of messages. Multiple threads may enter this
@@ -42,14 +50,14 @@
* cycle, requiring an external loop. This is useful when other work needs
* to be done in the service dispatch loop.
*/
- virtual int ReceiveAndDispatch() = 0;
+ int ReceiveAndDispatch();
/*
* Same as above with timeout in milliseconds. A negative value means
* infinite timeout, while a value of 0 means return immediately if no
* messages are available to receive.
*/
- virtual int ReceiveAndDispatch(int timeout) = 0;
+ int ReceiveAndDispatch(int timeout);
/*
* Receive and dispatch messages until canceled. When more than one thread
@@ -58,19 +66,39 @@
* hands Message instances (via move assignment) over to a queue of threads
* (or perhaps one of several) to handle.
*/
- virtual int EnterDispatchLoop() = 0;
+ int EnterDispatchLoop();
/*
* Sets the canceled state of the dispatcher. When canceled is true, any
* threads blocked waiting for messages will return. This method waits until
* all dispatch threads have exited the dispatcher.
*/
- virtual void SetCanceled(bool cancel) = 0;
+ void SetCanceled(bool cancel);
/*
* Gets the canceled state of the dispatcher.
*/
- virtual bool IsCanceled() const = 0;
+ bool IsCanceled() const;
+
+ private:
+ ServiceDispatcher();
+
+ // Internal thread accounting.
+ int ThreadEnter();
+ void ThreadExit();
+
+ std::mutex mutex_;
+ std::condition_variable condition_;
+ std::atomic<bool> canceled_{false};
+
+ std::vector<std::shared_ptr<Service>> services_;
+
+ int thread_count_ = 0;
+ LocalHandle event_fd_;
+ LocalHandle epoll_fd_;
+
+ ServiceDispatcher(const ServiceDispatcher&) = delete;
+ void operator=(const ServiceDispatcher&) = delete;
};
} // namespace pdx
diff --git a/libs/vr/libpdx/private/pdx/service_endpoint.h b/libs/vr/libpdx/private/pdx/service_endpoint.h
index 28bd6bc..d581894 100644
--- a/libs/vr/libpdx/private/pdx/service_endpoint.h
+++ b/libs/vr/libpdx/private/pdx/service_endpoint.h
@@ -136,6 +136,10 @@
// Cancels the endpoint, unblocking any receiver threads waiting for a
// message.
virtual Status<void> Cancel() = 0;
+
+ // Returns an fd that can be used with epoll() to wait for incoming messages
+ // from this endpoint.
+ virtual int epoll_fd() const = 0;
};
} // namespace pdx
diff --git a/libs/vr/libpdx/service.cpp b/libs/vr/libpdx/service.cpp
index fab4770..1d3b62a 100644
--- a/libs/vr/libpdx/service.cpp
+++ b/libs/vr/libpdx/service.cpp
@@ -1,4 +1,3 @@
-#define LOG_TAG "ServiceFramework"
#include "pdx/service.h"
#include <fcntl.h>
@@ -10,8 +9,6 @@
#include <pdx/trace.h>
-#define TRACE 0
-
namespace android {
namespace pdx {
diff --git a/libs/vr/libpdx/service_dispatcher.cpp b/libs/vr/libpdx/service_dispatcher.cpp
new file mode 100644
index 0000000..b112fa3
--- /dev/null
+++ b/libs/vr/libpdx/service_dispatcher.cpp
@@ -0,0 +1,192 @@
+#include <pdx/service_dispatcher.h>
+
+#include <errno.h>
+#include <log/log.h>
+#include <sys/epoll.h>
+#include <sys/eventfd.h>
+
+#include <pdx/service.h>
+#include <pdx/service_endpoint.h>
+
+static const int kMaxEventsPerLoop = 128;
+
+namespace android {
+namespace pdx {
+
+std::unique_ptr<ServiceDispatcher> ServiceDispatcher::Create() {
+ std::unique_ptr<ServiceDispatcher> dispatcher{new ServiceDispatcher()};
+ if (!dispatcher->epoll_fd_ || !dispatcher->event_fd_) {
+ dispatcher.reset();
+ }
+
+ return dispatcher;
+}
+
+ServiceDispatcher::ServiceDispatcher() {
+ event_fd_.Reset(eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK));
+ if (!event_fd_) {
+ ALOGE("Failed to create event fd because: %s\n", strerror(errno));
+ return;
+ }
+
+ epoll_fd_.Reset(epoll_create1(EPOLL_CLOEXEC));
+ if (!epoll_fd_) {
+ ALOGE("Failed to create epoll fd because: %s\n", strerror(errno));
+ return;
+ }
+
+ // Use "this" as a unique pointer to distinguish the event fd from all
+ // the other entries that point to instances of Service.
+ epoll_event event;
+ event.events = EPOLLIN;
+ event.data.ptr = this;
+
+ if (epoll_ctl(epoll_fd_.Get(), EPOLL_CTL_ADD, event_fd_.Get(), &event) < 0) {
+ ALOGE("Failed to add event fd to epoll fd because: %s\n", strerror(errno));
+
+ // Close the fds here and signal failure to the factory method.
+ event_fd_.Close();
+ epoll_fd_.Close();
+ }
+}
+
+ServiceDispatcher::~ServiceDispatcher() { SetCanceled(true); }
+
+int ServiceDispatcher::ThreadEnter() {
+ std::lock_guard<std::mutex> autolock(mutex_);
+
+ if (canceled_)
+ return -EBUSY;
+
+ thread_count_++;
+ return 0;
+}
+
+void ServiceDispatcher::ThreadExit() {
+ std::lock_guard<std::mutex> autolock(mutex_);
+ thread_count_--;
+ condition_.notify_one();
+}
+
+int ServiceDispatcher::AddService(const std::shared_ptr<Service>& service) {
+ std::lock_guard<std::mutex> autolock(mutex_);
+
+ epoll_event event;
+ event.events = EPOLLIN;
+ event.data.ptr = service.get();
+
+ if (epoll_ctl(epoll_fd_.Get(), EPOLL_CTL_ADD, service->endpoint()->epoll_fd(),
+ &event) < 0) {
+ ALOGE("Failed to add service to dispatcher because: %s\n", strerror(errno));
+ return -errno;
+ }
+
+ services_.push_back(service);
+ return 0;
+}
+
+int ServiceDispatcher::RemoveService(const std::shared_ptr<Service>& service) {
+ std::lock_guard<std::mutex> autolock(mutex_);
+
+ // It's dangerous to remove a service while other threads may be using it.
+ if (thread_count_ > 0)
+ return -EBUSY;
+
+ epoll_event dummy; // See BUGS in man 2 epoll_ctl.
+ if (epoll_ctl(epoll_fd_.Get(), EPOLL_CTL_DEL, service->endpoint()->epoll_fd(),
+ &dummy) < 0) {
+ ALOGE("Failed to remove service from dispatcher because: %s\n",
+ strerror(errno));
+ return -errno;
+ }
+
+ services_.erase(std::remove(services_.begin(), services_.end(), service),
+ services_.end());
+ return 0;
+}
+
+int ServiceDispatcher::ReceiveAndDispatch() { return ReceiveAndDispatch(-1); }
+
+int ServiceDispatcher::ReceiveAndDispatch(int timeout) {
+ int ret = ThreadEnter();
+ if (ret < 0)
+ return ret;
+
+ epoll_event events[kMaxEventsPerLoop];
+
+ int count = epoll_wait(epoll_fd_.Get(), events, kMaxEventsPerLoop, timeout);
+ if (count <= 0) {
+ ALOGE_IF(count < 0, "Failed to wait for epoll events because: %s\n",
+ strerror(errno));
+ ThreadExit();
+ return count < 0 ? -errno : -ETIMEDOUT;
+ }
+
+ for (int i = 0; i < count; i++) {
+ if (events[i].data.ptr == this) {
+ ThreadExit();
+ return -EBUSY;
+ } else {
+ Service* service = static_cast<Service*>(events[i].data.ptr);
+
+ ALOGI_IF(TRACE, "Dispatching message: fd=%d\n",
+ service->endpoint()->epoll_fd());
+ service->ReceiveAndDispatch();
+ }
+ }
+
+ ThreadExit();
+ return 0;
+}
+
+int ServiceDispatcher::EnterDispatchLoop() {
+ int ret = ThreadEnter();
+ if (ret < 0)
+ return ret;
+
+ epoll_event events[kMaxEventsPerLoop];
+
+ while (!IsCanceled()) {
+ int count = epoll_wait(epoll_fd_.Get(), events, kMaxEventsPerLoop, -1);
+ if (count < 0 && errno != EINTR) {
+ ALOGE("Failed to wait for epoll events because: %s\n", strerror(errno));
+ ThreadExit();
+ return -errno;
+ }
+
+ for (int i = 0; i < count; i++) {
+ if (events[i].data.ptr == this) {
+ ThreadExit();
+ return -EBUSY;
+ } else {
+ Service* service = static_cast<Service*>(events[i].data.ptr);
+
+ ALOGI_IF(TRACE, "Dispatching message: fd=%d\n",
+ service->endpoint()->epoll_fd());
+ service->ReceiveAndDispatch();
+ }
+ }
+ }
+
+ ThreadExit();
+ return 0;
+}
+
+void ServiceDispatcher::SetCanceled(bool cancel) {
+ std::unique_lock<std::mutex> lock(mutex_);
+ canceled_ = cancel;
+
+ if (canceled_ && thread_count_ > 0) {
+ eventfd_write(event_fd_.Get(), 1); // Signal threads to quit.
+
+ condition_.wait(lock, [this] { return !(canceled_ && thread_count_ > 0); });
+
+ eventfd_t value;
+ eventfd_read(event_fd_.Get(), &value); // Unsignal.
+ }
+}
+
+bool ServiceDispatcher::IsCanceled() const { return canceled_; }
+
+} // namespace pdx
+} // namespace android