openvidu-server: new Session lock protecting joinRoom/leaveRoom methods

openvidu-server/src/main/java/io/openvidu/server/core/Session.java

@@ -23,7 +23,9 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 
@@ -70,6 +72,12 @@ public class Session implements SessionInterface {
 	 */
 	public ReadWriteLock closingLock = new ReentrantReadWriteLock();
 
+	/**
+	 * This lock protects the operations of SessionManager#joinRoom and
+	 * SessionManager#leaveRoom
+	 */
+	public Lock joinLeaveLock = new ReentrantLock();
+
 	public final AtomicBoolean recordingManuallyStopped = new AtomicBoolean(false);
 
 	public Session(Session previousSession) {

openvidu-server/src/main/java/io/openvidu/server/kurento/core/KurentoSessionManager.java

@@ -88,11 +88,9 @@ public class KurentoSessionManager extends SessionManager {
 	/* Protected by Session.closingLock.readLock */
 	public void joinRoom(Participant participant, String sessionId, Integer transactionId) {
 		Set<Participant> existingParticipants = null;
-		boolean lockAcquired = false;
 		try {
 
 			KurentoSession kSession = (KurentoSession) sessions.get(sessionId);
-
 			if (kSession == null) {
 				// First user connecting to the session
 				Session sessionNotActive = sessionsNotActive.get(sessionId);
@@ -108,7 +106,11 @@ public class KurentoSessionManager extends SessionManager {
 
 				try {
 					if (KmsManager.selectAndRemoveKmsLock.tryLock(MS_MAX_LOCK_WAIT, TimeUnit.SECONDS)) {
-						lockAcquired = true;
+						try {
+							kSession = (KurentoSession) sessions.get(sessionId);
+
+							if (kSession == null) {
+								// Session still null. It was not created by other thread while waiting for lock
 								Kms lessLoadedKms = null;
 								try {
 									lessLoadedKms = this.kmsManager.getLessLoadedConnectedAndRunningKms();
@@ -117,11 +119,16 @@ public class KurentoSessionManager extends SessionManager {
 									this.cleanCollections(sessionId);
 									this.storeSessionNotActive(sessionNotActive);
 									throw new OpenViduException(Code.ROOM_CANNOT_BE_CREATED_ERROR_CODE,
-									"There is no available Media Node where to initialize session '" + sessionId + "'");
+											"There is no available Media Node where to initialize session '" + sessionId
+													+ "'");
 								}
 								log.info("KMS less loaded is {} with a load of {}", lessLoadedKms.getUri(),
 										lessLoadedKms.getLoad());
 								kSession = createSession(sessionNotActive, lessLoadedKms);
+							}
+						} finally {
+							KmsManager.selectAndRemoveKmsLock.unlock();
+						}
 
 					} else {
 						String error = "Timeout of " + MS_MAX_LOCK_WAIT + " seconds waiting to acquire lock";
@@ -145,19 +152,35 @@ public class KurentoSessionManager extends SessionManager {
 						+ "' is trying to join session '" + sessionId + "' but it is closing");
 			}
 
+			try {
+				if (kSession.joinLeaveLock.tryLock(15, TimeUnit.SECONDS)) {
+					try {
 						existingParticipants = getParticipants(sessionId);
 						kSession.join(participant);
+						sessionEventsHandler.onParticipantJoined(participant, sessionId, existingParticipants,
+								transactionId, null);
+					} finally {
+						kSession.joinLeaveLock.unlock();
+					}
+				} else {
+					log.error(
+							"Timeout waiting for join-leave Session lock to be available for participant {} of session {} in joinRoom",
+							participant.getParticipantPublicId(), sessionId);
+					sessionEventsHandler.onParticipantJoined(participant, sessionId, null, transactionId,
+							new OpenViduException(Code.GENERIC_ERROR_CODE, "Timeout waiting for Session lock"));
+				}
+			} catch (InterruptedException e) {
+				log.error(
+						"InterruptedException waiting for join-leave Session lock to be available for participant {} of session {} in joinRoom",
+						participant.getParticipantPublicId(), sessionId);
+				sessionEventsHandler.onParticipantJoined(participant, sessionId, null, transactionId,
+						new OpenViduException(Code.GENERIC_ERROR_CODE,
+								"InterruptedException waiting for Session lock"));
+			}
 		} catch (OpenViduException e) {
-			log.warn("PARTICIPANT {}: Error joining/creating session {}", participant.getParticipantPublicId(),
+			log.error("PARTICIPANT {}: Error joining/creating session {}", participant.getParticipantPublicId(),
 					sessionId, e);
 			sessionEventsHandler.onParticipantJoined(participant, sessionId, null, transactionId, e);
-		} finally {
-			if (lockAcquired) {
-				KmsManager.selectAndRemoveKmsLock.unlock();
-			}
-		}
-		if (existingParticipants != null) {
-			sessionEventsHandler.onParticipantJoined(participant, sessionId, existingParticipants, transactionId, null);
 		}
 	}
 
@@ -179,6 +202,11 @@ public class KurentoSessionManager extends SessionManager {
 			throw new OpenViduException(Code.ROOM_CLOSED_ERROR_CODE, "'" + participant.getParticipantPublicId()
 					+ "' is trying to leave from session '" + sessionId + "' but it is closing");
 		}
+
+		try {
+			if (session.joinLeaveLock.tryLock(15, TimeUnit.SECONDS)) {
+				try {
+
 					session.leave(participant.getParticipantPrivateId(), reason);
 
 					// Update control data structures
@@ -195,7 +223,8 @@ public class KurentoSessionManager extends SessionManager {
 						if (insecureUsers.containsKey(p.getParticipantPrivateId())) {
 							boolean stillParticipant = false;
 							for (Session s : sessions.values()) {
-					if (!s.isClosed() && (s.getParticipantByPrivateId(p.getParticipantPrivateId()) != null)) {
+								if (!s.isClosed()
+										&& (s.getParticipantByPrivateId(p.getParticipantPrivateId()) != null)) {
 									stillParticipant = true;
 									break;
 								}
@@ -215,8 +244,8 @@ public class KurentoSessionManager extends SessionManager {
 						log.info("Possible collision when closing the session '{}' (not found)", sessionId);
 						remainingParticipants = Collections.emptySet();
 					}
-		sessionEventsHandler.onParticipantLeft(participant, sessionId, remainingParticipants, transactionId, null,
+					sessionEventsHandler.onParticipantLeft(participant, sessionId, remainingParticipants, transactionId,
-				reason);
+							null, reason);
 
 					if (!EndReason.sessionClosedByServer.equals(reason)) {
 						// If session is closed by a call to "DELETE /api/sessions" do NOT stop the
@@ -239,7 +268,8 @@ public class KurentoSessionManager extends SessionManager {
 											if (session.isClosed()) {
 												return false;
 											}
-								log.info("No more participants in session '{}', removing it and closing it", sessionId);
+											log.info("No more participants in session '{}', removing it and closing it",
+													sessionId);
 											this.closeSessionAndEmptyCollections(session, reason, true);
 											sessionClosedByLastParticipant = true;
 										} finally {
@@ -262,7 +292,8 @@ public class KurentoSessionManager extends SessionManager {
 								&& ProtocolElements.RECORDER_PARTICIPANT_PUBLICID
 										.equals(remainingParticipants.iterator().next().getParticipantPublicId())) {
 							// RECORDER participant is the last one standing. Start countdown
-				log.info("Last participant left. Starting {} seconds countdown for stopping recording of session {}",
+							log.info(
+									"Last participant left. Starting {} seconds countdown for stopping recording of session {}",
 									this.openviduConfig.getOpenviduRecordingAutostopTimeout(), sessionId);
 							recordingManager.initAutomaticRecordingStopThread(session);
 						}
@@ -274,6 +305,22 @@ public class KurentoSessionManager extends SessionManager {
 					}
 
 					return sessionClosedByLastParticipant;
+
+				} finally {
+					session.joinLeaveLock.unlock();
+				}
+			} else {
+				log.error(
+						"Timeout waiting for join-leave Session lock to be available for participant {} of session {} in leaveRoom",
+						kParticipant.getParticipantPublicId(), session.getSessionId());
+				return false;
+			}
+		} catch (InterruptedException e) {
+			log.error(
+					"Timeout waiting for join-leave Session lock to be available for participant {} of session {} in leaveRoom",
+					kParticipant.getParticipantPublicId(), session.getSessionId());
+			return false;
+		}
 	}
 
 	/**

openvidu-server/src/main/java/io/openvidu/server/kurento/kms/KmsManager.java

@@ -206,13 +206,13 @@ public abstract class KmsManager {
 				kms.setKurentoClientConnected(false);
 				kms.setTimeOfKurentoClientDisconnection(System.currentTimeMillis());
 
-				if(kms.getKurentoClient().isClosed()) {
+				if (kms.getKurentoClient().isClosed()) {
 					log.info("Kurento Client \"disconnected\" event for KMS {} [{}]. Closed explicitely", kms.getUri(),
 							kms.getKurentoClient().toString());
 					return;
 				} else {
-					log.info("Kurento Client \"disconnected\" event for KMS {} [{}]. Waiting reconnection", kms.getUri(),
+					log.info("Kurento Client \"disconnected\" event for KMS {} [{}]. Waiting reconnection",
-							kms.getKurentoClient().toString());					
+							kms.getUri(), kms.getKurentoClient().toString());
 				}
 
 				// TODO: this is a fix for the lack of reconnected event
@@ -235,7 +235,7 @@ public abstract class KmsManager {
 									return;
 								}
 
-								if(kms.getKurentoClient().isClosed()) {
+								if (kms.getKurentoClient().isClosed()) {
 									log.info(
 											"Timer of KMS with uri {} and KurentoClient [{}] has been closed. Cancelling Timer",
 											kms.getUri(), kms.getKurentoClient().toString());