Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -232,9 +232,19 @@ public void run() {
// another thread closed our listener socket - that's expected during shutdown, but not in other circumstances
LOG.info("XceiverServerDomainSocket is closed", ace);
} catch (IOException ie) {
// usually when the xceiver count limit is hit.
LOG.warn("Got an exception. Peer {}", peer, ie);
IOUtils.closeQuietly(peer);
if (peer == null) {
// accept() itself failed, for example with no file descriptor left, and would fail again right away
Comment thread
Gargi-jais11 marked this conversation as resolved.
LOG.warn("accept() failed. Will retry after 1 second.", ie);
try {
Thread.sleep(TimeUnit.SECONDS.toMillis(1L));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should it be configurable? 1 second is a long wait...

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yandrey321 thes pause only happens when accept() itself throws, suppose when the datanode is out of file descriptors. Normal accepts, idle timeouts, and rejected connections are not delayed. In that state a retry fails again almost at once: without the pause the test saw about 14,600 accept() calls in 500 ms. One second matches the fixed 30-second pause the same loop already uses after an OutOfMemoryError, and Hadoop's DataXceiverServer uses a similar fixed backoff.

@ChenSammi, what do you prefer? I can keep the fixed 1 second, 2.) use a shorter fixed value such as 100 ms, or add a datanode config key for it. A config key seems like more than this case needs, but I'm happy to add one if you think it's useful.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I prefer the current fixed 1s, since it 's only to avoid bursting messages to the DN log file in case of DN run out of file descriptors, so I don't expect it will happen frequently, and when it happens, user is expected to tune the file descriptors or investigate whether there is FD leaks, instead of tuning this sleep interval.

} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
} else {
// usually when the xceiver count limit is hit.
LOG.warn("Got an exception. Peer {}", peer, ie);
IOUtils.closeQuietly(peer);
}
} catch (OutOfMemoryError ie) {
IOUtils.closeQuietly(peer);
// DataNode can run out of memory if there is too many transfers.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,10 @@
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.when;

import com.google.common.collect.Maps;
Expand All @@ -52,6 +55,7 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.RandomStringUtils;
Expand Down Expand Up @@ -95,6 +99,7 @@
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.MockedStatic;
import org.slf4j.LoggerFactory;

/**
Expand Down Expand Up @@ -636,6 +641,36 @@ public void testMaxXceiverCount() throws Exception {
}
}

/**
* A failed accept(), for example with no file descriptor left, is retried after a pause instead of at once.
*/
@Test
public void testAcceptFailureIsRetriedAfterPause() throws Exception {
conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, dir.getAbsolutePath());
AtomicInteger accepts = new AtomicInteger();
DomainSocket listener = mock(DomainSocket.class, RETURNS_DEEP_STUBS);
when(listener.accept()).thenAnswer(invocation -> {
accepts.incrementAndGet();
throw new SocketException("accept(2) error: Too many open files");
});
DomainSocketFactory factory = mock(DomainSocketFactory.class);
when(factory.isServiceEnabled()).thenReturn(true);
when(factory.isServiceReady()).thenReturn(true);
XceiverServerDomainSocket server;
try (MockedStatic<DomainSocket> domainSocket = mockStatic(DomainSocket.class)) {
domainSocket.when(() -> DomainSocket.bindAndListen(any())).thenReturn(listener);
server = new XceiverServerDomainSocket(randomDatanodeDetails(), conf, null, readExecutors, metrics, factory);
}
try {
server.start();
GenericTestUtils.waitFor(() -> accepts.get() > 0, 10, 5000);
Thread.sleep(500);
assertTrue(accepts.get() <= 2, "accept() was called " + accepts.get() + " times");
} finally {
server.stop();
}
}

/**
* When server receives any message which doesn't follow the version, request type, request body sequence, server
* will treat it as a critical error, close the connection.
Expand Down
Loading