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 #include "os/linux_generic/reactive_semaphore.h" 29 #include "os/log.h" 30 31 namespace bluetooth { 32 namespace os { 33 34 // See documentation for |Queue| 35 template <typename T> 36 class IQueueEnqueue { 37 public: 38 using EnqueueCallback = common::Callback<std::unique_ptr<T>()>; 39 virtual ~IQueueEnqueue() = default; 40 virtual void RegisterEnqueue(Handler* handler, EnqueueCallback callback) = 0; 41 virtual void UnregisterEnqueue() = 0; 42 }; 43 44 // See documentation for |Queue| 45 template <typename T> 46 class IQueueDequeue { 47 public: 48 using DequeueCallback = common::Callback<void()>; 49 virtual ~IQueueDequeue() = default; 50 virtual void RegisterDequeue(Handler* handler, DequeueCallback callback) = 0; 51 virtual void UnregisterDequeue() = 0; 52 virtual std::unique_ptr<T> TryDequeue() = 0; 53 }; 54 55 template <typename T> 56 class Queue : public IQueueEnqueue<T>, public IQueueDequeue<T> { 57 public: 58 // A function moving data from enqueue end buffer to queue, it will be continually be invoked until queue 59 // is full. Enqueue end should make sure buffer isn't empty and UnregisterEnqueue when buffer become empty. 60 using EnqueueCallback = common::Callback<std::unique_ptr<T>()>; 61 // A function moving data form queue to dequeue end buffer, it will be continually be invoked until queue 62 // is empty. TryDequeue should be use in this function to get data from queue. 63 using DequeueCallback = common::Callback<void()>; 64 // Create a queue with |capacity| is the maximum number of messages a queue can contain 65 explicit Queue(size_t capacity); 66 ~Queue(); 67 // Register |callback| that will be called on |handler| when the queue is able to enqueue one piece of data. 68 // This will cause a crash if handler or callback has already been registered before. 69 void RegisterEnqueue(Handler* handler, EnqueueCallback callback) override; 70 // Unregister current EnqueueCallback from this queue, this will cause a crash if not registered yet. 71 void UnregisterEnqueue() override; 72 // Register |callback| that will be called on |handler| when the queue has at least one piece of data ready 73 // for dequeue. This will cause a crash if handler or callback has already been registered before. 74 void RegisterDequeue(Handler* handler, DequeueCallback callback) override; 75 // Unregister current DequeueCallback from this queue, this will cause a crash if not registered yet. 76 void UnregisterDequeue() override; 77 78 // Try to dequeue an item from this queue. Return nullptr when there is nothing in the queue. 79 std::unique_ptr<T> TryDequeue() override; 80 81 private: 82 void EnqueueCallbackInternal(EnqueueCallback callback); 83 // An internal queue that holds at most |capacity| pieces of data 84 std::queue<std::unique_ptr<T>> queue_; 85 // A mutex that guards data in this queue 86 std::mutex mutex_; 87 88 class QueueEndpoint { 89 public: QueueEndpoint(unsigned int initial_value)90 explicit QueueEndpoint(unsigned int initial_value) 91 : reactive_semaphore_(initial_value), handler_(nullptr), reactable_(nullptr) {} 92 ReactiveSemaphore reactive_semaphore_; 93 Handler* handler_; 94 Reactor::Reactable* reactable_; 95 }; 96 97 QueueEndpoint enqueue_; 98 QueueEndpoint dequeue_; 99 }; 100 101 template <typename T> 102 class EnqueueBuffer { 103 public: EnqueueBuffer(IQueueEnqueue<T> * queue)104 EnqueueBuffer(IQueueEnqueue<T>* queue) : queue_(queue) {} 105 ~EnqueueBuffer()106 ~EnqueueBuffer() { 107 if (enqueue_registered_.exchange(false)) { 108 queue_->UnregisterEnqueue(); 109 } 110 } 111 Enqueue(std::unique_ptr<T> t,os::Handler * handler)112 void Enqueue(std::unique_ptr<T> t, os::Handler* handler) { 113 std::lock_guard<std::mutex> lock(mutex_); 114 buffer_.push(std::move(t)); 115 if (!enqueue_registered_.exchange(true)) { 116 queue_->RegisterEnqueue(handler, common::Bind(&EnqueueBuffer<T>::enqueue_callback, common::Unretained(this))); 117 } 118 } 119 Clear()120 void Clear() { 121 std::lock_guard<std::mutex> lock(mutex_); 122 if (enqueue_registered_.exchange(false)) { 123 queue_->UnregisterEnqueue(); 124 std::queue<std::unique_ptr<T>> empty; 125 std::swap(buffer_, empty); 126 } 127 } 128 Size()129 auto Size() const { 130 return buffer_.size(); 131 } 132 NotifyOnEmpty(common::OnceClosure callback)133 void NotifyOnEmpty(common::OnceClosure callback) { 134 std::lock_guard<std::mutex> lock(mutex_); 135 ASSERT(callback_on_empty_.is_null()); 136 callback_on_empty_ = std::move(callback); 137 } 138 139 private: enqueue_callback()140 std::unique_ptr<T> enqueue_callback() { 141 std::lock_guard<std::mutex> lock(mutex_); 142 std::unique_ptr<T> enqueued_t = std::move(buffer_.front()); 143 buffer_.pop(); 144 if (buffer_.empty() && enqueue_registered_.exchange(false)) { 145 queue_->UnregisterEnqueue(); 146 if (!callback_on_empty_.is_null()) { 147 std::move(callback_on_empty_).Run(); 148 } 149 } 150 return enqueued_t; 151 } 152 153 mutable std::mutex mutex_; 154 IQueueEnqueue<T>* queue_; 155 std::atomic_bool enqueue_registered_ = false; 156 std::queue<std::unique_ptr<T>> buffer_; 157 common::OnceClosure callback_on_empty_; 158 }; 159 160 #include "os/linux_generic/queue.tpp" 161 162 } // namespace os 163 } // namespace bluetooth 164