[master] Zil 5450 improvements to network and native tests

src/libNetwork/P2PServer.cpp

@@ -23,6 +23,7 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/read.hpp>
+#include <boost/container/small_vector.hpp>
 
 #include "Blacklist.h"
 #include "libUtils/Logger.h"
@@ -202,9 +203,10 @@ std::shared_ptr<P2PServer> P2PServer::CreateAndStart(AsioContext& asio,
   throw std::runtime_error(error_msg);
 }
 
-P2PServerConnection::P2PServerConnection(
-    std::weak_ptr<P2PServerImpl> owner, uint64_t this_id, Peer&& remote_peer,
-    TcpSocket socket, size_t max_message_size)
+P2PServerConnection::P2PServerConnection(std::weak_ptr<P2PServerImpl> owner,
+                                         uint64_t this_id, Peer&& remote_peer,
+                                         TcpSocket socket,
+                                         size_t max_message_size)
     : m_owner(std::move(owner)),
       m_id(this_id),
       m_remotePeer(std::move(remote_peer)),
@@ -229,6 +231,9 @@ void P2PServerConnection::ReadNextMessage() {
         if (!ec) {
           assert(n == HDR_LEN);
         }
+        if (ec) {
+          // LOG_GENERAL(WARNING, "Got error code: " << ec.message());
+        }
         if (ec != OPERATION_ABORTED) {
           self->OnHeaderRead(ec);
         }
@@ -237,10 +242,9 @@ void P2PServerConnection::ReadNextMessage() {
 
 void P2PServerConnection::OnHeaderRead(const ErrorCode& ec) {
   if (ec) {
-    if (ec != END_OF_FILE) {
-      LOG_GENERAL(INFO,
-                  "Peer " << m_remotePeer << " read error: " << ec.message());
-    }
+    LOG_GENERAL(INFO,
+                "Peer " << m_remotePeer << " read error: " << ec.message());
+    CloseSocket();
     OnConnectionClosed();
     return;
   }
@@ -254,7 +258,9 @@ void P2PServerConnection::OnHeaderRead(const ErrorCode& ec) {
                              << " Adding sending node "
                              << m_remotePeer.GetPrintableIPAddress()
                              << " as strictly blacklisted");
-    Blacklist::GetInstance().Add({m_remotePeer.GetIpAddress(),m_remotePeer.GetListenPortHost(),m_remotePeer.GetNodeIndentifier()});
+    Blacklist::GetInstance().Add({m_remotePeer.GetIpAddress(),
+                                  m_remotePeer.GetListenPortHost(),
+                                  m_remotePeer.GetNodeIndentifier()});
 
     CloseSocket();
     OnConnectionClosed();
@@ -269,6 +275,9 @@ void P2PServerConnection::OnHeaderRead(const ErrorCode& ec) {
         if (!ec) {
           assert(n == self->m_readBuffer.size() - HDR_LEN);
         }
+        if (ec) {
+          // LOG_GENERAL(WARNING, "Got error code: " << ec.message());
+        }
         if (ec != OPERATION_ABORTED) {
           self->OnBodyRead(ec);
         }
@@ -278,17 +287,19 @@ void P2PServerConnection::OnHeaderRead(const ErrorCode& ec) {
 void P2PServerConnection::OnBodyRead(const ErrorCode& ec) {
   if (ec) {
     LOG_GENERAL(INFO, "Read error: " << ec.message());
+    CloseSocket();
     OnConnectionClosed();
     return;
   }
-
   ReadMessageResult result;
   auto state = TryReadMessage(m_readBuffer.data(), m_readBuffer.size(), result);
 
   if (state != ReadState::SUCCESS) {
     LOG_GENERAL(WARNING, "Message deserialize error: blacklisting "
                              << m_remotePeer.GetPrintableIPAddress());
-    Blacklist::GetInstance().Add({m_remotePeer.GetIpAddress(),m_remotePeer.GetListenPortHost(),m_remotePeer.GetNodeIndentifier()});
+    Blacklist::GetInstance().Add({m_remotePeer.GetIpAddress(),
+                                  m_remotePeer.GetListenPortHost(),
+                                  m_remotePeer.GetNodeIndentifier()});
 
     CloseSocket();
     OnConnectionClosed();
@@ -298,6 +309,7 @@ void P2PServerConnection::OnBodyRead(const ErrorCode& ec) {
   auto owner = m_owner.lock();
   if (!owner || !owner->OnMessage(m_id, m_remotePeer, result)) {
     CloseSocket();
+    OnConnectionClosed();
     return;
   }
 
@@ -311,8 +323,44 @@ void P2PServerConnection::Close() {
 
 void P2PServerConnection::CloseSocket() {
   ErrorCode ec;
+  // both close and shutdown should be none blocking calls certainly on current linux
+  // shutdown marks the socket as blocked for both read and write
+  // shutdown tells the OS to begin the graceful closedown of the TCP connection.
+  // close() is a blocking call that waits for the OS to complete the closedown.
+  // close also frees the OS resources from the program so should be called even if
+  // an error condition is encountered
   m_socket.shutdown(boost::asio::socket_base::shutdown_both, ec);
+  if (ec) {
+    m_socket.close(ec);
+    if (ec) {
+      LOG_GENERAL(INFO, "Informational, not an issue - Error closing socket: " << ec.message());
+    }
+    return;
+  }
+  size_t unread = m_socket.available(ec);
+  if (ec) {
+    m_socket.close(ec);
+    return;
+  }
+  // On Linux, the  close()  function for sockets does not necessarily wait
+  // for the operating system to complete the operation before returning.
+  // It typically marks the socket as closed and releases any resources
+  // associated with it immediately. However, this does not guarantee that all
+  // pending data has been sent or received. It is important to handle any
+  // necessary error checking and ensure all data transmission is complete
+  // before calling  close()  on a socket.
+  if (unread > 0) {
+    do {
+      boost::container::small_vector<uint8_t, 4096> buf;
+      buf.resize(unread);
+      m_socket.read_some(boost::asio::mutable_buffer(buf.data(), unread), ec);
+      LOG_GENERAL(INFO, "Draining remaining IO before close"  << m_remotePeer.GetPrintableIPAddress());
+    } while (!ec && (unread = m_socket.available(ec)) > 0);
+  }
   m_socket.close(ec);
+  if (ec) {
+    LOG_GENERAL(INFO, "Informational, not an issue - Error closing socket: " << ec.message());
+  }
 }
 
 void P2PServerConnection::OnConnectionClosed() {

src/libNetwork/SendJobs.cpp

@@ -215,6 +215,8 @@ class GracefulCloseImpl
             LOG_GENERAL(DEBUG,
                         "Expected EOF, got ec=" << ec.message() << " n=" << n);
           }
+          ErrorCode ignored;
+          self->m_socket.close(ignored);
         });
   }
 };
@@ -226,10 +228,12 @@ void CloseGracefully(Socket&& socket) {
   }
   socket.shutdown(boost::asio::socket_base::shutdown_both, ec);
   if (ec) {
+    socket.close(ec);
     return;
   }
   size_t unread = socket.available(ec);
   if (ec) {
+    socket.close(ec);
     return;
   }
   if (unread > 0) {
@@ -239,6 +243,8 @@ void CloseGracefully(Socket&& socket) {
   }
   if (!ec) {
     std::make_shared<GracefulCloseImpl>(std::move(socket))->Close();
+  } else {
+    socket.close(ec);
   }
 }
 
@@ -378,13 +384,21 @@ class PeerSendQueue : public std::enable_shared_from_this<PeerSendQueue> {
     boost::asio::async_write(
         m_socket, boost::asio::const_buffer(msg.data.get(), msg.size),
         [self = shared_from_this()](const ErrorCode& ec, size_t) {
+          if (ec) {
+            // LOG_GENERAL(WARNING, "Got error code: " << ec.message());
+          }
           if (ec != OPERATION_ABORTED) {
             self->OnWritten(ec);
           }
         });
   }
 
   void OnWritten(const ErrorCode& ec) {
+    if (!ec && !m_closed) {
+      // LOG_GENERAL(INFO, "Successfully sent message to: "
+      //                       << m_peer.GetPrintableIPAddress()
+      //                       << " with size: " << m_queue.front().msg.size);
+    }
     if (m_closed) {
       return;
     }
@@ -415,11 +429,12 @@ class PeerSendQueue : public std::enable_shared_from_this<PeerSendQueue> {
       return;
     }
 
-    WaitTimer(m_timer, Milliseconds{RECONNECT_INTERVAL_IN_MS}, [this]() { Reconnect(); });
+    WaitTimer(m_timer, Milliseconds{RECONNECT_INTERVAL_IN_MS},
+              [this]() { Reconnect(); });
   }
 
   void Reconnect() {
-    LOG_GENERAL(DEBUG, "Peer " << m_peer << " reconnects");
+    LOG_GENERAL(INFO, "Peer " << m_peer << " reconnects");
     CloseGracefully(std::move(m_socket));
     m_socket = Socket(m_asioContext);
     Connect();
@@ -493,7 +508,8 @@ class SendJobsImpl : public SendJobs,
       return;
     }
 
-    LOG_GENERAL(DEBUG, "Enqueueing message, size=" << message.size);
+    LOG_GENERAL(DEBUG, "Enqueueing message, size=" << message.size
+                                                   << " peer = " << peer);
 
     // this fn enqueues the lambda to be executed on WorkerThread with
     // sequential guarantees for messages from every calling thread

src/libScilla/UnixDomainSocketServer.cpp

@@ -96,10 +96,12 @@ void UnixDomainSocketServer::WorkerThread() {
       boost::system::error_code ec;
       m_acceptor->accept(socket, ec);
       if (ec && m_started) {
+        socket.close(ec);
         throw std::runtime_error(ec.message());
       }
 
       if (!m_started) {
+        socket.close(ec);
         break;
       }
 
@@ -115,6 +117,7 @@ void UnixDomainSocketServer::WorkerThread() {
       }
 
       if (!m_started) {
+        socket.close(ec);
         break;
       }
 
@@ -130,6 +133,7 @@ void UnixDomainSocketServer::WorkerThread() {
       }
 
       if (!m_started) {
+        socket.close(ec);
         break;
       }
 
@@ -147,13 +151,15 @@ void UnixDomainSocketServer::WorkerThread() {
       if (ec) {
         LOG_GENERAL(WARNING,
                     "Write to " << m_path << " failed: " << ec.message());
+        socket.close(ec);
         continue;
       }
 
       socket.shutdown(
           boost::asio::local::stream_protocol::socket::shutdown_both, ec);
       if (ec) {
         LOG_GENERAL(WARNING, "Shutdown failed: " << ec.message());
+        socket.close(ec);
         continue;
       }
     }

src/libServer/APIServerImpl.cpp

@@ -93,6 +93,8 @@ class APIServerImpl::Connection
     if (!owner) {
       // server is closed
       m_stream.socket().shutdown(tcp::socket::shutdown_both, ec);
+      m_stream.socket().close(ec);
+      OnClosed();
       return;
     }
 
@@ -101,6 +103,8 @@ class APIServerImpl::Connection
         LOG_GENERAL(DEBUG, "Read error: " << ec.message());
         m_stream.socket().shutdown(tcp::socket::shutdown_both, ec);
       }
+      m_stream.socket().shutdown(tcp::socket::shutdown_both, ec);
+      m_stream.socket().close(ec);
       OnClosed();
       return;
     }
@@ -205,6 +209,8 @@ class APIServerImpl::Connection
     if (sz == 0 && m_owner.expired()) {
       beast::error_code ec;
       m_stream.socket().shutdown(tcp::socket::shutdown_both, ec);
+      m_stream.socket().close(ec);
+      OnClosed();
       return;
     }
 
@@ -346,7 +352,7 @@ bool APIServerImpl::DoListen() {
 
 #define CHECK_EC()                                                   \
   if (ec) {                                                          \
-    LOG_GENERAL(FATAL, "Cannot start API server: " << ec.message()); \
+    LOG_GENERAL(FATAL, "Cannot start API server: " << ec.message() << " port " << m_options.port); \
     return false;                                                    \
   }
 

src/libServer/LocalAPIServer.cpp

@@ -46,7 +46,7 @@ bool LocalAPIServer::StartListening() {
 
 #define CHECK_EC()                                                     \
   if (ec) {                                                            \
-    LOG_GENERAL(WARNING, "Cannot start API server: " << ec.message()); \
+    LOG_GENERAL(WARNING, "Cannot start API server: " << ec.message() << " port =" << m_port); \
     return false;                                                      \
   }
 
@@ -117,10 +117,12 @@ void LocalAPIServer::WorkerThread() {
       boost::system::error_code ec;
       m_acceptor->accept(socket, ec);
       if (ec && m_started) {
+        socket.close(ec);
         throw std::runtime_error(ec.message());
       }
 
       if (!m_started) {
+        socket.close(ec);
         break;
       }
 
@@ -131,10 +133,12 @@ void LocalAPIServer::WorkerThread() {
           DEFAULT_DELIMITER_CHAR, ec);
       if (ec || n <= 1) {
         LOG_GENERAL(WARNING, "Read (" << m_ip << ") failed: " << ec.message());
+        socket.close(ec);
         continue;
       }
 
       if (!m_started) {
+        socket.close(ec);
         break;
       }
 
@@ -150,6 +154,7 @@ void LocalAPIServer::WorkerThread() {
       }
 
       if (!m_started) {
+        socket.close(ec);
         break;
       }
 
@@ -166,15 +171,16 @@ void LocalAPIServer::WorkerThread() {
       boost::asio::write(socket, boost::asio::buffer(response), ec);
       if (ec) {
         LOG_GENERAL(WARNING, "Write (" << m_ip << ") failed: " << ec.message());
+        socket.close(ec);
         continue;
       }
 
       socket.shutdown(
           boost::asio::local::stream_protocol::socket::shutdown_both, ec);
       if (ec) {
         LOG_GENERAL(WARNING, "Shutdown failed: " << ec.message());
-        continue;
       }
+      socket.close(ec);
     }
   } catch (const std::exception& e) {
     LOG_GENERAL(WARNING, "Listening to " << m_ip << " failed: " << e.what());

tests/CMakeLists.txt

@@ -21,11 +21,15 @@ add_subdirectory (Server)
 add_subdirectory (EvmFiltersAPI)
 add_subdirectory (EthRpcMethods)
 add_subdirectory (Utils)
-#add_subdirectory (Zilliqa)
+add_subdirectory (Zilliqa)
+add_subdirectory (native)
 #add_subdirectory (RemoteStorageDB) Works only if you have a local mongo server running
 
 file(COPY ${CMAKE_SOURCE_DIR}/constants_local.xml DESTINATION ${CMAKE_BINARY_DIR})
-#FIXME: constants.xml is not used for local_run, thus shouldn't be copied
 # presently, it's a workaround to silence the error thrown by tests_zilliqa_local.py
 file(COPY ${CMAKE_SOURCE_DIR}/constants.xml DESTINATION ${CMAKE_BINARY_DIR})
 file(COPY Node DESTINATION ${CMAKE_CURRENT_BINARY_DIR})
+file(COPY ${CMAKE_SOURCE_DIR}/tests/Zilliqa/test_zilliqa_local.py DESTINATION ${CMAKE_SOURCE_DIR}/build/tests/zilliqa)
+file(COPY ${CMAKE_SOURCE_DIR}/tests/Zilliqa/test_zilliqa_lookup.py DESTINATION ${CMAKE_SOURCE_DIR}/build/tests/zilliqa)
+file(COPY ${CMAKE_SOURCE_DIR}/tests/Zilliqa/test_zilliqa_seedpub.py DESTINATION ${CMAKE_SOURCE_DIR}/build/tests/zilliqa)
+file(COPY ${CMAKE_SOURCE_DIR}/tests/Zilliqa/test_zilliqa_late.py DESTINATION ${CMAKE_SOURCE_DIR}/build/tests/zilliqa)