Fix ShellSubscriber concurrency issues

This CL creates a sendHeartbeat thread that ocassionally sends
heartbeats, consisting of a dataSize of 0, to perfd. perfd will discard
those heartbeats, recheck if the user has canceled the subscription, and
if not, wait for more data from statsd. Sending heartbeats solves two
big problems:

(1) Allows statsd to robustly check if writes to the socket fail because
the read end of the pipe has closed. Previously, if no atoms were pushed
or pulled, statsd never attempted to write to perfd, so statsd could
never detect the end of the subscription. However, now, writes are
regularly made regardless of if statsd receives data. Note that even if
atoms were pushed or pulled, there is no guarantee that they would have
matched the atom matchers sent in perfd's config.

(2) Allows perfd to escape a blocking read call and recheck whether the
user has canceled the subscription. If no data is sent to perfd, perfd
will block in this read call and the AndroidStudio UI will freeze up.

Heartbeats are only sent if statsd has not sent any data to perfd within
the last second, so we do not spam perfd with writes.

+ decomposes the startNewSubscription function
+ prevents startPull from holding the lock while sleeping

Test: atest stastd_test
Test: atest CtsStatsdHostTestCases
Test: manually confirm that AndroidStudio is not freezing
Bug: 153595161

Change-Id: I78f0818e8ed29bdadd02c151444ee7c9555623a4

1 files changed, 26 insertions, 20 deletions

diff --git a/cmds/statsd/tests/shell/ShellSubscriber_test.cpp b/cmds/statsd/tests/shell/ShellSubscriber_test.cpp
index 7b952d7a392e..363fcb4bf193 100644
--- a/cmds/statsd/tests/shell/ShellSubscriber_test.cpp
+++ b/cmds/statsd/tests/shell/ShellSubscriber_test.cpp
@@ -86,28 +86,34 @@ void runShellTest(ShellSubscription config, sp<MockUidMap> uidMap,
     // wait for the data to be written.
     std::this_thread::sleep_for(100ms);
 
-    int expected_data_size = expectedData.ByteSize();
-
-    // now read from the pipe. firstly read the atom size.
-    size_t dataSize = 0;
-    EXPECT_EQ((int)sizeof(dataSize), read(fds_data[0], &dataSize, sizeof(dataSize)));
-
-    EXPECT_EQ(expected_data_size, (int)dataSize);
-
-    // then read that much data which is the atom in proto binary format
-    vector<uint8_t> dataBuffer(dataSize);
-    EXPECT_EQ((int)dataSize, read(fds_data[0], dataBuffer.data(), dataSize));
-
-    // make sure the received bytes can be parsed to an atom
-    ShellData receivedAtom;
-    EXPECT_TRUE(receivedAtom.ParseFromArray(dataBuffer.data(), dataSize) != 0);
+    // Because we might receive heartbeats from statsd, consisting of data sizes
+    // of 0, encapsulate reads within a while loop.
+    bool readAtom = false;
+    while (!readAtom) {
+        // Read the atom size.
+        size_t dataSize = 0;
+        read(fds_data[0], &dataSize, sizeof(dataSize));
+        if (dataSize == 0) continue;
+        EXPECT_EQ(expectedData.ByteSize(), int(dataSize));
+
+        // Read that much data in proto binary format.
+        vector<uint8_t> dataBuffer(dataSize);
+        EXPECT_EQ((int)dataSize, read(fds_data[0], dataBuffer.data(), dataSize));
+
+        // Make sure the received bytes can be parsed to an atom.
+        ShellData receivedAtom;
+        EXPECT_TRUE(receivedAtom.ParseFromArray(dataBuffer.data(), dataSize) != 0);
+
+        // Serialize the expected atom to byte array and compare to make sure
+        // they are the same.
+        vector<uint8_t> expectedAtomBuffer(expectedData.ByteSize());
+        expectedData.SerializeToArray(expectedAtomBuffer.data(), expectedData.ByteSize());
+        EXPECT_EQ(expectedAtomBuffer, dataBuffer);
+
+        readAtom = true;
+    }
 
-    // serialze the expected atom to bytes. and compare. to make sure they are the same.
-    vector<uint8_t> atomBuffer(expected_data_size);
-    expectedData.SerializeToArray(&atomBuffer[0], expected_data_size);
-    EXPECT_EQ(atomBuffer, dataBuffer);
     close(fds_data[0]);
-
     if (reader.joinable()) {
         reader.join();
     }