CharacteristicBuffer: make it be a stream class; add locking

Author: dhalbert

ports/nrf/Makefile

@@ -142,6 +142,7 @@ SRC_C += \
 	peripherals/nrf/$(MCU_CHIP)/pins.c \
 	peripherals/nrf/$(MCU_CHIP)/power.c \
 	peripherals/nrf/timers.c \
+	sd_mutex.c \
 	supervisor/shared/memory.c
 
 

ports/nrf/common-hal/bleio/Characteristic.c

@@ -36,9 +36,9 @@
 #include "common-hal/bleio/Characteristic.h"
 #include "shared-module/bleio/Characteristic.h"
 
-// TODO - should these be per object?? *****
 STATIC volatile bleio_characteristic_obj_t *m_read_characteristic;
 STATIC volatile uint8_t m_tx_in_progress;
+// Serialize gattc writes that send a response. This might be done per object?
 STATIC nrf_mutex_t *m_write_mutex;
 
 STATIC uint16_t get_cccd(bleio_characteristic_obj_t *characteristic) {

ports/nrf/common-hal/bleio/CharacteristicBuffer.c

@@ -3,7 +3,7 @@
  *
  * The MIT License (MIT)
  *
- * Copyright (c) 2018 Artur Pacholec
+ * Copyright (c) 2019 Dan Halbert for Adafruit Industries
  *
  * Permission is hereby granted, free of charge, to any person obtaining a copy
  * of this software and associated documentation files (the "Software"), to deal
@@ -29,9 +29,13 @@
 
 #include "ble_drv.h"
 #include "ble_gatts.h"
-#include "nrf_soc.h"
+#include "sd_mutex.h"
 
+#include "lib/utils/interrupt_char.h"
 #include "py/runtime.h"
+#include "py/stream.h"
+
+#include "tick.h"
 
 #include "common-hal/bleio/__init__.h"
 #include "common-hal/bleio/CharacteristicBuffer.h"
@@ -43,33 +47,89 @@ STATIC void characteristic_buffer_on_ble_evt(ble_evt_t *ble_evt, void *param) {
         ble_gatts_evt_write_t *evt_write = &ble_evt->evt.gatts_evt.params.write;
         // Event handle must match the handle for my characteristic.
         if (evt_write->handle == self->characteristic->handle) {
-            // Push all the data onto the ring buffer.
+            // Push all the data onto the ring buffer, but wait for any reads to finish.
+            sd_mutex_acquire_wait_no_vm(&self->ringbuf_mutex);
             for (size_t i = 0; i < evt_write->len; i++) {
                 ringbuf_put(&self->ringbuf, evt_write->data[i]);
             }
+            // Don't check for errors: we're in an event handler.
+            sd_mutex_release(&self->ringbuf_mutex);
         break;
         }
     }
     }
 
 }
 
-// Assumes that buffer_size has been validated before call.
-void common_hal_bleio_characteristic_buffer_construct(bleio_characteristic_buffer_obj_t *self, bleio_characteristic_obj_t *characteristic, size_t buffer_size) {
+// Assumes that timeout and buffer_size have been validated before call.
+void common_hal_bleio_characteristic_buffer_construct(bleio_characteristic_buffer_obj_t *self,
+                                                      bleio_characteristic_obj_t *characteristic,
+                                                      mp_float_t timeout,
+                                                      size_t buffer_size) {
 
     self->characteristic = characteristic;
+    self->timeout_ms = timeout * 1000;
     // This is a macro.
-    ringbuf_alloc(&self->ringbuf, buffer_size);
+    // true means long-lived, so it won't be moved.
+    ringbuf_alloc(&self->ringbuf, buffer_size, true);
+    sd_mutex_new(&self->ringbuf_mutex);
 
     ble_drv_add_event_handler(characteristic_buffer_on_ble_evt, self);
 
 }
 
-// Returns a uint8_t byte value, or -1 if no data is available.
-int common_hal_bleio_characteristic_buffer_read(bleio_characteristic_buffer_obj_t *self) {
-    return ringbuf_get(&self->ringbuf);
+int common_hal_bleio_characteristic_buffer_read(bleio_characteristic_buffer_obj_t *self, uint8_t *data, size_t len, int *errcode) {
+    uint64_t start_ticks = ticks_ms;
+
+    // Wait for all bytes received or timeout
+    while ( (ringbuf_count(&self->ringbuf) < len) && (ticks_ms - start_ticks < self->timeout_ms) ) {
+#ifdef MICROPY_VM_HOOK_LOOP
+        MICROPY_VM_HOOK_LOOP ;
+        // Allow user to break out of a timeout with a KeyboardInterrupt.
+        if ( mp_hal_is_interrupted() ) {
+            return 0;
+        }
+#endif
+    }
+
+    // Copy received data. Lock out writes while copying.
+    sd_mutex_acquire_wait(&self->ringbuf_mutex);
+
+    size_t rx_bytes = MIN(ringbuf_count(&self->ringbuf), len);
+    for ( size_t i = 0; i < rx_bytes; i++ ) {
+        data[i] = ringbuf_get(&self->ringbuf);
+    }
+
+    // Writes now OK.
+    sd_mutex_release_check(&self->ringbuf_mutex);
+
+    return rx_bytes;
+}
+
+uint32_t common_hal_bleio_characteristic_buffer_rx_characters_available(bleio_characteristic_buffer_obj_t *self) {
+    return ringbuf_count(&self->ringbuf);
+}
+
+void common_hal_bleio_characteristic_buffer_clear_rx_buffer(bleio_characteristic_buffer_obj_t *self) {
+    // prevent conflict with uart irq
+    sd_mutex_acquire_wait(&self->ringbuf_mutex);
+    ringbuf_clear(&self->ringbuf);
+    sd_mutex_release_check(&self->ringbuf_mutex);
+}
+
+bool common_hal_bleio_characteristic_buffer_deinited(bleio_characteristic_buffer_obj_t *self) {
+    return self->characteristic == NULL;
 }
 
 void common_hal_bleio_characteristic_buffer_deinit(bleio_characteristic_buffer_obj_t *self) {
-    ble_drv_remove_event_handler(characteristic_buffer_on_ble_evt, self);
+    if (!common_hal_bleio_characteristic_buffer_deinited(self)) {
+        ble_drv_remove_event_handler(characteristic_buffer_on_ble_evt, self);
+    }
+}
+
+bool common_hal_bleio_characteristic_buffer_connected(bleio_characteristic_buffer_obj_t *self) {
+    return self->characteristic != NULL &&
+        self->characteristic->service != NULL &&
+        self->characteristic->service->device != NULL &&
+        common_hal_bleio_device_get_conn_handle(self->characteristic->service->device) != BLE_CONN_HANDLE_INVALID;
 }

ports/nrf/common-hal/bleio/CharacteristicBuffer.h

@@ -27,15 +27,18 @@
 #ifndef MICROPY_INCLUDED_COMMON_HAL_BLEIO_CHARACTERISTICBUFFER_H
 #define MICROPY_INCLUDED_COMMON_HAL_BLEIO_CHARACTERISTICBUFFER_H
 
-#include "py/ringbuf.h"
+#include "nrf_soc.h"
 
+#include "py/ringbuf.h"
 #include "shared-bindings/bleio/Characteristic.h"
 
 typedef struct {
     mp_obj_base_t base;
     bleio_characteristic_obj_t *characteristic;
+    uint32_t timeout_ms;
     // Ring buffer storing consecutive incoming values.
     ringbuf_t ringbuf;
+    nrf_mutex_t ringbuf_mutex;
 } bleio_characteristic_buffer_obj_t;
 
 #endif // MICROPY_INCLUDED_COMMON_HAL_BLEIO_CHARACTERISTICBUFFER_H

ports/nrf/common-hal/bleio/Device.c

@@ -262,11 +262,13 @@ STATIC bool discover_services(bleio_device_obj_t *device, uint16_t start_handle)
         mp_raise_OSError_msg(translate("Failed to discover services"));
     }
 
+    // Serialize discovery.
     err_code = sd_mutex_acquire(m_discovery_mutex);
     if (err_code != NRF_SUCCESS) {
         mp_raise_OSError_msg(translate("Failed to acquire mutex"));
     }
 
+    // Wait for someone else to release m_discovery_mutex.
     while (sd_mutex_acquire(m_discovery_mutex) == NRF_ERROR_SOC_MUTEX_ALREADY_TAKEN) {
 #ifdef MICROPY_VM_HOOK_LOOP
     MICROPY_VM_HOOK_LOOP

ports/nrf/common-hal/busio/UART.c

@@ -52,33 +52,6 @@
 
 static uint32_t get_nrf_baud (uint32_t baudrate);
 
-static uint16_t ringbuf_count(ringbuf_t *r)
-{
-    volatile int count = r->iput - r->iget;
-    if ( count < 0 ) {
-        count += r->size;
-    }
-
-    return (uint16_t) count;
-}
-
-static void ringbuf_clear(ringbuf_t *r)
-{
-    r->iput = r->iget = 0;
-}
-
-// will overwrite old data
-static void ringbuf_put_n(ringbuf_t* r, uint8_t* buf, uint8_t bufsize)
-{
-    for(uint8_t i=0; i < bufsize; i++) {
-        if ( ringbuf_put(r, buf[i]) < 0 ) {
-            // if full overwrite old data
-            (void) ringbuf_get(r);
-            ringbuf_put(r, buf[i]);
-        }
-    }
-}
-
 static void uart_callback_irq (const nrfx_uarte_event_t * event, void * context) {
     busio_uart_obj_t* self = (busio_uart_obj_t*) context;
 
@@ -145,16 +118,20 @@ void common_hal_busio_uart_construct (busio_uart_obj_t *self,
 
     // Init buffer for rx
     if ( rx != mp_const_none ) {
-        self->rbuf.buf = (uint8_t *) gc_alloc(receiver_buffer_size, false, false);
+        // Initially allocate the UART's buffer in the long-lived part of the
+        // heap.  UARTs are generally long-lived objects, but the "make long-
+        // lived" machinery is incapable of moving internal pointers like
+        // self->buffer, so do it manually.  (However, as long as internal
+        // pointers like this are NOT moved, allocating the buffer
+        // in the long-lived pool is not strictly necessary)
+        // (This is a macro.)
+        ringbuf_alloc(&self->rbuf, receiver_buffer_size, true);
 
         if ( !self->rbuf.buf ) {
             nrfx_uarte_uninit(&self->uarte);
             mp_raise_msg(&mp_type_MemoryError, translate("Failed to allocate RX buffer"));
         }
 
-        self->rbuf.size = receiver_buffer_size;
-        self->rbuf.iget = self->rbuf.iput = 0;
-
         self->rx_pin_number = rx->number;
         claim_pin(rx);
     }

ports/nrf/sd_mutex.c

@@ -0,0 +1,56 @@
+/*
+ * This file is part of the MicroPython project, http://micropython.org/
+ *
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2019 Dan Halbert for Adafruit Industries
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+#include "py/mpconfig.h"
+#include "py/runtime.h"
+#include "nrf_soc.h"
+
+void sd_mutex_acquire_check(nrf_mutex_t* p_mutex) {
+    uint32_t err_code = sd_mutex_acquire(p_mutex);
+    if (err_code != NRF_SUCCESS) {
+        mp_raise_OSError_msg_varg(translate("Failed to acquire mutex, err 0x%04x"), err_code);
+    }
+}
+
+void sd_mutex_acquire_wait(nrf_mutex_t* p_mutex) {
+    while (sd_mutex_acquire(p_mutex) == NRF_ERROR_SOC_MUTEX_ALREADY_TAKEN) {
+#ifdef MICROPY_VM_HOOK_LOOP
+    MICROPY_VM_HOOK_LOOP
+#endif
+    }
+}
+
+void sd_mutex_acquire_wait_no_vm(nrf_mutex_t* p_mutex) {
+    while (sd_mutex_acquire(p_mutex) == NRF_ERROR_SOC_MUTEX_ALREADY_TAKEN) {
+    }
+}
+
+void sd_mutex_release_check(nrf_mutex_t* p_mutex) {
+    uint32_t err_code = sd_mutex_release(p_mutex);
+    if (err_code != NRF_SUCCESS) {
+        mp_raise_OSError_msg_varg(translate("Failed to release mutex, err 0x%04x"), err_code);
+    }
+}

ports/nrf/sd_mutex.h

@@ -0,0 +1,46 @@
+/*
+ * This file is part of the MicroPython project, http://micropython.org/
+ *
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2019 Dan Halbert for Adafruit Industries
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+#ifndef MICROPY_INCLUDED_NRF_SD_MUTEX_H
+#define MICROPY_INCLUDED_NRF_SD_MUTEX_H
+
+#include "nrf_soc.h"
+
+// Helpers for common usage of nrf_mutex.
+
+// Try to acquire a mutex right now. Raise exception if we can't get it.
+void sd_mutex_acquire_check(nrf_mutex_t* p_mutex);
+
+// Wait for a mutex to become available. Run VM background tasks while waiting.
+void sd_mutex_acquire_wait(nrf_mutex_t* p_mutex);
+
+// Wait for a mutex to become available.. Block VM while waiting.
+void sd_mutex_acquire_wait_no_vm(nrf_mutex_t* p_mutex);
+
+// Release a mutex, and raise exception on error.
+void sd_mutex_release_check(nrf_mutex_t* p_mutex);
+
+#endif // MICROPY_INCLUDED_NRF_SD_MUTEX_H

py/ringbuf.h

@@ -26,6 +26,8 @@
 #ifndef MICROPY_INCLUDED_PY_RINGBUF_H
 #define MICROPY_INCLUDED_PY_RINGBUF_H
 
+#include "py/gc.h"
+
 #include <stdint.h>
 
 typedef struct _ringbuf_t {
@@ -40,9 +42,9 @@ typedef struct _ringbuf_t {
 // ringbuf_t buf = {buf_array, sizeof(buf_array)};
 
 // Dynamic initialization. This creates root pointer!
-#define ringbuf_alloc(r, sz) \
+#define ringbuf_alloc(r, sz, long_lived)                   \
 { \
-    (r)->buf = m_new(uint8_t, sz); \
+    (r)->buf = gc_alloc(sz, false, long_lived);   \
     (r)->size = sz; \
     (r)->iget = (r)->iput = 0; \
 }
@@ -71,4 +73,30 @@ static inline int ringbuf_put(ringbuf_t *r, uint8_t v) {
     return 0;
 }
 
+static inline uint16_t ringbuf_count(ringbuf_t *r)
+{
+    volatile int count = r->iput - r->iget;
+    if ( count < 0 ) {
+        count += r->size;
+    }
+
+    return (uint16_t) count;
+}
+
+static inline void ringbuf_clear(ringbuf_t *r)
+{
+    r->iput = r->iget = 0;
+}
+
+// will overwrite old data
+static inline void ringbuf_put_n(ringbuf_t* r, uint8_t* buf, uint8_t bufsize)
+{
+    for(uint8_t i=0; i < bufsize; i++) {
+        if ( ringbuf_put(r, buf[i]) < 0 ) {
+            // if full overwrite old data
+            (void) ringbuf_get(r);
+            ringbuf_put(r, buf[i]);
+        }
+    }
+}
 #endif // MICROPY_INCLUDED_PY_RINGBUF_H

py/stream.c

@@ -103,7 +103,7 @@ STATIC mp_obj_t stream_read_generic(size_t n_args, const mp_obj_t *args, byte fl
     // CPython does a readall, but here we silently let negatives through,
     // and they will cause a MemoryError.
     mp_int_t sz;
-    if (n_args == 1 || ((sz = mp_obj_get_int(args[1])) == -1)) {
+    if (n_args == 1 || args[1] == mp_const_none || ((sz = mp_obj_get_int(args[1])) == -1)) {
         return stream_readall(args[0]);
     }