Convert servers to cooperative multitasking

2 files changed, 281 insertions, 196 deletions

diff --git a/servers/auth_server_lin/src/auth_conn.cpp b/servers/auth_server_lin/src/auth_conn.cpp
index 28c8427..451b8d9 100644
--- a/servers/auth_server_lin/src/auth_conn.cpp
+++ b/servers/auth_server_lin/src/auth_conn.cpp
@@ -29,10 +29,6 @@
 #include "auth_conn.h"
 
 #define ABORT_SOCKET(s) 		s->close();						\
-					tqApp->processEvents();					\
-					while (s->state() == TQSocket::Closing) {		\
-						tqApp->processEvents();				\
-					}							\
 					s->disconnect();					\
 					delete s;						\
 					s = NULL;
@@ -49,7 +45,7 @@ struct exit_exception {
   instance of this class.
 */
 AuthSocket::AuthSocket(int sock, TQObject *parent, const char *name) :
-	TDEKerberosServerSocket(parent, name), m_criticalSection(0), m_stationID(-1), m_bound(false), m_config(static_cast<AuthServer*>(parent)->m_config), m_database(NULL), m_databaseStationsCursor(NULL),
+	TDEKerberosServerSocket(parent, name), m_criticalSection(0), m_stationID(-1), m_bound(false), m_servActive(false), m_servState(0), m_servClientSocket(NULL), m_servClientTimeout(NULL), m_config(static_cast<AuthServer*>(parent)->m_config), m_database(NULL), m_databaseStationsCursor(NULL),
 	m_databaseServicesCursor(NULL), m_databaseServiceTypesCursor(NULL), m_databasePermissionsCursor(NULL), m_databaseActivityCursor(NULL)
 {
 
@@ -80,6 +76,9 @@ AuthSocket::~AuthSocket() {
 	if (m_databaseActivityCursor) {
 		delete m_databaseActivityCursor;
 	}
+	if (m_servClientSocket) {
+		delete m_servClientSocket;
+	}
 }
 
 void AuthSocket::close() {
@@ -122,235 +121,312 @@ int AuthSocket::initiateKerberosHandshake() {
 	}
 }
 
-int AuthSocket::enterCommandLoop() {
-	m_criticalSection++;
-	try {
+void AuthSocket::servLoop() {
+	if (m_servActive) {
 		TQString command;
 		TQDataStream ds(this);
+
+		switch (m_servState) {
+			case 0:
+				if (!m_servClientTimeout) {
+					m_servClientTimeout = new TQTimer();
+					m_servClientTimeout->start(5000, TRUE);
+				}
+				if ((m_servClientSocket->state() == TQSocket::Connecting) || (m_servClientSocket->state() == TQSocket::HostLookup)) {
+					if (!m_servClientTimeout->isActive()) {
+						m_servClientSocket->close();
+						ds << TQString("ERRNOTAVL");
+						printf("[DEBUG] Connection failed to %s:%d for user %s@%s\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+						m_servActive = false;
+						delete m_servClientTimeout;
+						m_servClientTimeout = NULL;
+					}
+				}
+				else {
+					if (m_servClientTimeout) {
+						m_servClientTimeout->stop();
+						delete m_servClientTimeout;
+						m_servClientTimeout = NULL;
+					}
+					m_servState = 1;
+				}
+				break;
+			case 1:
+				if (m_servClientSocket->state() == TQSocket::Connected) {
+					if (m_servClientSocket->setUsingKerberos(true) != 0) {
+						m_servClientSocket->close();
+						ds << TQString("ERRNOTAVL");
+						printf("[DEBUG] Connection failed to %s:%d for user %s@%s due to Kerberos failure\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+						m_servActive = false;
+						delete m_servClientTimeout;
+						m_servClientTimeout = NULL;
+					}
+					m_servState = 2;
+				}
+				else {
+					m_servClientSocket->close();
+					ds << TQString("ERRNOTAVL");
+					printf("[DEBUG] Connection failed to %s:%d for user %s@%s\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+					m_servActive = false;
+					delete m_servClientTimeout;
+					m_servClientTimeout = NULL;
+				}
+				break;
+			case 2:
+				if (!m_servClientTimeout) {
+					m_servClientTimeout = new TQTimer();
+					m_servClientTimeout->start(5000, TRUE);
+				}
+				if (m_servClientSocket->state() == TQSocket::Connected) {
+					if (m_servClientSocket->canReadLine()) {
+						TQDataStream clientDS(m_servClientSocket);
+						TQString server_reply;
 	
-		while (state() == TQSocket::Connected) {
-			ds >> command;
-			if (command != "") {
-				printf("[DEBUG] Got command %s from user %s@%s\n\r", command.ascii(), m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
-				if (command == "LIST") {
-					// Send list of available servers...
-					m_slist.clear();
-	
-					// Get all stations from the database
-					m_databaseStationsCursor->select();
-					while (m_databaseStationsCursor->next()) {
-						bool authorized = false;
-						bool in_use = false;
-	
-						m_databasePermissionsCursor->select(TQString("station=%1").arg(m_databaseStationsCursor->value("pk").toInt()));
-						while (m_databasePermissionsCursor->next()) {
-							if (m_databasePermissionsCursor->value("username").toString() == m_authenticatedUserName) {
-								authorized = true;
-							}
+						clientDS >> server_reply;
+						if (server_reply == "OK") {
+							ds << TQString("OK");
+							m_servState = 3;
 						}
-						m_databaseActivityCursor->select(TQString("station=%1").arg(m_databaseStationsCursor->value("pk").toInt()));
-						while (m_databaseActivityCursor->next()) {
-							if (m_databaseActivityCursor->value("username").toString() != "") {
-								in_use = true;
-							}
+						else {
+							m_servClientSocket->close();
+							ds << TQString("ERRNOTAVL");
+							printf("[DEBUG] Connection failed to %s:%d for user %s@%s due to remote server returning %s\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii(), server_reply.ascii()); fflush(stdout);
+							m_servActive = false;
+							delete m_servClientTimeout;
+							m_servClientTimeout = NULL;
 						}
-	
-						if ((authorized) && (!in_use)) {
-							StationType st;
-							st.id = m_databaseStationsCursor->value("pk").toInt();
-							st.name = m_databaseStationsCursor->value("name").toString();
-							st.description = m_databaseStationsCursor->value("description").toString();
-							m_databaseServicesCursor->select(TQString("station=%1").arg(m_databaseStationsCursor->value("pk").toInt()));
-							while (m_databaseServicesCursor->next()) {
-								m_databaseServiceTypesCursor->select(TQString("serviceid=%1").arg(m_databaseServicesCursor->value("servicetype").toInt()));
-								ServiceType svt;
-								if (m_databaseServiceTypesCursor->next()) {
-									svt.name = m_databaseServiceTypesCursor->value("name").toString();
-									svt.description = m_databaseServiceTypesCursor->value("description").toString();
-									svt.clientLibrary = m_databaseServiceTypesCursor->value("client_library").toString();
-									svt.version = m_databaseServiceTypesCursor->value("version").toInt();
+					}
+					else {
+						if (!m_servClientTimeout->isActive()) {
+							// Timeout!
+							m_servClientSocket->close();
+							ds << TQString("ERRNOTAVL");
+							printf("[DEBUG] Connection failed to %s:%d for user %s@%s\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+							m_servActive = false;
+							delete m_servClientTimeout;
+							m_servClientTimeout = NULL;
+						}
+					}
+				}
+				else {
+					m_servClientSocket->close();
+					ds << TQString("ERRNOTAVL");
+					printf("[DEBUG] Connection failed to %s:%d for user %s@%s\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+					m_servActive = false;
+					delete m_servClientTimeout;
+					m_servClientTimeout = NULL;
+				}
+				break;
+			case 3:
+				if (m_servClientSocket->state() == TQSocket::Connected) {
+					TQByteArray ba(8192);
+					TQ_ULONG reclen;
+
+					if (canReadLine()) {
+						reclen = readBlock(ba.data(), 8192);
+						m_servClientSocket->writeBlock(ba.data(), reclen);
+					}
+					if (m_servClientSocket->canReadLine()) {
+						reclen = m_servClientSocket->readBlock(ba.data(), 8192);
+						writeBlock(ba.data(), reclen);
+					}
+				}
+				else {
+					m_servClientSocket->close();
+					ds << TQString("ERRNOTAVL");
+					printf("[DEBUG] Connection terminated by remote host %s:%d for user %s@%s\n\r", m_srvServiceHostName.ascii(), m_srvServicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+					m_servActive = false;
+				}
+				break;
+		}
+	}
+}
+
+void AuthSocket::commandLoop() {
+	if (m_servActive) {
+		servLoop();
+		TQTimer::singleShot(0, this, SLOT(commandLoop()));
+		return;
+	}
+
+	m_criticalSection++;
+	try {
+		if (state() == TQSocket::Connected) {
+			if (canReadLine()) {
+				TQString command;
+				TQDataStream ds(this);
+
+				ds >> command;
+				if (command != "") {
+					printf("[DEBUG] Got command %s from user %s@%s\n\r", command.ascii(), m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+					if (command == "LIST") {
+						// Send list of available servers...
+						m_slist.clear();
+		
+						// Get all stations from the database
+						m_databaseStationsCursor->select();
+						while (m_databaseStationsCursor->next()) {
+							bool authorized = false;
+							bool in_use = false;
+		
+							m_databasePermissionsCursor->select(TQString("station=%1").arg(m_databaseStationsCursor->value("pk").toInt()));
+							while (m_databasePermissionsCursor->next()) {
+								if (m_databasePermissionsCursor->value("username").toString() == m_authenticatedUserName) {
+									authorized = true;
 								}
-								if (svt.name == "") {
-									svt.name = i18n("<unknown>");
+							}
+							m_databaseActivityCursor->select(TQString("station=%1").arg(m_databaseStationsCursor->value("pk").toInt()));
+							while (m_databaseActivityCursor->next()) {
+								if (m_databaseActivityCursor->value("username").toString() != "") {
+									in_use = true;
 								}
-								if (svt.description == "") {
-									svt.description = i18n("<unknown>");
+							}
+		
+							if ((authorized) && (!in_use)) {
+								StationType st;
+								st.id = m_databaseStationsCursor->value("pk").toInt();
+								st.name = m_databaseStationsCursor->value("name").toString();
+								st.description = m_databaseStationsCursor->value("description").toString();
+								m_databaseServicesCursor->select(TQString("station=%1").arg(m_databaseStationsCursor->value("pk").toInt()));
+								while (m_databaseServicesCursor->next()) {
+									m_databaseServiceTypesCursor->select(TQString("serviceid=%1").arg(m_databaseServicesCursor->value("servicetype").toInt()));
+									ServiceType svt;
+									if (m_databaseServiceTypesCursor->next()) {
+										svt.name = m_databaseServiceTypesCursor->value("name").toString();
+										svt.description = m_databaseServiceTypesCursor->value("description").toString();
+										svt.clientLibrary = m_databaseServiceTypesCursor->value("client_library").toString();
+										svt.version = m_databaseServiceTypesCursor->value("version").toInt();
+									}
+									if (svt.name == "") {
+										svt.name = i18n("<unknown>");
+									}
+									if (svt.description == "") {
+										svt.description = i18n("<unknown>");
+									}
+									st.services.append(svt);
 								}
-								st.services.append(svt);
+		
+								m_slist.append(st);
 							}
-	
-							m_slist.append(st);
 						}
+		
+						ds << m_slist;
 					}
+					else if (command == "BIND") {
+						// Get desired Station Type from client
+						StationType st;
+						ds >> st;
+		
+						// Attempt to bind to station matching desired Service Type list...
+						m_stationID = -1;
 	
-					ds << m_slist;
-				}
-				else if (command == "BIND") {
-					// Get desired Station Type from client
-					StationType st;
-					ds >> st;
-	
-					// Attempt to bind to station matching desired Service Type list...
-					m_stationID = -1;
-
-					// Ensure that this user is not already connected
-					int activeID = -1;
-					m_databaseActivityCursor->select(TQString("username='%1' AND realmname='%2'").arg(m_authenticatedUserName).arg(m_authenticatedRealmName));
-					if (m_databaseActivityCursor->next()) {
-						activeID = m_databaseActivityCursor->value("station").toInt();
-					}
-					if (activeID < 0) {
-						for (StationList::Iterator it(m_slist.begin()); it != m_slist.end(); ++it) {
-							if ((*it).services == st.services) {
-								m_stationID = (*it).id;
-								break;
-							}
+						// Ensure that this user is not already connected
+						int activeID = -1;
+						m_databaseActivityCursor->select(TQString("username='%1' AND realmname='%2'").arg(m_authenticatedUserName).arg(m_authenticatedRealmName));
+						if (m_databaseActivityCursor->next()) {
+							activeID = m_databaseActivityCursor->value("station").toInt();
 						}
+						if (activeID < 0) {
+							for (StationList::Iterator it(m_slist.begin()); it != m_slist.end(); ++it) {
+								if ((*it).services == st.services) {
+									m_stationID = (*it).id;
+									break;
+								}
+							}
+			
+							if (m_stationID < 0) {
+								ds << TQString("ERRUNAVAL");
+							}
+							else {
+								m_bound = true;
 		
-						if (m_stationID < 0) {
-							ds << TQString("ERRUNAVAL");
+								// Update database
+								TQSqlRecord *buffer = m_databaseActivityCursor->primeInsert();
+								buffer->setValue("station", m_stationID);
+								buffer->setValue("username", m_authenticatedUserName);
+								buffer->setValue("realmname", m_authenticatedRealmName);
+								buffer->setValue("logontime", TQDateTime::currentDateTime().toTime_t());
+								m_databaseActivityCursor->insert();
+		
+								ds << TQString("OK");
+							}
 						}
 						else {
-							m_bound = true;
-	
-							// Update database
-							TQSqlRecord *buffer = m_databaseActivityCursor->primeInsert();
-							buffer->setValue("station", m_stationID);
-							buffer->setValue("username", m_authenticatedUserName);
-							buffer->setValue("realmname", m_authenticatedRealmName);
-							buffer->setValue("logontime", TQDateTime::currentDateTime().toTime_t());
-							m_databaseActivityCursor->insert();
-	
-							ds << TQString("OK");
+							ds << TQString("ERRPREVCN");
 						}
 					}
-					else {
-						ds << TQString("ERRPREVCN");
-					}
-				}
-				else if (command == "SERV") {
-					// Get client library name from the client
-					TQString libname;
-					ds >> libname;
-
-					m_databaseActivityCursor->select(TQString("username='%1' AND realmname='%2'").arg(m_authenticatedUserName).arg(m_authenticatedRealmName));
-					if (m_databaseActivityCursor->next()) {
-						m_stationID = m_databaseActivityCursor->value("station").toInt();
-					}
-
-					if (m_bound == true) {
-						ds << TQString("ERRINVCMD");
-					}
-					else {
-						if (m_stationID < 0) {
-							ds << TQString("ERRNOCONN");
+					else if (command == "SERV") {
+						// Get client library name from the client
+						TQString libname;
+						ds >> libname;
+	
+						m_databaseActivityCursor->select(TQString("username='%1' AND realmname='%2'").arg(m_authenticatedUserName).arg(m_authenticatedRealmName));
+						if (m_databaseActivityCursor->next()) {
+							m_stationID = m_databaseActivityCursor->value("station").toInt();
+						}
+	
+						if (m_bound == true) {
+							ds << TQString("ERRINVCMD");
 						}
 						else {
-							// Find the service ID for the specified client library name
-							TQ_INT32 sid = -1;
-							m_databaseServiceTypesCursor->select(TQString("client_library='%1'").arg(libname));
-							if (m_databaseServiceTypesCursor->next()) {
-								sid = m_databaseServiceTypesCursor->value("serviceid").toInt();
-							}
-							if (sid < 0) {
-								ds << TQString("ERRNOSERV");
+							if (m_stationID < 0) {
+								ds << TQString("ERRNOCONN");
 							}
 							else {
-								// Attempt to connect to the backend server
-								m_databaseServicesCursor->select(TQString("pk=%1 AND station=%2").arg(sid).arg(m_stationID));
-								if (m_databaseServicesCursor->next()) {
-									TQString serviceHostName = m_databaseServicesCursor->value("hostname").toString();
-									int servicePort = m_databaseServicesCursor->value("port").toInt();
-
-									TDEKerberosClientSocket clientSocket;
-									clientSocket.setServiceName("remotefpga");
-
-									clientSocket.setServerFQDN(serviceHostName);
-									clientSocket.connectToHost(serviceHostName, servicePort);
-
-									TQTimer connectionTimeout;
-									connectionTimeout.start(5000, TRUE);
-									while ((clientSocket.state() == TQSocket::Connecting) || (clientSocket.state() == TQSocket::HostLookup)) {
-										tqApp->processEvents();
-										if (!connectionTimeout.isActive()) {
-											break;
-										}
-									}
-									connectionTimeout.stop();
-									if (clientSocket.state() == TQSocket::Connected) {
-										if (clientSocket.setUsingKerberos(true) != 0) {
-											clientSocket.close();
-											ds << TQString("ERRNOTAVL");
-											printf("[DEBUG] Connection failed to %s:%d for user %s@%s due to Kerberos failure\n\r", serviceHostName.ascii(), servicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
-										}
-										else {
-											TQDataStream clientDS(&clientSocket);
-											TQString server_reply;
-											connectionTimeout.start(5000, TRUE);
-											while ((!clientSocket.canReadLine()) && (clientSocket.state() == TQSocket::Connected)) {
-												tqApp->processEvents();
-												if (!connectionTimeout.isActive()) {
-													break;
-												}
-											}
-											connectionTimeout.stop();
-											if ((clientSocket.canReadLine()) && (clientSocket.state() == TQSocket::Connected)) {
-												clientDS >> server_reply;
-											}
-											if (server_reply == "OK") {
-												ds << TQString("OK");
-												TQByteArray ba(8192);
-												TQ_ULONG reclen;
-												while ((state() == TQSocket::Connected) && (clientSocket.state() == TQSocket::Connected)) {
-													// RAJA FIXME
-													if (canReadLine()) {
-														reclen = readBlock(ba.data(), 8192);
-														clientSocket.writeBlock(ba.data(), reclen);
-													}
-													if (clientSocket.canReadLine()) {
-														reclen = clientSocket.readBlock(ba.data(), 8192);
-														writeBlock(ba.data(), reclen);
-													}
-													tqApp->processEvents();
-												}
-												clientSocket.close();
-											}
-											else {
-												clientSocket.close();
-												ds << TQString("ERRNOTAVL");
-												printf("[DEBUG] Connection failed to %s:%d for user %s@%s due to remote server returning %s\n\r", serviceHostName.ascii(), servicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii(), server_reply.ascii()); fflush(stdout);
-											}
-										}
+								// Find the service ID for the specified client library name
+								TQ_INT32 sid = -1;
+								m_databaseServiceTypesCursor->select(TQString("client_library='%1'").arg(libname));
+								if (m_databaseServiceTypesCursor->next()) {
+									sid = m_databaseServiceTypesCursor->value("serviceid").toInt();
+								}
+								if (sid < 0) {
+									ds << TQString("ERRNOSERV");
+								}
+								else {
+									// Attempt to connect to the backend server
+									m_databaseServicesCursor->select(TQString("pk=%1 AND station=%2").arg(sid).arg(m_stationID));
+									if (m_databaseServicesCursor->next()) {
+										m_srvServiceHostName = m_databaseServicesCursor->value("hostname").toString();
+										m_srvServicePort = m_databaseServicesCursor->value("port").toInt();
+
+										if (!m_servClientSocket) m_servClientSocket = new TDEKerberosClientSocket;
+										m_servClientSocket->setServiceName("remotefpga");
+	
+										m_servClientSocket->setServerFQDN(m_srvServiceHostName);
+										m_servClientSocket->connectToHost(m_srvServiceHostName, m_srvServicePort);
+
+										m_servState = 0;
+										m_servActive = true;
 									}
 									else {
-										clientSocket.close();
-										ds << TQString("ERRNOTAVL");
-										printf("[DEBUG] Connection failed to %s:%d for user %s@%s\n\r", serviceHostName.ascii(), servicePort, m_authenticatedUserName.ascii(), m_authenticatedRealmName.ascii()); fflush(stdout);
+										ds << TQString("ERRNOSERV");
 									}
 								}
-								else {
-									ds << TQString("ERRNOSERV");
-								}
 							}
 						}
 					}
-				}
-				else {
-					ds << TQString("ERRINVCMD");
+					else {
+						ds << TQString("ERRINVCMD");
+					}
 				}
 			}
-			tqApp->processEvents();
-		}
 
-		m_criticalSection--;
-		return 0;
+			m_criticalSection--;
+			TQTimer::singleShot(0, this, SLOT(commandLoop()));
+			return;
+		}
 	}
 	catch (...) {
 		m_criticalSection--;
-		return -1;
+		return;
 	}
 }
 
+int AuthSocket::enterCommandLoop() {
+	TQTimer::singleShot(0, this, SLOT(commandLoop()));
+	return 0;
+}
+
 int AuthSocket::connectToDatabase() {
 	if (m_database) {
 		return -2;
diff --git a/servers/auth_server_lin/src/auth_conn.h b/servers/auth_server_lin/src/auth_conn.h
index 55bb5de..f1b3295 100644
--- a/servers/auth_server_lin/src/auth_conn.h
+++ b/servers/auth_server_lin/src/auth_conn.h
@@ -57,6 +57,8 @@ class AuthSocket : public TDEKerberosServerSocket
 	private slots:
 		int connectToDatabase();
 		void connectionClosedHandler();
+		void commandLoop();
+		void servLoop();
 	
 	private:
 		int line;
@@ -65,6 +67,13 @@ class AuthSocket : public TDEKerberosServerSocket
 		int m_stationID;
 		bool m_bound;
 
+		bool m_servActive;
+		int m_servState;
+		TDEKerberosClientSocket* m_servClientSocket;
+		TQTimer* m_servClientTimeout;
+		TQString m_srvServiceHostName;
+		int m_srvServicePort;
+
 		KSimpleConfig* m_config;
 		TQSqlDatabase* m_database;
 		TQSqlCursor* m_databaseStationsCursor;