1 /* 2 * Copyright 2019 The Android Open Source Project 3 * 4 * Licensed under the Apache License, Version 2.0 (the "License"); 5 * you may not use this file except in compliance with the License. 6 * You may obtain a copy of the License at 7 * 8 * http://www.apache.org/licenses/LICENSE-2.0 9 * 10 * Unless required by applicable law or agreed to in writing, software 11 * distributed under the License is distributed on an "AS IS" BASIS, 12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 * See the License for the specific language governing permissions and 14 * limitations under the License. 15 */ 16 17 #pragma once 18 19 #include <unistd.h> 20 21 #include <functional> 22 #include <mutex> 23 #include <queue> 24 25 #include "common/bind.h" 26 #include "common/callback.h" 27 #include "os/handler.h" 28 #ifdef OS_LINUX_GENERIC 29 #include "os/linux_generic/reactive_semaphore.h" 30 #endif 31 #include "os/log.h" 32 33 namespace bluetooth { 34 namespace os { 35 36 // See documentation for |Queue| 37 template <typename T> 38 class IQueueEnqueue { 39 public: 40 using EnqueueCallback = common::Callback<std::unique_ptr<T>()>; 41 virtual ~IQueueEnqueue() = default; 42 virtual void RegisterEnqueue(Handler* handler, EnqueueCallback callback) = 0; 43 virtual void UnregisterEnqueue() = 0; 44 }; 45 46 // See documentation for |Queue| 47 template <typename T> 48 class IQueueDequeue { 49 public: 50 using DequeueCallback = common::Callback<void()>; 51 virtual ~IQueueDequeue() = default; 52 virtual void RegisterDequeue(Handler* handler, DequeueCallback callback) = 0; 53 virtual void UnregisterDequeue() = 0; 54 virtual std::unique_ptr<T> TryDequeue() = 0; 55 }; 56 57 template <typename T> 58 class Queue : public IQueueEnqueue<T>, public IQueueDequeue<T> { 59 public: 60 // A function moving data from enqueue end buffer to queue, it will be continually be invoked until queue 61 // is full. Enqueue end should make sure buffer isn't empty and UnregisterEnqueue when buffer become empty. 62 using EnqueueCallback = common::Callback<std::unique_ptr<T>()>; 63 // A function moving data form queue to dequeue end buffer, it will be continually be invoked until queue 64 // is empty. TryDequeue should be use in this function to get data from queue. 65 using DequeueCallback = common::Callback<void()>; 66 // Create a queue with |capacity| is the maximum number of messages a queue can contain 67 explicit Queue(size_t capacity); 68 ~Queue(); 69 // Register |callback| that will be called on |handler| when the queue is able to enqueue one piece of data. 70 // This will cause a crash if handler or callback has already been registered before. 71 void RegisterEnqueue(Handler* handler, EnqueueCallback callback) override; 72 // Unregister current EnqueueCallback from this queue, this will cause a crash if not registered yet. 73 void UnregisterEnqueue() override; 74 // Register |callback| that will be called on |handler| when the queue has at least one piece of data ready 75 // for dequeue. This will cause a crash if handler or callback has already been registered before. 76 void RegisterDequeue(Handler* handler, DequeueCallback callback) override; 77 // Unregister current DequeueCallback from this queue, this will cause a crash if not registered yet. 78 void UnregisterDequeue() override; 79 80 // Try to dequeue an item from this queue. Return nullptr when there is nothing in the queue. 81 std::unique_ptr<T> TryDequeue() override; 82 83 private: 84 void EnqueueCallbackInternal(EnqueueCallback callback); 85 // An internal queue that holds at most |capacity| pieces of data 86 std::queue<std::unique_ptr<T>> queue_; 87 // A mutex that guards data in this queue 88 std::mutex mutex_; 89 90 class QueueEndpoint { 91 public: 92 #ifdef OS_LINUX_GENERIC QueueEndpoint(unsigned int initial_value)93 explicit QueueEndpoint(unsigned int initial_value) 94 : reactive_semaphore_(initial_value), handler_(nullptr), reactable_(nullptr) {} 95 ReactiveSemaphore reactive_semaphore_; 96 #endif 97 Handler* handler_; 98 Reactor::Reactable* reactable_; 99 }; 100 101 QueueEndpoint enqueue_; 102 QueueEndpoint dequeue_; 103 }; 104 105 template <typename T> 106 class EnqueueBuffer { 107 public: EnqueueBuffer(IQueueEnqueue<T> * queue)108 EnqueueBuffer(IQueueEnqueue<T>* queue) : queue_(queue) {} 109 ~EnqueueBuffer()110 ~EnqueueBuffer() { 111 if (enqueue_registered_.exchange(false)) { 112 queue_->UnregisterEnqueue(); 113 } 114 } 115 Enqueue(std::unique_ptr<T> t,os::Handler * handler)116 void Enqueue(std::unique_ptr<T> t, os::Handler* handler) { 117 std::lock_guard<std::mutex> lock(mutex_); 118 buffer_.push(std::move(t)); 119 if (!enqueue_registered_.exchange(true)) { 120 queue_->RegisterEnqueue(handler, common::Bind(&EnqueueBuffer<T>::enqueue_callback, common::Unretained(this))); 121 } 122 } 123 Clear()124 void Clear() { 125 std::lock_guard<std::mutex> lock(mutex_); 126 if (enqueue_registered_.exchange(false)) { 127 queue_->UnregisterEnqueue(); 128 std::queue<std::unique_ptr<T>> empty; 129 std::swap(buffer_, empty); 130 } 131 } 132 Size()133 auto Size() const { 134 return buffer_.size(); 135 } 136 NotifyOnEmpty(common::OnceClosure callback)137 void NotifyOnEmpty(common::OnceClosure callback) { 138 std::lock_guard<std::mutex> lock(mutex_); 139 ASSERT(callback_on_empty_.is_null()); 140 callback_on_empty_ = std::move(callback); 141 } 142 143 private: enqueue_callback()144 std::unique_ptr<T> enqueue_callback() { 145 std::lock_guard<std::mutex> lock(mutex_); 146 std::unique_ptr<T> enqueued_t = std::move(buffer_.front()); 147 buffer_.pop(); 148 if (buffer_.empty() && enqueue_registered_.exchange(false)) { 149 queue_->UnregisterEnqueue(); 150 if (!callback_on_empty_.is_null()) { 151 std::move(callback_on_empty_).Run(); 152 } 153 } 154 return enqueued_t; 155 } 156 157 mutable std::mutex mutex_; 158 IQueueEnqueue<T>* queue_; 159 std::atomic_bool enqueue_registered_ = false; 160 std::queue<std::unique_ptr<T>> buffer_; 161 common::OnceClosure callback_on_empty_; 162 }; 163 164 #ifdef OS_LINUX_GENERIC 165 #include "os/linux_generic/queue.tpp" 166 #endif 167 168 } // namespace os 169 } // namespace bluetooth 170