SensorService performance improvements.

i) Send ack for wake_up sensors on the socket connection instead of using Binder RPC.
  ii) Cache events per connection in case there are write failures. Compute cache size
      from FIFO counts of sensors.
 iii) Send FlushCompleteEvent only for apps that explicitly called flush().

Change-Id: I018969736b7794b1b930529586f2294a03ee8667

include/gui/BitTube.h

@@ -48,6 +48,9 @@ public:
     // get receive file-descriptor
     int getFd() const;
 
+    // get the send file-descriptor.
+    int getSendFd() const;
+
     // send objects (sized blobs). All objects are guaranteed to be written or the call fails.
     template <typename T>
     static ssize_t sendObjects(const sp<BitTube>& tube,

include/gui/ISensorEventConnection.h

@@ -40,7 +40,6 @@ public:
                                    nsecs_t maxBatchReportLatencyNs, int reservedFlags) = 0;
     virtual status_t setEventRate(int handle, nsecs_t ns) = 0;
     virtual status_t flush() = 0;
-    virtual void decreaseWakeLockRefCount() = 0;
 };
 
 // ----------------------------------------------------------------------------

libs/gui/BitTube.cpp

@@ -99,6 +99,11 @@ int BitTube::getFd() const
     return mReceiveFd;
 }
 
+int BitTube::getSendFd() const
+{
+    return mSendFd;
+}
+
 ssize_t BitTube::write(void const* vaddr, size_t size)
 {
     ssize_t err, len;

libs/gui/ISensorEventConnection.cpp

@@ -34,8 +34,7 @@ enum {
     GET_SENSOR_CHANNEL = IBinder::FIRST_CALL_TRANSACTION,
     ENABLE_DISABLE,
     SET_EVENT_RATE,
-    FLUSH_SENSOR,
+    FLUSH_SENSOR
-    DECREASE_WAKE_LOCK_REFCOUNT
 };
 
 class BpSensorEventConnection : public BpInterface<ISensorEventConnection>
@@ -84,13 +83,6 @@ public:
         remote()->transact(FLUSH_SENSOR, data, &reply);
         return reply.readInt32();
     }
-
-    virtual void decreaseWakeLockRefCount() {
-        Parcel data, reply;
-        data.writeInterfaceToken(ISensorEventConnection::getInterfaceDescriptor());
-        remote()->transact(DECREASE_WAKE_LOCK_REFCOUNT, data, &reply, IBinder::FLAG_ONEWAY);
-        return;
-    }
 };
 
 IMPLEMENT_META_INTERFACE(SensorEventConnection, "android.gui.SensorEventConnection");
@@ -133,11 +125,6 @@ status_t BnSensorEventConnection::onTransact(
             reply->writeInt32(result);
             return NO_ERROR;
         } break;
-        case DECREASE_WAKE_LOCK_REFCOUNT: {
-            CHECK_INTERFACE(ISensorEventConnection, data, reply);
-            decreaseWakeLockRefCount();
-            return NO_ERROR;
-        } break;
     }
     return BBinder::onTransact(code, data, reply, flags);
 }

libs/gui/SensorEventQueue.cpp

@@ -18,6 +18,7 @@
 
 #include <stdint.h>
 #include <sys/types.h>
+#include <sys/socket.h>
 
 #include <utils/Errors.h>
 #include <utils/RefBase.h>
@@ -147,7 +148,14 @@ status_t SensorEventQueue::setEventRate(Sensor const* sensor, nsecs_t ns) const
 void SensorEventQueue::sendAck(const ASensorEvent* events, int count) {
     for (int i = 0; i < count; ++i) {
         if (events[i].flags & WAKE_UP_SENSOR_EVENT_NEEDS_ACK) {
-            mSensorEventConnection->decreaseWakeLockRefCount();
+            // Send just a byte of data to acknowledge for the wake up sensor events
+            // received
+            char buf = '1';
+            ssize_t size = ::send(mSensorChannel->getFd(), &buf, sizeof(buf),
+                                             MSG_DONTWAIT | MSG_NOSIGNAL);
+            if (size < 0) {
+                ALOGE("sendAck failure %d", size);
+            }
         }
     }
     return;

services/sensorservice/SensorService.cpp

@@ -18,6 +18,7 @@
 #include <math.h>
 #include <stdint.h>
 #include <sys/types.h>
+#include <sys/socket.h>
 
 #include <cutils/properties.h>
 
@@ -165,6 +166,7 @@ void SensorService::onFirstRef()
 
             mWakeLockAcquired = false;
             run("SensorService", PRIORITY_URGENT_DISPLAY);
+            mLooper = new Looper(false);
             mInitCheck = NO_ERROR;
         }
     }
@@ -340,6 +342,8 @@ bool SensorService::threadLoop()
     SensorDevice& device(SensorDevice::getInstance());
     const size_t vcount = mVirtualSensorList.size();
 
+    SensorEventAckReceiver sender(this);
+    sender.run("SensorEventAckReceiver", PRIORITY_URGENT_DISPLAY);
     ssize_t count;
     const int halVersion = device.getHalDeviceVersion();
     do {
@@ -348,6 +352,11 @@ bool SensorService::threadLoop()
             ALOGE("sensor poll failed (%s)", strerror(-count));
             break;
         }
+
+        // Reset sensors_event_t.flags to zero for all events in the buffer.
+        for (int i = 0; i < count; i++) {
+             buffer[i].flags = 0;
+        }
         Mutex::Autolock _l(mLock);
         // Poll has returned. Hold a wakelock if one of the events is from a wake up sensor. The
         // rest of this loop is under a critical section protected by mLock. Acquiring a wakeLock,
@@ -443,6 +452,19 @@ bool SensorService::threadLoop()
     return false;
 }
 
+sp<Looper> SensorService::getLooper() const {
+    return mLooper;
+}
+
+bool SensorService::SensorEventAckReceiver::threadLoop() {
+    ALOGD("new thread SensorEventAckReceiver");
+    do {
+        sp<Looper> looper = mService->getLooper();
+        looper->pollOnce(-1);
+    } while(!Thread::exitPending());
+    return false;
+}
+
 void SensorService::recordLastValueLocked(
         const sensors_event_t* buffer, size_t count) {
     const sensors_event_t* last = NULL;
@@ -656,6 +678,12 @@ status_t SensorService::enable(const sp<SensorEventConnection>& connection,
         err = sensor->activate(connection.get(), true);
     }
 
+    if (err == NO_ERROR && sensor->getSensor().isWakeUpSensor()) {
+        // Add the file descriptor to the Looper for receiving acknowledgments;
+        int ret = mLooper->addFd(connection->getSensorChannel()->getSendFd(), 0,
+                                        ALOOPER_EVENT_INPUT, connection.get(), NULL);
+    }
+
     if (err != NO_ERROR) {
         // batch/activate has failed, reset our state.
         cleanupWithoutDisableLocked(connection, handle);
@@ -752,9 +780,8 @@ status_t SensorService::flushSensor(const sp<SensorEventConnection>& connection,
 
 
 bool SensorService::canAccessSensor(const Sensor& sensor) {
-    String16 permissionString(sensor.getRequiredPermission());
+    return (sensor.getRequiredPermission().isEmpty()) ||
-    return permissionString.size() == 0 ||
+            PermissionCache::checkCallingPermission(String16(sensor.getRequiredPermission()));
-            PermissionCache::checkCallingPermission(permissionString);
 }
 
 bool SensorService::verifyCanAccessSensor(const Sensor& sensor, const char* operation) {
@@ -797,7 +824,6 @@ void SensorService::checkWakeLockStateLocked() {
     }
 }
 // ---------------------------------------------------------------------------
-
 SensorService::SensorRecord::SensorRecord(
         const sp<SensorEventConnection>& connection)
 {
@@ -828,25 +854,30 @@ bool SensorService::SensorRecord::removeConnection(
 
 SensorService::SensorEventConnection::SensorEventConnection(
         const sp<SensorService>& service, uid_t uid)
-    : mService(service), mUid(uid), mWakeLockRefCount(0)
+    : mService(service), mUid(uid), mWakeLockRefCount(0), mEventCache(NULL), mCacheSize(0),
-{
+      mMaxCacheSize(0) {
     const SensorDevice& device(SensorDevice::getInstance());
     if (device.getHalDeviceVersion() >= SENSORS_DEVICE_API_VERSION_1_1) {
-        // Increase socket buffer size to 1MB for batching capabilities.
+        // Increase socket buffer size to a max of 100 KB for batching capabilities.
-        mChannel = new BitTube(service->mSocketBufferSize);
+        mChannel = new BitTube(mService->mSocketBufferSize);
     } else {
         mChannel = new BitTube(SOCKET_BUFFER_SIZE_NON_BATCHED);
     }
+#if DEBUG_CONNECTIONS
+    mEventsReceived = mEventsSentFromCache = mEventsSent = 0;
+#endif
 }
 
-SensorService::SensorEventConnection::~SensorEventConnection()
+SensorService::SensorEventConnection::~SensorEventConnection() {
-{
     ALOGD_IF(DEBUG_CONNECTIONS, "~SensorEventConnection(%p)", this);
+    if (mEventCache != NULL) {
+        delete mEventCache;
+    }
     mService->cleanupConnection(this);
 }
 
-void SensorService::SensorEventConnection::onFirstRef()
+void SensorService::SensorEventConnection::onFirstRef() {
-{
+    LooperCallback::onFirstRef();
 }
 
 bool SensorService::SensorEventConnection::needsWakeLock() {
@@ -856,15 +887,28 @@ bool SensorService::SensorEventConnection::needsWakeLock() {
 
 void SensorService::SensorEventConnection::dump(String8& result) {
     Mutex::Autolock _l(mConnectionLock);
-    result.appendFormat("%d WakeLockRefCount\n", mWakeLockRefCount);
+    result.appendFormat("\t %d WakeLockRefCount \n", mWakeLockRefCount);
     for (size_t i = 0; i < mSensorInfo.size(); ++i) {
         const FlushInfo& flushInfo = mSensorInfo.valueAt(i);
-        result.appendFormat("\t %s | status: %s | pending flush events %d | uid %d\n",
+        result.appendFormat("\t %s | status: %s | pending flush events %d | flush calls %d| uid %d|"
+                            "cache size: %d max cache size %d\n",
                             mService->getSensorName(mSensorInfo.keyAt(i)).string(),
                             flushInfo.mFirstFlushPending ? "First flush pending" :
                                                            "active",
                             flushInfo.mPendingFlushEventsToSend,
-                            mUid);
+                            flushInfo.mNumFlushCalls,
+                            mUid,
+                            mCacheSize,
+                            mMaxCacheSize);
+#if DEBUG_CONNECTIONS
+        result.appendFormat("\t events recvd: %d | sent %d | cache %d | dropped %d\n",
+                                        mEventsReceived,
+                                        mEventsSent,
+                                        mEventsSentFromCache,
+                                        mEventsReceived - (mEventsSentFromCache +
+                                                           mEventsSent + mCacheSize));
+#endif
+
     }
 }
 
@@ -910,8 +954,7 @@ void SensorService::SensorEventConnection::setFirstFlushPending(int32_t handle,
 
 status_t SensorService::SensorEventConnection::sendEvents(
         sensors_event_t const* buffer, size_t numEvents,
-        sensors_event_t* scratch)
+        sensors_event_t* scratch) {
-{
     // filter out events not for this connection
     size_t count = 0;
     Mutex::Autolock _l(mConnectionLock);
@@ -927,30 +970,127 @@ status_t SensorService::SensorEventConnection::sendEvents(
                 curr = buffer[i].meta_data.sensor;
             }
             ssize_t index = mSensorInfo.indexOfKey(curr);
-            if (index >= 0 && mSensorInfo[index].mFirstFlushPending == true &&
+            // Check if this connection has registered for this sensor. If not continue to the
-                buffer[i].type == SENSOR_TYPE_META_DATA) {
+            // next sensor_event.
-                // This is the first flush before activate is called. Events can now be sent for
+            if (index < 0) {
-                // this sensor on this connection.
+                ++i;
-                ALOGD_IF(DEBUG_CONNECTIONS, "First flush event for sensor==%d ",
+                continue;
-                         buffer[i].meta_data.sensor);
-                mSensorInfo.editValueAt(index).mFirstFlushPending = false;
             }
-            if (index >= 0 && mSensorInfo[index].mFirstFlushPending == false)  {
+
-                do {
+            // Check if there is a pending flush_complete event for this sensor on this connection.
+            FlushInfo& flushInfo = mSensorInfo.editValueAt(index);
+            if (buffer[i].type == SENSOR_TYPE_META_DATA) {
+                if (flushInfo.mFirstFlushPending == true) {
+                    // This is the first flush before activate is called. Events can now be sent for
+                    // this sensor on this connection.
+                    ALOGD_IF(DEBUG_CONNECTIONS, "First flush event for sensor==%d ",
+                                                    buffer[i].meta_data.sensor);
+                    flushInfo.mFirstFlushPending = false;
+                    ++i;
+                    continue;
+                }
+            }
+
+            // If there is a pending flush complete event for this sensor on this connection,
+            // ignore the event and proceed to the next.
+            if (flushInfo.mFirstFlushPending) {
+                ++i;
+                continue;
+            }
+
+            do {
+                if (buffer[i].type == SENSOR_TYPE_META_DATA) {
+                    // Send flush complete event only if flush() has been explicitly called by
+                    // this app else ignore.
+                    if (flushInfo.mNumFlushCalls > 0) {
+                        scratch[count++] = buffer[i];
+                        flushInfo.mNumFlushCalls--;
+                    }
+                    ++i;
+                } else {
+                    // Regular sensor event, just copy it to the scratch buffer.
                     scratch[count++] = buffer[i++];
-                } while ((i<numEvents) && ((buffer[i].sensor == curr) ||
+                }
-                         (buffer[i].type == SENSOR_TYPE_META_DATA  &&
+            } while ((i<numEvents) && ((buffer[i].sensor == curr) ||
-                          buffer[i].meta_data.sensor == curr)));
+                                       (buffer[i].type == SENSOR_TYPE_META_DATA  &&
-            } else {
+                                        buffer[i].meta_data.sensor == curr)));
-                i++;
-            }
         }
     } else {
         scratch = const_cast<sensors_event_t *>(buffer);
         count = numEvents;
     }
 
-    // Send pending flush events (if any) before sending events from the cache.
+    // Early return if there are no events for this connection.
+    if (count == 0) {
+        return status_t(NO_ERROR);
+    }
+
+#if DEBUG_CONNECTIONS
+     mEventsReceived += count;
+#endif
+    if (mCacheSize != 0) {
+        // There are some events in the cache which need to be sent first. Copy this buffer to
+        // the end of cache.
+        if (mCacheSize + count <= mMaxCacheSize) {
+            memcpy(&mEventCache[mCacheSize], scratch, count * sizeof(sensors_event_t));
+            mCacheSize += count;
+        } else {
+            // Some events need to be dropped.
+            int remaningCacheSize = mMaxCacheSize - mCacheSize;
+            if (remaningCacheSize != 0) {
+                memcpy(&mEventCache[mCacheSize], scratch,
+                                                remaningCacheSize * sizeof(sensors_event_t));
+            }
+            int numEventsDropped = count - remaningCacheSize;
+            countFlushCompleteEventsLocked(mEventCache, numEventsDropped);
+            // Drop the first "numEventsDropped" in the cache.
+            memmove(mEventCache, &mEventCache[numEventsDropped],
+                    (mCacheSize - numEventsDropped) * sizeof(sensors_event_t));
+
+            // Copy the remainingEvents in scratch buffer to the end of cache.
+            memcpy(&mEventCache[mCacheSize - numEventsDropped], scratch + remaningCacheSize,
+                                            numEventsDropped * sizeof(sensors_event_t));
+        }
+        return status_t(NO_ERROR);
+    }
+
+    int numWakeUpSensorEvents = countWakeUpSensorEventsLocked(scratch, count);
+    mWakeLockRefCount += numWakeUpSensorEvents;
+
+    // NOTE: ASensorEvent and sensors_event_t are the same type.
+    ssize_t size = SensorEventQueue::write(mChannel,
+                                    reinterpret_cast<ASensorEvent const*>(scratch), count);
+    if (size < 0) {
+        // Write error, copy events to local cache.
+        mWakeLockRefCount -= numWakeUpSensorEvents;
+        if (mEventCache == NULL) {
+            mMaxCacheSize = computeMaxCacheSizeLocked();
+            mEventCache = new sensors_event_t[mMaxCacheSize];
+            mCacheSize = 0;
+        }
+        memcpy(&mEventCache[mCacheSize], scratch, count * sizeof(sensors_event_t));
+        mCacheSize += count;
+
+        // Add this file descriptor to the looper to get a callback when this fd is available for
+        // writing.
+        mService->getLooper()->addFd(mChannel->getSendFd(), 0,
+                ALOOPER_EVENT_OUTPUT | ALOOPER_EVENT_INPUT, this, NULL);
+        return size;
+    }
+
+#if DEBUG_CONNECTIONS
+    if (size > 0) {
+        mEventsSent += count;
+    }
+#endif
+
+    return size < 0 ? status_t(size) : status_t(NO_ERROR);
+}
+
+void SensorService::SensorEventConnection::writeToSocketFromCacheLocked() {
+    // At a time write at most half the size of the receiver buffer in SensorEventQueue.
+    const int maxWriteSize = SensorEventQueue::MAX_RECEIVE_BUFFER_EVENT_COUNT/2;
+    // Send pending flush events (if any) before sending events from the buffer.
     {
         ASensorEvent flushCompleteEvent;
         flushCompleteEvent.type = SENSOR_TYPE_META_DATA;
@@ -963,35 +1103,44 @@ status_t SensorService::SensorEventConnection::sendEvents(
                 flushCompleteEvent.meta_data.sensor = mSensorInfo.keyAt(i);
                 ssize_t size = SensorEventQueue::write(mChannel, &flushCompleteEvent, 1);
                 if (size < 0) {
-                    // ALOGW("dropping %d events on the floor", count);
+                    return;
-                    countFlushCompleteEventsLocked(scratch, count);
-                    return size;
                 }
                 ALOGD_IF(DEBUG_CONNECTIONS, "sent dropped flush complete event==%d ",
-                         flushCompleteEvent.meta_data.sensor);
+                                                flushCompleteEvent.meta_data.sensor);
                 flushInfo.mPendingFlushEventsToSend--;
             }
         }
     }
+    // Write "count" events at a time.
+    for (int numEventsSent = 0; numEventsSent < mCacheSize;) {
+        const int count = (mCacheSize - numEventsSent) < maxWriteSize ?
+                                        mCacheSize - numEventsSent : maxWriteSize;
+        int numWakeUpSensorEvents =
+                  countWakeUpSensorEventsLocked(mEventCache + numEventsSent, count);
+        mWakeLockRefCount += numWakeUpSensorEvents;
 
-    // Early return if there are no events for this connection.
+        ssize_t size = SensorEventQueue::write(mChannel,
-    if (count == 0) {
+                          reinterpret_cast<ASensorEvent const*>(mEventCache + numEventsSent),
-        return status_t(NO_ERROR);
+                          count);
+        if (size < 0) {
+            memmove(mEventCache, &mEventCache[numEventsSent],
+                                 (mCacheSize - numEventsSent) * sizeof(sensors_event_t));
+            ALOGD_IF(DEBUG_CONNECTIONS, "wrote %d events from cache size==%d ",
+                                            numEventsSent, mCacheSize);
+            mCacheSize -= numEventsSent;
+            mWakeLockRefCount -= numWakeUpSensorEvents;
+            return;
+        }
+        numEventsSent += count;
+#if DEBUG_CONNECTIONS
+        mEventsSentFromCache += count;
+#endif
     }
-
+    ALOGD_IF(DEBUG_CONNECTIONS, "wrote all events from cache size=%d ", mCacheSize);
-    int numWakeUpSensorEvents = countWakeUpSensorEventsLocked(scratch, count);
+    // All events from the cache have been sent. Reset cache size to zero.
-    // NOTE: ASensorEvent and sensors_event_t are the same type
+    mCacheSize = 0;
-    ssize_t size = SensorEventQueue::write(mChannel,
+    // Poll only for ALOOPER_EVENT_INPUT(read) on the file descriptor.
-            reinterpret_cast<ASensorEvent const*>(scratch), count);
+    mService->getLooper()->addFd(mChannel->getSendFd(), 0, ALOOPER_EVENT_INPUT, this, NULL);
-    if (size == -EAGAIN) {
-        // the destination doesn't accept events anymore, it's probably
-        // full. For now, we just drop the events on the floor.
-        // ALOGW("dropping %d events on the floor", count);
-        countFlushCompleteEventsLocked(scratch, count);
-        mWakeLockRefCount -= numWakeUpSensorEvents;
-        return size;
-    }
-    return size < 0 ? status_t(size) : status_t(NO_ERROR);
 }
 
 void SensorService::SensorEventConnection::countFlushCompleteEventsLocked(
@@ -1015,7 +1164,6 @@ int SensorService::SensorEventConnection::countWakeUpSensorEventsLocked(
     for (int i = 0; i < count; ++i) {
         if (mService->isWakeUpSensorEvent(scratch[i])) {
             scratch[i].flags |= WAKE_UP_SENSOR_EVENT_NEEDS_ACK;
-            ++mWakeLockRefCount;
             return 1;
         }
     }
@@ -1035,6 +1183,7 @@ status_t SensorService::SensorEventConnection::enableDisable(
     if (enabled) {
         err = mService->enable(this, handle, samplingPeriodNs, maxBatchReportLatencyNs,
                                reservedFlags);
+
     } else {
         err = mService->disable(this, handle);
     }
@@ -1055,14 +1204,16 @@ status_t  SensorService::SensorEventConnection::flush() {
     // Loop through all sensors for this connection and call flush on each of them.
     for (size_t i = 0; i < mSensorInfo.size(); ++i) {
         const int handle = mSensorInfo.keyAt(i);
+        FlushInfo& flushInfo = mSensorInfo.editValueFor(handle);
         if (halVersion < SENSORS_DEVICE_API_VERSION_1_1 || mService->isVirtualSensor(handle)) {
             // For older devices just increment pending flush count which will send a trivial
             // flush complete event.
-            FlushInfo& flushInfo = mSensorInfo.editValueFor(handle);
             flushInfo.mPendingFlushEventsToSend++;
         } else {
             status_t err_flush = mService->flushSensor(this, handle);
-            if (err_flush != NO_ERROR) {
+            if (err_flush == NO_ERROR) {
+                flushInfo.mNumFlushCalls++;
+            } else {
                 ALOGE("Flush error handle=%d %s", handle, strerror(-err_flush));
             }
             err = (err_flush != NO_ERROR) ? err_flush : err;
@@ -1071,15 +1222,69 @@ status_t  SensorService::SensorEventConnection::flush() {
     return err;
 }
 
-void SensorService::SensorEventConnection::decreaseWakeLockRefCount() {
+int SensorService::SensorEventConnection::handleEvent(int fd, int events, void* data) {
-    {
+    if (events & ALOOPER_EVENT_HANGUP || events & ALOOPER_EVENT_ERROR) {
+        return 0;
+    }
+
+    if (events & ALOOPER_EVENT_INPUT) {
+        char buf;
+        ssize_t ret = ::recv(fd, &buf, sizeof(buf), MSG_DONTWAIT);
+
+        {
+           Mutex::Autolock _l(mConnectionLock);
+           --mWakeLockRefCount;
+        }
+        // Check if wakelock can be released by sensorservice. mConnectionLock needs to be released
+        // here as checkWakeLockState() will need it.
+        if (mWakeLockRefCount == 0) {
+            mService->checkWakeLockState();
+        }
+        // continue getting callbacks.
+        return 1;
+    }
+
+    if (events & ALOOPER_EVENT_OUTPUT) {
+        // send sensor data that is stored in mEventCache.
         Mutex::Autolock _l(mConnectionLock);
-        --mWakeLockRefCount;
+        writeToSocketFromCacheLocked();
-    }
-    // Release the lock before calling checkWakeLockState which also needs the same connectionLock.
-    if (mWakeLockRefCount == 0) {
-        mService->checkWakeLockState();
     }
+    return 1;
+}
+
+int SensorService::SensorEventConnection::computeMaxCacheSizeLocked() const {
+    int fifoWakeUpSensors = 0;
+    int fifoNonWakeUpSensors = 0;
+    for (size_t i = 0; i < mSensorInfo.size(); ++i) {
+        const Sensor& sensor = mService->getSensorFromHandle(mSensorInfo.keyAt(i));
+        if (sensor.getFifoReservedEventCount() == sensor.getFifoMaxEventCount()) {
+            // Each sensor has a reserved fifo. Sum up the fifo sizes for all wake up sensors and
+            // non wake_up sensors.
+            if (sensor.isWakeUpSensor()) {
+                fifoWakeUpSensors += sensor.getFifoReservedEventCount();
+            } else {
+                fifoNonWakeUpSensors += sensor.getFifoReservedEventCount();
+            }
+        } else {
+            // Shared fifo. Compute the max of the fifo sizes for wake_up and non_wake up sensors.
+            if (sensor.isWakeUpSensor()) {
+                fifoWakeUpSensors = fifoWakeUpSensors > sensor.getFifoMaxEventCount() ?
+                                          fifoWakeUpSensors : sensor.getFifoMaxEventCount();
+
+            } else {
+                fifoNonWakeUpSensors = fifoNonWakeUpSensors > sensor.getFifoMaxEventCount() ?
+                                          fifoNonWakeUpSensors : sensor.getFifoMaxEventCount();
+
+            }
+        }
+   }
+   if (fifoWakeUpSensors + fifoNonWakeUpSensors == 0) {
+       // It is extremely unlikely that there is a write failure in non batch mode. Return a cache
+       // size of 100.
+       ALOGI("Write failure in non-batch mode");
+       return 100;
+   }
+   return fifoWakeUpSensors + fifoNonWakeUpSensors;
 }
 
 // ---------------------------------------------------------------------------

services/sensorservice/SensorService.h

@@ -24,7 +24,9 @@
 #include <utils/SortedVector.h>
 #include <utils/KeyedVector.h>
 #include <utils/threads.h>
+#include <utils/AndroidThreads.h>
 #include <utils/RefBase.h>
+#include <utils/Looper.h>
 
 #include <binder/BinderService.h>
 
@@ -38,10 +40,11 @@
 // ---------------------------------------------------------------------------
 
 #define DEBUG_CONNECTIONS   false
-// Max size is 1 MB which is enough to accept a batch of about 10k events.
+// Max size is 100 KB which is enough to accept a batch of about 1000 events.
-#define MAX_SOCKET_BUFFER_SIZE_BATCHED 1024 * 1024
+#define MAX_SOCKET_BUFFER_SIZE_BATCHED 100 * 1024
+// For older HALs which don't support batching, use a smaller socket buffer size.
 #define SOCKET_BUFFER_SIZE_NON_BATCHED 4 * 1024
-#define WAKE_UP_SENSOR_EVENT_NEEDS_ACK (1 << 31)
+#define WAKE_UP_SENSOR_EVENT_NEEDS_ACK (1U << 31)
 
 struct sensors_poll_device_t;
 struct sensors_module_t;
@@ -72,7 +75,8 @@ class SensorService :
     virtual sp<ISensorEventConnection> createSensorEventConnection();
     virtual status_t dump(int fd, const Vector<String16>& args);
 
-    class SensorEventConnection : public BnSensorEventConnection {
+    class SensorEventConnection : public BnSensorEventConnection, public LooperCallback {
+        friend class SensorService;
         virtual ~SensorEventConnection();
         virtual void onFirstRef();
         virtual sp<BitTube> getSensorChannel() const;
@@ -80,7 +84,6 @@ class SensorService :
                                        nsecs_t maxBatchReportLatencyNs, int reservedFlags);
         virtual status_t setEventRate(int handle, nsecs_t samplingPeriodNs);
         virtual status_t flush();
-        void decreaseWakeLockRefCount();
         // Count the number of flush complete events which are about to be dropped in the buffer.
         // Increment mPendingFlushEventsToSend in mSensorInfo. These flush complete events will be
         // sent separately before the next batch of events.
@@ -90,7 +93,20 @@ class SensorService :
         // Increment it by exactly one unit for each packet sent on the socket. SOCK_SEQPACKET for
         // the socket ensures that either the entire packet is read or dropped.
         // Return 1 if mWakeLockRefCount has been incremented, zero if not.
-        int countWakeUpSensorEventsLocked(sensors_event_t* scratch, const int count);
+        int countWakeUpSensorEventsLocked(sensors_event_t* scratch, int count);
+
+        // Writes events from mEventCache to the socket.
+        void writeToSocketFromCacheLocked();
+
+        // Compute the approximate cache size from the FIFO sizes of various sensors registered for
+        // this connection. Wake up and non-wake up sensors have separate FIFOs but FIFO may be
+        // shared amongst wake-up sensors and non-wake up sensors.
+        int computeMaxCacheSizeLocked() const;
+
+        // LooperCallback method. If there is data to read on this fd, it is an ack from the
+        // app that it has read events from a wake up sensor, decrement mWakeLockRefCount.
+        // If this fd is available for writing send the data from the cache.
+        virtual int handleEvent(int fd, int events, void* data);
 
         sp<SensorService> const mService;
         sp<BitTube> mChannel;
@@ -108,10 +124,20 @@ class SensorService :
             // Every activate is preceded by a flush. Only after the first flush complete is
             // received, the events for the sensor are sent on that *connection*.
             bool mFirstFlushPending;
-            FlushInfo() : mPendingFlushEventsToSend(0), mFirstFlushPending(false) {}
+            // Number of time flush() was called on this connection. This is incremented every time
+            // flush() is called and decremented when flush_complete_event is received.
+            int mNumFlushCalls;
+            FlushInfo() : mPendingFlushEventsToSend(0), mFirstFlushPending(false),
+                                            mNumFlushCalls(0) {}
         };
         // protected by SensorService::mLock. Key for this vector is the sensor handle.
         KeyedVector<int, FlushInfo> mSensorInfo;
+        sensors_event_t *mEventCache;
+        int mCacheSize, mMaxCacheSize;
+
+#if DEBUG_CONNECTIONS
+        int mEventsReceived, mEventsSent, mEventsSentFromCache;
+#endif
 
     public:
         SensorEventConnection(const sp<SensorService>& service, uid_t uid);
@@ -138,6 +164,13 @@ class SensorService :
         size_t getNumConnections() const { return mConnections.size(); }
     };
 
+    class SensorEventAckReceiver : public Thread {
+        sp<SensorService> const mService;
+    public:
+        virtual bool threadLoop();
+        SensorEventAckReceiver(const sp<SensorService>& service): mService(service) {}
+    };
+
     String8 getSensorName(int handle) const;
     bool isVirtualSensor(int handle) const;
     Sensor getSensorFromHandle(int handle) const;
@@ -160,6 +193,9 @@ class SensorService :
     void checkWakeLockState();
     void checkWakeLockStateLocked();
     bool isWakeUpSensorEvent(const sensors_event_t& event) const;
+
+    sp<Looper> getLooper() const;
+
     // constants
     Vector<Sensor> mSensorList;
     Vector<Sensor> mUserSensorListDebug;
@@ -168,6 +204,7 @@ class SensorService :
     Vector<SensorInterface *> mVirtualSensorList;
     status_t mInitCheck;
     size_t mSocketBufferSize;
+    sp<Looper> mLooper;
 
     // protected by mLock
     mutable Mutex mLock;