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