Revert "Deep build refactoring phase 2 (#3564)" (#3678)

This reverts commit 449029a827.

utils/messageqcpp/CMakeLists.txt

@@ -8,6 +8,7 @@ set(messageqcpp_LIB_SRCS
     iosocket.cpp
     messagequeue.cpp
     messagequeuepool.cpp
+    samenodepseudosocket.cpp
     socketparms.cpp
 )
 

utils/messageqcpp/messagequeuepool.cpp

@@ -25,58 +25,41 @@
 #include <new>
 #include <type_traits>
 
+
 namespace messageqcpp
 {
 
 using ClientMapType = std::multimap<std::string, std::unique_ptr<ClientObject>>;
 
-template <class T>
-struct Immortal
+struct LockedClientMap
 {
-  template <class... Args>
-  Immortal(Args&&... args)
-  {
-    ::new (space) T(std::forward<Args>(args)...);
-  }
-
-  operator T&() & noexcept
-  {
-    return reinterpret_cast<T&>(space);
-  }
-
- private:
-  alignas(T) unsigned char space[sizeof(T)];
-};
-
-class LockedClientMap
-{
-  struct KeyToUsePrivateCtor
-  {
-    explicit KeyToUsePrivateCtor() = default;
-  };
-
- public:
-  LockedClientMap(const LockedClientMap&) = delete;
-  LockedClientMap& operator=(const LockedClientMap&) = delete;
-  ~LockedClientMap() = delete;
-
-  static LockedClientMap& getInstance()
-  {
-    static Immortal<LockedClientMap> instance(KeyToUsePrivateCtor{});
-    return instance;
-  }
-
-  LockedClientMap(KeyToUsePrivateCtor)
+  LockedClientMap()
+  {
+  }
+  ~LockedClientMap()
   {
   }
-
- private:
   ClientMapType clientMap;
   std::mutex queueMutex;
-
-  friend class MessageQueueClientPool;
 };
 
+static int clientMapNiftyCounter;
+
+static typename std::aligned_storage<sizeof(LockedClientMap), alignof(LockedClientMap)>::type clientMapBuf;
+
+auto& lockedMap = reinterpret_cast<LockedClientMap&>(clientMapBuf);
+
+
+LockedClientMapInitilizer::LockedClientMapInitilizer ()
+{
+  if (clientMapNiftyCounter++ == 0) new (&lockedMap) LockedClientMap (); // placement new
+}
+LockedClientMapInitilizer::~LockedClientMapInitilizer ()
+{
+  if (--clientMapNiftyCounter == 0) (&lockedMap)->~LockedClientMap();
+}
+
+
 // 300 seconds idle until cleanup
 #define MAX_IDLE_TIME 300
 
@@ -87,7 +70,7 @@ static uint64_t TimeSpecToSeconds(struct timespec* ts)
 
 MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrIp, uint64_t port)
 {
-  auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
+  auto lock = std::scoped_lock(lockedMap.queueMutex);
 
   std::ostringstream oss;
   oss << dnOrIp << "_" << port;
@@ -110,13 +93,14 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrI
   newClientObject->client.reset(new MessageQueueClient(dnOrIp, port));
   newClientObject->inUse = true;
   newClientObject->lastUsed = nowSeconds;
-  LockedClientMap::getInstance().clientMap.emplace(std::move(searchString), std::move(newClientObject));
+  lockedMap.clientMap.emplace(std::move(searchString), std::move(newClientObject));
   return newClientObject->client.get();
 }
 
 MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& module)
 {
-  auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
+  auto lock = std::scoped_lock(lockedMap.queueMutex);
+
 
   MessageQueueClient* returnClient = MessageQueueClientPool::findInPool(module);
 
@@ -132,11 +116,13 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& modul
   clock_gettime(CLOCK_MONOTONIC, &now);
   uint64_t nowSeconds = TimeSpecToSeconds(&now);
 
+
+
   newClientObject->client.reset(new MessageQueueClient(module));
   newClientObject->inUse = true;
   newClientObject->lastUsed = nowSeconds;
   auto result = newClientObject->client.get();
-  LockedClientMap::getInstance().clientMap.emplace(std::move(module), std::move(newClientObject));
+  lockedMap.clientMap.emplace(std::move(module), std::move(newClientObject));
   return result;
 }
 
@@ -147,10 +133,11 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
   uint64_t nowSeconds = TimeSpecToSeconds(&now);
   MessageQueueClient* returnClient = NULL;
 
-  auto it = LockedClientMap::getInstance().clientMap.begin();
+  auto it = lockedMap.clientMap.begin();
+
 
   // Scan pool
-  while (it != LockedClientMap::getInstance().clientMap.end())
+  while (it != lockedMap.clientMap.end())
   {
     ClientObject* clientObject = it->second.get();
     uint64_t elapsedTime = nowSeconds - clientObject->lastUsed;
@@ -162,7 +149,7 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
       // Do this so we don't invalidate current interator
       auto toDelete = it;
       it++;
-      LockedClientMap::getInstance().clientMap.erase(toDelete);
+      lockedMap.clientMap.erase(toDelete);
       continue;
     }
 
@@ -176,7 +163,7 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
         // Do this so we don't invalidate current interator
         auto toDelete = it;
         it++;
-        LockedClientMap::getInstance().clientMap.erase(toDelete);
+        lockedMap.clientMap.erase(toDelete);
         continue;
       }
     }
@@ -206,10 +193,10 @@ void MessageQueueClientPool::releaseInstance(MessageQueueClient* client)
   if (client == NULL)
     return;
 
-  auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
-  auto it = LockedClientMap::getInstance().clientMap.begin();
+  auto lock = std::scoped_lock(lockedMap.queueMutex);
+  auto it = lockedMap.clientMap.begin();
 
-  while (it != LockedClientMap::getInstance().clientMap.end())
+  while (it != lockedMap.clientMap.end())
   {
     if (it->second->client.get() == client)
     {
@@ -234,14 +221,15 @@ void MessageQueueClientPool::deleteInstance(MessageQueueClient* client)
   if (client == NULL)
     return;
 
-  auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
-  auto it = LockedClientMap::getInstance().clientMap.begin();
 
-  while (it != LockedClientMap::getInstance().clientMap.end())
+  auto lock = std::scoped_lock(lockedMap.queueMutex);
+  auto it = lockedMap.clientMap.begin();
+
+  while (it != lockedMap.clientMap.end())
   {
     if (it->second->client.get() == client)
     {
-      LockedClientMap::getInstance().clientMap.erase(it);
+      lockedMap.clientMap.erase(it);
       return;
     }
 

utils/messageqcpp/messagequeuepool.h

@@ -26,6 +26,12 @@
 namespace messageqcpp
 {
 
+
+static struct LockedClientMapInitilizer {
+  LockedClientMapInitilizer ();
+  ~LockedClientMapInitilizer ();
+} clientMapInitilizer; // static initializer for every translation unit
+
 struct ClientObject
 {
   std::unique_ptr<MessageQueueClient> client;
@@ -43,8 +49,8 @@ class MessageQueueClientPool
   static MessageQueueClient* findInPool(const std::string& search);
 
  private:
-  MessageQueueClientPool() {};
-  ~MessageQueueClientPool() {};
+  MessageQueueClientPool(){};
+  ~MessageQueueClientPool(){};
 };
 
 }  // namespace messageqcpp

utils/messageqcpp/samenodepseudosocket.cpp

@@ -0,0 +1,127 @@
+/* Copyright (C) 2024 MariaDB Corp.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#include <string>
+
+#include "samenodepseudosocket.h"
+#include "iosocket.h"
+
+namespace messageqcpp
+{
+SameNodePseudoSocket::SameNodePseudoSocket(joblist::DistributedEngineComm* exeMgrDecPtr) : dec_(exeMgrDecPtr)
+{
+  assert(dec_);
+}
+
+SameNodePseudoSocket::~SameNodePseudoSocket()
+{
+}
+
+void SameNodePseudoSocket::open()
+{
+}
+
+void SameNodePseudoSocket::close()
+{
+}
+
+Socket* SameNodePseudoSocket::clone() const
+{
+  return nullptr;
+}
+
+SameNodePseudoSocket::SameNodePseudoSocket(const SameNodePseudoSocket& /*rhs*/)
+{
+}
+
+SameNodePseudoSocket& SameNodePseudoSocket::operator=(const SameNodePseudoSocket& /*rhs*/)
+{
+  return *this;
+}
+
+const SBS SameNodePseudoSocket::read(const struct ::timespec* /*timeout*/, bool* /*isTimeOut*/, Stats* /*stats*/) const
+{
+  return nullptr;
+}
+
+// This is the only working method of this class. It puts SBS directly into DEC queue.
+void SameNodePseudoSocket::write(SBS msg, Stats* /*stats*/)
+{
+  dec_->addDataToOutput(msg);
+}
+
+void SameNodePseudoSocket::write(const ByteStream& /*msg*/, Stats* /*stats*/)
+{
+}
+
+void SameNodePseudoSocket::write_raw(const ByteStream& /*msg*/, Stats* /*stats*/) const
+{
+}
+
+void SameNodePseudoSocket::connect(const sockaddr* /*serv_addr*/)
+{
+}
+
+void SameNodePseudoSocket::bind(const sockaddr* /*serv_addr*/)
+{
+}
+
+const IOSocket SameNodePseudoSocket::accept(const struct timespec* /*timeout*/)
+{
+  return IOSocket();
+}
+
+void SameNodePseudoSocket::listen(int /*backlog*/)
+{
+}
+
+const std::string SameNodePseudoSocket::toString() const
+{
+  return "";
+}
+
+const std::string SameNodePseudoSocket::addr2String() const
+{
+  return "";
+}
+
+bool SameNodePseudoSocket::isSameAddr(const Socket* /*rhs*/) const
+{
+  return false;
+}
+
+bool SameNodePseudoSocket::isSameAddr(const struct in_addr& /*ipv4Addr*/) const
+{
+  return false;
+}
+
+int SameNodePseudoSocket::ping(const std::string& /*ipaddr*/, const struct timespec* /*timeout*/)
+{
+  return 0;
+}
+
+bool SameNodePseudoSocket::isConnected() const
+{
+  return true;
+}
+
+bool SameNodePseudoSocket::hasData() const
+{
+  return false;
+}
+
+}  // namespace messageqcpp

utils/messageqcpp/samenodepseudosocket.h

@@ -0,0 +1,99 @@
+/* Copyright (C) 2024 MariaDB Corp.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#pragma once
+
+#include "../../dbcon/joblist/distributedenginecomm.h"
+
+#include "socket.h"
+#include "socketparms.h"
+#include "bytestream.h"
+
+namespace messageqcpp
+{
+class IOSocket;
+
+// This class is a dummy replacement for a TCP socket
+// wrapper to communicate with the same node.
+class SameNodePseudoSocket : public Socket
+{
+ public:
+  explicit SameNodePseudoSocket(joblist::DistributedEngineComm* exeMgrDecPtr);
+  ~SameNodePseudoSocket() override;
+  void write(SBS msg, Stats* stats = nullptr) override;
+
+ private:
+  void bind(const sockaddr* serv_addr) override;
+  SameNodePseudoSocket(const SameNodePseudoSocket& rhs);
+  virtual SameNodePseudoSocket& operator=(const SameNodePseudoSocket& rhs);
+
+  void connectionTimeout(const struct ::timespec* /*timeout*/) override
+  {
+  }
+
+  void syncProto(bool /*use*/) override
+  {
+  }
+
+  int getConnectionNum() const override
+  {
+    return 1;
+  }
+
+  inline void socketParms(const SocketParms& /*socket*/) override
+  {
+  }
+
+  inline const SocketParms socketParms() const override
+  {
+    return SocketParms();
+  }
+
+  // all these virtual methods are to stay inaccessable.
+  inline void sa(const sockaddr* sa) override;
+  void open() override;
+  void close() override;
+  inline bool isOpen() const override;
+  const SBS read(const struct timespec* timeout = nullptr, bool* isTimeOut = nullptr,
+                 Stats* stats = nullptr) const override;
+  void write(const ByteStream& msg, Stats* stats = nullptr) override;
+  void write_raw(const ByteStream& msg, Stats* stats = nullptr) const override;
+  void listen(int backlog = 5) override;
+  const IOSocket accept(const struct timespec* timeout = nullptr) override;
+  void connect(const sockaddr* serv_addr) override;
+  Socket* clone() const override;
+  virtual const std::string toString() const;
+  const std::string addr2String() const override;
+  bool isSameAddr(const Socket* rhs) const override;
+  bool isSameAddr(const struct in_addr& ipv4Addr) const override;
+  static int ping(const std::string& ipaddr, const struct timespec* timeout = nullptr);
+  bool isConnected() const override;
+  bool hasData() const override;
+
+  joblist::DistributedEngineComm* dec_ = nullptr;
+};
+
+inline bool SameNodePseudoSocket::isOpen() const
+{
+  return true;
+}
+
+inline void SameNodePseudoSocket::sa(const sockaddr* /*sa*/)
+{
+}
+
+}  // namespace messageqcpp