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/7] 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/7] 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/7] 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;

From 4da4ac7a2f392c2b3d3c2bb4fc5adec7fd704879 Mon Sep 17 00:00:00 2001
From: Marcus O'Flaherty <moflaherty1@sheffield.ac.uk>
Date: Thu, 14 Nov 2024 19:47:32 +0000
Subject: [PATCH 4/7] replace sleep on construction with poll on services
 backend socket

---
 src/ServiceDiscovery/Services.cpp | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/src/ServiceDiscovery/Services.cpp b/src/ServiceDiscovery/Services.cpp
index 7e8cca1..e952e78 100644
--- a/src/ServiceDiscovery/Services.cpp
+++ b/src/ServiceDiscovery/Services.cpp
@@ -27,17 +27,16 @@ bool Services::Init(Store &m_variables, zmq::context_t* context_in, SlowControlC
   if(!m_variables.Get("service_name",m_name)) m_name="test_service";
   if(!m_variables.Get("db_name",m_dbname)) m_dbname="daq";
   
-  
-    
   if(!m_backend_client.Initialise(m_variables)){
     std::clog<<"error initialising slowcontrol client"<<std::endl;
     return false;
   }
    
-  // TODO FIXME with better mechanism
-  // after Initilising the slow control client needs ~15 seconds for the middleman to connect
-  std::this_thread::sleep_for(std::chrono::seconds(15));
-  // hopefully the middleman has found us by now
+  // wait for the middleman to connect. ServiceDiscovery broadcasts are sent out intermittently,
+  // so we need to wait for the middleman to receive one & connect before we can communicate with it.
+  int pub_period=5;
+  m_variables.Get("service_publish_sec",pub_period);
+  Ready(pub_period*1000);
   
   return true;
 }

From a4db12f7a51c42773a3ee04987b181cef4a860b2 Mon Sep 17 00:00:00 2001
From: root <root@Mon-01.cern.ch>
Date: Fri, 15 Nov 2024 01:01:40 +0000
Subject: [PATCH 5/7] simplify ServicesBackend by not running call to DoCommand
 in a separate thread

---
 src/ServiceDiscovery/ServicesBackend.cpp | 114 ++++++++++-------------
 src/ServiceDiscovery/ServicesBackend.h   |   2 +-
 2 files changed, 51 insertions(+), 65 deletions(-)

diff --git a/src/ServiceDiscovery/ServicesBackend.cpp b/src/ServiceDiscovery/ServicesBackend.cpp
index 3fbe98f..16e48f6 100644
--- a/src/ServiceDiscovery/ServicesBackend.cpp
+++ b/src/ServiceDiscovery/ServicesBackend.cpp
@@ -384,15 +384,6 @@ bool ServicesBackend::SendMulticast(std::string command, std::string* err){
 	return true;
 }
 
-//////////////////////////////////////////////////////////////////////////////////////////////////////////
-// FIXME ok, i think the combination of SendCommand and DoCommand are an over-complication.             //
-// SendCommand is basically a pointless wrapper that spawns a thread to do what it could do itself.     //
-// This would remove the surplus promise/future layer. The only change to DoCommand required            //
-// would be that the timeouts from each stage (wait for send, wait for reply)                           //
-// would need to be appropriately modified -.e.g if total timeout is 300ms and it takes 100ms to send,  //
-// then we would only wait for 200ms for the reply.                                                     //
-//////////////////////////////////////////////////////////////////////////////////////////////////////////
-
 bool ServicesBackend::SendCommand(const std::string& topic, const std::string& command, std::vector<std::string>* results, const unsigned int* timeout_ms, std::string* err){
 	// send a command and receive response.
 	// This is a wrapper that ensures we always return within the requested timeout.
@@ -428,50 +419,18 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 	// (for now) go over a dealer/router combination that cannot filter on the topic.
 	Command cmd{command, type, topic.substr(2,std::string::npos)};
 	
-	// submit the command asynchrously.
-	// This way we have control over how long we wait for the response
-	// The response will be a Command object with remaining members populated.
-	//std::future<Command> response = std::async(std::launch::async, &ServicesBackend::DoCommand, this, cmd);
-	// std::async returns a std::future that will block on destruction until the promise returns.
-	// if we don't want that to happen, i.e. we want to abandon it if it times out,
-	// we instead need to obtain a future from a promise (which is somehow not blocking?),
-	// and run our code in a detached thread, using the promise to pass back the result.
-	// tbh i don't quite get why this is different but there we go.
-	// see https://stackoverflow.com/a/23454840/3544936 and https://stackoverflow.com/a/23460094/3544936
-	std::promise<Command> returnval;
-	std::future<Command> response = returnval.get_future();
-	if(verbosity>10) std::cout<<"ServicesBackend::SendCommand spinning up new thread"<<std::endl;
-	std::thread{&ServicesBackend::DoCommand, this, cmd, std::move(returnval)}.detach();
-	
-	// the return from a std::async call is a 'future' object
-	// this object will be populated with the return value when it becomes available,
-	// but we can wait for a given timeout and then bail if it hasn't resolved in time.
-	
 	int timeout=command_timeout;            // default timeout for submission of command and receipt of response
 	if(timeout_ms) timeout=*timeout_ms;     // override by user if a custom timeout is given
-	std::chrono::milliseconds span(timeout);
-	// wrap our attempt to get the future in a try-catch, in case of exception
+	
+	// wrap our attempt to get the response in try/catch, just in case?
 	try {
-		// wait_for will return either when the result is ready, or when it times out
-		if(verbosity>10) std::cout<<"ServicesBackend::SendCommand waiting for response"<<std::endl;
-		if(response.wait_for(span)!=std::future_status::timeout){
-			// we got a response in time. retrieve and parse return value
-			if(verbosity>10) std::cout<<"ServicesBackend::SendCommand fetching response"<<std::endl;
-			cmd = response.get();
-			if(verbosity>10) std::cout<<"ServicesBackend::SendCommand response is "<<cmd.response.size()
-			                          <<" parts"<<std::endl;
-			if(results) *results = cmd.response;
-			if(err) *err = cmd.err;
-			return cmd.success;
-		} else {
-			// timed out
-			std::string errmsg="Timed out after waiting "+std::to_string(timeout)+"ms for response "
-			                   "from command '"+command+"'";
-			if(verbosity>3) std::cerr<<errmsg<<std::endl;
-			if(err) *err=errmsg;
-			//std::cout<<"SendCommand returning false"<<std::endl;
-			return false;
-		}
+		if(verbosity>10) std::cout<<"ServicesBackend::SendCommand calling DoCommand"<<std::endl;
+		DoCommand(cmd, timeout);
+		if(verbosity>10) std::cout<<"ServicesBackend::SendCommand response is "<<cmd.response.size()
+		                          <<" parts"<<std::endl;
+		if(results) *results = cmd.response;
+		if(err) *err = cmd.err;
+		return cmd.success;
 	} catch(std::exception& e){
 		// one thing that can cause an exception is if we terminate the application
 		// before the promise is fulfilled (i.e. the response came, or zmq timed out)
@@ -495,7 +454,7 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 	return ret;
 }
 
-bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
+bool ServicesBackend::DoCommand(Command& cmd, int timeout_ms){
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand received command"<<std::endl;
 	// submit a command, wait for the response and return it
 	
@@ -503,6 +462,18 @@ bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
 	uint32_t thismsgid = ++msg_id;
 	cmd.msg_id = thismsgid;
 	
+	// we submit the command asynchrously.
+	// This way we control how long we wait for the response, ensuring we don't block the caller indefinitely.
+	// The response will be a Command object with remaining members populated.
+	// initial guess would have been to use:
+	//std::future<Command> response = std::async(std::launch::async, &ServicesBackend::DoCommand, this, cmd);
+	// but std::async returns a std::future that will block on destruction until the function returns.
+	// if we don't want that to happen, i.e. we want to abandon it if it times out,
+	// we instead need to obtain a future from a promise (which is somehow not blocking?),
+	// and run our code in a detached thread, using the promise to pass back the result.
+	// tbh i don't quite get why this is different but there we go.
+	// see https://stackoverflow.com/a/23454840/3544936 and https://stackoverflow.com/a/23460094/3544936
+	
 	// zmq sockets aren't thread-safe, so we have one central sender.
 	// we submit our command and keep a ticket to retrieve the return status on completion.
 	// similarly, the next response received may not be for us, so a central dealer receives
@@ -527,9 +498,27 @@ bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
 	waiting_senders.emplace(cmd, std::move(send_ticket));
 	send_queue_mutex.unlock();
 	
-	// wait for our number to come up. loooong timeout, but don't hang forever.
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand waiting for send confirmation"<<std::endl;
-	if(send_receipt.wait_for(std::chrono::seconds(30))==std::future_status::timeout){
+	// make a note of the time
+	auto send_start = std::chrono::high_resolution_clock::now();
+	// wait for confirmation of sending
+	bool timedout=false;
+	if(send_receipt.wait_for(std::chrono::milliseconds(timeout_ms))==std::future_status::timeout){
+		// wait_for timed out.
+		timedout=true;
+	} else {
+		// got the response before timeout!
+		// See how long that took to send, and subtract that from our total time allotment
+		// to see how long we have to wait for the response to arrive.
+		auto send_end = std::chrono::high_resolution_clock::now();
+		int send_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(send_end - send_start).count();
+		timeout_ms -= send_time_ms;
+		// juuuust in case, ensure our remaining time is not negative. :)
+		if(timeout_ms<0) timedout=true;
+	}
+	
+	// did we get a response in time?
+	if(timedout){
 		if(verbosity>10) std::cerr<<"ServicesBackend::DoCommand timeout"<<std::endl;
 		// sending timed out
 		if(cmd.type=='w') ++write_commands_failed;
@@ -537,20 +526,19 @@ bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
 		Log("Timed out sending command "+std::to_string(thismsgid),v_warning,verbosity);
 		cmd.success = false;
 		cmd.err = "Timed out sending command";
-		result.set_value(cmd);
 		
-		// since we are giving up waiting for the response, remove ourselves from
-		// the list of waiting recipients
+		// since we are giving up waiting for the response, remove ourselves
+		// from the list of recipients awaiting response
 		if(verbosity>10) std::cout<<"ServicesBackend::DoCommand de-registering for response on id "<<thismsgid<<std::endl;
 		send_queue_mutex.lock();
 		waiting_recipients.erase(thismsgid);
 		send_queue_mutex.unlock();
 		
-		return true;
-	} // else got a return value
+		return false;
+	} // else got send confirmation in time
 	
 	// so we got response about our send request, but did it go through?
-	// check for errors sending
+	// check for errors in sending
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand got send confirmation"<<std::endl;
 	int ret = send_receipt.get();
 	std::string errmsg;
@@ -565,7 +553,6 @@ bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
 		Log(errmsg,v_debug,verbosity);
 		cmd.success = false;
 		cmd.err = errmsg;
-		result.set_value(cmd);
 		
 		// since the send failed we don't expect a response, so remove ourselves
 		// from the list of recipients awaiting response
@@ -581,7 +568,7 @@ bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
 	
 	// if we succeeded in sending the message, we now need to wait for a repsonse.
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand waiting for response"<<std::endl;
-	if(response_reciept.wait_for(std::chrono::seconds(30))==std::future_status::timeout){
+	if(response_reciept.wait_for(std::chrono::milliseconds(timeout_ms))==std::future_status::timeout){
 		if(verbosity>10) std::cout<<"ServicesBackend::DoCommand response timeout"<<std::endl;
 		// timed out
 		if(cmd.type=='w') ++write_commands_failed;
@@ -589,19 +576,17 @@ bool ServicesBackend::DoCommand(Command cmd, std::promise<Command> result){
 		Log("Timed out waiting for response for command "+std::to_string(thismsgid),v_warning,verbosity);
 		cmd.success = false;
 		cmd.err = "Timed out waiting for response";
-		result.set_value(cmd);
 		return false;
 	} else {
 		// got a response!
 		if(verbosity>10) std::cout<<"ServicesBackend::DoCommand got a response for command "<<cmd.msg_id
 		                          <<", passing back to caller"<<std::endl;
 		try{
-			result.set_value(response_reciept.get());
+			cmd = response_reciept.get();
 		} catch(std::exception& e){
 			Log("ServicesBackend response for command "+std::to_string(cmd.msg_id)+" was exception "+e.what(),v_error,verbosity);
 			cmd.err=e.what();
 			cmd.success=false;
-			result.set_value(cmd);
 			return false;
 		}
 		return true;
@@ -622,6 +607,7 @@ bool ServicesBackend::GetNextResponse(){
 	
 	// check return status
 	if(ret==-2) return true;      // no messages waiting to be received
+	printf("GNR poll returned %d\n",ret);
 	
 	if(verbosity>10) std::cout<<"ServicesBackend::GetNextResponse had response in socket"<<std::endl;
 	if(ret==-3){
diff --git a/src/ServiceDiscovery/ServicesBackend.h b/src/ServiceDiscovery/ServicesBackend.h
index 805a6c7..6bdd0fa 100644
--- a/src/ServiceDiscovery/ServicesBackend.h
+++ b/src/ServiceDiscovery/ServicesBackend.h
@@ -102,7 +102,7 @@ class ServicesBackend {
 	bool InitMulticast(); // private
 	bool RegisterServices(); //private
 	// wrapper funtion; add command to outgoing queue, receive response. ~30s timeout.
-	bool DoCommand(Command cmd, std::promise<Command>); //private
+	bool DoCommand(Command& cmd, int timeout_ms); //private
 	// actual send/receive functions
 	bool SendNextCommand(); //private
 	bool GetNextResponse(); //priavte

From 5ef89828b9c5790b8f46cda0d2290bfc61129b83 Mon Sep 17 00:00:00 2001
From: Marcus O'Flaherty <marcus.o-flaherty@warwick.ac.uk>
Date: Tue, 19 Nov 2024 16:38:46 +0000
Subject: [PATCH 6/7] Propagate timeout of ServicesBackend::SendCommand to
 underlying PollAndSend so that poll time is timeout (rather than class
 member). Could be problematic as it means a user calling SendCommand with a
 long timeout could cause backend to get stuck, but only if the poll fails (in
 which case it can't do anything anyway)...

---
 src/ServiceDiscovery/Services.cpp        |  4 ++-
 src/ServiceDiscovery/Services.h          |  1 +
 src/ServiceDiscovery/ServicesBackend.cpp | 31 ++++++++++++++++--------
 src/ServiceDiscovery/ServicesBackend.h   | 13 +++++++++-
 4 files changed, 37 insertions(+), 12 deletions(-)

diff --git a/src/ServiceDiscovery/Services.cpp b/src/ServiceDiscovery/Services.cpp
index e952e78..05cc7e1 100644
--- a/src/ServiceDiscovery/Services.cpp
+++ b/src/ServiceDiscovery/Services.cpp
@@ -265,7 +265,8 @@ bool Services::GetDeviceConfig(std::string& json_data, const int version, const
 
 // get a run configuration via configuration ID
 bool Services::GetRunConfig(std::string& json_data, const int config_id, const unsigned int timeout){
-
+  
+  printf("GetRunConfig with timeout %d\n",timeout);
   json_data="";
   
   std::string cmd_string = "{ \"config_id\":"+std::to_string(config_id) + "}";
@@ -343,6 +344,7 @@ bool Services::GetRunConfig(std::string& json_data, const std::string& name, con
 
 bool Services::GetRunDeviceConfig(std::string& json_data, const int runconfig_id, const std::string& device, int* version, unsigned int timeout){
   
+  printf("GetRunDeviceConfig with timeout %d\n",timeout);
   json_data="";
   
   const std::string& name = (device=="") ? m_name : device;
diff --git a/src/ServiceDiscovery/Services.h b/src/ServiceDiscovery/Services.h
index ffef3d4..c764dbb 100644
--- a/src/ServiceDiscovery/Services.h
+++ b/src/ServiceDiscovery/Services.h
@@ -30,6 +30,7 @@ namespace ToolFramework {
     Store info;
   };
   
+  
   class Services{
     
     
diff --git a/src/ServiceDiscovery/ServicesBackend.cpp b/src/ServiceDiscovery/ServicesBackend.cpp
index 16e48f6..a0e6b9e 100644
--- a/src/ServiceDiscovery/ServicesBackend.cpp
+++ b/src/ServiceDiscovery/ServicesBackend.cpp
@@ -2,7 +2,7 @@
 
 using namespace ToolFramework;
 
-Command::Command(std::string command_in, char type_in, std::string topic_in){
+Command::Command(std::string command_in, char type_in, std::string topic_in, const unsigned int timeout_ms_in){
 	command = command_in;
 	type = type_in;
 	topic=topic_in;
@@ -10,6 +10,7 @@ Command::Command(std::string command_in, char type_in, std::string topic_in){
 	response=std::vector<std::string>{};
 	err="";
 	msg_id=0;
+	timeout_ms=timeout_ms_in;
 }
 
 Command::Command(){
@@ -20,12 +21,14 @@ Command::Command(){
 	response=std::vector<std::string>{};
 	err="";
 	msg_id=0;
+	timeout_ms=0;
 }
 
 void Command::Print() const {
 	std::cout<<"command="<<command<<std::endl;
 	std::cout<<"type="<<type<<std::endl;
 	std::cout<<"topic="<<topic<<std::endl;
+	std::cout<<"timeout="<<timeout_ms<<std::endl;
 	std::cout<<"success="<<success<<std::endl;
 	std::cout<<"err="<<err<<std::endl;
 	std::cout<<"id="<<msg_id<<std::endl;
@@ -43,6 +46,7 @@ Command::Command(const Command& cmd_in){
 	response = cmd_in.response;
 	err = cmd_in.err;
 	msg_id = cmd_in.msg_id;
+	timeout_ms = cmd_in.timeout_ms;
 }
 
 Command::Command(Command&& cmd_in){
@@ -53,12 +57,14 @@ Command::Command(Command&& cmd_in){
 	response = std::move(cmd_in.response);
 	err = cmd_in.err;
 	msg_id = cmd_in.msg_id;
+	timeout_ms = cmd_in.timeout_ms;
 	cmd_in.command="";
 	cmd_in.type='\0';
 	cmd_in.topic="";
 	cmd_in.success=0;
 	cmd_in.msg_id=0;
 	cmd_in.response=std::vector<std::string>{};
+	cmd_in.timeout_ms=0;
 }
 
 Command& Command::operator=(Command&& cmd_in){
@@ -70,12 +76,14 @@ Command& Command::operator=(Command&& cmd_in){
 		response = std::move(cmd_in.response);
 		err = cmd_in.err;
 		msg_id = cmd_in.msg_id;
+		timeout_ms = cmd_in.timeout_ms;
 		cmd_in.command="";
 		cmd_in.type='\0';
 		cmd_in.topic="";
 		cmd_in.success=0;
 		cmd_in.msg_id=0;
 		cmd_in.response=std::vector<std::string>{};
+		cmd_in.timeout_ms=0;
 	}
 	return *this;
 }
@@ -389,6 +397,9 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 	// This is a wrapper that ensures we always return within the requested timeout.
 	if(verbosity>10) std::cout<<"ServicesBackend::SendCommand invoked with command '"<<command<<"'"<<std::endl;
 	
+	int timeout=command_timeout;            // default timeout for submission of command and receipt of response
+	if(timeout_ms) timeout=*timeout_ms;     // override by user if a custom timeout is given
+	
 	// encapsulate the command in an object.
 	// We need this since we can only get one return value from an asynchronous function call,
 	// and we want both a response string and error flag.
@@ -417,10 +428,9 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 	// can use with ZMQ_SUBSCRIBE to filter out particular messages.
 	// In fact it's useful to indicate a topic in all cases, even when the actual message will
 	// (for now) go over a dealer/router combination that cannot filter on the topic.
-	Command cmd{command, type, topic.substr(2,std::string::npos)};
-	
-	int timeout=command_timeout;            // default timeout for submission of command and receipt of response
-	if(timeout_ms) timeout=*timeout_ms;     // override by user if a custom timeout is given
+	// forward the timeout to the Command (and thus zmq::poll in PollAndSend...) ... is this sensible? HMMMMM FIXME
+	Command cmd{command, type, topic.substr(2,std::string::npos),timeout};
+	printf("formed command with timeout %d ms from timeout at %p of value %d\n",cmd.timeout_ms,timeout_ms,(timeout_ms ? *timeout_ms : 0));
 	
 	// wrap our attempt to get the response in try/catch, just in case?
 	try {
@@ -455,6 +465,7 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 }
 
 bool ServicesBackend::DoCommand(Command& cmd, int timeout_ms){
+	printf("DoCommand with timeout %d ms\n",timeout_ms);
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand received command"<<std::endl;
 	// submit a command, wait for the response and return it
 	
@@ -542,6 +553,7 @@ bool ServicesBackend::DoCommand(Command& cmd, int timeout_ms){
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand got send confirmation"<<std::endl;
 	int ret = send_receipt.get();
 	std::string errmsg;
+	if(ret==-4) errmsg="No connection on out socket in PollAndSend!";
 	if(ret==-3) errmsg="Error polling out socket in PollAndSend! Is socket closed?";
 	if(ret==-2) errmsg="No listener on out socket in PollAndSend!";
 	if(ret==-1) errmsg="Error sending in PollAndSend!";
@@ -708,9 +720,6 @@ bool ServicesBackend::SendNextCommand(){
 		//Log(logmsg.str(),v_debug,verbosity);
 	}
 	
-	// write commands go to the pub socket, read commands to the dealer
-	zmq::socket_t* thesocket = (cmd.type=='w') ? clt_pub_socket : clt_dlr_socket;
-	
 	// send out the command
 	// commands should be formatted as 4 parts:
 	// 0. pub topic (used by ZMQ_SUBSCRIBE for pub, but send it always)
@@ -723,11 +732,13 @@ bool ServicesBackend::SendNextCommand(){
 	dlr_socket_mutex.lock();
 	if(verbosity>10) std::cout<<"ServicesBackend::SendNextCommand calling PollAndSend"
 	                          <<", message type: "<<cmd.type<<", topic '"<<cmd.topic<<"'"<<std::endl;
+	printf("calling PollAndSend with timeout %d ms\n",cmd.timeout_ms);
 	if(cmd.type=='w'){
-		ret = PollAndSend(thesocket, out_polls.at(1), outpoll_timeout, cmd.topic, clt_ID, cmd.msg_id, cmd.command);
+		// write commands go to the pub socket, read commands to the dealer
+		ret = PollAndSend(clt_pub_socket, out_polls.at(0), cmd.timeout_ms, cmd.topic, clt_ID, cmd.msg_id, cmd.command);
 	} else {
 		// clt_ID already added by dealer socket
-		ret = PollAndSend(thesocket, out_polls.at(1), outpoll_timeout, cmd.topic, cmd.msg_id, cmd.command);
+		ret = PollAndSend(clt_dlr_socket, out_polls.at(1), cmd.timeout_ms, cmd.topic, cmd.msg_id, cmd.command);
 	}
 	if(verbosity>10) std::cout<<"ServicesBackend::SendNextCommand send returned "<<ret<<", passing to recipient"<<std::endl;
 	dlr_socket_mutex.unlock();
diff --git a/src/ServiceDiscovery/ServicesBackend.h b/src/ServiceDiscovery/ServicesBackend.h
index 6bdd0fa..2e2c252 100644
--- a/src/ServiceDiscovery/ServicesBackend.h
+++ b/src/ServiceDiscovery/ServicesBackend.h
@@ -27,7 +27,7 @@
 namespace ToolFramework {
 
 struct Command {
-	Command(std::string command_in, char cmd_type_in, std::string topic_in);
+	Command(std::string command_in, char cmd_type_in, std::string topic_in, const unsigned int timeout_ms_in);
 	
 	Command(const Command& cmd_in);  // copy constructor
 	Command(Command&& cmd_in);       // move constructor
@@ -41,6 +41,7 @@ struct Command {
 	uint32_t success;            // middleman treats this as BOOL: 1=success, 0=fail
 	std::string err;
 	uint32_t msg_id;
+	uint32_t timeout_ms;
 };
 
 
@@ -198,15 +199,20 @@ class ServicesBackend {
 		int send_ok=0;
 		// check for listener
 		int ret=0;
+		printf("polling out socket for %d ms\n",timeout);
 		try {
 			ret = zmq::poll(&poll, 1, timeout);
 		} catch (zmq::error_t& err){
 			std::cerr<<"ServicesBackend::PollAndSend poller caught "<<err.what()<<std::endl;
 			ret = -1;
 		}
+		printf("ret %d\n",ret);
 		if(ret<0){
 			// error polling - is the socket closed?
 			send_ok = -3;
+		} else if(ret==0){
+			// 'resource temporarily unavailable'...? no connections on this socket?
+			send_ok = -4;
 		} else if(poll.revents & ZMQ_POLLOUT){
 			bool success = Send(sock, false, std::forward<T>(message));
 			send_ok = success ? 0 : -1;
@@ -226,15 +232,20 @@ class ServicesBackend {
 		int send_ok = 0;
 		// check for listener
 		int ret = 0;
+		printf("polling out socket for %d ms\n",timeout);
 		try {
 			ret = zmq::poll(&poll, 1, timeout);
 		} catch (zmq::error_t& err){
 			std::cerr<<"ServicesBackend::PollAndSend poller caught "<<err.what()<<std::endl;
 			ret = -1;
 		}
+		printf("ret %d\n",ret);
 		if(ret<0){
 			// error polling - is the socket closed?
 			send_ok = -3;
+		} else if(ret==0){
+			// 'resource temporarily unavailable'...? no connections on this socket?
+			send_ok = -4;
 		} else if(poll.revents & ZMQ_POLLOUT){
 			bool success = Send(sock, false, std::forward<T>(message), std::forward<Rest>(rest)...);
 			send_ok = success ? 0 : -1;

From 32702671560a76312738956fb8b1af62232b3657 Mon Sep 17 00:00:00 2001
From: root <root@Mon-01.cern.ch>
Date: Thu, 28 Nov 2024 13:03:45 +0000
Subject: [PATCH 7/7] remove printfs

---
 src/ServiceDiscovery/ServiceDiscovery.cpp | 2 +-
 src/ServiceDiscovery/Services.cpp         | 7 +------
 src/ServiceDiscovery/Services.h           | 1 -
 src/ServiceDiscovery/ServicesBackend.cpp  | 7 -------
 src/ServiceDiscovery/ServicesBackend.h    | 4 ----
 5 files changed, 2 insertions(+), 19 deletions(-)

diff --git a/src/ServiceDiscovery/ServiceDiscovery.cpp b/src/ServiceDiscovery/ServiceDiscovery.cpp
index 5ea9601..86cfe1d 100644
--- a/src/ServiceDiscovery/ServiceDiscovery.cpp
+++ b/src/ServiceDiscovery/ServiceDiscovery.cpp
@@ -446,7 +446,7 @@ void* ServiceDiscovery::MulticastListenThread(void* arg){
   mreq.imr_interface.s_addr = htonl(INADDR_ANY);         
   if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,&mreq, sizeof(mreq)) < 0) {
     perror("setsockopt mreq");
-    printf("Failed to goin multicast group listen thread");
+    printf("Failed to join multicast group listen thread");
     exit(1);
   }         
   
diff --git a/src/ServiceDiscovery/Services.cpp b/src/ServiceDiscovery/Services.cpp
index 05cc7e1..70e3eca 100644
--- a/src/ServiceDiscovery/Services.cpp
+++ b/src/ServiceDiscovery/Services.cpp
@@ -265,8 +265,7 @@ bool Services::GetDeviceConfig(std::string& json_data, const int version, const
 
 // get a run configuration via configuration ID
 bool Services::GetRunConfig(std::string& json_data, const int config_id, const unsigned int timeout){
-  
-  printf("GetRunConfig with timeout %d\n",timeout);
+
   json_data="";
   
   std::string cmd_string = "{ \"config_id\":"+std::to_string(config_id) + "}";
@@ -344,7 +343,6 @@ bool Services::GetRunConfig(std::string& json_data, const std::string& name, con
 
 bool Services::GetRunDeviceConfig(std::string& json_data, const int runconfig_id, const std::string& device, int* version, unsigned int timeout){
   
-  printf("GetRunDeviceConfig with timeout %d\n",timeout);
   json_data="";
   
   const std::string& name = (device=="") ? m_name : device;
@@ -361,14 +359,11 @@ bool Services::GetRunDeviceConfig(std::string& json_data, const int runconfig_id
     return false;
   }
   
-printf("GetRunConfig from GetRunDev returned: '%s'\n",run_config.c_str());
   // 2. extract the device's configuration id
   Store tmp;
   tmp.JsonParser(run_config);
   int device_config_id;
   get_ok = tmp.Get(name, device_config_id);
-printf("tmp store contains:\n");
-tmp.Print();
   
   if(!get_ok){
     std::string err= "GetRunDeviceConfig error getting device config; device '"+name
diff --git a/src/ServiceDiscovery/Services.h b/src/ServiceDiscovery/Services.h
index c764dbb..ffef3d4 100644
--- a/src/ServiceDiscovery/Services.h
+++ b/src/ServiceDiscovery/Services.h
@@ -30,7 +30,6 @@ namespace ToolFramework {
     Store info;
   };
   
-  
   class Services{
     
     
diff --git a/src/ServiceDiscovery/ServicesBackend.cpp b/src/ServiceDiscovery/ServicesBackend.cpp
index a0e6b9e..2b609a0 100644
--- a/src/ServiceDiscovery/ServicesBackend.cpp
+++ b/src/ServiceDiscovery/ServicesBackend.cpp
@@ -430,7 +430,6 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 	// (for now) go over a dealer/router combination that cannot filter on the topic.
 	// forward the timeout to the Command (and thus zmq::poll in PollAndSend...) ... is this sensible? HMMMMM FIXME
 	Command cmd{command, type, topic.substr(2,std::string::npos),timeout};
-	printf("formed command with timeout %d ms from timeout at %p of value %d\n",cmd.timeout_ms,timeout_ms,(timeout_ms ? *timeout_ms : 0));
 	
 	// wrap our attempt to get the response in try/catch, just in case?
 	try {
@@ -465,7 +464,6 @@ bool ServicesBackend::SendCommand(const std::string& topic, const std::string& c
 }
 
 bool ServicesBackend::DoCommand(Command& cmd, int timeout_ms){
-	printf("DoCommand with timeout %d ms\n",timeout_ms);
 	if(verbosity>10) std::cout<<"ServicesBackend::DoCommand received command"<<std::endl;
 	// submit a command, wait for the response and return it
 	
@@ -619,7 +617,6 @@ bool ServicesBackend::GetNextResponse(){
 	
 	// check return status
 	if(ret==-2) return true;      // no messages waiting to be received
-	printf("GNR poll returned %d\n",ret);
 	
 	if(verbosity>10) std::cout<<"ServicesBackend::GetNextResponse had response in socket"<<std::endl;
 	if(ret==-3){
@@ -732,7 +729,6 @@ bool ServicesBackend::SendNextCommand(){
 	dlr_socket_mutex.lock();
 	if(verbosity>10) std::cout<<"ServicesBackend::SendNextCommand calling PollAndSend"
 	                          <<", message type: "<<cmd.type<<", topic '"<<cmd.topic<<"'"<<std::endl;
-	printf("calling PollAndSend with timeout %d ms\n",cmd.timeout_ms);
 	if(cmd.type=='w'){
 		// write commands go to the pub socket, read commands to the dealer
 		ret = PollAndSend(clt_pub_socket, out_polls.at(0), cmd.timeout_ms, cmd.topic, clt_ID, cmd.msg_id, cmd.command);
@@ -816,7 +812,6 @@ bool ServicesBackend::Send(zmq::socket_t* sock, bool more, std::string messageda
 	if(verbosity>10) std::cout<<__PRETTY_FUNCTION__<<" called"<<std::endl;
 	// form the zmq::message_t
 	zmq::message_t message(messagedata.size());
-	//snprintf((char*)message.data(), messagedata.size()+1, "%s", messagedata.c_str());
 	memcpy(message.data(), messagedata.data(), message.size());
 	
 	// send it with given SNDMORE flag
@@ -844,7 +839,6 @@ bool ServicesBackend::Send(zmq::socket_t* sock, bool more, std::vector<std::stri
 		// form zmq::message_t
 		zmq::message_t message(messages.at(i).size());
 		memcpy(message.data(), messages.at(i).data(), messages.at(i).size());
-		//snprintf((char*)message.data(), messages.at(i).size()+1, "%s", messages.at(i).c_str());
 		
 		// send this part
 		bool send_ok = sock->send(message, ZMQ_SNDMORE);
@@ -859,7 +853,6 @@ bool ServicesBackend::Send(zmq::socket_t* sock, bool more, std::vector<std::stri
 	// form the zmq::message_t for the last part
 	zmq::message_t message(messages.back().size());
 	memcpy(message.data(), messages.back().data(), messages.back().size());
-	//snprintf((char*)message.data(), messages.back().size()+1, "%s", messages.back().c_str());
 	
 	// send it with, or without SNDMORE flag as requested
 	bool send_ok;
diff --git a/src/ServiceDiscovery/ServicesBackend.h b/src/ServiceDiscovery/ServicesBackend.h
index 2e2c252..116f548 100644
--- a/src/ServiceDiscovery/ServicesBackend.h
+++ b/src/ServiceDiscovery/ServicesBackend.h
@@ -199,14 +199,12 @@ class ServicesBackend {
 		int send_ok=0;
 		// check for listener
 		int ret=0;
-		printf("polling out socket for %d ms\n",timeout);
 		try {
 			ret = zmq::poll(&poll, 1, timeout);
 		} catch (zmq::error_t& err){
 			std::cerr<<"ServicesBackend::PollAndSend poller caught "<<err.what()<<std::endl;
 			ret = -1;
 		}
-		printf("ret %d\n",ret);
 		if(ret<0){
 			// error polling - is the socket closed?
 			send_ok = -3;
@@ -232,14 +230,12 @@ class ServicesBackend {
 		int send_ok = 0;
 		// check for listener
 		int ret = 0;
-		printf("polling out socket for %d ms\n",timeout);
 		try {
 			ret = zmq::poll(&poll, 1, timeout);
 		} catch (zmq::error_t& err){
 			std::cerr<<"ServicesBackend::PollAndSend poller caught "<<err.what()<<std::endl;
 			ret = -1;
 		}
-		printf("ret %d\n",ret);
 		if(ret<0){
 			// error polling - is the socket closed?
 			send_ok = -3;