sensors: make the HAL thread safe

According to the Sensors HAL interface documentation, the implementation
must be thread safe and allow functions defined in sensors.h to be called
from different threads at the same time.
Use a mutex to synchronize the HAL. In addition, initialize the acquisition
pipes with the O_NONBLOCK flag to return immediately in case another tread
which calls poll() reads the data.
Put the read pipe call in a loop as the read system call might return a number
less than the number of requested bytes.

BUG=none
Change-Id: I6ee3f8e0c2ba19942da8b82da1660c9163c5d237
Tracked-On: https://jira01.devtools.intel.com/browse/BP-247
Signed-off-by: Constantin Musca <constantin.musca@intel.com>

4 files changed, 48 insertions, 10 deletions

diff --git a/peripheral/sensors/mraa/AcquisitionThread.cpp b/peripheral/sensors/mraa/AcquisitionThread.cpp
index c1c9206..8485508 100644
--- a/peripheral/sensors/mraa/AcquisitionThread.cpp
+++ b/peripheral/sensors/mraa/AcquisitionThread.cpp
@@ -121,7 +121,7 @@ bool AcquisitionThread::init() {
   }
 
   /* create pipe to signal events to the main thread */
-  rc = pipe(pipeFds);
+  rc = pipe2(pipeFds, O_NONBLOCK);
   if (rc != 0) {
     ALOGE("%s: Cannot initialize pipe", __func__);
     goto pipe_err;
@@ -175,7 +175,12 @@ bool AcquisitionThread::generateFlushCompleteEvent() {
   data.meta_data.sensor = sensor->getHandle();
   data.meta_data.what = META_DATA_FLUSH_COMPLETE;
 
-  /* send the event via the associated pipe */
+  /*
+   * Send the event via the associated pipe. It doesn't need to be in a loop
+   * as O_NONBLOCK is enabled and the number of bytes is <= PIPE_BUF.
+   * If there is room to write n bytes to the pipe, then write succeeds
+   * immediately, writing all n bytes; otherwise write fails.
+   */
   rc = write(getWritePipeFd(), &data, sizeof(sensors_event_t));
   if (rc != sizeof(sensors_event_t)) {
     ALOGE("%s: not all data has been sent over the pipe", __func__);
diff --git a/peripheral/sensors/mraa/Sensor.cpp b/peripheral/sensors/mraa/Sensor.cpp
index db76bda..849f380 100644
--- a/peripheral/sensors/mraa/Sensor.cpp
+++ b/peripheral/sensors/mraa/Sensor.cpp
@@ -36,7 +36,9 @@ Sensor::~Sensor() {
 int Sensor::activate(int handle, int enabled) { return 0; }
 
 bool Sensor::readOneEvent(sensors_event_t *event) {
-  int rc;
+  int bytes_read = 0, bytes_to_read = sizeof(sensors_event_t);
+  int fd = -1;
+  char *ptr = (char *)event;
 
   if (acquisitionThread == nullptr) {
     ALOGE("%s: sensor %d doesn't have an acquisition thread", __func__, handle);
@@ -44,8 +46,17 @@ bool Sensor::readOneEvent(sensors_event_t *event) {
   }
 
   /* read one event from the pipe read endpoint */
-  rc = read(acquisitionThread->getReadPipeFd(), event, sizeof(sensors_event_t));
-  if (rc != sizeof(sensors_event_t)) {
+  fd = acquisitionThread->getReadPipeFd();
+  do {
+    bytes_read = read(fd, ptr, bytes_to_read);
+    if (bytes_read <= 0) {
+      break;
+    }
+    bytes_to_read -= bytes_read;
+    ptr += bytes_read;
+  } while (bytes_to_read > 0);
+
+  if (bytes_to_read != 0) {
     return false;
   }
 
diff --git a/peripheral/sensors/mraa/SensorsHAL.cpp b/peripheral/sensors/mraa/SensorsHAL.cpp
index 94f768d..eebcd45 100644
--- a/peripheral/sensors/mraa/SensorsHAL.cpp
+++ b/peripheral/sensors/mraa/SensorsHAL.cpp
@@ -24,6 +24,7 @@
 Sensor * (*SensorContext::sensorFactoryFuncs[MAX_DEVICES])(int);
 struct sensor_t SensorContext::sensorDescs[MAX_DEVICES];
 int SensorContext::sensorsNum = 0;
+android::Mutex SensorContext::mutex;
 
 SensorContext::SensorContext(const hw_module_t *module) {
   /* create the epoll fd used to register the incoming fds */
@@ -65,6 +66,8 @@ SensorContext::~SensorContext() {
 
 int SensorContext::addSensorModule(struct sensor_t *sensorDesc,
     Sensor * (*sensorFactoryFunc)(int)) {
+  android::Mutex::Autolock autolock(mutex);
+
   if ((sensorDesc == nullptr) || (sensorFactoryFunc == nullptr)) {
     ALOGE("%s: cannot add a null sensor", __func__);
     return -EINVAL;
@@ -102,6 +105,8 @@ int SensorContext::OpenWrapper(const struct hw_module_t *module,
 
 int SensorContext::GetSensorsListWrapper(struct sensors_module_t *module,
                             struct sensor_t const **list) {
+  android::Mutex::Autolock autolock(mutex);
+
   if (!list || (sensorsNum == 0)) {
     return 0;
   }
@@ -188,27 +193,32 @@ int SensorContext::pollEvents(sensors_event_t *data, int count) {
       return nfds;
     }
 
+    mutex.lock();
     for(i = 0; i < nfds && returnedEvents < count; i++) {
       if (ev[i].events == EPOLLIN) {
         sensorIndex = ev[i].data.u32;
         if ((sensorIndex < 0) || (sensorIndex > sensorsNum)) {
           ALOGE("%s: Invalid sensor index", __func__);
+          mutex.unlock();
           return -1;
         }
 
         if (sensors[sensorIndex] == nullptr) {
-          ALOGE("%s: Sensor %d is not activated", __func__, sensorIndex);
-          return -1;
+          /* The sensor might have been deactivated by another thread */
+          continue;
         }
 
+        /*
+         * The read operation might fail if the data is read by another
+         * pollEvents call executed by another thread.
+         */
         if (sensors[sensorIndex]->readOneEvent(data + returnedEvents)) {
           returnedEvents++;
-        } else {
-          ALOGE("%s: Cannot read event from sensor %d", __func__, sensorIndex);
-          return -1;
         }
       }
     }
+    mutex.unlock();
+
     if (returnedEvents > 0) {
       return returnedEvents;
     }
@@ -248,6 +258,7 @@ int SensorContext::flush(int handle) {
 
 int SensorContext::CloseWrapper(hw_device_t *dev) {
   SensorContext *sensorContext = reinterpret_cast<SensorContext *>(dev);
+  android::Mutex::Autolock autolock(mutex);
 
   if (sensorContext != nullptr) {
     delete sensorContext;
@@ -258,11 +269,15 @@ int SensorContext::CloseWrapper(hw_device_t *dev) {
 
 int SensorContext::ActivateWrapper(sensors_poll_device_t *dev,
                                    int handle, int enabled) {
+  android::Mutex::Autolock autolock(mutex);
+
   return reinterpret_cast<SensorContext *>(dev)->activate(handle, enabled);
 }
 
 int SensorContext::SetDelayWrapper(sensors_poll_device_t *dev,
                                    int handle, int64_t ns) {
+  android::Mutex::Autolock autolock(mutex);
+
   return reinterpret_cast<SensorContext *>(dev)->setDelay(handle, ns);
 }
 
@@ -273,12 +288,16 @@ int SensorContext::PollEventsWrapper(sensors_poll_device_t *dev,
 
 int SensorContext::BatchWrapper(sensors_poll_device_1_t *dev, int handle,
                                 int flags, int64_t period_ns, int64_t timeout) {
+  android::Mutex::Autolock autolock(mutex);
+
   return reinterpret_cast<SensorContext *>(dev)->batch(handle, flags, period_ns,
                                                       timeout);
 }
 
 int SensorContext::FlushWrapper(sensors_poll_device_1_t *dev,
                                 int handle) {
+  android::Mutex::Autolock autolock(mutex);
+
   return reinterpret_cast<SensorContext *>(dev)->flush(handle);
 }
 
diff --git a/peripheral/sensors/mraa/SensorsHAL.hpp b/peripheral/sensors/mraa/SensorsHAL.hpp
index 74bc8a4..0e7ef14 100644
--- a/peripheral/sensors/mraa/SensorsHAL.hpp
+++ b/peripheral/sensors/mraa/SensorsHAL.hpp
@@ -18,6 +18,7 @@
 #define SENSORS_HAL_HPP
 
 #include <hardware/sensors.h>
+#include <utils/Mutex.h>
 #include "Sensor.hpp"
 #include "SensorUtils.hpp"
 
@@ -109,6 +110,8 @@ class SensorContext {
     static struct sensor_t sensorDescs[MAX_DEVICES];
     /* Number of registered sensors */
     static int sensorsNum;
+    /* Mutex used to synchronize the Sensor Context */
+    static android::Mutex mutex;
 };
 
 #endif  // SENSORS_HAL_HPP