Bug#35703454 NdbSocket refactoring - use moves [NOCLOSE]

Back-ported for 8.0: Some parts of "WL#15524 Patch #16 Use TLS
for Event Listener" have been merged into this patch.

Use rvalue references (NdbSocket&&) when passing ownership of NdbSocket to
function, use const references (const NdbSocket&) otherwise.

When function creates a new instance of NdbSocket return it as a return
value rather than via an out parameter.

Implementation of struct ndb_logevent_handle is rewritten from being a
plain C struct to have proper constructors for both taking ownership of
socket and only borrow socket, and instead of using malloc and free
function new and delete are used.

Refactoring made SecureSocketOutputStream, BufferedSecureOutputStream,
and SocketInputStream redundant and they are removed.

Change-Id: I60caae52f2b0997b78ae179e5e9b47b6c99073b8

Author: zmur

storage/ndb/include/transporter/TransporterRegistry.hpp

@@ -110,7 +110,7 @@ class TransporterService : public SocketServer::Service {
   {
     m_transporter_registry= t;
   }
-  SocketServer::Session * newSession(ndb_socket_t socket) override;
+  SocketServer::Session * newSession(NdbSocket&& socket) override;
 };
 
 /**
@@ -253,7 +253,7 @@ class TransporterRegistry
 
      @returns false on failure and true on success
   */
-  bool connect_server(NdbSocket & sockfd,
+  bool connect_server(NdbSocket&& sockfd,
                       BaseString& msg,
                       bool& log_failure);
 

storage/ndb/include/util/InputStream.hpp

@@ -57,27 +57,18 @@ class FileInputStream : public InputStream {
 
 extern FileInputStream Stdin;
 
-class SecureSocketInputStream : public InputStream {
+class SocketInputStream : public InputStream {
   const NdbSocket & m_socket;
   unsigned m_timeout_ms;
   unsigned m_timeout_remain;
   bool m_startover;
   bool m_timedout;
 public:
-  SecureSocketInputStream(const NdbSocket &, unsigned read_timeout_ms = 3000);
-  ~SecureSocketInputStream() override {}
+  SocketInputStream(const NdbSocket &, unsigned read_timeout_ms = 3000);
+  ~SocketInputStream() override {}
   char* gets(char * buf, int bufLen) override;
   bool timedout() { return m_timedout; }
   void reset_timeout() override { m_timedout= false; m_timeout_remain= m_timeout_ms;}
 };
 
-class SocketInputStream : public SecureSocketInputStream {
-  NdbSocket m_owned_socket;
-public:
-  SocketInputStream(ndb_socket_t s, unsigned timeout_ms = 3000) :
-    SecureSocketInputStream(m_owned_socket, timeout_ms),
-    m_owned_socket(s, NdbSocket::From::Existing) {}
-  ~SocketInputStream() override {}
-};
-
 #endif

storage/ndb/include/util/NdbSocket.h

@@ -39,7 +39,7 @@ class NdbSocket {
 
   NdbSocket() = default;
   NdbSocket(NdbSocket&& oth);
-  NdbSocket(ndb_socket_t ndbsocket, From fromType) {
+  NdbSocket(ndb_socket_t ndbsocket, From fromType = From::New) {
     ssl = socket_table_get_ssl(ndbsocket.s, (fromType == From::Existing));
     init_from_native(ndbsocket.s);
   }
@@ -52,6 +52,7 @@ class NdbSocket {
   bool has_tls() const               { return ssl; }
   ndb_socket_t ndb_socket() const    { return s; }
   socket_t native_socket() const     { return ndb_socket_get_native(s); }
+  socket_t release_native_socket();
   std::string to_string() const;
 
   /* Get an SSL for client-side TLS.
@@ -239,6 +240,15 @@ void NdbSocket::invalidate_socket_handle() {
   ndb_socket_invalidate(&s);
 }
 
+inline
+socket_t NdbSocket::release_native_socket() {
+  require(ssl == nullptr);
+  require(mutex == nullptr);
+  socket_t sock = s.s;
+  invalidate_socket_handle();
+  return sock;
+}
+
 inline
 std::string NdbSocket::to_string() const {
   std::string str = ndb_socket_to_string(s);

storage/ndb/include/util/OutputStream.hpp

@@ -83,15 +83,15 @@ class FileOutputStream : public OutputStream {
   void flush() override { fflush(f); }
 };
 
-class SecureSocketOutputStream : public OutputStream {
+class SocketOutputStream : public OutputStream {
 protected:
   const NdbSocket & m_socket;
   unsigned m_timeout_ms;
   bool m_timedout;
   unsigned m_timeout_remain;
 public:
-  SecureSocketOutputStream(const NdbSocket &, unsigned write_timeout_ms = 1000);
-  ~SecureSocketOutputStream() override {}
+  SocketOutputStream(const NdbSocket &, unsigned write_timeout_ms = 1000);
+  ~SocketOutputStream() override {}
   bool timedout() { return m_timedout; }
   void reset_timeout() override { m_timedout= false; m_timeout_remain= m_timeout_ms;}
 
@@ -102,24 +102,13 @@ class SecureSocketOutputStream : public OutputStream {
   int write(const void * buf, size_t len) override;
 };
 
-class SocketOutputStream : public SecureSocketOutputStream {
-public:
-  SocketOutputStream(ndb_socket_t s, unsigned timeout_ms = 1000) :
-    SecureSocketOutputStream(m_owned_socket, timeout_ms),
-    m_owned_socket(s, NdbSocket::From::Existing) {}
-  ~SocketOutputStream() override {}
-
-private:
-  NdbSocket m_owned_socket;
-};
-
-class BufferedSecureOutputStream : public SecureSocketOutputStream {
+class BufferSocketOutputStream : public SocketOutputStream {
 protected:
   class UtilBuffer& m_buffer;
 public:
-  BufferedSecureOutputStream(const NdbSocket & socket,
-                             unsigned write_timeout_ms = 1000);
-  ~BufferedSecureOutputStream() override;
+  BufferSocketOutputStream(const NdbSocket & socket,
+                           unsigned write_timeout_ms = 1000);
+  ~BufferSocketOutputStream() override;
 
   int print(const char * fmt, ...) override
     ATTRIBUTE_FORMAT(printf, 2, 3);
@@ -130,18 +119,6 @@ class BufferedSecureOutputStream : public SecureSocketOutputStream {
   void flush() override;
 };
 
-class BufferedSockOutputStream : public BufferedSecureOutputStream {
-public:
-  BufferedSockOutputStream(ndb_socket_t s, unsigned timeout_msec = 1000) :
-    BufferedSecureOutputStream(m_owned_socket, timeout_msec),
-    m_owned_socket(s, NdbSocket::From::Existing) {}
-
-  ~BufferedSockOutputStream() override {}
-private:
-  NdbSocket m_owned_socket;
-};
-
-
 class NullOutputStream : public OutputStream {
 public:
   NullOutputStream() {}

storage/ndb/include/util/SocketAuthenticator.hpp

@@ -37,8 +37,8 @@ class SocketAuthenticator
 public:
   SocketAuthenticator() {}
   virtual ~SocketAuthenticator() {}
-  virtual int client_authenticate(NdbSocket &) = 0;
-  virtual int server_authenticate(NdbSocket &) = 0;
+  virtual int client_authenticate(const NdbSocket &) = 0;
+  virtual int server_authenticate(const NdbSocket &) = 0;
 
   static constexpr int AuthOk = 0;
   static const char * error(int);   // returns error message for code
@@ -50,8 +50,8 @@ class SocketAuthSimple : public SocketAuthenticator
 public:
   SocketAuthSimple() {}
   ~SocketAuthSimple() override {}
-  int client_authenticate(NdbSocket &) override;
-  int server_authenticate(NdbSocket &) override;
+  int client_authenticate(const NdbSocket &) override;
+  int server_authenticate(const NdbSocket &) override;
 };
 
 

storage/ndb/include/util/SocketClient.hpp

@@ -44,9 +44,8 @@ class SocketClient
     m_connect_timeout_millisec = timeout_millisec;
   }
   int bind(ndb_sockaddr local);
-  ndb_socket_t connect(ndb_sockaddr server_addr);
-  void connect(NdbSocket &, ndb_sockaddr server_addr);
-  int authenticate(NdbSocket &);
+  NdbSocket connect(ndb_sockaddr server_addr);
+  int authenticate(const NdbSocket &);
 
   ndb_socket_t m_sockfd;
 };

storage/ndb/include/util/SocketServer.hpp

@@ -29,6 +29,7 @@
 #include "portlib/NdbThread.h"
 #include "portlib/ndb_socket_poller.h"
 #include "portlib/ndb_sockaddr.h"
+#include "util/NdbSocket.h"
 
 #include <Vector.hpp>
 
@@ -51,21 +52,21 @@ class SocketServer {
   protected:
     friend class SocketServer;
     friend void* sessionThread_C(void*);
-    Session(ndb_socket_t sock) :
+    Session(const NdbSocket& sock) :
       m_stop(false),
       m_refCount(0),
       m_socket(sock),
       m_thread_stopped(false)
       {
 	DBUG_ENTER("SocketServer::Session");
 	DBUG_PRINT("enter",("NDB_SOCKET: %s",
-                            ndb_socket_to_string(m_socket).c_str()));
+                            m_socket.to_string().c_str()));
 	DBUG_VOID_RETURN;
       }
     bool m_stop;    // Has the session been ordered to stop?
     unsigned m_refCount;
   private:
-    ndb_socket_t m_socket;
+    const NdbSocket& m_socket;
     bool m_thread_stopped; // Has the session thread stopped?
   };
 
@@ -82,7 +83,7 @@ class SocketServer {
      *
      * To manage threads self, just return NULL
      */
-    virtual Session * newSession(ndb_socket_t theSock) = 0;
+    virtual Session * newSession(NdbSocket&& theSock) = 0;
     virtual void stopSessions(){}
   };
 

storage/ndb/src/common/transporter/Multi_Transporter.cpp

@@ -68,12 +68,12 @@ Multi_Transporter::resetBuffers()
   send_checksum_state.init();
 }
 
-bool Multi_Transporter::connect_server_impl(NdbSocket &)
+bool Multi_Transporter::connect_server_impl(NdbSocket&&)
 {
   return true;
 }
 
-bool Multi_Transporter::connect_client_impl(NdbSocket &)
+bool Multi_Transporter::connect_client_impl(NdbSocket&&)
 {
   return true;
 }

storage/ndb/src/common/transporter/Multi_Transporter.hpp

@@ -143,8 +143,8 @@ class Multi_Transporter : public Transporter {
    * A client connects to the remote server
    * A server accepts any new connections
    */
-  bool connect_server_impl(NdbSocket & sockfd) override;
-  bool connect_client_impl(NdbSocket & sockfd) override;
+  bool connect_server_impl(NdbSocket&& sockfd) override;
+  bool connect_client_impl(NdbSocket&& sockfd) override;
 
   /**
    * Disconnects a TCP/IP node, possibly blocking.

storage/ndb/src/common/transporter/SHM_Transporter.cpp

@@ -321,13 +321,13 @@ SHM_Transporter::setupBuffers()
 }
 
 bool
-SHM_Transporter::connect_server_impl(NdbSocket & sockfd)
+SHM_Transporter::connect_server_impl(NdbSocket&& sockfd)
 {
   DBUG_ENTER("SHM_Transporter::connect_server_impl");
   DEBUG_FPRINTF((stderr, "(%u)connect_server_impl(%u)\n",
                  localNodeId, remoteNodeId));
-  SecureSocketOutputStream s_output(sockfd);
-  SecureSocketInputStream s_input(sockfd);
+  SocketOutputStream s_output(sockfd);
+  SocketInputStream s_input(sockfd);
 
   // Create
   if (!_shmSegCreated)
@@ -419,12 +419,12 @@ SHM_Transporter::connect_server_impl(NdbSocket & sockfd)
   }
   DEBUG_FPRINTF((stderr, "(%u)set_socket()(%u)\n",
                  localNodeId, remoteNodeId));
-  set_socket(sockfd);
+  set_socket(std::move(sockfd));
   DBUG_RETURN(r);
 }
 
 void
-SHM_Transporter::set_socket(NdbSocket & sock)
+SHM_Transporter::set_socket(NdbSocket&& sock)
 {
   set_get(sock.ndb_socket(), IPPROTO_TCP, TCP_NODELAY, "TCP_NODELAY", 1);
   set_get(sock.ndb_socket(), SOL_SOCKET, SO_KEEPALIVE, "SO_KEEPALIVE", 1);
@@ -436,13 +436,13 @@ SHM_Transporter::set_socket(NdbSocket & sock)
 }
 
 bool
-SHM_Transporter::connect_client_impl(NdbSocket & sockfd)
+SHM_Transporter::connect_client_impl(NdbSocket&& sockfd)
 {
   DBUG_ENTER("SHM_Transporter::connect_client_impl");
   DEBUG_FPRINTF((stderr, "(%u)connect_client_impl(%u)\n",
                  localNodeId, remoteNodeId));
-  SecureSocketInputStream s_input(sockfd);
-  SecureSocketOutputStream s_output(sockfd);
+  SocketInputStream s_input(sockfd);
+  SocketOutputStream s_output(sockfd);
   char buf[256];
 
   // Wait for server to create and attach
@@ -550,7 +550,7 @@ SHM_Transporter::connect_client_impl(NdbSocket & sockfd)
                            localNodeId, __LINE__, remoteNodeId));
     detach_shm(false);
   }
-  set_socket(sockfd);
+  set_socket(std::move(sockfd));
   DEBUG_FPRINTF((stderr, "(%u)set_socket(%u)\n",
                  localNodeId, remoteNodeId));
   DBUG_RETURN(r);

storage/ndb/src/common/transporter/SHM_Transporter.hpp

@@ -111,7 +111,7 @@ class SHM_Transporter : public Transporter {
    *            i.e., both agrees that the other one has setup the segment.
    *            Otherwise false.
    */
-  bool connect_server_impl(NdbSocket &) override;
+  bool connect_server_impl(NdbSocket&&) override;
 
   /**
    * Blocking
@@ -124,15 +124,15 @@ class SHM_Transporter : public Transporter {
    *            i.e., both agrees that the other one has setup the segment.
    *            Otherwise false.
    */
-  bool connect_client_impl(NdbSocket &) override;
+  bool connect_client_impl(NdbSocket&&) override;
 
   bool connect_common();
 
   bool ndb_shm_create();
   bool ndb_shm_get();
   bool ndb_shm_attach();
   void ndb_shm_destroy();
-  void set_socket(NdbSocket &);
+  void set_socket(NdbSocket&&);
 
   /**
    * Check if there are two processes attached to the segment (a connection)

storage/ndb/src/common/transporter/TCP_Transporter.cpp

@@ -201,19 +201,19 @@ TCP_Transporter::resetBuffers()
   send_checksum_state.init();
 }
 
-bool TCP_Transporter::connect_server_impl(NdbSocket & socket)
+bool TCP_Transporter::connect_server_impl(NdbSocket&& socket)
 {
   DBUG_ENTER("TCP_Transpporter::connect_server_impl");
-  DBUG_RETURN(connect_common(socket));
+  DBUG_RETURN(connect_common(std::move(socket)));
 }
 
-bool TCP_Transporter::connect_client_impl(NdbSocket & socket)
+bool TCP_Transporter::connect_client_impl(NdbSocket&& socket)
 {
   DBUG_ENTER("TCP_Transpporter::connect_client_impl");
-  DBUG_RETURN(connect_common(socket));
+  DBUG_RETURN(connect_common(std::move(socket)));
 }
 
-bool TCP_Transporter::connect_common(NdbSocket & socket)
+bool TCP_Transporter::connect_common(NdbSocket&& socket)
 {
   setSocketOptions(socket.ndb_socket());
   socket.set_nonblocking(true);

storage/ndb/src/common/transporter/TCP_Transporter.hpp

@@ -109,9 +109,9 @@ class TCP_Transporter : public Transporter {
    * A client connects to the remote server
    * A server accepts any new connections
    */
-  bool connect_server_impl(NdbSocket & sockfd) override;
-  bool connect_client_impl(NdbSocket & sockfd) override;
-  bool connect_common(NdbSocket & sockfd);
+  bool connect_server_impl(NdbSocket&& sockfd) override;
+  bool connect_client_impl(NdbSocket&& sockfd) override;
+  bool connect_common(NdbSocket&& sockfd);
 
   /**
    * Disconnects a TCP/IP node, possibly blocking.