Fix sockfd leakage in SensorService.

i) Call removeFd() only if the fd in the BitTube has been
previously added to the Looper. Use a flag to determine whether the fd
has been previously added or not.
ii) Increment mPendingFlushEventsToSend after holding a connectionLock.
iii) Store the number of acks that are pending in SensorEventQueue
 and send them all at once.

Bug: 17472228
Change-Id: I1ec834fea1112a9cfbd9cddd2198438793698502

include/gui/SensorEventQueue.h

@@ -86,6 +86,7 @@ private:
     ASensorEvent* mRecBuffer;
     size_t mAvailable;
     size_t mConsumed;
+    uint32_t mNumAcksToSend;
 };
 
 // ----------------------------------------------------------------------------

libs/gui/SensorEventQueue.cpp

@@ -36,7 +36,8 @@ namespace android {
 // ----------------------------------------------------------------------------
 
 SensorEventQueue::SensorEventQueue(const sp<ISensorEventConnection>& connection)
-    : mSensorEventConnection(connection), mRecBuffer(NULL), mAvailable(0), mConsumed(0) {
+    : mSensorEventConnection(connection), mRecBuffer(NULL), mAvailable(0), mConsumed(0),
+      mNumAcksToSend(0) {
     mRecBuffer = new ASensorEvent[MAX_RECEIVE_BUFFER_EVENT_COUNT];
 }
 
@@ -148,14 +149,17 @@ 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) {
-            // Send just a byte of data to acknowledge for the wake up sensor events
+            ++mNumAcksToSend;
-            // received
+        }
-            char buf = '1';
+    }
-            ssize_t size = ::send(mSensorChannel->getFd(), &buf, sizeof(buf),
+    // Send mNumAcksToSend to acknowledge for the wake up sensor events received.
-                                             MSG_DONTWAIT | MSG_NOSIGNAL);
+    if (mNumAcksToSend > 0) {
-            if (size < 0) {
+        ssize_t size = ::send(mSensorChannel->getFd(), &mNumAcksToSend, sizeof(mNumAcksToSend),
-                ALOGE("sendAck failure %d", size);
+                MSG_DONTWAIT | MSG_NOSIGNAL);
-            }
+        if (size < 0) {
+            ALOGE("sendAck failure %d %d", size, mNumAcksToSend);
+        } else {
+            mNumAcksToSend = 0;
         }
     }
     return;

services/sensorservice/SensorService.cpp

@@ -66,7 +66,8 @@ namespace android {
 const char* SensorService::WAKE_LOCK_NAME = "SensorService";
 
 SensorService::SensorService()
-    : mInitCheck(NO_INIT)
+    : mInitCheck(NO_INIT), mSocketBufferSize(SOCKET_BUFFER_SIZE_NON_BATCHED),
+      mWakeLockAcquired(false)
 {
 }
 
@@ -433,7 +434,6 @@ bool SensorService::threadLoop()
         if (bufferHasWakeUpEvent && !mWakeLockAcquired) {
             acquire_wake_lock(PARTIAL_WAKE_LOCK, WAKE_LOCK_NAME);
             mWakeLockAcquired = true;
-            ALOGD_IF(DEBUG_CONNECTIONS, "acquired wakelock %s", WAKE_LOCK_NAME);
         }
         recordLastValueLocked(mSensorEventBuffer, count);
 
@@ -523,7 +523,6 @@ bool SensorService::threadLoop()
         if (mWakeLockAcquired && !needsWakeLock) {
             release_wake_lock(WAKE_LOCK_NAME);
             mWakeLockAcquired = false;
-            ALOGD_IF(DEBUG_CONNECTIONS, "released wakelock %s", WAKE_LOCK_NAME);
         }
     } while (!Thread::exitPending());
 
@@ -650,6 +649,7 @@ void SensorService::cleanupConnection(SensorEventConnection* c)
             if (sensor) {
                 sensor->activate(c, false);
             }
+            c->removeSensor(handle);
         }
         SensorRecord* rec = mActiveSensors.valueAt(i);
         ALOGE_IF(!rec, "mActiveSensors[%zu] is null (handle=0x%08x)!", i, handle);
@@ -667,6 +667,7 @@ void SensorService::cleanupConnection(SensorEventConnection* c)
             i++;
         }
     }
+    c->updateLooperRegistration(mLooper);
     mActiveConnections.remove(connection);
     BatteryService::cleanup(c->getUid());
     if (c->needsWakeLock()) {
@@ -770,10 +771,8 @@ status_t SensorService::enable(const sp<SensorEventConnection>& connection,
         err = sensor->activate(connection.get(), true);
     }
 
-    if (err == NO_ERROR && sensor->getSensor().isWakeUpSensor()) {
+    if (err == NO_ERROR) {
-        // Add the file descriptor to the Looper for receiving acknowledgments;
+        connection->updateLooperRegistration(mLooper);
-        int ret = mLooper->addFd(connection->getSensorChannel()->getSendFd(), 0,
-                                        ALOOPER_EVENT_INPUT, connection.get(), NULL);
     }
 
     if (err != NO_ERROR) {
@@ -813,6 +812,7 @@ status_t SensorService::cleanupWithoutDisableLocked(
             BatteryService::disableSensor(connection->getUid(), handle);
         }
         if (connection->hasAnySensor() == false) {
+            connection->updateLooperRegistration(mLooper);
             mActiveConnections.remove(connection);
         }
         // see if this sensor becomes inactive
@@ -866,11 +866,10 @@ status_t SensorService::flushSensor(const sp<SensorEventConnection>& connection)
             err = INVALID_OPERATION;
             continue;
         }
-        SensorEventConnection::FlushInfo& flushInfo = connection->mSensorInfo.editValueFor(handle);
         if (halVersion <= SENSORS_DEVICE_API_VERSION_1_0 || isVirtualSensor(handle)) {
             // For older devices just increment pending flush count which will send a trivial
             // flush complete event.
-            flushInfo.mPendingFlushEventsToSend++;
+            connection->incrementPendingFlushCount(handle);
         } else {
             status_t err_flush = sensor->flush(connection.get(), handle);
             if (err_flush == NO_ERROR) {
@@ -922,7 +921,6 @@ void SensorService::checkWakeLockStateLocked() {
         }
     }
     if (releaseLock) {
-        ALOGD_IF(DEBUG_CONNECTIONS, "releasing wakelock %s", WAKE_LOCK_NAME);
         release_wake_lock(WAKE_LOCK_NAME);
         mWakeLockAcquired = false;
     }
@@ -955,6 +953,7 @@ bool SensorService::SensorRecord::removeConnection(
     // Remove this connections from the queue of flush() calls made on this sensor.
     for (Vector< wp<SensorEventConnection> >::iterator it =
             mPendingFlushConnections.begin(); it != mPendingFlushConnections.end();) {
+
         if (it->unsafe_get() == connection.unsafe_get()) {
             it = mPendingFlushConnections.erase(it);
         } else {
@@ -987,9 +986,8 @@ SensorService::SensorRecord::getFirstPendingFlushConnection() {
 
 SensorService::SensorEventConnection::SensorEventConnection(
         const sp<SensorService>& service, uid_t uid)
-    : mService(service), mUid(uid), mWakeLockRefCount(0), mEventCache(NULL), mCacheSize(0),
+    : mService(service), mUid(uid), mWakeLockRefCount(0), mHasLooperCallbacks(false),
-      mMaxCacheSize(0) {
+      mDead(false), mEventCache(NULL), mCacheSize(0), mMaxCacheSize(0) {
-    const SensorDevice& device(SensorDevice::getInstance());
     mChannel = new BitTube(mService->mSocketBufferSize);
 #if DEBUG_CONNECTIONS
     mEventsReceived = mEventsSentFromCache = mEventsSent = 0;
@@ -1011,7 +1009,7 @@ void SensorService::SensorEventConnection::onFirstRef() {
 
 bool SensorService::SensorEventConnection::needsWakeLock() {
     Mutex::Autolock _l(mConnectionLock);
-    return mWakeLockRefCount > 0;
+    return !mDead && mWakeLockRefCount > 0;
 }
 
 void SensorService::SensorEventConnection::dump(String8& result) {
@@ -1090,6 +1088,65 @@ void SensorService::SensorEventConnection::setFirstFlushPending(int32_t handle,
     }
 }
 
+void SensorService::SensorEventConnection::updateLooperRegistration(const sp<Looper>& looper) {
+    Mutex::Autolock _l(mConnectionLock);
+    updateLooperRegistrationLocked(looper);
+}
+
+void SensorService::SensorEventConnection::updateLooperRegistrationLocked(
+        const sp<Looper>& looper) {
+    bool isConnectionActive = mSensorInfo.size() > 0;
+    // If all sensors are unregistered OR Looper has encountered an error, we
+    // can remove the Fd from the Looper if it has been previously added.
+    if (!isConnectionActive || mDead) {
+        if (mHasLooperCallbacks) {
+            ALOGD_IF(DEBUG_CONNECTIONS, "%p removeFd fd=%d", this, mChannel->getSendFd());
+            looper->removeFd(mChannel->getSendFd());
+            mHasLooperCallbacks = false;
+        }
+        return;
+    }
+
+    int looper_flags = 0;
+    if (mCacheSize > 0) looper_flags |= ALOOPER_EVENT_OUTPUT;
+    for (size_t i = 0; i < mSensorInfo.size(); ++i) {
+        const int handle = mSensorInfo.keyAt(i);
+        if (mService->getSensorFromHandle(handle).isWakeUpSensor()) {
+            looper_flags |= ALOOPER_EVENT_INPUT;
+            break;
+        }
+    }
+    // If flags is still set to zero, we don't need to add this fd to the Looper, if
+    // the fd has already been added, remove it. This is likely to happen when ALL the
+    // events stored in the cache have been sent to the corresponding app.
+    if (looper_flags == 0) {
+        if (mHasLooperCallbacks) {
+            ALOGD_IF(DEBUG_CONNECTIONS, "removeFd fd=%d", mChannel->getSendFd());
+            looper->removeFd(mChannel->getSendFd());
+            mHasLooperCallbacks = false;
+        }
+        return;
+    }
+    // Add the file descriptor to the Looper for receiving acknowledegments if the app has
+    // registered for wake-up sensors OR for sending events in the cache.
+    int ret = looper->addFd(mChannel->getSendFd(), 0, looper_flags, this, NULL);
+    if (ret == 1) {
+        ALOGD_IF(DEBUG_CONNECTIONS, "%p addFd fd=%d", this, mChannel->getSendFd());
+        mHasLooperCallbacks = true;
+    } else {
+        ALOGE("Looper::addFd failed ret=%d fd=%d", ret, mChannel->getSendFd());
+    }
+}
+
+void SensorService::SensorEventConnection::incrementPendingFlushCount(int32_t handle) {
+    Mutex::Autolock _l(mConnectionLock);
+    ssize_t index = mSensorInfo.indexOfKey(handle);
+    if (index >= 0) {
+        FlushInfo& flushInfo = mSensorInfo.editValueAt(index);
+        flushInfo.mPendingFlushEventsToSend++;
+    }
+}
+
 status_t SensorService::SensorEventConnection::sendEvents(
         sensors_event_t const* buffer, size_t numEvents,
         sensors_event_t* scratch,
@@ -1104,8 +1161,9 @@ status_t SensorService::SensorEventConnection::sendEvents(
             if (buffer[i].type == SENSOR_TYPE_META_DATA) {
                 ALOGD_IF(DEBUG_CONNECTIONS, "flush complete event sensor==%d ",
                         buffer[i].meta_data.sensor);
-                // Setting sensor_handle to the correct sensor to ensure the sensor events per connection are
+                // Setting sensor_handle to the correct sensor to ensure the sensor events per
-                // filtered correctly. buffer[i].sensor is zero for meta_data events.
+                // connection are filtered correctly.  buffer[i].sensor is zero for meta_data
+                // events.
                 sensor_handle = buffer[i].meta_data.sensor;
             }
             ssize_t index = mSensorInfo.indexOfKey(sensor_handle);
@@ -1233,8 +1291,7 @@ status_t SensorService::SensorEventConnection::sendEvents(
 
         // Add this file descriptor to the looper to get a callback when this fd is available for
         // writing.
-        mService->getLooper()->addFd(mChannel->getSendFd(), 0,
+        updateLooperRegistrationLocked(mService->getLooper());
-                ALOOPER_EVENT_OUTPUT | ALOOPER_EVENT_INPUT, this, NULL);
         return size;
     }
 
@@ -1340,8 +1397,9 @@ void SensorService::SensorEventConnection::writeToSocketFromCacheLocked() {
     ALOGD_IF(DEBUG_CONNECTIONS, "wrote all events from cache size=%d ", mCacheSize);
     // All events from the cache have been sent. Reset cache size to zero.
     mCacheSize = 0;
-    // Poll only for ALOOPER_EVENT_INPUT(read) on the file descriptor.
+    // There are no more events in the cache. We don't need to poll for write on the fd.
-    mService->getLooper()->addFd(mChannel->getSendFd(), 0, ALOOPER_EVENT_INPUT, this, NULL);
+    // Update Looper registration.
+    updateLooperRegistrationLocked(mService->getLooper());
 }
 
 void SensorService::SensorEventConnection::countFlushCompleteEventsLocked(
@@ -1400,22 +1458,38 @@ status_t  SensorService::SensorEventConnection::flush() {
     return  mService->flushSensor(this);
 }
 
-int SensorService::SensorEventConnection::handleEvent(int fd, int events, void* data) {
+int SensorService::SensorEventConnection::handleEvent(int fd, int events, void* /*data*/) {
     if (events & ALOOPER_EVENT_HANGUP || events & ALOOPER_EVENT_ERROR) {
-        return 0;
+        {
+            // If the Looper encounters some error, set the flag mDead, reset mWakeLockRefCount,
+            // and remove the fd from Looper. Call checkWakeLockState to know if SensorService
+            // can release the wake-lock.
+            ALOGD_IF(DEBUG_CONNECTIONS, "%p Looper error %d", this, fd);
+            Mutex::Autolock _l(mConnectionLock);
+            mDead = true;
+            mWakeLockRefCount = 0;
+            updateLooperRegistrationLocked(mService->getLooper());
+        }
+        mService->checkWakeLockState();
+        return 1;
     }
 
     if (events & ALOOPER_EVENT_INPUT) {
-        char buf;
+        uint32_t numAcks = 0;
-        ssize_t ret = ::recv(fd, &buf, sizeof(buf), MSG_DONTWAIT);
+        ssize_t ret = ::recv(fd, &numAcks, sizeof(numAcks), MSG_DONTWAIT);
-
         {
            Mutex::Autolock _l(mConnectionLock);
-           if (mWakeLockRefCount > 0) {
+           // Sanity check to ensure  there are no read errors in recv, numAcks is always
-               --mWakeLockRefCount;
+           // within the range and not zero. If any of the above don't hold reset mWakeLockRefCount
+           // to zero.
+           if (ret != sizeof(numAcks) || numAcks > mWakeLockRefCount || numAcks == 0) {
+               ALOGE("Looper read error ret=%d numAcks=%d", ret, numAcks);
+               mWakeLockRefCount = 0;
+           } else {
+               mWakeLockRefCount -= numAcks;
            }
 #if DEBUG_CONNECTIONS
-           ++mTotalAcksReceived;
+           mTotalAcksReceived += numAcks;
 #endif
         }
         // Check if wakelock can be released by sensorservice. mConnectionLock needs to be released

services/sensorservice/SensorService.h

@@ -117,6 +117,20 @@ class SensorService :
         // If this fd is available for writing send the data from the cache.
         virtual int handleEvent(int fd, int events, void* data);
 
+        // Increment mPendingFlushEventsToSend for the given sensor handle.
+        void incrementPendingFlushCount(int32_t handle);
+
+        // Add or remove the file descriptor associated with the BitTube to the looper. If mDead is
+        // set to true or there are no more sensors for this connection, the file descriptor is
+        // removed if it has been previously added to the Looper. Depending on the state of the
+        // connection FD may be added to the Looper. The flags to set are determined by the internal
+        // state of the connection. FDs are added to the looper when wake-up sensors are registered
+        // (to poll for acknowledgements) and when write fails on the socket when there are too many
+        // events (to poll when the FD is available for writing). FDs are removed when there is an
+        // error and the other end hangs up or when this client unregisters for this connection.
+        void updateLooperRegistration(const sp<Looper>& looper);
+        void updateLooperRegistrationLocked(const sp<Looper>& looper);
+
         sp<SensorService> const mService;
         sp<BitTube> mChannel;
         uid_t mUid;
@@ -124,8 +138,17 @@ class SensorService :
         // Number of events from wake up sensors which are still pending and haven't been delivered
         // to the corresponding application. It is incremented by one unit for each write to the
         // socket.
-        int mWakeLockRefCount;
+        uint32_t mWakeLockRefCount;
 
+        // If this flag is set to true, it means that the file descriptor associated with the
+        // BitTube has been added to the Looper in SensorService. This flag is typically set when
+        // this connection has wake-up sensors associated with it or when write has failed on this
+        // connection and we're storing some events in the cache.
+        bool mHasLooperCallbacks;
+        // If there are any errors associated with the Looper this flag is set to true and
+        // mWakeLockRefCount is reset to zero. needsWakeLock method will always return false, if
+        // this flag is set.
+        bool mDead;
         struct FlushInfo {
             // The number of flush complete events dropped for this sensor is stored here.
             // They are sent separately before the next batch of events.
@@ -222,7 +245,7 @@ class SensorService :
     status_t mInitCheck;
     // Socket buffersize used to initialize BitTube. This size depends on whether batching is
     // supported or not.
-    size_t mSocketBufferSize;
+    uint32_t mSocketBufferSize;
     sp<Looper> mLooper;
 
     // protected by mLock