fix a few problems with BitTube

BitTube used to send objects one at a time and didn't
handle errors properly.

We now send all the objects in one call, which means they
have to be read as a single batch as well. This changes the
BitTube API.

Update SensorService to the new API.

Also added an API to set the size of the send buffer.

Bug: 10641596
Change-Id: I77c70d35e351fdba0416fae4b7ca3b1d56272251

include/gui/BitTube.h

@@ -33,30 +33,49 @@ class BitTube : public RefBase
 {
 public:
 
-            BitTube();
+    // creates a BitTube with a default (4KB) send buffer
-            BitTube(const Parcel& data);
+    BitTube();
+
+    // creates a BitTube with a a specified send and receive buffer size
+    explicit BitTube(size_t bufsize);
+
+    explicit BitTube(const Parcel& data);
     virtual ~BitTube();
 
+    // check state after construction
     status_t initCheck() const;
+
+    // get receive file-descriptor
     int getFd() const;
-    ssize_t write(void const* vaddr, size_t size);
-    ssize_t read(void* vaddr, size_t size);
-
-    status_t writeToParcel(Parcel* reply) 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,
             T const* events, size_t count) {
         return sendObjects(tube, events, count, sizeof(T));
     }
 
+    // receive objects (sized blobs). If the receiving buffer isn't large enough,
+    // excess messages are silently discarded.
     template <typename T>
     static ssize_t recvObjects(const sp<BitTube>& tube,
             T* events, size_t count) {
         return recvObjects(tube, events, count, sizeof(T));
     }
 
+    // parcels this BitTube
+    status_t writeToParcel(Parcel* reply) const;
+
 private:
+    void init(size_t rcvbuf, size_t sndbuf);
+
+    // send a message. The write is guaranteed to send the whole message or fail.
+    ssize_t write(void const* vaddr, size_t size);
+
+    // receive a message. the passed buffer must be at least as large as the
+    // write call used to send the message, excess data is silently discarded.
+    ssize_t read(void* vaddr, size_t size);
+
     int mSendFd;
     mutable int mReceiveFd;
 

include/gui/SensorEventQueue.h

@@ -49,6 +49,9 @@ class Looper;
 class SensorEventQueue : public ASensorEventQueue, public RefBase
 {
 public:
+
+    enum { MAX_RECEIVE_BUFFER_EVENT_COUNT = 256 };
+
             SensorEventQueue(const sp<ISensorEventConnection>& connection);
     virtual ~SensorEventQueue();
     virtual void onFirstRef();
@@ -79,6 +82,9 @@ private:
     sp<BitTube> mSensorChannel;
     mutable Mutex mLock;
     mutable sp<Looper> mLooper;
+    ASensorEvent* mRecBuffer;
+    size_t mAvailable;
+    size_t mConsumed;
 };
 
 // ----------------------------------------------------------------------------

libs/gui/BitTube.cpp

@@ -32,39 +32,26 @@ namespace android {
 
 // Socket buffer size.  The default is typically about 128KB, which is much larger than
 // we really need.  So we make it smaller.
-static const size_t SOCKET_BUFFER_SIZE = 4 * 1024;
+static const size_t DEFAULT_SOCKET_BUFFER_SIZE = 4 * 1024;
 
 
 BitTube::BitTube()
     : mSendFd(-1), mReceiveFd(-1)
 {
-    int sockets[2];
+    init(DEFAULT_SOCKET_BUFFER_SIZE, DEFAULT_SOCKET_BUFFER_SIZE);
-    if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, sockets) == 0) {
+}
-        int size = SOCKET_BUFFER_SIZE;
+
-        setsockopt(sockets[0], SOL_SOCKET, SO_SNDBUF, &size, sizeof(size));
+BitTube::BitTube(size_t bufsize)
-        setsockopt(sockets[0], SOL_SOCKET, SO_RCVBUF, &size, sizeof(size));
+    : mSendFd(-1), mReceiveFd(-1)
-        setsockopt(sockets[1], SOL_SOCKET, SO_SNDBUF, &size, sizeof(size));
+{
-        setsockopt(sockets[1], SOL_SOCKET, SO_RCVBUF, &size, sizeof(size));
+    init(bufsize, bufsize);
-        fcntl(sockets[0], F_SETFL, O_NONBLOCK);
-        fcntl(sockets[1], F_SETFL, O_NONBLOCK);
-        mReceiveFd = sockets[0];
-        mSendFd = sockets[1];
-    } else {
-        mReceiveFd = -errno;
-        ALOGE("BitTube: pipe creation failed (%s)", strerror(-mReceiveFd));
-    }
 }
 
 BitTube::BitTube(const Parcel& data)
     : mSendFd(-1), mReceiveFd(-1)
 {
     mReceiveFd = dup(data.readFileDescriptor());
-    if (mReceiveFd >= 0) {
+    if (mReceiveFd < 0) {
-        int size = SOCKET_BUFFER_SIZE;
-        setsockopt(mReceiveFd, SOL_SOCKET, SO_SNDBUF, &size, sizeof(size));
-        setsockopt(mReceiveFd, SOL_SOCKET, SO_RCVBUF, &size, sizeof(size));
-        fcntl(mReceiveFd, F_SETFL, O_NONBLOCK);
-    } else {
         mReceiveFd = -errno;
         ALOGE("BitTube(Parcel): can't dup filedescriptor (%s)",
                 strerror(-mReceiveFd));
@@ -80,6 +67,25 @@ BitTube::~BitTube()
         close(mReceiveFd);
 }
 
+void BitTube::init(size_t rcvbuf, size_t sndbuf) {
+    int sockets[2];
+    if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, sockets) == 0) {
+        size_t size = DEFAULT_SOCKET_BUFFER_SIZE;
+        setsockopt(sockets[0], SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf));
+        setsockopt(sockets[1], SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf));
+        // sine we don't use the "return channel", we keep it small...
+        setsockopt(sockets[0], SOL_SOCKET, SO_SNDBUF, &size, sizeof(size));
+        setsockopt(sockets[1], SOL_SOCKET, SO_RCVBUF, &size, sizeof(size));
+        fcntl(sockets[0], F_SETFL, O_NONBLOCK);
+        fcntl(sockets[1], F_SETFL, O_NONBLOCK);
+        mReceiveFd = sockets[0];
+        mSendFd = sockets[1];
+    } else {
+        mReceiveFd = -errno;
+        ALOGE("BitTube: pipe creation failed (%s)", strerror(-mReceiveFd));
+    }
+}
+
 status_t BitTube::initCheck() const
 {
     if (mReceiveFd < 0) {
@@ -98,10 +104,10 @@ ssize_t BitTube::write(void const* vaddr, size_t size)
     ssize_t err, len;
     do {
         len = ::send(mSendFd, vaddr, size, MSG_DONTWAIT | MSG_NOSIGNAL);
+        // cannot return less than size, since we're using SOCK_SEQPACKET
         err = len < 0 ? errno : 0;
     } while (err == EINTR);
     return err == 0 ? len : -err;
-
 }
 
 ssize_t BitTube::read(void* vaddr, size_t size)
@@ -134,39 +140,31 @@ status_t BitTube::writeToParcel(Parcel* reply) const
 ssize_t BitTube::sendObjects(const sp<BitTube>& tube,
         void const* events, size_t count, size_t objSize)
 {
-    ssize_t numObjects = 0;
+    const char* vaddr = reinterpret_cast<const char*>(events);
-    for (size_t i=0 ; i<count ; i++) {
+    ssize_t size = tube->write(vaddr, count*objSize);
-        const char* vaddr = reinterpret_cast<const char*>(events) + objSize * i;
+
-        ssize_t size = tube->write(vaddr, objSize);
+    // should never happen because of SOCK_SEQPACKET
-        if (size < 0) {
+    LOG_ALWAYS_FATAL_IF((size >= 0) && (size % objSize),
-            // error occurred
+            "BitTube::sendObjects(count=%d, size=%d), res=%d (partial events were sent!)",
-            return size;
+            count, objSize, size);
-        } else if (size == 0) {
+
-            // no more space
+    //ALOGE_IF(size<0, "error %d sending %d events", size, count);
-            break;
+    return size < 0 ? size : size / objSize;
-        }
-        numObjects++;
-    }
-    return numObjects;
 }
 
 ssize_t BitTube::recvObjects(const sp<BitTube>& tube,
         void* events, size_t count, size_t objSize)
 {
-    ssize_t numObjects = 0;
+    char* vaddr = reinterpret_cast<char*>(events);
-    for (size_t i=0 ; i<count ; i++) {
+    ssize_t size = tube->read(vaddr, count*objSize);
-        char* vaddr = reinterpret_cast<char*>(events) + objSize * i;
+
-        ssize_t size = tube->read(vaddr, objSize);
+    // should never happen because of SOCK_SEQPACKET
-        if (size < 0) {
+    LOG_ALWAYS_FATAL_IF((size >= 0) && (size % objSize),
-            // error occurred
+            "BitTube::recvObjects(count=%d, size=%d), res=%d (partial events were received!)",
-            return size;
+            count, objSize, size);
-        } else if (size == 0) {
+
-            // no more messages
+    //ALOGE_IF(size<0, "error %d receiving %d events", size, count);
-            break;
+    return size < 0 ? size : size / objSize;
-        }
-        numObjects++;
-    }
-    return numObjects;
 }
 
 // ----------------------------------------------------------------------------

libs/gui/SensorEventQueue.cpp

@@ -35,12 +35,12 @@ namespace android {
 // ----------------------------------------------------------------------------
 
 SensorEventQueue::SensorEventQueue(const sp<ISensorEventConnection>& connection)
-    : mSensorEventConnection(connection)
+    : mSensorEventConnection(connection), mRecBuffer(NULL), mAvailable(0), mConsumed(0) {
-{
+    mRecBuffer = new ASensorEvent[MAX_RECEIVE_BUFFER_EVENT_COUNT];
 }
 
-SensorEventQueue::~SensorEventQueue()
+SensorEventQueue::~SensorEventQueue() {
-{
+    delete [] mRecBuffer;
 }
 
 void SensorEventQueue::onFirstRef()
@@ -59,9 +59,21 @@ ssize_t SensorEventQueue::write(const sp<BitTube>& tube,
     return BitTube::sendObjects(tube, events, numEvents);
 }
 
-ssize_t SensorEventQueue::read(ASensorEvent* events, size_t numEvents)
+ssize_t SensorEventQueue::read(ASensorEvent* events, size_t numEvents) {
-{
+    if (mAvailable == 0) {
-    return BitTube::recvObjects(mSensorChannel, events, numEvents);
+        ssize_t err = BitTube::recvObjects(mSensorChannel,
+                mRecBuffer, MAX_RECEIVE_BUFFER_EVENT_COUNT);
+        if (err < 0) {
+            return err;
+        }
+        mAvailable = err;
+        mConsumed = 0;
+    }
+    size_t count = numEvents < mAvailable ? numEvents : mAvailable;
+    memcpy(events, mRecBuffer + mConsumed, count*sizeof(ASensorEvent));
+    mAvailable -= count;
+    mConsumed += count;
+    return count;
 }
 
 sp<Looper> SensorEventQueue::getLooper() const

services/sensorservice/SensorService.cpp

@@ -292,8 +292,12 @@ bool SensorService::threadLoop()
 {
     ALOGD("nuSensorService thread starting...");
 
-    const size_t numEventMax = 16;
+    // each virtual sensor could generate an event per "real" event, that's why we need
-    const size_t minBufferSize = numEventMax + numEventMax * mVirtualSensorList.size();
+    // to size numEventMax much smaller than MAX_RECEIVE_BUFFER_EVENT_COUNT.
+    // in practice, this is too aggressive, but guaranteed to be enough.
+    const size_t minBufferSize = SensorEventQueue::MAX_RECEIVE_BUFFER_EVENT_COUNT;
+    const size_t numEventMax = minBufferSize / (1 + mVirtualSensorList.size());
+
     sensors_event_t buffer[minBufferSize];
     sensors_event_t scratch[minBufferSize];
     SensorDevice& device(SensorDevice::getInstance());