Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

OF-2921: Prevent deadlock by not broadcasting synchronously #2635

Merged
merged 3 commits into from
Jan 4, 2025
Merged

OF-2921: Prevent deadlock by not broadcasting synchronously #2635

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b98a83d
OF-2921: Prevent deadlock by not broadcasting synchronously
guusdk Dec 17, 2024
debf053
OF-2921: Reduce Websocket operations under mutex
guusdk Jan 2, 2025
6428f38
OF-2921: Don't process websocket error under mutex
guusdk Jan 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions xmppserver/src/main/java/org/jivesoftware/openfire/SessionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,7 +291,9 @@ public synchronized void terminateDetached(LocalSession session) {
Presence presence = new Presence();
presence.setType(Presence.Type.unavailable);
presence.setFrom(session.getAddress());
router.route(presence);

// Broadcast asynchronously, to reduce the likelihood of the broadcast introducing a deadlock (OF-2921).
TaskEngine.getInstance().submit(() -> router.route(presence));
}

session.getStreamManager().onClose(router, serverAddress);
Expand Down Expand Up @@ -1360,7 +1362,9 @@ public void onConnectionClose(Object handback) {
Presence presence = new Presence();
presence.setType(Presence.Type.unavailable);
presence.setFrom(session.getAddress());
router.route(presence);

// Broadcast asynchronously, to reduce the likelihood of the broadcast introducing a deadlock (OF-2921).
TaskEngine.getInstance().submit(() -> router.route(presence));
}

session.getStreamManager().onClose(router, serverAddress);
Expand Down
17 changes: 14 additions & 3 deletions xmppserver/src/main/java/org/jivesoftware/openfire/session/LocalClientSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -944,14 +944,25 @@ public void deliver(Packet queueOrPushStanza) throws UnauthorizedException {
if (stanzasToPush.isEmpty()) {
return;
}
synchronized (streamManager)
{

// When stream management is enabled, deliver and record stanzas under a mutex. If it's not enabled, don't
// acquire the lock to reduce lock contention (OF-2921).
if (!streamManager.isEnabled()) {
// Push stanzas to the client.
for (final Packet stanzaToPush : stanzasToPush) {
if (conn != null) {
conn.deliver(stanzaToPush);
}
streamManager.sentStanza(stanzaToPush);
}
} else {
synchronized (streamManager) {
// Push stanzas to the client.
for (final Packet stanzaToPush : stanzasToPush) {
if (conn != null) {
conn.deliver(stanzaToPush);
}
streamManager.sentStanza(stanzaToPush);
}
}
}
}
Expand Down
24 changes: 11 additions & 13 deletions ...r/src/main/java/org/jivesoftware/openfire/websocket/WebSocketClientConnectionHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,20 +170,18 @@ public void onTextMethod(String stanza)
public void onError(Throwable error)
{
Log.debug("Error detected; connection: {}, session: {}", wsConnection, wsSession, error);
synchronized (this) {
try {
if (isWebSocketOpen()) {
Log.warn("Attempting to close connection on which an error occurred: {}", wsConnection, error);
wsConnection.close(new StreamError(StreamError.Condition.internal_server_error), !isWebSocketOpen());
} else {
Log.debug("Error detected on websocket that isn't open (any more):", error);
wsConnection.close(null, !isWebSocketOpen());
}
} catch (Exception e) {
Log.error("Error disconnecting websocket", e);
} finally {
wsSession = null;
try {
if (isWebSocketOpen()) {
Log.warn("Attempting to close connection on which an error occurred: {}", wsConnection, error);
wsConnection.close(new StreamError(StreamError.Condition.internal_server_error), !isWebSocketOpen());
} else {
Log.debug("Error detected on websocket that isn't open (any more):", error);
wsConnection.close(null, !isWebSocketOpen());
}
} catch (Exception e) {
Log.error("Error disconnecting websocket", e);
} finally {
wsSession = null;
}
}

Expand Down
Loading