rs close sockets when relinquish primary so clients arent surprised

4 files changed, 17 insertions, 11 deletions

diff --git a/db/instance.cpp b/db/instance.cpp
index eeaadac7511..f7511fac5fe 100644
--- a/db/instance.cpp
+++ b/db/instance.cpp
@@ -62,7 +62,6 @@ namespace mongo {
     bool useCursors = true;
     bool useHints = true;
     
-    void closeAllSockets();
     void flushOpLog( stringstream &ss ) {
         if( _diaglog.f && _diaglog.f->is_open() ) {
             ss << "flushing op log and files\n";
@@ -730,7 +729,7 @@ namespace mongo {
 
         /* must do this before unmapping mem or you may get a seg fault */
         log() << "shutdown: going to close sockets..." << endl;
-        boost::thread close_socket_thread(closeAllSockets);
+        boost::thread close_socket_thread( boost::bind(MessagingPort::closeAllSockets, 0) );
 
         // wait until file preallocation finishes
         // we would only hang here if the file_allocator code generates a
diff --git a/db/repl/rs.cpp b/db/repl/rs.cpp
index 116e33e1a48..f8d72085e28 100644
--- a/db/repl/rs.cpp
+++ b/db/repl/rs.cpp
@@ -66,6 +66,12 @@ namespace mongo {
         if( box.getState().primary() ) {
             log() << "replSet relinquishing primary state" << rsLog;
             changeState(MemberState::RS_RECOVERING);
+            
+            /* close sockets that were talking to us */
+            log() << "replSet closing sockets after reqlinquishing primary" << rsLog;
+            MessagingPort::closeAllSockets(1);
+
+            // todo: >
             //changeState(MemberState::RS_SECONDARY);
         }
         else if( box.getState().startup2() ) {
diff --git a/util/message.cpp b/util/message.cpp
index 459a9c7c0e1..ddd7b160c09 100644
--- a/util/message.cpp
+++ b/util/message.cpp
@@ -255,14 +255,10 @@ namespace mongo {
             _cur = _buf;
         }
 
-        int len() {
-            return _cur - _buf;
-        }
+        int len() const { return _cur - _buf; }
 
     private:
-
         MessagingPort* _port;
-
         char * _buf;
         char * _cur;
     };
@@ -272,10 +268,13 @@ namespace mongo {
         mongo::mutex m;
     public:
         Ports() : ports(), m("Ports") {}
-        void closeAll() {
+        void closeAll(unsigned skip_mask) {
             scoped_lock bl(m);
-            for ( set<MessagingPort*>::iterator i = ports.begin(); i != ports.end(); i++ )
+            for ( set<MessagingPort*>::iterator i = ports.begin(); i != ports.end(); i++ ) {
+                if( (*i)->tag & skip_mask )
+                    continue;
                 (*i)->shutdown();
+            }
         }
         void insert(MessagingPort* p) { 
             scoped_lock bl(m);
@@ -291,8 +290,8 @@ namespace mongo {
     // are being destructed during termination.
     Ports& ports = *(new Ports());
 
-    void closeAllSockets() {
-        ports.closeAll();
+    void MessagingPort::closeAllSockets(unsigned mask) {
+        ports.closeAll(mask);
     }
 
     MessagingPort::MessagingPort(int _sock, const SockAddr& _far) : sock(_sock), piggyBackData(0), farEnd(_far), _timeout(), tag(0) {
diff --git a/util/message.h b/util/message.h
index 975240f3ec5..2b9119e0ad2 100644
--- a/util/message.h
+++ b/util/message.h
@@ -136,6 +136,8 @@ namespace mongo {
         int _timeout;
         int _logLevel; // passed to log() when logging errors
 
+        static void closeAllSockets(unsigned tagMask = 0xffffffff);
+
         /* ports can be tagged with various classes.  see closeAllSockets(tag). defaults to 0. */
         unsigned tag;