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