SERVER-28239 fix double close in ticket holder

The legacy transport layer allowed end() to be called multiple times on
the same session, which would incorrectly double close + double
increment the ticket holder.  This occured during primary step down, as
endAllSessions was called (which called end()) and then the owning
threads called end() as they finished.

Fix involves a mutex per connection and a check for _closed in
closeConnection.

(cherry picked from commit b1ec4cfbb535ddc7ee741a76132fee67a3ab1d1e)

5 files changed, 126 insertions, 1 deletions

diff --git a/src/mongo/transport/SConscript b/src/mongo/transport/SConscript
index 1307514466e..8e414610b2c 100644
--- a/src/mongo/transport/SConscript
+++ b/src/mongo/transport/SConscript
@@ -126,3 +126,15 @@ env.CppUnitTest(
     ]
 )
 
+env.CppUnitTest(
+    target='transport_layer_legacy_test',
+    source=[
+        'transport_layer_legacy_test.cpp',
+    ],
+    LIBDEPS=[
+        'transport_layer_legacy',
+        '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/db/service_context_noop_init',
+    ],
+)
+
diff --git a/src/mongo/transport/transport_layer_legacy.cpp b/src/mongo/transport/transport_layer_legacy.cpp
index a3b9c7306b0..d5da7a53d52 100644
--- a/src/mongo/transport/transport_layer_legacy.cpp
+++ b/src/mongo/transport/transport_layer_legacy.cpp
@@ -226,6 +226,12 @@ void TransportLayerLegacy::end(const SessionHandle& session) {
 }
 
 void TransportLayerLegacy::_closeConnection(Connection* conn) {
+    stdx::lock_guard<stdx::mutex> lk(conn->closeMutex);
+
+    if (conn->closed) {
+        return;
+    }
+
     conn->closed = true;
     conn->amp->shutdown();
     Listener::globalTicketHolder.release();
diff --git a/src/mongo/transport/transport_layer_legacy.h b/src/mongo/transport/transport_layer_legacy.h
index 6ef7b73a7c1..2d4ea57b446 100644
--- a/src/mongo/transport/transport_layer_legacy.h
+++ b/src/mongo/transport/transport_layer_legacy.h
@@ -114,6 +114,8 @@ private:
         Connection(std::unique_ptr<AbstractMessagingPort> port)
             : amp(std::move(port)), connectionId(amp->connectionId()) {}
 
+        stdx::mutex closeMutex;
+
         std::unique_ptr<AbstractMessagingPort> amp;
 
         const long long connectionId;
diff --git a/src/mongo/transport/transport_layer_legacy_test.cpp b/src/mongo/transport/transport_layer_legacy_test.cpp
new file mode 100644
index 00000000000..30627a9a453
--- /dev/null
+++ b/src/mongo/transport/transport_layer_legacy_test.cpp
@@ -0,0 +1,103 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/transport/service_entry_point.h"
+#include "mongo/transport/transport_layer_legacy.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/assert_util.h"
+
+namespace mongo {
+namespace {
+
+class ServiceEntryPointUtil : public ServiceEntryPoint {
+public:
+    void startSession(transport::SessionHandle session) {
+        Message m;
+        Status s = session->sourceMessage(&m).wait();
+
+        ASSERT_NOT_OK(s);
+
+        tll->end(session);
+    }
+
+    transport::TransportLayerLegacy* tll = nullptr;
+};
+
+// This test verifies a fix for SERVER-28239.  The actual service entry point in use by mongod and
+// mongos calls end() manually, which occasionally was the second call to end() if after a primary
+// stepdown.  This tests verifies our fix (making end() safe to call multiple times)
+TEST(TransportLayerLegacy, endSessionsDoesntDoubleClose) {
+    ServiceEntryPointUtil sepu;
+
+    transport::TransportLayerLegacy::Options opts{};
+    opts.port = 27017;
+    transport::TransportLayerLegacy tll(opts, &sepu);
+
+    sepu.tll = &tll;
+
+    tll.setup();
+    tll.start();
+
+    stdx::mutex mutex;
+    bool end = false;
+    stdx::condition_variable cv;
+
+    stdx::thread thr{[&] {
+        Socket s;
+        SockAddr sa{"localhost", 27017};
+        s.connect(sa);
+
+        stdx::unique_lock<stdx::mutex> lk(mutex);
+        cv.wait(lk, [&] { return end; });
+    }};
+
+    while (Listener::globalTicketHolder.used() == 0) {
+    }
+
+    tll.endAllSessions(transport::Session::TagMask{});
+
+    while (Listener::globalTicketHolder.used() == 1) {
+    }
+
+    {
+        stdx::lock_guard<stdx::mutex> lk(mutex);
+        end = true;
+        cv.notify_one();
+    }
+
+    thr.join();
+
+    ASSERT(Listener::globalTicketHolder.used() == 0);
+
+    tll.shutdown();
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/util/net/listen.cpp b/src/mongo/util/net/listen.cpp
index 1842a091b41..7f76fbf8fac 100644
--- a/src/mongo/util/net/listen.cpp
+++ b/src/mongo/util/net/listen.cpp
@@ -455,7 +455,9 @@ void Listener::initAndListen() {
         events[count] = ev->get();
     }
 
-    while (!inShutdown()) {
+    // The check against _finished allows us to actually stop the listener by signalling it through
+    // the _finished flag.
+    while (!inShutdown() && !_finished.load()) {
         // Turn on listening for accept-ready sockets
         for (size_t count = 0; count < _socks.size(); ++count) {
             int status = WSAEventSelect(_socks[count], events[count], FD_ACCEPT | FD_CLOSE);