Skip to content

[#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued #1057

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

Open
vharseko wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued #1057

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
84 changes: 75 additions & 9 deletions opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
*
* Copyright 2006-2009 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
* Portions Copyright 2026 3A Systems, LLC.
*/
package org.opends.server.replication.protocol;

Expand All @@ -36,6 +37,7 @@

import javax.net.ssl.SSLSocket;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.api.DirectoryThread;
import org.opends.server.types.HostPort;
Expand Down Expand Up @@ -96,7 +98,20 @@ public final class Session extends DirectoryThread implements Closeable
*/
private BufferedOutputStream output;

private final LinkedBlockingQueue<byte[]> sendQueue = new LinkedBlockingQueue<>(4000);
/** A message queued for the thread of this session, and what to run once it is written. */
private static final class Outgoing
{
private final byte[] buffer;
private final Runnable whenWritten;

private Outgoing(byte[] buffer, Runnable whenWritten)
{
this.buffer = buffer;
this.whenWritten = whenWritten;
}
}

private final LinkedBlockingQueue<Outgoing> sendQueue = new LinkedBlockingQueue<>(4000);
private AtomicBoolean isRunning = new AtomicBoolean(false);
private final CountDownLatch latch = new CountDownLatch(1);

Expand Down Expand Up @@ -140,6 +155,10 @@ public Session(final Socket socket,
/**
* This method is called when the session with the remote must be closed.
* This object won't be used anymore after this method is called.
* <p>
* The thread of this session is stopped where it is: whatever is still queued for it is not
* written, and the callbacks of those messages never run - see
* {@link #publish(ReplicationMsg, Runnable)}.
*/
@Override
public void close()
Expand Down Expand Up @@ -306,23 +325,51 @@ public boolean isEncrypted()
* If an IO error occurred.
*/
public void publish(final ReplicationMsg msg) throws IOException
{
publish(msg, null);
}

/**
* Sends a replication message to the remote peer, and runs the provided callback once the
* message has been written to the socket.
* <p>
* While the thread of this session runs, a message published is queued for it and written
* later, so the return of this method says only that the message is queued. The callback is
* the only word that the message has left this server: it runs once, on the thread which wrote
* the message, after the write returned - and never for a message which was not written, which
* is what becomes of a message the write of which fails, and of everything still queued when
* the session is closed. It must be short and must not block: the session writes nothing else
* until it returns.
*
* @param msg
* The message to be sent.
* @param whenWritten
* What to run once the message has been written, or null.
* @return whether the message was written or queued to be written; false when it was neither,
* because it has no encoding for the protocol version of the peer or because the
* session is being closed - the callback then never runs.
* @throws IOException
* If an IO error occurred.
*/
public boolean publish(final ReplicationMsg msg, final Runnable whenWritten) throws IOException
{
final byte[] buffer = msg.getBytes(protocolVersion);
if (buffer == null)
{
// skip anything that cannot be encoded for this peer.
return;
return false;
}
if (isRunning.get())
{
final Outgoing outgoing = new Outgoing(buffer, whenWritten);
while (!closeInitiated)
{
try
{
// Avoid blocking forever so that we can check for session closure.
if (sendQueue.offer(buffer, 100, TimeUnit.MILLISECONDS))
if (sendQueue.offer(outgoing, 100, TimeUnit.MILLISECONDS))
{
return;
return true;
}
}
catch (final InterruptedException e)
Expand All @@ -331,10 +378,27 @@ public void publish(final ReplicationMsg msg) throws IOException
throw new IOException(e.getMessage());
}
}
return false;
}
else
send(buffer);
written(whenWritten);
return true;
}

/** Runs what was to run once a message is written; a callback which fails takes nothing down. */
private void written(final Runnable whenWritten)
{
if (whenWritten != null)
{
send(buffer);
try
{
whenWritten.run();
}
catch (final RuntimeException e)
{
logger.error(LocalizableMessage.raw("The callback of a message written to %s failed: %s",
readableRemoteAddress, stackTraceToSingleLineString(e)));
}
}
}

Expand Down Expand Up @@ -535,24 +599,26 @@ public void run()
boolean needClosing = false;
while (!closeInitiated)
{
byte[] buffer;
Outgoing outgoing;
try
{
buffer = sendQueue.take();
outgoing = sendQueue.take();
}
catch (InterruptedException ie)
{
break;
}
try
{
send(buffer);
send(outgoing.buffer);
}
catch (IOException e)
{
setSessionError(e);
needClosing = true;
continue;
}
written(outgoing.whenWritten);
}
isRunning.set(false);
if (needClosing)
Expand Down
51 changes: 37 additions & 14 deletions opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,11 @@
*/
package org.opends.server.replication.server;

import java.io.IOException;
import java.net.SocketException;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.api.DirectoryThread;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.common.ServerStatus;
Expand Down Expand Up @@ -120,24 +122,14 @@ public void run()
replicationServerDomain.getBaseDN(), handler.getServerId());
}
}
else if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
{
forwardReplicaOfflineMsg((ReplicaOfflineMsg) updateMsg);
}
else
{
// Publish the update to the remote server using a protocol version it supports
session.publish(updateMsg);
/*
* Only the forward to a peer RS ends the wait of the shutdown: what the grace period
* buys is the rest of the topology learning that the replica went offline.
* ReplicationServerDomain.put() never queues this message for a directory server - its
* isUpdateMsgFiltered() drops it there - but a directory server which is catching up
* reads its updates from the changelog, where ReplicaCursor synthesizes a
* ReplicaOfflineMsg from the offline CSN of the replica. Publishing that one says
* nothing about the peer RSs the shutdown is waiting for.
*/
if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
{
dsrsShutdownSync.replicaOfflineMsgForwarded(
replicationServerDomain.getBaseDN(), updateMsg.getCSN(), handler.getServerId());
}
}
}
}
Expand Down Expand Up @@ -170,6 +162,37 @@ public void run()
}
}

/**
* Publishes a ReplicaOfflineMsg to the peer replication server, and reports the forward to the
* shutdown which may be waiting for it.
* <p>
* Only the forward to a peer RS ends the wait of the shutdown: what the grace period buys is
* the rest of the topology learning that the replica went offline.
* ReplicationServerDomain.put() never queues this message for a directory server - its
* isUpdateMsgFiltered() drops it there - but a directory server which is catching up reads its
* updates from the changelog, where ReplicaCursor synthesizes a ReplicaOfflineMsg from the
* offline CSN of the replica. Publishing that one says nothing about the peer RSs the shutdown
* is waiting for, so it goes the way of every other update.
* <p>
* The forward is reported once the message has been written to the peer, not once it is queued
* for the thread of the session: the shutdown closes the session as soon as its wait ends, and
* Session.close() drops whatever is still queued, so a message reported forwarded while it
* was queued behind one the peer had not read yet would never reach the peer. A message the
* session refuses - one the protocol version of the peer cannot carry, or one published while
* the session is being closed - will never be written, and the shutdown must not wait for it.
*/
private void forwardReplicaOfflineMsg(final ReplicaOfflineMsg msg) throws IOException
{
final DN baseDN = replicationServerDomain.getBaseDN();
final int serverId = handler.getServerId();
final boolean accepted = session.publish(msg,
() -> dsrsShutdownSync.replicaOfflineMsgForwarded(baseDN, msg.getCSN(), serverId));
if (!accepted)
{
dsrsShutdownSync.replicaOfflineMsgNotForwarded(baseDN, serverId);
}
}

private boolean isUpdateMsgFiltered(UpdateMsg updateMsg)
{
if (handler.isDataServer())
Expand Down
Loading
Loading