Fix ThreadSanitiser issue found with TapProducer

The problem occurs because we're reading expiry time unlocked
but set it with the queueLock held.

It seems that the disconnect logic should really look at the 3 "elements" it cares
about in a consistent way, so the new disconnect method locks the producer and
evaluates all 3 parts.

TSAN output (shortened):

WARNING: ThreadSanitizer: data race (pid=139855)
  Read of size 4 at 0x7d600000f068 by thread T18 (mutexes: write M2532):
    #0 ConnHandler::getExpiryTime() /ep-engine/src/tapconnection.h:350
    couchbase#1 ConnManager::run() /ep-engine/src/connmap.cc:150
    couchbase#2 ExecutorThread::run() /ep-engine/src/executorthread.cc:115
    couchbase#3 launch_executor_thread(void*) /ep-engine/src/executorthread.cc:33
    couchbase#4 platform_thread_wrap(void*) /platform/src/cb_pthreads.cc:54

  Previous write of size 4 at 0x7d600000f068 by main thread (mutexes: write M22798, write M22800):
    #0 TapProducer::processAck(...) /ep-engine/src/tapconnection.h:346
    couchbase#1 EventuallyPersistentEngine::processTapAck(v..) /ep-engine/src/ep_engine.cc:3038
    couchbase#2 EventuallyPersistentEngine::tapNotify(...) /ep-engine/src/ep_engine.cc:2761
    couchbase#3 EvpTapNotify(...) /ep-engine/src/ep_engine.cc:1446
    couchbase#4 mock_tap_notify(...) /memcached/programs/engine_testapp/engine_testapp.cc:442
    couchbase#5 test_tap_ack_stream(...) /ep-engine/tests/ep_testsuite.cc:7851
    #6 execute_test(...) /memcached/programs/engine_testapp/engine_testapp.cc:1091

Change-Id: Ibf13c2958aac3da745506e8ac6995ce76bd03223
Reviewed-on: http://review.couchbase.org/57234
Reviewed-by: Dave Rigby <daver@couchbase.com>
Tested-by: buildbot <build@couchbase.com>
Reviewed-by: Chiyoung Seo <chiyoung@couchbase.com>

Author: jimwwalker

src/connmap.cc

@@ -489,7 +489,7 @@ void TapConnMap::manageConnections() {
     for (iter = map_.begin(); iter != map_.end(); ++iter) {
         TapProducer *tp = dynamic_cast<TapProducer*>(iter->second.get());
         if (tp != NULL) {
-            if (tp->supportsAck() && (tp->getExpiryTime() < now) && tp->windowIsFull()) {
+            if (tp->shouldDisconnect(now)) {
                 LOG(EXTENSION_LOG_WARNING,
                     "%s Expired and ack windows is full. Disconnecting...",
                     tp->logHeader());

src/tapconnection.h

@@ -940,6 +940,11 @@ class TapProducer : public Producer {
         clearQueues_UNLOCKED();
     }
 
+    bool shouldDisconnect(rel_time_t now) {
+        LockHolder lh(queueLock);
+        return supportsAck() && (getExpiryTime() < now) && windowIsFull();
+    }
+
     static const char* opaqueCmdToString(uint32_t opaque_code);
 
 protected: