Merge "Implement Queue" (e21413ba) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/os/Android.bp

+2 −0

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ filegroup {
		"linux_generic/handler.cc",
		"linux_generic/reactor.cc",
		"linux_generic/repeating_alarm.cc",
		"linux_generic/reactive_semaphore.cc",
		"linux_generic/thread.cc",
		]
		}
		@@ -14,6 +15,7 @@ filegroup {
		srcs: [
		"linux_generic/alarm_unittest.cc",
		"linux_generic/handler_unittest.cc",
		"linux_generic/queue_unittest.cc",
		"linux_generic/reactor_unittest.cc",
		"linux_generic/repeating_alarm_unittest.cc",
		"linux_generic/thread_unittest.cc",

system/gd/os/handler.h

+3 −0

Original line number	Diff line number	Diff line
		@@ -46,6 +46,9 @@ class Handler {
		// Remove all pending events from the queue of this handler
		void Clear();

		template <typename T>
		friend class Queue;

		private:
		std::queue<Closure> tasks_;
		Thread* thread_;

system/gd/os/linux_generic/linux.h

0 → 100644

+21 −0

Original line number	Diff line number	Diff line
		/*
		* Copyright 2019 The Android Open Source Project
		*
		* 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

		#ifndef EFD_SEMAPHORE
		#define EFD_SEMAPHORE 1
		#endif

system/gd/os/linux_generic/queue.tpp

0 → 100644

+101 −0

Original line number	Diff line number	Diff line
		/*
		* Copyright 2019 The Android Open Source Project
		*
		* 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.
		*/

		template <typename T>
		Queue<T>::Queue(size_t capacity)
		: enqueue_callback_(nullptr), dequeue_callback_(nullptr), enqueue_(capacity), dequeue_(0){};

		template <typename T>
		Queue<T>::~Queue() {
		ASSERT(enqueue_callback_ == nullptr);
		ASSERT(dequeue_callback_ == nullptr);
		};

		template <typename T>
		void Queue<T>::RegisterEnqueue(Handler* handler, EnqueueCallback callback) {
		std::lock_guard<std::mutex> lock(mutex_);
		ASSERT(enqueue_.handler_ == nullptr);
		ASSERT(enqueue_callback_ == nullptr);
		ASSERT(enqueue_.reactable_ == nullptr);
		enqueue_.handler_ = handler;
		enqueue_callback_ = callback;
		enqueue_.reactable_ = enqueue_.handler_->thread_->GetReactor()->Register(
		enqueue_.reactive_semaphore_.GetFd(), [this] { EnqueueCallbackInternal(); }, nullptr);
		}

		template <typename T>
		void Queue<T>::UnregisterEnqueue() {
		std::lock_guard<std::mutex> lock(mutex_);
		ASSERT(enqueue_.reactable_ != nullptr);
		enqueue_.handler_->thread_->GetReactor()->Unregister(enqueue_.reactable_);
		enqueue_.reactable_ = nullptr;
		enqueue_callback_ = nullptr;
		enqueue_.handler_ = nullptr;
		}

		template <typename T>
		void Queue<T>::RegisterDequeue(Handler* handler, DequeueCallback callback) {
		std::lock_guard<std::mutex> lock(mutex_);
		ASSERT(dequeue_.handler_ == nullptr);
		ASSERT(dequeue_callback_ == nullptr);
		ASSERT(dequeue_.reactable_ == nullptr);
		dequeue_.handler_ = handler;
		dequeue_callback_ = callback;
		dequeue_.reactable_ = dequeue_.handler_->thread_->GetReactor()->Register(dequeue_.reactive_semaphore_.GetFd(),
		[this] { dequeue_callback_(); }, nullptr);
		}

		template <typename T>
		void Queue<T>::UnregisterDequeue() {
		std::lock_guard<std::mutex> lock(mutex_);
		ASSERT(dequeue_.reactable_ != nullptr);
		dequeue_.handler_->thread_->GetReactor()->Unregister(dequeue_.reactable_);
		dequeue_.reactable_ = nullptr;
		dequeue_callback_ = nullptr;
		dequeue_.handler_ = nullptr;
		}

		template <typename T>
		std::unique_ptr<T> Queue<T>::TryDequeue() {
		std::lock_guard<std::mutex> lock(mutex_);

		if (queue_.empty()) {
		return nullptr;
		}

		dequeue_.reactive_semaphore_.Decrease();

		std::unique_ptr<T> data = std::move(queue_.front());
		queue_.pop();

		enqueue_.reactive_semaphore_.Increase();

		return data;
		}

		template <typename T>
		void Queue<T>::EnqueueCallbackInternal() {
		enqueue_.reactive_semaphore_.Decrease();

		{
		std::unique_ptr<T> data = enqueue_callback_();
		ASSERT(data != nullptr);
		std::lock_guard<std::mutex> lock(mutex_);
		queue_.push(std::move(data));
		}

		dequeue_.reactive_semaphore_.Increase();
		}