Bluetooth: Disable eager reader threads in RX path

- Disabled eager reader threads in RX path for both
  ROME and PRONTO targets.

- Exising machamisn is to poll for incoming data
  (event, ACL data) using dedicated thread called
  eager reader thread.

- Now the responsibility of polling for data moved
  to existing HCI worker thread.

- Enhanced the read mechanism from hci hal interface.

CRs-Fixed: 934414
Change-Id: Icfd27b7b93f83033b2230e61a23df69aa3c33bdf

4 files changed, 296 insertions, 6 deletions

diff --git a/hci/include/hci_internals.h b/hci/include/hci_internals.h
index 6b9fb014b..e28de6e70 100644
--- a/hci/include/hci_internals.h
+++ b/hci/include/hci_internals.h
@@ -18,6 +18,14 @@
 
 #pragma once
 
+#define REMOVE_EAGER_THREADS TRUE
+
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+#include "allocator.h"
+#include "thread.h"
+#include "reactor.h"
+#endif
+
 // 2 bytes for opcode, 1 byte for parameter length (Volume 2, Part E, 5.4.1)
 #define HCI_COMMAND_PREAMBLE_SIZE 3
 // 2 bytes for handle, 2 bytes for data length (Volume 2, Part E, 5.4.2)
@@ -26,3 +34,31 @@
 #define HCI_SCO_PREAMBLE_SIZE 3
 // 1 byte for event code, 1 byte for parameter length (Volume 2, Part E, 5.4.4)
 #define HCI_EVENT_PREAMBLE_SIZE 2
+
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+struct hci_reader_t {
+  int inbound_fd;
+
+  const allocator_t *allocator;
+  size_t buffer_size;
+  uint8_t *data_buffer;
+  int rd_ptr;
+  int wr_ptr;
+
+  thread_t *inbound_read_thread;
+  reactor_object_t *inbound_read_object;
+};
+typedef  struct hci_reader_t hci_reader_t;
+typedef void (*hci_reader_cb)(void *context);
+
+hci_reader_t *hci_reader_new(
+    int fd_to_read,
+    size_t buffer_size,
+    size_t max_buffer_count,
+    thread_t *thread,
+    hci_reader_cb read_cb
+    );
+
+size_t hci_reader_read(hci_reader_t *reader, uint8_t *buffer, size_t max_size);
+void hci_reader_free(hci_reader_t *reader);
+#endif
diff --git a/hci/src/hci_hal.c b/hci/src/hci_hal.c
index fa526ce10..a50d7eac6 100644
--- a/hci/src/hci_hal.c
+++ b/hci/src/hci_hal.c
@@ -17,6 +17,14 @@
  ******************************************************************************/
 
 #include "hci_hal.h"
+#include "hci_internals.h"
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+#include <assert.h>
+#include <string.h>
+#include "eager_reader.h"
+#include "osi.h"
+#include "log.h"
+#endif
 
 const hci_hal_t *hci_hal_get_interface() {
 #if HCI_USE_MCT
@@ -26,3 +34,87 @@ const hci_hal_t *hci_hal_get_interface() {
 #endif
 }
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+void hci_reader_free(hci_reader_t *reader) {
+  if (!reader)
+    return;
+
+  // Only unregister from the input if we actually did register
+  if (reader->inbound_read_object)
+    reactor_unregister(reader->inbound_read_object);
+
+  // Free the current buffer, because it's not in the queue
+  // and won't be freed below
+  if (reader->data_buffer)
+    osi_free(reader->data_buffer);
+
+  osi_free(reader);
+}
+
+hci_reader_t *hci_reader_new(
+    int fd_to_read,
+    size_t buffer_size,
+    size_t max_buffer_count,
+    thread_t *thread,
+    hci_reader_cb read_cb
+    ) {
+
+  assert(fd_to_read != INVALID_FD);
+  assert(buffer_size > 0);
+  assert(max_buffer_count > 0);
+
+
+  hci_reader_t *ret = osi_calloc(sizeof(hci_reader_t));
+  if (!ret) {
+    LOG_ERROR("%s unable to allocate memory for new hci_reader.", __func__);
+    goto error;
+  }
+
+  ret->inbound_fd = fd_to_read;
+
+  ret->data_buffer = osi_calloc(buffer_size);
+  ret->buffer_size = buffer_size;
+  ret->rd_ptr = 0;
+  ret->wr_ptr = 0;
+
+  ret->inbound_read_thread = thread;
+  if (!ret->inbound_read_thread) {
+    LOG_ERROR("%s unable to make reading thread.", __func__);
+    goto error;
+  }
+
+  ret->inbound_read_object = reactor_register(
+      thread_get_reactor(ret->inbound_read_thread),
+      fd_to_read,
+      ret,
+      read_cb,
+      NULL
+      );
+
+  return ret;
+
+error:;
+  hci_reader_free(ret);
+  return NULL;
+}
+
+size_t hci_reader_read(hci_reader_t *reader, uint8_t *buffer, size_t req_size) {
+  size_t bytes_read = 0;
+  assert(reader != NULL);
+  assert(buffer != NULL);
+
+  // If the caller wants nonblocking behavior, poll to see if we have
+  // any bytes available before reading.
+  if (reader->rd_ptr < reader->wr_ptr) {
+    bytes_read = reader->wr_ptr - reader->rd_ptr;
+    if (bytes_read > req_size)
+      bytes_read = req_size;
+    memcpy(buffer, reader->data_buffer+reader->rd_ptr, bytes_read);
+    reader->rd_ptr += bytes_read;
+  } else {
+    bytes_read = read(reader->inbound_fd, buffer, req_size);
+  }
+
+  return bytes_read;
+}
+#endif
diff --git a/hci/src/hci_hal_h4.c b/hci/src/hci_hal_h4.c
index 332c1a2ba..d4dc105f5 100644
--- a/hci/src/hci_hal_h4.c
+++ b/hci/src/hci_hal_h4.c
@@ -45,13 +45,23 @@ static const vendor_t *vendor;
 static thread_t *thread; // Not owned by us
 
 static int uart_fd;
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static hci_reader_t *uart_stream;
+#else
 static eager_reader_t *uart_stream;
+#endif
 static serial_data_type_t current_data_type;
 static bool stream_has_interpretation;
 static bool stream_corruption_detected;
 static uint8_t stream_corruption_bytes_to_ignore;
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static void event_uart_has_bytes(void *context);
+#else
 static void event_uart_has_bytes(eager_reader_t *reader, void *context);
+#endif
+
+static bool stream_corrupted_during_le_scan_workaround(const uint8_t byte_read);
 
 // Interface functions
 
@@ -82,20 +92,28 @@ static bool hal_open() {
     goto error;
   }
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  uart_stream = hci_reader_new(uart_fd, HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX, thread, event_uart_has_bytes);
+  if (!uart_stream) {
+    LOG_ERROR("%s unable to create hci reader for the uart serial port.", __func__);
+    goto error;
+  }
+#else
   uart_stream = eager_reader_new(uart_fd, &allocator_malloc, HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX, "hci_single_channel");
   if (!uart_stream) {
     LOG_ERROR("%s unable to create eager reader for the uart serial port.", __func__);
     goto error;
   }
+  eager_reader_register(uart_stream, thread_get_reactor(thread), event_uart_has_bytes, NULL);
+  thread_set_priority(eager_reader_get_read_thread(uart_stream), HCI_THREAD_PRIORITY);
+#endif
 
   stream_has_interpretation = false;
   stream_corruption_detected = false;
   stream_corruption_bytes_to_ignore = 0;
-  eager_reader_register(uart_stream, thread_get_reactor(thread), event_uart_has_bytes, NULL);
 
   // Raise thread priorities to keep up with audio
   thread_set_priority(thread, HCI_THREAD_PRIORITY);
-  thread_set_priority(eager_reader_get_read_thread(uart_stream), HCI_THREAD_PRIORITY);
 
   return true;
 
@@ -107,8 +125,12 @@ error:
 
 static void hal_close() {
   LOG_INFO("%s", __func__);
-
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  hci_reader_free(uart_stream);
+#else
   eager_reader_free(uart_stream);
+#endif
+
   vendor->send_command(VENDOR_CLOSE_USERIAL, NULL);
   uart_fd = INVALID_FD;
 }
@@ -125,7 +147,11 @@ static size_t read_data(serial_data_type_t type, uint8_t *buffer, size_t max_siz
     return 0;
   }
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  return hci_reader_read(uart_stream, buffer, max_size);
+#else
   return eager_reader_read(uart_stream, buffer, max_size, block);
+#endif
 }
 
 static void packet_finished(serial_data_type_t type) {
@@ -134,7 +160,24 @@ static void packet_finished(serial_data_type_t type) {
   else if (current_data_type != type)
     LOG_ERROR("%s with different type than existing interpretation.", __func__);
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  if (uart_stream->rd_ptr == uart_stream->wr_ptr) {
+    uart_stream->rd_ptr = uart_stream->wr_ptr = 0;
+    stream_has_interpretation = false;
+  } else {
+    uint8_t type_byte;
+    type_byte = uart_stream->data_buffer[uart_stream->rd_ptr++];
+    if (type_byte < DATA_TYPE_ACL || type_byte > DATA_TYPE_EVENT) {
+      LOG_ERROR("%s Unknown HCI message type. Dropping this byte 0x%x, min %x, max %x", __func__,
+                                                    type_byte, DATA_TYPE_ACL, DATA_TYPE_EVENT);
+      return;
+    }
+    current_data_type = type_byte;
+    callbacks->data_ready(current_data_type);
+  }
+#else
   stream_has_interpretation = false;
+#endif
 }
 
 static uint16_t transmit_data(serial_data_type_t type, uint8_t *data, uint16_t length) {
@@ -225,6 +268,33 @@ static bool stream_corrupted_during_le_scan_workaround(const uint8_t byte_read)
 }
 
 // See what data is waiting, and notify the upper layer
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static void event_uart_has_bytes(void *context) {
+  uint8_t type_byte;
+  size_t bytes_read;
+  hci_reader_t *reader = (hci_reader_t *) context;
+  bytes_read = read(reader->inbound_fd, reader->data_buffer + reader->wr_ptr, reader->buffer_size - reader->wr_ptr);
+
+  if (bytes_read <= 0)  {
+    LOG_ERROR("%s could not read HCI message type", __func__);
+    return;
+  }
+  reader->wr_ptr += bytes_read;
+  if (!stream_has_interpretation) {
+    type_byte = reader->data_buffer[reader->rd_ptr++];
+
+    if (type_byte < DATA_TYPE_ACL || type_byte > DATA_TYPE_EVENT) {
+      LOG_ERROR("%s Unknown HCI message type. Dropping this byte 0x%x, min %x, max %x", __func__, type_byte, DATA_TYPE_ACL, DATA_TYPE_EVENT);
+      return;
+    }
+
+    stream_has_interpretation = true;
+    current_data_type = type_byte;
+  }
+
+  callbacks->data_ready(current_data_type);
+}
+#else
 static void event_uart_has_bytes(eager_reader_t *reader, UNUSED_ATTR void *context) {
   if (stream_has_interpretation) {
     callbacks->data_ready(current_data_type);
@@ -247,6 +317,7 @@ static void event_uart_has_bytes(eager_reader_t *reader, UNUSED_ATTR void *conte
     current_data_type = type_byte;
   }
 }
+#endif
 
 static const hci_hal_t interface = {
   hal_init,
diff --git a/hci/src/hci_hal_mct.c b/hci/src/hci_hal_mct.c
index adc17c3ad..2c7a32336 100644
--- a/hci/src/hci_hal_mct.c
+++ b/hci/src/hci_hal_mct.c
@@ -45,12 +45,22 @@ static const vendor_t *vendor;
 static thread_t *thread; // Not owned by us
 
 static int uart_fds[CH_MAX];
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static hci_reader_t *event_stream;
+static hci_reader_t *acl_stream;
+#else
 static eager_reader_t *event_stream;
 static eager_reader_t *acl_stream;
+#endif
 
 static uint16_t transmit_data_on(int fd, uint8_t *data, uint16_t length);
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static void event_event_stream_has_bytes(void *context);
+static void event_acl_stream_has_bytes(void *context);
+#else
 static void event_event_stream_has_bytes(eager_reader_t *reader, void *context);
 static void event_acl_stream_has_bytes(eager_reader_t *reader, void *context);
+#endif
 
 // Interface functions
 
@@ -97,6 +107,21 @@ static bool hal_open() {
     goto error;
   }
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  event_stream = hci_reader_new(uart_fds[CH_EVT], HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX,
+                                                 thread, event_event_stream_has_bytes);
+  if (!event_stream) {
+    LOG_ERROR("%s unable to create hci reader for the event uart serial port.", __func__);
+    goto error;
+  }
+
+  acl_stream = hci_reader_new(uart_fds[CH_ACL_IN], HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX,
+                                                    thread, event_acl_stream_has_bytes);
+  if (!acl_stream) {
+    LOG_ERROR("%s unable to create hci reader for the acl-in uart serial port.", __func__);
+    goto error;
+  }
+#else
   event_stream = eager_reader_new(uart_fds[CH_EVT], &allocator_malloc, HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX, "hci_mct");
   if (!event_stream) {
     LOG_ERROR("%s unable to create eager reader for the event uart serial port.", __func__);
@@ -104,7 +129,7 @@ static bool hal_open() {
   }
 
   acl_stream = eager_reader_new(uart_fds[CH_ACL_IN], &allocator_malloc, HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX, "hci_mct");
-  if (!event_stream) {
+  if (!acl_stream) {
     LOG_ERROR("%s unable to create eager reader for the acl-in uart serial port.", __func__);
     goto error;
   }
@@ -112,6 +137,8 @@ static bool hal_open() {
   eager_reader_register(event_stream, thread_get_reactor(thread), event_event_stream_has_bytes, NULL);
   eager_reader_register(acl_stream, thread_get_reactor(thread), event_acl_stream_has_bytes, NULL);
 
+#endif
+
   return true;
 
 error:;
@@ -122,8 +149,14 @@ error:;
 static void hal_close() {
   LOG_INFO("%s", __func__);
 
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  hci_reader_free(event_stream);
+  hci_reader_free(acl_stream);
+#else
   eager_reader_free(event_stream);
   eager_reader_free(acl_stream);
+#endif
+
   vendor->send_command(VENDOR_CLOSE_USERIAL, NULL);
 
   for (int i = 0; i < CH_MAX; i++)
@@ -156,11 +189,19 @@ static bool hal_dev_in_reset()
 #endif
 
 static size_t read_data(serial_data_type_t type, uint8_t *buffer, size_t max_size, bool block) {
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  if (type == DATA_TYPE_ACL) {
+    return hci_reader_read(acl_stream, buffer, max_size);
+  } else if (type == DATA_TYPE_EVENT) {
+    return hci_reader_read(event_stream, buffer, max_size);
+  }
+#else
   if (type == DATA_TYPE_ACL) {
     return eager_reader_read(acl_stream, buffer, max_size, block);
   } else if (type == DATA_TYPE_EVENT) {
     return eager_reader_read(event_stream, buffer, max_size, block);
   }
+#endif
 
   LOG_ERROR("%s invalid data type: %d", __func__, type);
   return 0;
@@ -168,6 +209,25 @@ static size_t read_data(serial_data_type_t type, uint8_t *buffer, size_t max_siz
 
 static void packet_finished(UNUSED_ATTR serial_data_type_t type) {
   // not needed by this protocol
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+  hci_reader_t *stream = NULL;
+  if (type == DATA_TYPE_ACL) {
+    stream = acl_stream;
+  } else if (type == DATA_TYPE_EVENT) {
+    stream = event_stream;
+  }
+
+  if(!stream) {
+    LOG_ERROR("%s invalid data type: %d", __func__, type);
+    return;
+  }
+
+  if (stream->rd_ptr == stream->wr_ptr) {
+    stream->rd_ptr = stream->wr_ptr = 0;
+  } else {
+    callbacks->data_ready(type);
+  }
+#endif
 }
 
 static uint16_t transmit_data(serial_data_type_t type, uint8_t *data, uint16_t length) {
@@ -208,14 +268,45 @@ static uint16_t transmit_data_on(int fd, uint8_t *data, uint16_t length) {
   return transmitted_length;
 }
 
-static void event_event_stream_has_bytes(UNUSED_ATTR eager_reader_t *reader, UNUSED_ATTR void *context) {
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static void event_event_stream_has_bytes(void *context) {
+  size_t bytes_read;
+  hci_reader_t *reader = (hci_reader_t *) context;
+  bytes_read = read(reader->inbound_fd, reader->data_buffer+reader->wr_ptr,
+                                    reader->buffer_size - reader->wr_ptr);
+  if (bytes_read <= 0) {
+    LOG_ERROR("%s could not read HCI message type", __func__);
+    return;
+  }
+  reader->wr_ptr += bytes_read;
   callbacks->data_ready(DATA_TYPE_EVENT);
 }
+#else
+static void event_event_stream_has_bytes(UNUSED_ATTR eager_reader_t *reader, UNUSED_ATTR void *context) {
+   callbacks->data_ready(DATA_TYPE_EVENT);
+}
+#endif
 
-static void event_acl_stream_has_bytes(UNUSED_ATTR eager_reader_t *reader, UNUSED_ATTR void *context) {
+#if (defined(REMOVE_EAGER_THREADS) && (REMOVE_EAGER_THREADS == TRUE))
+static void event_acl_stream_has_bytes(void *context) {
   // No real concept of incoming SCO typed data, just ACL
+  size_t bytes_read;
+  hci_reader_t *reader = (hci_reader_t *) context;
+  bytes_read = read(reader->inbound_fd, reader->data_buffer+reader->wr_ptr,
+                                  reader->buffer_size - reader->wr_ptr);
+  if (bytes_read <= 0) {
+    LOG_ERROR("%s could not read HCI message type", __func__);
+    return;
+  }
+  reader->wr_ptr += bytes_read;
   callbacks->data_ready(DATA_TYPE_ACL);
 }
+#else
+static void event_acl_stream_has_bytes(UNUSED_ATTR eager_reader_t *reader, UNUSED_ATTR void *context) {
+   // No real concept of incoming SCO typed data, just ACL
+   callbacks->data_ready(DATA_TYPE_ACL);
+}
+#endif
 
 static const hci_hal_t interface = {
   hal_init,