From 0927bd88f040a3868042c521b00f0d973f17f7de Mon Sep 17 00:00:00 2001
From: Marcus O'Flaherty <marcus.o-flaherty@warwick.ac.uk>
Date: Thu, 14 Nov 2024 11:29:55 +0000
Subject: [PATCH 1/3] Add Ready function to Services(Backend) to check for
 connections to backend sockets

---
 src/ServiceDiscovery/Services.cpp        |  4 ++++
 src/ServiceDiscovery/Services.h          |  1 +
 src/ServiceDiscovery/ServicesBackend.cpp | 24 ++++++++++++++++++++++++
 src/ServiceDiscovery/ServicesBackend.h   |  1 +
 4 files changed, 30 insertions(+)

diff --git a/src/ServiceDiscovery/Services.cpp b/src/ServiceDiscovery/Services.cpp
index a4be2fc..7e8cca1 100644
--- a/src/ServiceDiscovery/Services.cpp
+++ b/src/ServiceDiscovery/Services.cpp
@@ -42,6 +42,10 @@ bool Services::Init(Store &m_variables, zmq::context_t* context_in, SlowControlC
   return true;
 }
 
+bool Services::Ready(const unsigned int timeout){
+  return m_backend_client.Ready(timeout);
+}
+
 // ===========================================================================
 // Write Functions
 // ---------------
diff --git a/src/ServiceDiscovery/Services.h b/src/ServiceDiscovery/Services.h
index 86c6b5f..ffef3d4 100644
--- a/src/ServiceDiscovery/Services.h
+++ b/src/ServiceDiscovery/Services.h
@@ -38,6 +38,7 @@ namespace ToolFramework {
     Services();
     ~Services();
     bool Init(Store &m_variables, zmq::context_t* context_in, SlowControlCollection* sc_vars_in, bool new_service=false);
+    bool Ready(const unsigned int timeout=10000); // default service discovery broadcast period is 5s, middleman also checks intermittently, compound total time should be <10s...
     
     bool SQLQuery(const std::string& database, const std::string& query, std::vector<std::string>& responses, const unsigned int timeout=SERVICES_DEFAULT_TIMEOUT);
     bool SQLQuery(const std::string& database, const std::string& query, std::string& response, const unsigned int timeout=SERVICES_DEFAULT_TIMEOUT);
diff --git a/src/ServiceDiscovery/ServicesBackend.cpp b/src/ServiceDiscovery/ServicesBackend.cpp
index aff94e2..47e0e67 100644
--- a/src/ServiceDiscovery/ServicesBackend.cpp
+++ b/src/ServiceDiscovery/ServicesBackend.cpp
@@ -936,3 +936,27 @@ bool ServicesBackend::Receive(zmq::socket_t* sock, std::vector<zmq::message_t>&
 	return true;
 }
 
+bool ServicesBackend::Ready(int timeout){
+	
+	// poll the output sockets for listeners
+	int ret;
+	try {
+		ret = zmq::poll(out_polls.data(), out_polls.size(), timeout);
+	} catch (zmq::error_t& err){
+		std::cerr<<"ServicesBackend::Ready caught "<<err.what()<<std::endl;
+		return false;
+	}
+	if(ret<0){
+		// error polling - is the socket closed?
+		std::cerr<<"ServicesBackend::Ready error: "<<zmq_strerror(errno)<<std::endl;
+		return false;
+	} else if(ret==0){
+		// 'resource temoprarily unavailable' - no-one connected.
+	} else if((out_polls.at(0).revents & ZMQ_POLLOUT) &&   // pub socket (writes)
+	          (out_polls.at(1).revents & ZMQ_POLLOUT) ){   // dealer socket (reads)
+		return true;
+	}
+	
+	return false;
+	
+}
diff --git a/src/ServiceDiscovery/ServicesBackend.h b/src/ServiceDiscovery/ServicesBackend.h
index a776dc9..805a6c7 100644
--- a/src/ServiceDiscovery/ServicesBackend.h
+++ b/src/ServiceDiscovery/ServicesBackend.h
@@ -51,6 +51,7 @@ class ServicesBackend {
 	ServicesBackend(){};
 	~ServicesBackend(){};
 	void SetUp(zmq::context_t* in_context, std::function<void(std::string msg, int msg_verb, int verbosity)> log=nullptr); // possibly move to constructor
+	bool Ready(int timeout); // check if zmq sockets have connections
 	bool Initialise(std::string configfile);
 	bool Initialise(Store &varaibles_in);
 	bool Finalise();

From 0cfbe771adbcb809c2eec3b1333fea431de462cb Mon Sep 17 00:00:00 2001
From: root <root@Mon-01.cern.ch>
Date: Thu, 28 Nov 2024 12:47:21 +0000
Subject: [PATCH 2/3] don't poll PUB socket as it always returns true
 immediately, making the timeout ineffective

---
 src/ServiceDiscovery/ServicesBackend.cpp | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/src/ServiceDiscovery/ServicesBackend.cpp b/src/ServiceDiscovery/ServicesBackend.cpp
index 47e0e67..d560f12 100644
--- a/src/ServiceDiscovery/ServicesBackend.cpp
+++ b/src/ServiceDiscovery/ServicesBackend.cpp
@@ -939,9 +939,11 @@ bool ServicesBackend::Receive(zmq::socket_t* sock, std::vector<zmq::message_t>&
 bool ServicesBackend::Ready(int timeout){
 	
 	// poll the output sockets for listeners
+	// only poll dealer socket, pub sockets always return true immediately so ignore the timeout
+	// polling the input socket checks for a message, so don't do that.
 	int ret;
 	try {
-		ret = zmq::poll(out_polls.data(), out_polls.size(), timeout);
+		ret = zmq::poll(&out_polls.at(1), 1, timeout);
 	} catch (zmq::error_t& err){
 		std::cerr<<"ServicesBackend::Ready caught "<<err.what()<<std::endl;
 		return false;
@@ -952,8 +954,7 @@ bool ServicesBackend::Ready(int timeout){
 		return false;
 	} else if(ret==0){
 		// 'resource temoprarily unavailable' - no-one connected.
-	} else if((out_polls.at(0).revents & ZMQ_POLLOUT) &&   // pub socket (writes)
-	          (out_polls.at(1).revents & ZMQ_POLLOUT) ){   // dealer socket (reads)
+	} else if(out_polls.at(1).revents & ZMQ_POLLOUT){
 		return true;
 	}
 	

From 2a035bc7f6ece9fb41068a04cc9388caf00610e6 Mon Sep 17 00:00:00 2001
From: root <root@Mon-01.cern.ch>
Date: Thu, 28 Nov 2024 18:45:37 +0000
Subject: [PATCH 3/3] need mutex lock as socket is not thread safe

---
 src/ServiceDiscovery/ServicesBackend.cpp | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/ServiceDiscovery/ServicesBackend.cpp b/src/ServiceDiscovery/ServicesBackend.cpp
index d560f12..3fbe98f 100644
--- a/src/ServiceDiscovery/ServicesBackend.cpp
+++ b/src/ServiceDiscovery/ServicesBackend.cpp
@@ -943,7 +943,9 @@ bool ServicesBackend::Ready(int timeout){
 	// polling the input socket checks for a message, so don't do that.
 	int ret;
 	try {
+		dlr_socket_mutex.lock();
 		ret = zmq::poll(&out_polls.at(1), 1, timeout);
+		dlr_socket_mutex.unlock();
 	} catch (zmq::error_t& err){
 		std::cerr<<"ServicesBackend::Ready caught "<<err.what()<<std::endl;
 		return false;