Modify eager reader to support multi-byte reads

10 files changed, 147 insertions, 113 deletions

diff --git a/hci/include/hci_hal.h b/hci/include/hci_hal.h
index b86b3e7bc..9a9413097 100644
--- a/hci/include/hci_hal.h
+++ b/hci/include/hci_hal.h
@@ -56,12 +56,11 @@ typedef struct hci_hal_interface_t {
   // "Daisy, Daisy..."
   void (*close)(void);
 
-  // Retrieve bytes for ACL, SCO, or EVENT data packets.
+  // Retrieve up to |max_size| bytes for ACL, SCO, or EVENT data packets into
+  // |buffer|, blocking until max_size bytes read if |block| is true.
   // Only guaranteed to be correct in the context of a data_ready callback
   // of the corresponding type.
-  uint8_t (*read_byte)(serial_data_type_t type);
-  // Returns true if there is a byte available of the specified |type|.
-  bool (*has_byte)(serial_data_type_t type);
+  size_t (*read_data)(serial_data_type_t type, uint8_t *buffer, size_t max_size, bool block);
   // The upper layer must call this to notify the HAL that it has finished
   // reading a packet of the specified |type|. Underlying implementations that
   // use shared channels for multiple data types depend on this to know when
diff --git a/hci/src/hci_hal_h4.c b/hci/src/hci_hal_h4.c
index 3e0b34403..adad3a24b 100644
--- a/hci/src/hci_hal_h4.c
+++ b/hci/src/hci_hal_h4.c
@@ -96,41 +96,26 @@ static void hal_close() {
   uart_fd = INVALID_FD;
 }
 
-static uint8_t read_byte(serial_data_type_t type) {
+static size_t read_data(serial_data_type_t type, uint8_t *buffer, size_t max_size, bool block) {
   if (type < DATA_TYPE_ACL || type > DATA_TYPE_EVENT) {
     ALOGE("%s invalid data type: %d", __func__, type);
     return 0;
   } else if (!stream_has_interpretation) {
-    ALOGE("%s read byte with no valid stream intepretation.", __func__);
+    ALOGE("%s with no valid stream intepretation.", __func__);
     return 0;
   } else if (current_data_type != type) {
-    ALOGE("%s read byte with different type than existing interpretation.", __func__);
+    ALOGE("%s with different type than existing interpretation.", __func__);
     return 0;
   }
 
-  return eager_reader_read_byte(uart_stream);
-}
-
-static bool has_byte(serial_data_type_t type) {
-  if (type < DATA_TYPE_ACL || type > DATA_TYPE_EVENT) {
-    ALOGE("%s invalid data type: %d", __func__, type);
-    return 0;
-  } else if (!stream_has_interpretation) {
-    ALOGE("%s has byte with no valid stream intepretation.", __func__);
-    return 0;
-  } else if (current_data_type != type) {
-    ALOGE("%s has byte with different type than existing interpretation.", __func__);
-    return 0;
-  }
-
-  return eager_reader_has_byte(uart_stream);
+  return eager_reader_read(uart_stream, buffer, max_size, block);
 }
 
 static void packet_finished(serial_data_type_t type) {
   if (!stream_has_interpretation)
-    ALOGE("%s packet finished with no existing stream interpretation.", __func__);
+    ALOGE("%s with no existing stream interpretation.", __func__);
   else if (current_data_type != type)
-    ALOGE("%s packed finished with different type than existing interpretation.", __func__);
+    ALOGE("%s with different type than existing interpretation.", __func__);
 
   stream_has_interpretation = false;
 }
@@ -183,15 +168,11 @@ done:;
 
 // See what data is waiting, and notify the upper layer
 static void event_uart_has_bytes(eager_reader_t *reader, UNUSED_ATTR void *context) {
-  uint8_t type_byte;
-
   if (stream_has_interpretation) {
-    // Reentry in the the case that the upper layer couldn't read the entire
-    // packet the first time around
-    type_byte = current_data_type;
-  }
-  else {
-    type_byte = eager_reader_read_byte(reader);
+    callbacks->data_ready(current_data_type);
+  } else {
+    uint8_t type_byte;
+    eager_reader_read(reader, &type_byte, 1, true);
     if (type_byte < DATA_TYPE_ACL || type_byte > DATA_TYPE_EVENT) {
       ALOGE("[h4] Unknown HCI message type. Dropping this byte 0x%x, min %x, max %x", type_byte, DATA_TYPE_ACL, DATA_TYPE_EVENT);
       return;
@@ -200,9 +181,6 @@ static void event_uart_has_bytes(eager_reader_t *reader, UNUSED_ATTR void *conte
     stream_has_interpretation = true;
     current_data_type = type_byte;
   }
-
-  if (has_byte(current_data_type))
-    callbacks->data_ready(type_byte);
 }
 
 static const hci_hal_interface_t interface = {
@@ -211,8 +189,7 @@ static const hci_hal_interface_t interface = {
   hal_open,
   hal_close,
 
-  read_byte,
-  has_byte,
+  read_data,
   packet_finished,
   transmit_data,
 };
diff --git a/hci/src/hci_hal_mct.c b/hci/src/hci_hal_mct.c
index e1d855759..4e0835110 100644
--- a/hci/src/hci_hal_mct.c
+++ b/hci/src/hci_hal_mct.c
@@ -122,28 +122,17 @@ static void hal_close() {
     uart_fds[i] = INVALID_FD;
 }
 
-static uint8_t read_byte(serial_data_type_t type) {
+static size_t read_data(serial_data_type_t type, uint8_t *buffer, size_t max_size, bool block) {
   if (type == DATA_TYPE_ACL) {
-    return eager_reader_read_byte(acl_stream);
+    return eager_reader_read(acl_stream, buffer, max_size, block);
   } else if (type == DATA_TYPE_EVENT) {
-    return eager_reader_read_byte(event_stream);
+    return eager_reader_read(event_stream, buffer, max_size, block);
   }
 
   ALOGE("%s invalid data type: %d", __func__, type);
   return 0;
 }
 
-static bool has_byte(serial_data_type_t type) {
-  if (type == DATA_TYPE_ACL) {
-    return eager_reader_has_byte(acl_stream);
-  } else if (type == DATA_TYPE_EVENT) {
-    return eager_reader_has_byte(event_stream);
-  }
-
-  ALOGE("%s invalid data type: %d", __func__, type);
-  return false;
-}
-
 static void packet_finished(UNUSED_ATTR serial_data_type_t type) {
   // not needed by this protocol
 }
@@ -201,8 +190,7 @@ static const hci_hal_interface_t interface = {
   hal_open,
   hal_close,
 
-  read_byte,
-  has_byte,
+  read_data,
   packet_finished,
   transmit_data,
 };
diff --git a/hci/src/hci_layer.c b/hci/src/hci_layer.c
index ed32e84b3..7e1920cfe 100644
--- a/hci/src/hci_layer.c
+++ b/hci/src/hci_layer.c
@@ -479,9 +479,8 @@ static void event_packet_ready(fixed_queue_t *queue, UNUSED_ATTR void *context)
 static void hal_says_data_ready(serial_data_type_t type) {
   packet_receive_data_t *incoming = &incoming_packets[PACKET_TYPE_TO_INBOUND_INDEX(type)];
 
-  while (hal->has_byte(type)) {
-    uint8_t byte = hal->read_byte(type);
-
+  uint8_t byte;
+  while (hal->read_data(type, &byte, 1, false) != 0) {
     switch (incoming->state) {
       case BRAND_NEW:
         // Initialize and prepare to jump to the preamble reading state
@@ -524,7 +523,9 @@ static void hal_says_data_ready(serial_data_type_t type) {
         incoming->index++;
         incoming->bytes_remaining--;
 
-        // TODO(zachoverflow): consider reading in available bytes up to the length, instead of byte by byte
+        size_t bytes_read = hal->read_data(type, (incoming->buffer->data + incoming->index), incoming->bytes_remaining, false);
+        incoming->index += bytes_read;
+        incoming->bytes_remaining -= bytes_read;
 
         incoming->state = incoming->bytes_remaining == 0 ? FINISHED : incoming->state;
         break;
diff --git a/hci/test/hci_hal_h4_test.cpp b/hci/test/hci_hal_h4_test.cpp
index ec7ae7365..8055d807a 100644
--- a/hci/test/hci_hal_h4_test.cpp
+++ b/hci/test/hci_hal_h4_test.cpp
@@ -59,7 +59,9 @@ static semaphore_t *reentry_semaphore;
 static void expect_packet_synchronous(serial_data_type_t type, char *packet_data) {
   int length = strlen(packet_data);
   for (int i = 0; i < length; i++) {
-    EXPECT_EQ(packet_data[i], hal->read_byte(type));
+    uint8_t byte;
+    EXPECT_EQ((size_t)1, hal->read_data(type, &byte, 1, true));
+    EXPECT_EQ(packet_data[i], byte);
   }
 
   hal->packet_finished(type);
@@ -105,8 +107,10 @@ STUB_FUNCTION(void, data_ready_callback, (serial_data_type_t type))
   DURING(read_async_reentry) {
     EXPECT_EQ(DATA_TYPE_ACL, type);
 
-    while (hal->has_byte(type)) {
-      EXPECT_EQ(sample_data3[reentry_i], hal->read_byte(type));
+    uint8_t byte;
+    size_t bytes_read;
+    while ((bytes_read = hal->read_data(type, &byte, 1, false)) != 0) {
+      EXPECT_EQ(sample_data3[reentry_i], byte);
       semaphore_post(reentry_semaphore);
       reentry_i++;
       if (reentry_i == (int)strlen(sample_data3)) {
diff --git a/hci/test/hci_hal_mct_test.cpp b/hci/test/hci_hal_mct_test.cpp
index ea324384f..60fb1c004 100644
--- a/hci/test/hci_hal_mct_test.cpp
+++ b/hci/test/hci_hal_mct_test.cpp
@@ -60,7 +60,9 @@ static semaphore_t *reentry_semaphore;
 static void expect_packet_synchronous(serial_data_type_t type, char *packet_data) {
   int length = strlen(packet_data);
   for (int i = 0; i < length; i++) {
-    EXPECT_EQ(packet_data[i], hal->read_byte(type));
+    uint8_t byte;
+    EXPECT_EQ((size_t)1, hal->read_data(type, &byte, 1, true));
+    EXPECT_EQ(packet_data[i], byte);
   }
 
   hal->packet_finished(type);
@@ -104,8 +106,10 @@ STUB_FUNCTION(void, data_ready_callback, (serial_data_type_t type))
   DURING(read_async_reentry) {
     EXPECT_EQ(DATA_TYPE_ACL, type);
 
-    while (hal->has_byte(type)) {
-      EXPECT_EQ(sample_data3[reentry_i], hal->read_byte(type));
+    uint8_t byte;
+    size_t bytes_read;
+    while ((bytes_read = hal->read_data(type, &byte, 1, false)) != 0) {
+      EXPECT_EQ(sample_data3[reentry_i], byte);
       semaphore_post(reentry_semaphore);
       reentry_i++;
       if (reentry_i == (int)strlen(sample_data3)) {
diff --git a/hci/test/hci_layer_test.cpp b/hci/test/hci_layer_test.cpp
index 2ac9012d2..ec86254ef 100644
--- a/hci/test/hci_layer_test.cpp
+++ b/hci/test/hci_layer_test.cpp
@@ -172,26 +172,22 @@ STUB_FUNCTION(uint16_t, hal_transmit_data, (serial_data_type_t type, uint8_t *da
   return 0;
 }
 
-STUB_FUNCTION(uint8_t, hal_read_byte, (serial_data_type_t type))
+STUB_FUNCTION(size_t, hal_read_data, (serial_data_type_t type, uint8_t *buffer, size_t max_size, UNUSED_ATTR bool block))
   DURING(receive_simple) {
     EXPECT_EQ(DATA_TYPE_ACL, type);
-    if (data_to_receive->offset < data_to_receive->len) {
-      return data_to_receive->data[data_to_receive->offset++];
-    }
-  }
+    for (size_t i = 0; i < max_size; i++) {
+      if (data_to_receive->offset >= data_to_receive->len)
+        break;
 
-  UNEXPECTED_CALL;
-  return 0;
-}
+      buffer[i] = data_to_receive->data[data_to_receive->offset++];
 
-STUB_FUNCTION(bool, hal_has_byte, (serial_data_type_t type))
-  DURING(receive_simple) {
-    EXPECT_EQ(DATA_TYPE_ACL, type);
-    return data_to_receive->offset < data_to_receive->len;
+      if (i == (max_size - 1))
+        return i + 1; // We return the length, not the index;
+    }
   }
 
   UNEXPECTED_CALL;
-  return false;
+  return 0;
 }
 
 STUB_FUNCTION(void, hal_packet_finished, (serial_data_type_t type))
@@ -399,8 +395,7 @@ static void reset_for(TEST_MODES_T next) {
   RESET_CALL_COUNT(hal_init);
   RESET_CALL_COUNT(hal_open);
   RESET_CALL_COUNT(hal_close);
-  RESET_CALL_COUNT(hal_read_byte);
-  RESET_CALL_COUNT(hal_has_byte);
+  RESET_CALL_COUNT(hal_read_data);
   RESET_CALL_COUNT(hal_packet_finished);
   RESET_CALL_COUNT(hal_transmit_data);
   RESET_CALL_COUNT(btsnoop_open);
@@ -443,8 +438,7 @@ class HciLayerTest : public AlarmTestHarness {
       hal.init = hal_init;
       hal.open = hal_open;
       hal.close = hal_close;
-      hal.read_byte = hal_read_byte;
-      hal.has_byte = hal_has_byte;
+      hal.read_data = hal_read_data;
       hal.packet_finished = hal_packet_finished;
       hal.transmit_data = hal_transmit_data;
       btsnoop.open = btsnoop_open;
@@ -540,7 +534,6 @@ TEST_F(HciLayerTest, test_transmit_simple) {
   free(packet);
 }
 
-// TODO finish test
 TEST_F(HciLayerTest, test_receive_simple) {
   reset_for(receive_simple);
   data_to_receive = manufacture_packet(MSG_STACK_TO_HC_HCI_ACL, small_sample_data);
@@ -566,11 +559,13 @@ TEST_F(HciLayerTest, test_send_internal_command) {
   EXPECT_CALL_COUNT(low_power_wake_assert, 1);
   EXPECT_CALL_COUNT(callback_transmit_finished, 1);
 
-  // TODO send a response and make sure the response ends up at the callback
+  // TODO(zachoverflow): send a response and make sure the response ends up at the callback
 
   free(data_to_receive);
 }
 
+// TODO(zachoverflow): test post-reassembly better, stub out fragmenter instead of using it
+
 TEST_F(HciLayerTest, test_turn_on_logging) {
   reset_for(turn_on_logging);
   hci->turn_on_logging(logging_path);
diff --git a/osi/include/eager_reader.h b/osi/include/eager_reader.h
index a99e120a7..3bf1e6edf 100644
--- a/osi/include/eager_reader.h
+++ b/osi/include/eager_reader.h
@@ -55,13 +55,9 @@ void eager_reader_register(eager_reader_t *reader, reactor_t *reactor, eager_rea
 // function is idempotent.
 void eager_reader_unregister(eager_reader_t *reader);
 
-// Returns the next byte in the stream, blocks if not available yet.
-// |reader| may not be NULL.
+// Reads up to |max_size| bytes into |buffer|. If |block| is true, blocks until
+// |max_size| bytes are read. Otherwise only reads from currently available bytes.
 // NOT SAFE FOR READING FROM MULTIPLE THREADS
 // but you should probably only be reading from one thread anyway,
 // otherwise the byte stream probably doesn't make sense.
-uint8_t eager_reader_read_byte(eager_reader_t *reader);
-
-// Returns true if there is at least one byte to read in the stream.
-// |reader| may not be NULL.
-bool eager_reader_has_byte(eager_reader_t *reader);
+size_t eager_reader_read(eager_reader_t *reader, uint8_t *buffer, size_t max_size, bool block);
diff --git a/osi/src/eager_reader.c b/osi/src/eager_reader.c
index 8eec87255..c275003fc 100644
--- a/osi/src/eager_reader.c
+++ b/osi/src/eager_reader.c
@@ -57,6 +57,7 @@ struct eager_reader_t {
   void *outbound_context;
 };
 
+static bool has_byte(const eager_reader_t *reader);
 static void inbound_data_waiting(void *context);
 static void internal_outbound_read_ready(void *context);
 
@@ -163,29 +164,38 @@ void eager_reader_unregister(eager_reader_t *reader) {
 }
 
 // SEE HEADER FOR THREAD SAFETY NOTE
-uint8_t eager_reader_read_byte(eager_reader_t *reader) {
+size_t eager_reader_read(eager_reader_t *reader, uint8_t *buffer, size_t max_size, bool block) {
   assert(reader != NULL);
+  assert(buffer != NULL);
 
-  eventfd_t value;
-  if (eventfd_read(reader->bytes_available_fd, &value) == -1)
-    ALOGE("%s unable to read semaphore for output data.", __func__);
+  size_t bytes_read = 0;
 
-  if (!reader->current_buffer)
-    reader->current_buffer = fixed_queue_dequeue(reader->buffers);
+  while (bytes_read < max_size) {
+    if (!block && !has_byte(reader))
+      return bytes_read;
 
-  uint8_t byte = reader->current_buffer->data[reader->current_buffer->offset];
-  reader->current_buffer->offset++;
+    eventfd_t value;
+    if (eventfd_read(reader->bytes_available_fd, &value) == -1)
+      ALOGE("%s unable to read semaphore for output data.", __func__);
 
-  // Prep for next byte request
-  if (reader->current_buffer->offset >= reader->current_buffer->length) {
-    reader->allocator->free(reader->current_buffer);
-    reader->current_buffer = NULL;
+    if (!reader->current_buffer)
+      reader->current_buffer = fixed_queue_dequeue(reader->buffers);
+
+    buffer[bytes_read] = reader->current_buffer->data[reader->current_buffer->offset];
+    reader->current_buffer->offset++;
+    bytes_read++;
+
+    // Prep for next byte
+    if (reader->current_buffer->offset >= reader->current_buffer->length) {
+      reader->allocator->free(reader->current_buffer);
+      reader->current_buffer = NULL;
+    }
   }
 
-  return byte;
+  return bytes_read;
 }
 
-bool eager_reader_has_byte(eager_reader_t *reader) {
+static bool has_byte(const eager_reader_t *reader) {
   assert(reader != NULL);
 
   fd_set read_fds;
diff --git a/osi/test/eager_reader_test.cpp b/osi/test/eager_reader_test.cpp
index 630611510..84ad4dd3a 100644
--- a/osi/test/eager_reader_test.cpp
+++ b/osi/test/eager_reader_test.cpp
@@ -32,7 +32,38 @@ extern "C" {
 
 #define BUFFER_SIZE 32
 
-static char *small_data = (char *)"white chocolate lindor truffles";
+static const char *small_data = "white chocolate lindor truffles";
+static const char *large_data =
+  "Let him make him examine and thoroughly sift everything he reads, and "
+  "lodge nothing in his fancy upon simple authority and upon trust. "
+  "Aristotle's principles will then be no more principles to him, than those "
+  "of Epicurus and the Stoics: let this diversity of opinions be propounded "
+  "to, and laid before him; he will himself choose, if he be able; if not, "
+  "he will remain in doubt. "
+  ""
+  "   \"Che non men the saver, dubbiar m' aggrata.\" "
+  "   [\"I love to doubt, as well as to know.\"--Dante, Inferno, xi. 93] "
+  ""
+  "for, if he embrace the opinions of Xenophon and Plato, by his own reason, "
+  "they will no more be theirs, but become his own.  Who follows another, "
+  "follows nothing, finds nothing, nay, is inquisitive after nothing. "
+  ""
+  "   \"Non sumus sub rege; sibi quisque se vindicet.\" "
+  "   [\"We are under no king; let each vindicate himself.\" --Seneca, Ep.,33] "
+  ""
+  "let him, at least, know that he knows.  it will be necessary that he "
+  "imbibe their knowledge, not that he be corrupted with their precepts; "
+  "and no matter if he forget where he had his learning, provided he know "
+  "how to apply it to his own use.  truth and reason are common to every "
+  "one, and are no more his who spake them first, than his who speaks them "
+  "after: 'tis no more according to plato, than according to me, since both "
+  "he and i equally see and understand them.  bees cull their several sweets "
+  "from this flower and that blossom, here and there where they find them, "
+  "but themselves afterwards make the honey, which is all and purely their "
+  "own, and no more thyme and marjoram: so the several fragments he borrows "
+  "from others, he will transform and shuffle together to compile a work "
+  "that shall be absolutely his own; that is to say, his judgment: "
+  "his instruction, labour and study, tend to nothing else but to form that. ";
 
 static semaphore_t *done;
 
@@ -51,34 +82,50 @@ class EagerReaderTest : public ::testing::Test {
 };
 
 static void expect_data(eager_reader_t *reader, void *context) {
-  EXPECT_TRUE(eager_reader_has_byte(reader)) << "if we got a callback we expect there to be data";
+  char *data = (char *)context;
+  int length = strlen(data);
 
+  for (int i = 0; i < length; i++) {
+    uint8_t byte;
+    EXPECT_EQ((size_t)1, eager_reader_read(reader, &byte, 1, true));
+    EXPECT_EQ(data[i], byte);
+  }
+
+  semaphore_post(done);
+}
+
+static void expect_data_multibyte(eager_reader_t *reader, void *context) {
   char *data = (char *)context;
   int length = strlen(data);
-  int i;
 
-  for (i = 0; i < length; i++) {
-    EXPECT_EQ(data[i], eager_reader_read_byte(reader));
+  for (int i = 0; i < length;) {
+    uint8_t buffer[28];
+    int bytes_to_read = (length - i) > 28 ? 28 : (length - i);
+    int bytes_read = eager_reader_read(reader, buffer, bytes_to_read, false);
+    EXPECT_EQ(bytes_to_read, bytes_read);
+    for (int j = 0; j < bytes_read && i < length; j++, i++) {
+      EXPECT_EQ(data[i], buffer[j]);
+    }
   }
 
   semaphore_post(done);
 }
 
 TEST_F(EagerReaderTest, test_new_simple) {
-  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, (char *)"test_thread");
+  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, "test_thread");
   ASSERT_TRUE(reader != NULL);
 }
 
 TEST_F(EagerReaderTest, test_free_simple) {
-  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, (char *)"test_thread");
+  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, "test_thread");
   eager_reader_free(reader);
 }
 
 TEST_F(EagerReaderTest, test_small_data) {
-  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, (char *)"test_thread");
+  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, "test_thread");
 
   thread_t *read_thread = thread_new("read_thread");
-  eager_reader_register(reader, thread_get_reactor(read_thread), expect_data, small_data);
+  eager_reader_register(reader, thread_get_reactor(read_thread), expect_data, (void *)small_data);
 
   write(pipefd[1], small_data, strlen(small_data));
 
@@ -86,3 +133,16 @@ TEST_F(EagerReaderTest, test_small_data) {
   eager_reader_free(reader);
   thread_free(read_thread);
 }
+
+TEST_F(EagerReaderTest, test_large_data_multibyte) {
+  eager_reader_t *reader = eager_reader_new(pipefd[0], &allocator_malloc, BUFFER_SIZE, SIZE_MAX, "test_thread");
+
+  thread_t *read_thread = thread_new("read_thread");
+  eager_reader_register(reader, thread_get_reactor(read_thread), expect_data_multibyte, (void *)large_data);
+
+  write(pipefd[1], large_data, strlen(large_data));
+
+  semaphore_wait(done);
+  eager_reader_free(reader);
+  thread_free(read_thread);
+}