Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit f0a819c5 authored by Chienyuan's avatar Chienyuan
Browse files

Implement Queue 

* Implement Queue which provides a flow control mechanism for data
  transmission between modules. This queue streaming data with
  EnqueueCallback and DequeueCallback based on the reactor pattern.
  Enqueue end should register callback when data is ready to be sent
  and unregister when no data ready to send. Dequeue end should
  register callback when ready to handle data and unregister when not
  ready to handle data.

* Implement ReactiveSemaphore, a wrapper for an event_fd work in
  non-blocking and Semaphore mode used by Queue to handle readable
  signal in the reactor pattern.

Test: run bluetooth_test_gd
Change-Id: Ia7019cdbe271d193c92f1a0b405ecced41a2d84b
parent 5cdea69a

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();

}

system/gd/os/linux_generic/queue_unittest.cc

0 → 100644
+683 −0

File added.

Preview size limit exceeded, changes collapsed.