Forem: Juan Olivares

Java Split Text File Efficiently

Juan Olivares — Fri, 09 May 2025 14:27:43 +0000

The problem

When facing the task of writing java code that splits a text file into chunks of a maximum size we might be tempted to write code like the following

public void split(Path inputFile, Path outputDir) throws IOException {
    Files.createDirectories(outputDir);

    try (BufferedReader reader = Files.newBufferedReader(inputFile)) {
        int fileIndex = 0;
        long currentSize = 0;
        BufferedWriter writer = newWriter(outputDir, fileIndex++);

        String line;
        while ((line = reader.readLine()) != null) {
            byte[] lineBytes = (line + System.lineSeparator()).getBytes();
            if (currentSize + lineBytes.length > maxFileSizeBytes) {
                writer.close();
                writer = newWriter(outputDir, fileIndex++);
                currentSize = 0;
            }
            writer.write(line);
            writer.newLine();
            currentSize += lineBytes.length;
        }
        writer.close();
    }
}

private BufferedWriter newWriter(Path dir, int index) throws IOException {
    Path filePath = dir.resolve("part_" + index + ".txt");
    return Files.newBufferedWriter(filePath);
}

This code is technically ok, however it’s quite inefficient for splitting large files into many chunks.

It performs many heap allocations (lines), resulting in lots of of temporary objects created and discarded (Strings, byte arrays).

There’s also a less obvious issue, it’s copying data to several buffers and performing context switches between user and kernel modes.

Here’s what happens in the current code under the hood:

BufferedReader:

Calls read() on an underlying FileReader or InputStreamReader.
Data is copied from kernel space → user space buffer.
Then parsed into Java Strings (heap-allocated).

getBytes():

Converts String to a new byte[] → more heap allocations.

BufferedWriter:

Takes the byte/char data from user space.
Calls write() which again involves copying user space → kernel space.
Eventually flushed to disk.

So data moves back and forth between kernel and user space multiple times, with extra heap churn. Apart from the garbage collection pressure, it has the following consequences:

Memory bandwidth is wasted copying between buffers.
High CPU utilization for what could be a disk-to-disk transfer.
The OS could have handled the bulk copy directly (via DMA or optimized I/O), but Java code hijacks that efficiency by involving user-space logic.

The solution

So, how can we avoid the above issues? The answer is using zero copy as much as possible i.e. avoid leaving the kernel space as much as possible. This can be done in java by using the FileChannel method long transferTo(long position, long count, WritableByteChannel target) . It will essentially perform a disk-to-disk transfer and will leverage any IO optimization of the operating system.

A key challenge arises: the described method operates on byte chunks, potentially disrupting line integrity. To address this, we need a strategy that ensures lines are kept intact even when the file is processed by moving byte segments.

Solving the problem without the above challenge would be easy, just call transferTo for every chunk, incrementing position as position = position + maxFileSize until no more data can be transferred.

To preserve line integrity, we need to determine the end of the last complete line within each byte chunk. We'll achieve this by initially seeking to the expected end of the chunk and then scanning backwards to locate the preceding line break. This will give us the accurate byte count for the chunk, ensuring the inclusion of the final, unbroken line. This will be the sole section of code performing buffer allocations and copies, and since these operations should be minimal, the performance impact is expected to be negligible.

The following code illustrates the techniques discussed

private static final int LINE_ENDING_SEARCH_WINDOW = 8 * 1024;

private long maxSizePerFileInBytes;
private Path outputDirectory;
private Path tempDir;

private void split(Path fileToSplit) throws IOException {
    try (RandomAccessFile raf = new RandomAccessFile(fileToSplit.toFile(), "r");
            FileChannel inputChannel = raf.getChannel()) {

        long fileSize = raf.length();
        long position = 0;
        int fileCounter = 1;

        while (position < fileSize) {
            // Calculate end position (try to get close to max size)
            long targetEndPosition = Math.min(position + maxSizePerFileInBytes, fileSize);

            // If we're not at the end of the file, find the last line ending before max size
            long endPosition = targetEndPosition;
            if (endPosition < fileSize) {
                endPosition = findLastLineEndBeforePosition(raf, position, targetEndPosition);
            }

            long chunkSize = endPosition - position;
            var outputFilePath = tempDir.resolve("_part" + fileCounter);
            try (FileOutputStream fos = new FileOutputStream(outputFilePath.toFile());
                    FileChannel outputChannel = fos.getChannel()) {
                inputChannel.transferTo(position, chunkSize, outputChannel);
            }

            position = endPosition;
            fileCounter++;
        }

    }
}

private long findLastLineEndBeforePosition(RandomAccessFile raf, long startPosition, long maxPosition)
        throws IOException {
    long originalPosition = raf.getFilePointer();

    try {
        int bufferSize = LINE_ENDING_SEARCH_WINDOW;
        long chunkSize = maxPosition - startPosition;

        if (chunkSize < bufferSize) {
            bufferSize = (int) chunkSize;
        }

        byte[] buffer = new byte[bufferSize];
        long searchPos = maxPosition;

        while (searchPos > startPosition) {
            long distanceToStart = searchPos - startPosition;
            int bytesToRead = (int) Math.min(bufferSize, distanceToStart);

            long readStartPos = searchPos - bytesToRead;
            raf.seek(readStartPos);

            int bytesRead = raf.read(buffer, 0, bytesToRead);
            if (bytesRead <= 0)
                break;

            // Search backwards through the buffer for newline
            for (int i = bytesRead - 1; i >= 0; i--) {
                if (buffer[i] == '\n') {
                    return readStartPos + i + 1;
                }
            }

            searchPos -= bytesRead;
        }

        throw new IllegalArgumentException(
                "File " + fileToSplit + " cannot be split. No newline found within the limits.");
    } finally {
        raf.seek(originalPosition);
    }
}

The findLastLineEndBeforePosition method, has certain limitations. Specifically, it only works on Unix-like systems (\n), very long lines can lead to numerous backward read iterations, and files containing lines exceeding maxSizePerFileInBytes cannot be split. However, it's well-suited for scenarios like splitting access log files, which typically have short lines and a high volume of entries.

Benchmarking and Performance Analysis

Theoretically our almost zero copy split file should be faster than the initial implementation, now it’s time to measure how fast it can be. For that, I’ve run some benchmarks for the two implementations, these are the results.

Benchmark                                                    Mode  Cnt           Score      Error   Units
FileSplitterBenchmark.splitFile                              avgt   15        1179.429 ±   54.271   ms/op
FileSplitterBenchmark.splitFile:·gc.alloc.rate               avgt   15        1349.613 ±   60.903  MB/sec
FileSplitterBenchmark.splitFile:·gc.alloc.rate.norm          avgt   15  1694927403.481 ± 6060.581    B/op
FileSplitterBenchmark.splitFile:·gc.count                    avgt   15         718.000             counts
FileSplitterBenchmark.splitFile:·gc.time                     avgt   15         317.000                 ms
FileSplitterBenchmark.splitFileZeroCopy                      avgt   15          77.352 ±    1.339   ms/op
FileSplitterBenchmark.splitFileZeroCopy:·gc.alloc.rate       avgt   15          23.759 ±    0.465  MB/sec
FileSplitterBenchmark.splitFileZeroCopy:·gc.alloc.rate.norm  avgt   15     2555608.877 ± 8644.153    B/op
FileSplitterBenchmark.splitFileZeroCopy:·gc.count            avgt   15          10.000             counts
FileSplitterBenchmark.splitFileZeroCopy:·gc.time             avgt   15           5.000                 ms

Following is the benchmark code and the file size used for the above results.

int maxSizePerFileInBytes = 1024 * 1024 // 1 MB chunks

@Setup(Level.Trial) public void setup() throws Exception {
    inputFile = Paths.get("/tmp/large_input.txt");
    outputDir = Paths.get("/tmp/split_output");
    // Create a large file for benchmarking if it doesn't exist
    if (!Files.exists(inputFile)) {
        try (BufferedWriter writer = Files.newBufferedWriter(inputFile)) {
            for (int i = 0; i < 10_000_000; i++) {
                writer.write("This is line number " + i);
                writer.newLine();
            }
        }
    }
}

@Benchmark public void splitFile() throws Exception {
    splitter.split(inputFile, outputDir);
}

@Benchmark public void splitFileZeroCopy() throws Exception {
    zeroCopySplitter.split(inputFile);
}

$ du -h /tmp/large_input.txt 
266M    /tmp/large_input.txt

Analyzing the results, the zeroCopySplitter demonstrates a considerable speedup, taking only 77 ms compared to the initial splitter's 1179 ms for this specific case. This performance advantage can be crucial when dealing with large volumes of data or many files.

Conclusion

Efficiently splitting large text files demands system-level performance considerations beyond mere logic. While a basic approach highlights the problem with excessive memory operations, a redesigned solution leveraging zero-copy techniques and preserving line integrity yields substantial performance improvements.

This demonstrates the impact of system-aware programming and understanding I/O mechanics in creating significantly faster and more resource-efficient tools for handling large textual data like logs or datasets.

Testcontainers MongoDB ReplicaSet

Juan Olivares — Thu, 24 Oct 2024 13:34:57 +0000

For Java applications using MongoDB, we can leverage Testcontainers to set up and manage our test environment. This tool allows us to create and configure lightweight, disposable containers for our tests, offering a more isolated and reliable testing experience.

Testcontainers provides a built-in MongoDB container that consists of a single-node replica set. This is convenient for simple tests. However, for scenarios requiring a multi-node MongoDB cluster, we encounter a challenge. While Testcontainers recommends an external library—https://github.com/silaev/mongodb-replica-set—this solution appears to be abandoned at the time of writing.

In this article, we'll explore how to create a simple multi-node MongoDB replica set using Testcontainers and discuss the challenges we'll face along the way.

The Challenges

Creating a multi-node MongoDB replica set looks like a straightforward task. We start several MongoDB instances using the parameter --replSet <replSetID>. Once all servers are running, we execute the rs.initiate(<rsConfig>) command on one of the servers. The rsConfig we pass to this command depends on the number of servers we've started. For a three-server replica set, it would be something as follows:

rs.initiate({
    "_id": "rs0",
    "members": [
        {"_id": 0, "host": "mongo1:27017"},
        {"_id": 1, "host": "mongo2:27017"},
        {"_id": 2, "host": "mongo3:27017"}
    ]
});

Challenge 1

In this example, we have three MongoDB servers started with the parameter--replSet rs0. Each server has a different DNS name:mongo1,mongo2, andmongo3, all listening on the default MongoDB port27017. When we run thers.initiate(<rsConfig>)command, MongoDB checks that the MongoDB address (host:port) of the service executing the command is part of the members list, it will also check that all address in the list are resolvable from that server. It won't allow the creation of the replica set if this precondition is not met.

This is the first thing we need to take into account when creating a multi node MongoDB replica set.

Challenge 2

Once we have our MongoDB replica set ready we can connect to it. When connecting to our MongoDB replica set we have to use a connection string like the following: mongodb://mongo1:27017,mongo2:27017,mongo3:27017/?replicaSet=rs0 . The driver using that connection string will pickup one of the servers in the list, connect to it and ask for the members list, once got the list, the driver will try to resolve the address of each member of the list. So the host connecting to the replica set must be able to resolve the addresses used to configure the replica set (hosts in members ).

This is the second thing we need to take into account when creating a multi node MongoDB replica set.

Approaching the problem using Docker

Docker custom netwok and aliases

Docker networking and aliases allow containers to communicate with each other within isolated networks, making it possible to create clusters of servers. Containers in the same network can communicate using their aliases as if they were hostnames, enabling easy inter-container communication for clustering. It will allow to overcome the first limitation we found when initiating the replica set.

However, it's important to note that the host network is typically not part of the custom network where the MongoDB containers reside. Consequently, when the driver running on the host attempts to connect to the replica set, it may fail to resolve the member list addresses due to its lack of access to the custom network.

Docker host network driver

Docker offers the option to allow containers to share the host's network, providing direct access to the host's network interfaces. While this feature eliminates the need for port mapping, it can lead to conflicts if multiple containers attempt to use the same port. To mitigate this risk, it's essential to carefully assign unique ports to each container. By doing so, you can minimize the likelihood of encountering port conflicts and ensure smooth container operation.

One significant advantage of using the host network is the ability for containers to communicate directly with each other and for the host to resolve their addresses. This can simplify network configuration and facilitate communication between containers and the host.

This feature is supported out of the box in Linux. For Mac/Windows, Docker Desktop supports it from version 4.34 and later, however it needs to be explicitly enabled.

Testing our solution

Before jumping into implementing our multi-node MongoDB replica set in java using Testcontainers, let’s use docker-compose to try to create a PoC using the Docker host network driver approach.

version: "3.8"

services:
  mongo1:
    container_name: mongo1
    image: mongo:6.0
    command: ["--replSet", "rs0", "--bind_ip_all", "--port", "27020"]
    restart: always
    network_mode: "host"
    healthcheck:
      test: echo "try { rs.status() } catch (err) { rs.initiate({_id:'rs0',members:[{_id:0,host:'localhost:27020',priority:1},{_id:1,host:'localhost:27021',priority:0.5},{_id:2,host:'localhost:27022',priority:0.5}]}) }" | mongosh --port 27020 --quiet
      interval: 5s
      timeout: 30s
      start_period: 0s
      start_interval: 1s
      retries: 30

  mongo2:
    image: mongo:6.0
    container_name: mongo2
    command: ["--replSet", "rs0" ,"--bind_ip_all", "--port", "27021"]
    restart: always
    network_mode: "host"

  mongo3:
    image: mongo:6.0
    container_name: mongo3
    command: ["--replSet", "rs0" ,"--bind_ip_all", "--port", "27022"]
    restart: always
    network_mode: "host"

Observe the use of network_mode: "host" for each container, which allows them to share the host's network. This configuration, in conjunction with the specified ports (--port 27020, --port 27021, --port 27022), enables the containers to communicate directly using their respective port numbers. These port assignments are reflected in the replica set configuration: rs.initiate({_id:'rs0',members:[{_id:0,host:'localhost:27020',priority:1},{_id:1,host:'localhost:27021',priority:0.5},{_id:2,host:'localhost:27022',priority:0.5}]})

Let’s start and test our MongoDB replica set using docker-compose

$ docker-compose up -d
[+] Running 3/3
 ✔ Container mongo1  Started                                                                                                                                                                                     0.1s
 ✔ Container mongo2  Started                                                                                                                                                                                     0.1s
 ✔ Container mongo3  Started

$ mongosh "mongodb://localhost:27020,localhost:27021,localhost:27022/?replicaSet=rs0"
Current Mongosh Log ID: 6719fb60b65eaef268d51880
Connecting to:      mongodb://localhost:27020,localhost:27021,localhost:27022/?replicaSet=rs0&serverSelectionTimeoutMS=2000&appName=mongosh+2.3.2
Using MongoDB:      6.0.18
Using Mongosh:      2.3.2

For mongosh info see: https://www.mongodb.com/docs/mongodb-shell/

------
   The server generated these startup warnings when booting
   2024-10-24T06:48:05.687+00:00: Using the XFS filesystem is strongly recommended with the WiredTiger storage engine. See http://dochub.mongodb.org/core/prodnotes-filesystem
   2024-10-24T06:48:06.371+00:00: Access control is not enabled for the database. Read and write access to data and configuration is unrestricted
   2024-10-24T06:48:06.371+00:00: /sys/kernel/mm/transparent_hugepage/enabled is 'always'. We suggest setting it to 'never' in this binary version
   2024-10-24T06:48:06.371+00:00: vm.max_map_count is too low
------

rs0 [primary] test>

The java code

Once we’ve proven our approach works using docker-compose let’s write the Testcontainers java version. It will use jbang to make prototyping easier.

///usr/bin/env jbang "$0" "$@" ; exit $?
//JAVA 21

//DEPS org.testcontainers:testcontainers:1.20.2
//DEPS org.slf4j:slf4j-simple:1.7.36

import static java.lang.String.format;

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;

import java.io.IOException;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class MongoDBReplicaSet {
    private static final String MONGO_IMAGE = "mongo:6.0.13";
    private static final String RS_NAME = "rs0";

    private List<PortAndContainer> nodes = new ArrayList<>();

    record PortAndContainer(int port, GenericContainer container) {}

    public MongoDBReplicaSet() { this(3);}

    public MongoDBReplicaSet(int n) {
        if (n < 1) {
            throw new IllegalArgumentException("At least one node is required");
        }

        for (int i = 0; i < n; i++) {
            var port = findAvailableTcpPort();
            var node = new GenericContainer(MONGO_IMAGE)//
                    .withNetworkMode("host")//
                    .waitingFor(Wait.forLogMessage(".*Waiting for connections.*\\n", 1))//
                    .withCommand("mongod", "--replSet", RS_NAME, "--bind_ip_all", "--port", String.valueOf(port));
            nodes.add(new PortAndContainer(port, node));
        }
    }

    public String getConnectionString() {
        return nodes.stream()//
                .map(node -> String.format("%s:%d", node.container().getHost(), node.port()))
                .collect(Collectors.joining(",", "mongodb://", "/?replicaSet=" + RS_NAME));
    }

    public void start() {
        nodes.forEach(node -> node.container().start());

        var replicaMembersCfg = nodes.stream()//
                .map(node -> format("{\"_id\": %d, \"host\": \"%s:%d\"}", nodes.indexOf(node),
                        node.container().getHost(), node.port()))
                .collect(Collectors.joining(","));
        var rsInitiate = format("rs.initiate({\"_id\": \"%s\", \"members\": [%s]})", RS_NAME, replicaMembersCfg);

        var node = nodes.get(0);
        var nodeUrl = node.container().getHost() + ":" + node.port();

        execInContainer(node.container(), "mongosh", nodeUrl + "/admin", "--quiet", "--eval", rsInitiate);
        execInContainer(node.container(), "/bin/bash", "-c",
                "until mongosh " + nodeUrl + "/admin"
                        + " --eval \"printjson(rs.isMaster())\" | grep ismaster | grep true > /dev/null 2>&1;"
                        + "do sleep 1;done");
    }

    private void execInContainer(GenericContainer container, String... command) {
        try {
            var result = container.execInContainer(command);
            if (result.getExitCode() != 0) {
                throw new RuntimeException(format("Failed execution of command. Code: %s, output: %s ",
                        result.getExitCode(), result.getStdout() + "\n" + result.getStderr()));
            }
        } catch (UnsupportedOperationException | IOException | InterruptedException e) {
            throw new RuntimeException("Failed to execute command", e);
        }
    }

    public void stop() {
        nodes.forEach(node -> node.container().stop());
    }

    private int findAvailableTcpPort() {
        try (ServerSocket socket = new ServerSocket(0)) {
            return socket.getLocalPort();
        } catch (IOException e) {
            throw new IllegalStateException("Failed to find available port", e);
        }
    }

    public static void main(String[] args) throws Exception {
        var replicaSet = new MongoDBReplicaSet();
        try {
            replicaSet.start();
            System.out.println("MongoDB Replica Set started. Connection string: " + replicaSet.getConnectionString());
            Thread.currentThread().join();
        } finally {
            replicaSet.stop();
        }
    }
}

In the code, we employed the withNetworkMode("host") configuration to allow the containers to utilize the host's network. Additionally, we dynamically determined an available port for each container using var port = findAvailableTcpPort(). This port was then incorporated into the server startup command as "--port", String.valueOf(port). Subsequently, when initiating the MongoDB replica set, we utilized these port values to populate the members list for each container.

Let’s start and test our Testcontainers MongoDB replica set using jbang .

$ jbang MongoDBReplicaSet.java

MongoDB Replica Set started. Connection string: mongodb://localhost:54303,localhost:54304,localhost:54305/?replicaSet=rs0

$ mongosh "mongodb://localhost:55039,localhost:55040,localhost:55041/?replicaSet=rs0"
Current Mongosh Log ID: 6719fbc79230eaae68149f3a
Connecting to:      mongodb://localhost:55039,localhost:55040,localhost:55041/?replicaSet=rs0&serverSelectionTimeoutMS=2000&appName=mongosh+2.3.2
Using MongoDB:      6.0.13
Using Mongosh:      2.3.2

For mongosh info see: https://www.mongodb.com/docs/mongodb-shell/

------
   The server generated these startup warnings when booting
   2024-10-24T07:47:35.720+00:00: Using the XFS filesystem is strongly recommended with the WiredTiger storage engine. See http://dochub.mongodb.org/core/prodnotes-filesystem
   2024-10-24T07:47:36.383+00:00: Access control is not enabled for the database. Read and write access to data and configuration is unrestricted
   2024-10-24T07:47:36.383+00:00: /sys/kernel/mm/transparent_hugepage/enabled is 'always'. We suggest setting it to 'never'
   2024-10-24T07:47:36.383+00:00: vm.max_map_count is too low
------

rs0 [primary] test>

Summing up

We have successfully demonstrated the feasibility of creating a Testcontainers multi-node MongoDB replica set using Docker host network driver approach. However, this is merely a foundational example. While the provided code may be suitable for certain use cases, it might require further refinement or customization to meet specific requirements.

References

https://java.testcontainers.org/modules/databases/mongodb/

https://github.com/silaev/mongodb-replica-set

https://medium.com/workleap/the-only-local-mongodb-replica-set-with-docker-compose-guide-youll-ever-need-2f0b74dd8384

Metrics Can Fool You: Measuring Execution Time in Connection-Pooled Environments

Juan Olivares — Wed, 14 Aug 2024 13:35:35 +0000

Measuring the execution time of requests to external services is critical for performance monitoring and optimization. However, when connections to these external services are pooled, you might be inadvertently measuring more than just the request time. Specifically, if requests are taking too long and you run out of available connections, your custom logic may start including the waiting time to get a connection from the pool. This can lead to misleading metrics, causing you to misinterpret the performance of your system. Let's delve into how this happens and how you can avoid being fooled by your own metrics.

The Pitfall: Including Waiting Time in Metrics

When all connections in the pool are in use, additional requests must wait until a connection becomes available. This waiting time can skew your metrics if not measured separately from the actual request time.

Scenario: Running Out of Connections

Initial State: Your connection pool has a fixed number of connections, all of which are in use.
New Request: A new request comes in but must wait for a connection to become available.
Wait Time: The request waits (possibly for a significant amount of time) until a connection is free.
Request Time: Once a connection is obtained, the actual request is made.

If your custom logic measures the total time from when the request was made until a response is received, you are including both the waiting time and the request time.

Practical Example: Reproducing the Problem in Spring Boot with Apache HttpClient 5

To illustrate how you can be fooled by your own metrics in a connection-pooled environment, let's walk through a practical example using Spring Boot and Apache HttpClient 5. We'll set up a simple Spring Boot application that makes HTTP requests to an external service, measure the execution time of these requests, and demonstrate how connection pool exhaustion can lead to misleading metrics.

To simulate delays in the external service, we will use the httpbin Docker image. Httpbin provides an easy-to-use HTTP request and response service, which we can use to create artificial delays in our requests.

@SpringBootApplication
@RestController
public class Server {

    public static void main(String... args) {
        SpringApplication.run(Server.class, args);
    }

    class TimeClientHttpRequestInterceptor implements ClientHttpRequestInterceptor {

        @Override
        public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)
                throws IOException {
            var t0 = System.currentTimeMillis();
            try {
                return execution.execute(request, body);
            } finally {
                System.out.println("Request took: " + (System.currentTimeMillis() - t0) + "ms");
            }
        }
    }

    @Bean
    public RestClient restClient() {
        var connectionManager = new PoolingHttpClientConnectionManager();
        connectionManager.setMaxTotal(2); // Max number of connections in the pool
        connectionManager.setDefaultMaxPerRoute(2); // Max number of connections per route

        return RestClient.builder()//
                .requestFactory(new HttpComponentsClientHttpRequestFactory(
                        HttpClients.custom().setConnectionManager(connectionManager).build()))
                .baseUrl("http://localhost:9091")//
                .requestInterceptor(new TimeClientHttpRequestInterceptor()).build();
    }

    @GetMapping("/")
    String hello() {
        return restClient().get().uri("/delay/2").retrieve().body(String.class);
    }
}

In the above code we created a request interceptor (ClientHttpRequestInterceptor) to measure what we thought would be the execution time of requests to the external service backed by httpbin.

We also explicitly set the pool to a very small size of 2 connections to make it easy to reproduce the problem.

Now we just need to start httpbin, run our spring boot app and conduct a simple test using ab

$ docker run -p 9091:80 kennethreitz/httpbin

ab -n 10 -c 4 http://localhost:8080/
...
Percentage of the requests served within a certain time (ms)
  50%   4049
  66%   4054
  75%   4055
  80%   4055
  90%   4057
  95%   4057
  98%   4057
  99%   4057
 100%   4057 (longest request)

Request took: 2021ms
Request took: 2016ms
Request took: 2022ms
Request took: 4040ms
Request took: 4047ms
Request took: 4030ms
Request took: 4037ms
Request took: 4043ms
Request took: 4050ms
Request took: 4034ms

If we look at the numbers, we can see that even though we set an artificial delay of 2 seconds for the external server, we're actually getting a delay of 4 seconds for most requests. Moreover, we notice that only the first requests honor the configured delay of 2 seconds, while subsequent requests result in a delay of 4 seconds.

Time to profile

Profiling is essential when encountering strange code behavior because it identifies performance bottlenecks, uncovers hidden issues like memory leaks, and shows how your application uses system resources.

This time we’ll profile the running app using JFR while conducting the ab load testing.

$ jcmd <pid> JFR.start name=app-profile  duration=60s filename=app-profile-$(date +%FT%H-%M-%S).jfr

$ ab -n 50 -c 4 http://localhost:8080/
...
Percentage of the requests served within a certain time (ms)
  50%   4043
  66%   4051
  75%   4057
  80%   4060
  90%   4066
  95%   4068
  98%   4077
  99%   4077
 100%   4077 (longest request)

If we open the JFR file and look at the flamegraph, we can see that most of the execution time is spent by our HTTP client. The client's execution time is divided between waiting for our external service to respond and waiting to get a connection from the pool.

That explains why the response times we see are double the expected fixed delay of 2 seconds we set for our external server. We configured a pool of 2 connections. However, in our test, we're performing 4 concurrent requests. So, only the first 2 requests will be served in the expected time of 2 seconds. Subsequent requests will have to wait for the pool to release a connection, thus increasing the observed response time.

If we look at the flamegraph again we can also find out why the time measured by our ClientHttpRequestInterceptor does not reflect the time the external server takes to respond but the time it takes to get a connection from the pool plus the time it takes to perform the actual request to the external server. Our interceptor is actually wrapping a stack trace that ends up calling a pool manager to get a connection: PoolingHttpClientConnectionManager

Monitoring the response time of any HTTP client is best done using its built-in metrics because these metrics are specifically designed to capture precise timing information. They account for all aspects of the HTTP request lifecycle, including connection acquisition, data transmission, and response handling. This ensures that the measurements are accurate and consistent with the actual performance of the client.

EC2 real network bandwidth

Juan Olivares — Fri, 31 May 2024 13:49:12 +0000

EC2 instances does not always provide exact numbers on their network capacity. In some cases where we need to optimize our network traffic we might need to know the exact numbers for our own instances.

iperf & Terraform

One way to achieve this is using iperf3. This tool uses a client-server model, which is ideal for testing network performance between two hosts. The server listens for incoming test requests, and the client initiates the test, making it easy to measure the capacity of the network link between them.

The next step involves setting up two EC2 instances. The first will act as a server, which will help us determine its actual network capacity. The second will act as a client. The client instance should have a higher network capacity than the server to ensure that the server's network link is fully utilized during the test.

To implement this, we need to create a pair of EC2 instances along with their corresponding resources, such as roles and security groups, in our AWS account. Doing this manually for every EC2 instance type we need to measure could be tedious, so we'll use Terraform for this task.

Following are the important parts of terraform code to setup this test environment to measure network capacity of m3.medium instances.

resource "aws_instance" "iperf_server" {
  ami           = data.aws_ami.amazon_linux_2.id
  instance_type = "m3.medium"

  subnet_id       = data.aws_subnet.current.id
  security_groups = [aws_security_group.iperf.id]

  iam_instance_profile = aws_iam_instance_profile.iperf.name

  tags = {
    Name = "${var.name}-iperf-server"
  }

  key_name = aws_key_pair.iperf.key_name
  connection {
    host        = coalesce(self.public_ip, self.private_ip)
    type        = "ssh"
    user        = var.ssh_user
    private_key = tls_private_key.iperf.private_key_pem
  }

  provisioner "remote-exec" {
    inline = [
      "sudo yum -y update",
      "sudo yum -y install iperf3"
    ]
  }
}

resource "aws_instance" "iperf_client" {
  ami           = data.aws_ami.amazon_linux_2.id
  instance_type = "m5.12xlarge"

  subnet_id       = data.aws_subnet.current.id
  security_groups = [aws_security_group.iperf.id]

  iam_instance_profile = aws_iam_instance_profile.iperf.name

  tags = {
    Name = "${var.name}-iperf-client"
  }

  key_name = aws_key_pair.iperf.key_name
  connection {
    host        = coalesce(self.public_ip, self.private_ip)
    type        = "ssh"
    user        = var.ssh_user
    private_key = tls_private_key.iperf.private_key_pem
  }

  provisioner "remote-exec" {
    inline = [
      "sudo yum -y update",
      "sudo yum -y install iperf3"
    ]
  }
}

resource "aws_security_group" "iperf" {

  name = "${var.name}-iperf-seg"
  description = "Allow inbound traffic for iperf"

  vpc_id = data.aws_vpc.current.id

  ingress {
    from_port = 0
    to_port   = 0
    protocol  = "-1"
    self      = true
  }

  ingress {
    from_port   = 5001  # Assuming iperf server listens on port 5001
    to_port     = 5001
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
  }

  ingress {
    description = "port 22"
    from_port   = 22
    to_port     = 22
    protocol    = "tcp"
    cidr_blocks = var.ssh_cidr_ingress_blocks
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }

  tags = {
    "Name" : "${var.name}-iperf-seg"
  }

}

output "server_private_ip" {
  value       = aws_instance.iperf_server.private_ip
}

output "server_public_ip" {
  value       = aws_instance.iperf_server.public_ip
}

output "client_public_ip" {
  value       = aws_instance.iperf_client.public_ip
}

Conducting the test

Once we have created our test env, we just need to ssh into the server instance and start iperf in server mode

$ iperf -s

Then we’ll ssh into the client and start the test attaching to the server using its private ip to reduce extra network overhead.

$ iperf -c <server_private_ip>

If we plan to test multiple instance types, we can create a script to run the tests as follows.

terraform_path=.

vars=$(terraform -chdir=$terraform_path output -json)
client_public_ip=$(echo $vars | jq -r '.client_public_ip.value')
server_public_ip=$(echo $vars | jq -r '.server_public_ip.value')
server_private_ip=$(echo $vars | jq -r '.server_private_ip.value')

keypair="$terraform_path/iperf-keypair.pem"

ssh -i $keypair \
    -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
    ec2-user@$server_public_ip "iperf3 -s -D"

ssh -i $keypair \
    -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
    ec2-user@$client_public_ip -t "iperf3 -c $server_private_ip -t 150 -i 60"

Real figures

I ran this test for some instance types. The results are as follows:

Instance Type	Bandwidth
m3.medium	681 Mbits/sec
c4.large	628 Mbits/sec
m4.large	768 Mbits/sec
r4.large	9.53 Gbits/sec
i3.large	4.97 Gbits/sec
t2.nano	1.04 Gbits/sec
t3.medium	4.52 Gbits/sec

Remember that burstable instances, such as the t2 and t3 families, incorporate the concept of network credits. These credits permit them to "burst" their network performance beyond the baseline level for short periods, as needed. Therefore, the figures in the table are not directly comparable between burstable and non-burstable instance types.

To establish the baseline network performance of burstable instances, we need to conduct the test for a duration sufficient to use up the accumulated network credits.

JVM Profiling in Action

Juan Olivares — Thu, 18 Apr 2024 09:14:02 +0000

Unlocking the full potential Java applications requires a deep understanding of their behavior and performance characteristics. JVM profiling serves as a powerful toolset for uncovering bottlenecks, identifying resource inefficiencies, and optimizing the performance of Java applications.

In this tutorial, We'll explore real-world code examples, iteratively applying optimizations based on the insights gained from profiling tools. Our goal is simple: to show how to maximize the efficiency, responsiveness, and scalability of Java applications through targeted performance enhancements.

The code

We'll be optimizing a piece of code that processes an access log containing JSON-formatted entries. Each entry in the log represents an access to a specific application, providing information such as the app_id and timestamp of the access.

The objective of the code is to calculate the number of accesses per second, offering valuable insights into the application's usage patterns and performance metrics.

class AccessLogAggregator {
    Map<OffsetDateTime, Long> accessBySecond(Path path) throws IOException {
        return Files.lines(path) //
                .map(this::parseLineJson) //
                .filter(AccessLogLine::isValid) //
                .collect(Collectors.groupingBy(a -> a.date.truncatedTo(ChronoUnit.SECONDS), Collectors.counting()));
    }

    AccessLogLine parseLineJson(String line) {
        var jsonLine = new org.json.JSONObject(line);
        var appID = jsonLine.getString("app_id");
        var date = OffsetDateTime.parse(jsonLine.getString("timestamp"));
        return new AccessLogLine(appID, date);
    }

    record AccessLogLine(String appID, OffsetDateTime date) {
        private static final Pattern APP_ID_VALID_REGEX = Pattern.compile("^[a-zA-Z0-9]+$");

        boolean isValid() {
            return APP_ID_VALID_REGEX.matcher(appID).matches();
        }
    }
}

As our site gains popularity, processing these files in parallel no longer suffices, as it still leads to longer processing times. Therefore, we must focus on optimizing this particular piece of code to ensure efficient handling of the increasing workload.

The profiling

Let's begin by measuring the current process time. To do this, we'll repeatedly process a single access log file, profiling the execution and measuring its execution time.

The profiling code will ran with the following specs:

openjdk version "21.0.1" 2023-10-17 LTS

OpenJDK Runtime Environment Temurin-21.0.1+12 (build 21.0.1+12-LTS)
OpenJDK 64-Bit Server VM Temurin-21.0.1+12 (build 21.0.1+12-LTS, mixed mode)

Chip: Apple M1 Pro
Total Number of Cores: 8 (6 performance and 2 efficiency)
Memory: 32 GB

We'll use async-profiler and flame graphs for profiling. To simplify the process, we'll run the code using JBang.

public static void main(String[] args) throws IOException {
    var path = Path.of("access.log");
    var aggregator = new AccessLogAggregator();

    var t0 = System.currentTimeMillis();
    for (int i = 0; i < 10; i++) {
        aggregator.accessBySecond(path);
    }
    System.out.println("Elapsed time: " + (System.currentTimeMillis() - t0) + " ms");
}

$ jbang --javaagent=ap-loader@jvm-profiling-tools/ap-loader=start,event=cpu,file=profile.html AccessLogAggregator.java

Profiling started
Elapsed time: 1810 ms

Now that we have established a baseline, let's review the generated flame graph to begin optimization.

Most of the time is spent in JSON parsing, specifically in AccessLogAggregator.parseLineJson. Within parsing, the majority of time is spent in org/json/JSONObject.<init>. Since there are faster JSON libraries available, we'll switch to Gson and repeat the tests. To do this, we'll add the following method to our code:

AccessLogLine parseLineGson(String line) {
    var jsonLine = JsonParser.parseString(line).getAsJsonObject();
    var appID = jsonLine.get("app_id").getAsString();
    var date = OffsetDateTime.parse(jsonLine.get("timestamp").getAsString());
    return new AccessLogLine(appID, date);
}

$ jbang --javaagent=ap-loader@jvm-profiling-tools/ap-loader=start,event=cpu,file=profile.html AccessLogAggregator.java

Profiling started
Elapsed time: 876 ms

We achieved a 2x performance improvement by switching JSON libraries. Now, let's examine our updated flame graph.

Now that the JSON parsing time has decreased, we see that timestamp parsing, specifically java/time/OffsetDateTime.parse, is taking a significant amount of the total parsing time. Java's built-in timestamp parsing is notoriously slow. Luckily, a faster alternative exists: https://github.com/ethlo/itu.

Let’s modify the code again replacing the timestamp parsing

AccessLogLine parseLineGson(String line) {
    var jsonLine = JsonParser.parseString(line).getAsJsonObject();
    var appID = jsonLine.get("app_id").getAsString();
    var date = ITU.parseDateTime(jsonLine.get("time").getAsString());
    return new AccessLogLine(appID, date);
}

$ jbang --javaagent=ap-loader@jvm-profiling-tools/ap-loader=start,event=cpu,file=profile.html AccessLogAggregator.java

Profiling started
Elapsed time: 651 ms

This time, we achieved a smaller but still important performance improvement of 1.35x, particularly when processing thousands of files. Let's examine the flame graph again.

It's clear that the proportion of total parsing attributed to timestamp parsing has decreased after its replacement.

Wrap up

After implementing the two improvements, we achieved an overall performance enhancement of 2.8 times compared to the initial code.

If you look at the code again you’ll notice that we validate each app_id to be alphanumeric. We employ a regex for this, which is typically known for its slow speed. Before profiling, we might have considered replacing regex validation with a manual method using Character.isLetterOrDigit. However, the flame graph reveals that this would have been a clear case of premature optimization because the validation's impact on the total time is minimal.

By using a regex, we maintain the flexibility to further modify the pattern for what we consider a valid app_id without having to modify our code.

In summary, profiling gave us valuable data-driven insights. This enabled us to optimize effectively, avoid unnecessary changes, measure performance improvements, and facilitate an iterative optimization process.

References

https://xam.dk/blog/jbang-reproducers/

https://someth2say.wordpress.com/2023/06/04/jbang-and-flame-graphs/

GraphQL Java Data Loader

Juan Olivares — Thu, 21 Dec 2023 10:23:24 +0000

Intro

GraphQL is a powerful and flexible query language for APIs that enables clients to request exactly the data they need, eliminating over-fetching and under-fetching of information. However, as GraphQL queries become more complex and involve multiple data sources, it can be challenging to efficiently retrieve and serve data to clients. This is where GraphQL data loaders come into play.

GraphQL data loaders are a critical component in optimizing GraphQL APIs, designed to tackle the notorious N+1 query problem, which occurs when a GraphQL server fetches the same data repeatedly for a list of related items. Data loaders help streamline the process of fetching data from various sources, such as databases, APIs, or even local caches, by batching and caching requests. By doing so, they significantly improve the efficiency and performance of GraphQL queries.

In this tutorial we will take a deep dive on the batching feature, for what we will explore how it does its magic having a look at the java implementation of the data loader.

Batching

Batching is the process of collecting multiple individual data retrieval requests into a single batch request, thus reducing the number of calls made to data sources. This is especially crucial when dealing with relationships in GraphQL queries.

Consider a typical scenario where a GraphQL query requests a list of items and, for each item, additional related data such as user information. Without batching, this would result in a separate database query or API request for each item, leading to the N+1 query problem. With batching, these individual requests can be efficiently combined into a single request, drastically reducing the number of round-trips to the data source

Java Data Loader Batching

Let’s say we have a graphql query like the below one

{
  user {
    name
    friends {
      name
    }
  }
}

It generates the following query result

{
  "user": {
    "name": "John",
    “friends”: [
      {
        "name": "Jane",
      },
      {
        "name": "Bob",
      },
      {
        "name": "Alice",
      }
    ]
  }
}

A naive implementation would perform a call to retrieve a user object for every user in the query response, i.e. 4 calls, one for the root object and one for each friend in the list.

However the DataLoader does not immediately perform the remote calls, it just enqueues the calls and returns a promise (CompletableFuture) to deliver a user object. Once we have enqueued all the calls that build our query result we must request the DataLoader to start executing them. At this point is where the magic happens. The DataLoader will start to extract the user id for each call and put it into a list which will be used to query the backend we configured and retrieve the user list using just one request.

The batching usually happens by levels, in this case we have 2 levels. The root user and his friends. By using the DataLoader batchig this response will require just 2 calls.

The code

Let’s put some code to show how it would be used.

First thing we need to have is a BatchLoader. It will load users from the user backend in batches, thus reducing the amount of API calls to that backend.

List<User> loadUsersById(List<Long> userIds) {
   System.out.println("Api call to load users = " + userIds);
   return users.stream().filter(u -> userIds.contains(u.id())).toList();
}

BatchLoader<Long, User> userBatchLoader = new BatchLoader<>() {
   @Override
   public CompletionStage<List<User>> load(List<Long> userIds) {
      return CompletableFuture.supplyAsync(() -> {
         return loadUsersById(userIds);
      });
   }
};

Then we need to create a DataLoader which will use the previous BatchLoader to perform the loading of the whole user tree.

var userLoader = DataLoaderFactory.newDataLoader(userBatchLoader);

var userDTO = new UserDTO();
userLoader.load(1L).thenAccept(user -> {
   userDTO.id = user.id();
   userDTO.name = user.name();
   user.friends().forEach(friendId -> {
      userLoader.load(friendId).thenAccept(friend -> {
         userDTO.friends.add(new FriendDTO(friend.id(), friend.name()));
      });
   });
});

userLoader.dispatchAndJoin();
System.out.println(userDTO);

It will produce the following debug output

Api call to load users = [1]
Api call to load users = [2, 3, 4]
UserDTO{id=1, name='John', friends=[FriendDTO[id=2, name=Jane], FriendDTO[id=3, name=Bob], FriendDTO[id=4, name=Alice]]}

If you are curious about how this internally works I will show you one custom implementation of the user DataLoader. Not the real one. Just one simplified version to help you in getting the whole picture.

static class UserLoader {
   BatchLoader<Long, User> userBatchLoader;

   record QueueEntry(long id, CompletableFuture<User> value) { }
   List<QueueEntry> loaderQueue = new ArrayList<>();

   UserLoader(BatchLoader<Long, User> userBatchLoader) {
      this.userBatchLoader = userBatchLoader;
   }

   CompletableFuture<User> load(long userId) {
      var future = new CompletableFuture<User>();
      loaderQueue.add(new QueueEntry(userId, future));
      return future;
   }

   List<User> dispatchAndJoin() {
      List<User> joinedResults = dispatch().join();
      List<User> results = new ArrayList<>(joinedResults);
      while (loaderQueue.size() > 0) {
         joinedResults = dispatch().join();
         results.addAll(joinedResults);
      }
      return results;
   }

   CompletableFuture<List<User>> dispatch() {
      var userIds = new ArrayList<Long>();
      final List<CompletableFuture<User>> queuedFutures = new ArrayList<>();

      loaderQueue.forEach(qe -> {
         userIds.add(qe.id());
         queuedFutures.add(qe.value());
      });

      loaderQueue.clear();

      var userFutures = userBatchLoader.load(userIds).toCompletableFuture();

      return userFutures.thenApply(users -> {
         for (int i = 0; i < queuedFutures.size(); i++) {
            var userId = userIds.get(i);
            var user = users.get(i);
            var future = queuedFutures.get(i);
            future.complete(user);
         }
         return users;
      });
   }
}

So, first look at CompletableFuture<User> load(long userId), it does not perform any userId lookup, it just:

Enqueues the lookup
Produces a CompletableFuture to let you chain further lookups based on the one you provided. So, the lookups are deferred until we actually request its execution using dispatchAndJoin()

Now, look at List<User> dispatchAndJoin(). That will be called once we are ready to retrieve the user list. It will:

Call CompletableFuture<List<User>> dispatch() which will perform the following actions:
1. Group all userIds into one list and send it to the underlying BatchLoader which performs the actual API call to the backend.
2. Complete the CompletableFuture that was provided when we registered the lookup (when we called CompletableFuture<User> load(long userId)), thus adding more elements to loaderQueue. At this point userId lookups for the next level got enqueued.
Repeat the process while there are elements remaining in loaderQueue.

References

https://www.graphql-java.com/documentation/batching/
https://github.com/graphql-java/java-dataloader

Benchmarking Java Zero-copy

Juan Olivares — Mon, 17 Jul 2023 09:53:15 +0000

The purpose of this document is to understand why zero-copy strategy performs better from an empirical point of view.

I’ll cover this with an example of some java code that merges large files into one single target file. For the merge code I will use 2 different approaches:

Using NIO API (zero copy)
Using IO API

To dig into the reason of the better performance of the zero copy I will benchmark these two approaches using jmh. By looking at the results I will point some numbers which shows why the zero copy approach performs better.

The Code

NIO

It will relay on the NIO API FileChannel#transfer which will use the syscall sendfile(). By using this call the kernel will take care of reading data from the source and writing data to the target without leaving the kernel space, thus the zero-copy.

@Benchmark
public void mergeNIO() throws Exception {
    var outFile = Files.createTempFile("benchmark", ".file");
    try (var out = FileChannel.open(outFile, CREATE, WRITE, DELETE_ON_CLOSE)) {
        for (var f : files) {
            try (FileChannel in = FileChannel.open(f.toPath(), READ)) {
                for (long p = 0, l = in.size(); p < l; ) {
                    p += in.transferTo(p, l - p, out);
                }
            }
        }
    }
}

IO

It will read data from one InputStream chunk by chunk into buffer and feed them into the OutputStream. Under the hood a read() syscall and a write() syscall will be performed as well as buffer copy from user space to kernel space and vice versa.

@Benchmark
public void mergeIO() throws Exception {
    var outFile = Files.createTempFile("benchmark", ".file");
    try (var out = Files.newOutputStream(outFile, CREATE, WRITE, DELETE_ON_CLOSE)) {
        for (var f : files) {
            try (var is = Files.newInputStream(f.toPath())) {
                byte[] buffer = new byte[16*1024];
                int read;
                while ((read = is.read(buffer, 0, 16*1024)) >= 0) {
                    out.write(buffer, 0, read);
                }
            }
        }
    }
}

Benchmark results

For the jmh execution I used the gradle plugin me.champeau.jmh with the following profilers configuration:

jmh {
    profilers = ["gc", "perf", "perfasm"]
}

NIO

Summary


Benchmark                                   Mode  Cnt       Score      Error   Units
FileBenchmark.mergeNIO                      avgt   25       3.733 ±  0.145       s/op
FileBenchmark.mergeNIO:·cpi                 avgt            0.872           clks/insn
FileBenchmark.mergeNIO:·gc.alloc.rate       avgt   25       0.002 ±  0.001     MB/sec
FileBenchmark.mergeNIO:·gc.alloc.rate.norm  avgt   25    7559.760 ± 44.878       B/op
FileBenchmark.mergeNIO:·gc.count            avgt   25         ≈ 0              counts
FileBenchmark.mergeNIO:·ipc                 avgt            1.147           insns/clk

Perf

Secondary result "org.example.FileBenchmark.mergeNIO:·perf":
Perf stats:
--------------------------------------------------

      56527,762660      task-clock (msec)         #    0,720 CPUs utilized          
             4.268      context-switches          #    0,076 K/sec                  
               378      cpu-migrations            #    0,007 K/sec                  
               385      page-faults               #    0,007 K/sec                  
    44.583.533.420      cycles                    #    0,789 GHz                      (30,73%)
    51.560.575.148      instructions              #    1,16  insn per cycle           (38,44%)
     9.135.306.805      branches                  #  161,607 M/sec                    (38,46%)
       110.707.654      branch-misses             #    1,21% of all branches          (38,45%)
    15.420.453.382      L1-dcache-loads           #  272,794 M/sec                    (24,72%)
     1.013.435.761      L1-dcache-load-misses     #    6,57% of all L1-dcache hits    (15,43%)
       283.399.957      LLC-loads                 #    5,013 M/sec                    (15,45%)
        98.132.015      LLC-load-misses           #   34,63% of all LL-cache hits     (23,11%)
   <not supported>      L1-icache-loads                                             
       303.670.304      L1-icache-load-misses                                         (30,80%)
    15.177.454.388      dTLB-loads                #  268,496 M/sec                    (29,93%)
         3.568.314      dTLB-load-misses          #    0,02% of all dTLB cache hits   (16,71%)
           453.513      iTLB-loads                #    0,008 M/sec                    (15,37%)
           195.070      iTLB-load-misses          #   43,01% of all iTLB cache hits   (23,03%)
   <not supported>      L1-dcache-prefetches                                        
   <not supported>      L1-dcache-prefetch-misses                                   

      78,491809508 seconds time elapsed

PerfASM


....[Hottest Methods (after inlining)]..........................................................
  98.59%   [kernel.kallsyms]  [unknown] 
   0.19%           libjvm.so  ElfSymbolTable::lookup 
   0.14%                      <unknown> 
   0.12%        libc-2.27.so  vfprintf 
   0.07%        libc-2.27.so  _IO_fwrite 
   0.04%        libc-2.27.so  _IO_default_xsputn 
   0.03%           libjvm.so  outputStream::do_vsnprintf_and_write_with_automatic_buffer 
   0.03%  libpthread-2.27.so  __libc_write 
   0.03%           libjvm.so  xmlStream::write_text 
   0.03%           libjvm.so  defaultStream::hold 
   0.02%           libjvm.so  stringStream::write 
   0.02%           libjvm.so  outputStream::update_position 
   0.02%        libc-2.27.so  syscall 
   0.02%           libjvm.so  defaultStream::write 
   0.02%          ld-2.27.so  __tls_get_addr 
   0.02%           libjvm.so  fileStream::write 
   0.02%        libc-2.27.so  vsnprintf 
   0.02%           libjvm.so  RelocIterator::initialize 
   0.02%        libc-2.27.so  [unknown] 
   0.02%           libjvm.so  outputStream::print 
   0.55%  <...other 207 warm methods...>
................................................................................................
 100.00%  <totals>

IO

Summary

Benchmark                                   Mode  Cnt       Score      Error   Units
FileBenchmark.mergeIO                       avgt   25       5.028 ±  0.183       s/op
FileBenchmark.mergeIO:·cpi                  avgt            0.966           clks/insn
FileBenchmark.mergeIO:·gc.alloc.rate        avgt   25       0.029 ±  0.001     MB/sec
FileBenchmark.mergeIO:·gc.alloc.rate.norm   avgt   25  154191.947 ± 51.517       B/op
FileBenchmark.mergeIO:·gc.count             avgt   25         ≈ 0              counts
FileBenchmark.mergeIO:·ipc                  avgt            1.035           insns/clk

Perf

Secondary result "org.example.FileBenchmark.mergeIO:·perf":
Perf stats:
--------------------------------------------------

      72972,795879      task-clock (msec)         #    0,772 CPUs utilized          
             4.971      context-switches          #    0,068 K/sec                  
               546      cpu-migrations            #    0,007 K/sec                  
               911      page-faults               #    0,012 K/sec                  
    57.575.878.331      cycles                    #    0,789 GHz                      (30,73%)
    60.626.232.521      instructions              #    1,05  insn per cycle           (38,43%)
    10.920.368.865      branches                  #  149,650 M/sec                    (38,40%)
       129.224.774      branch-misses             #    1,18% of all branches          (38,44%)
    18.526.168.088      L1-dcache-loads           #  253,878 M/sec                    (25,30%)
     2.885.163.789      L1-dcache-load-misses     #   15,57% of all L1-dcache hits    (17,75%)
       306.897.902      LLC-loads                 #    4,206 M/sec                    (17,32%)
       103.565.396      LLC-load-misses           #   33,75% of all LL-cache hits     (23,08%)
   <not supported>      L1-icache-loads                                             
       438.074.938      L1-icache-load-misses                                         (30,77%)
    18.537.122.526      dTLB-loads                #  254,028 M/sec                    (26,65%)
         8.112.288      dTLB-load-misses          #    0,04% of all dTLB cache hits   (18,35%)
         2.404.472      iTLB-loads                #    0,033 M/sec                    (15,39%)
         8.051.716      iTLB-load-misses          #  334,86% of all iTLB cache hits   (23,05%)
   <not supported>      L1-dcache-prefetches                                        
   <not supported>      L1-dcache-prefetch-misses                                   

      94,524422154 seconds time elapsed

PerfASM

....[Hottest Methods (after inlining)]..........................................................
  88.57%   [kernel.kallsyms]  [unknown] 
   6.70%        runtime stub  StubRoutines::jlong_disjoint_arraycopy 
   0.97%         c2, level 4  java.nio.channels.Channels::writeFully, version 2, compile id 697 
   0.92%         c2, level 4  sun.nio.ch.ChannelInputStream::read, version 2, compile id 706 
   0.24%    Unknown, level 0  sun.nio.ch.NativeThread::current, version 1, compile id 617 
   0.23%    Unknown, level 0  sun.nio.ch.FileDispatcherImpl::write0, version 1, compile id 656 
   0.22%         c2, level 4  org.example.FileBenchmark::mergeIO, version 4, compile id 725 
   0.19%                      <unknown> 
   0.18%  libpthread-2.27.so  __libc_write 
   0.18%  libpthread-2.27.so  __pthread_disable_asynccancel 
   0.15%  libpthread-2.27.so  __pthread_enable_asynccancel 
   0.13%           libjvm.so  ElfSymbolTable::lookup 
   0.13%  libpthread-2.27.so  __libc_read 
   0.11%    Unknown, level 0  sun.nio.ch.FileDispatcherImpl::read0, version 1, compile id 654 
   0.09%        libc-2.27.so  vfprintf 
   0.08%           libnio.so  fdval 
   0.06%           libnio.so  Java_sun_nio_ch_FileDispatcherImpl_write0 
   0.04%        libc-2.27.so  _IO_fwrite 
   0.03%        libc-2.27.so  _IO_default_xsputn 
   0.03%           libjvm.so  xmlStream::write_text 
   0.76%  <...other 201 warm methods...>
................................................................................................
 100.00%  <totals>

Conclusion

If we look at both summary reports we can see that the NIO approach is faster: 3.733 s/op vs 5.028 s/op and it requires barely no heap allocations (0.002 MB/sec vs 0.029 MB/sec) compared to the IO approach.

By having a look at the perf and perfASM reports we can see why NIO is faster.

It incurs fewer page faults 385 vs 911 due to the lack of read() syscall
It doesn't spend time in array copy operations (StubRoutines::jlong_disjoint_arraycopy) thus honoring the zero-copy name 😄

References

Java GC Pressure In Action

Juan Olivares — Tue, 20 Jun 2023 09:29:39 +0000

The purpose of this document is to show how GC pressure must be taken into account when designing java batch systems that move significant amounts of data.

When moving data from source to target, we might think intuitively that creating bigger batches of data might be faster because it will reduce the total number of iterations we must perform. It can be true until a certain point, however once we reach some threshold we will start having dismissing returns because of GC pressure.

I’ll cover this with an example of some java code that generates a given number of lines in batches and writes them to a file. I'll run the code using different parameters to get execution times, and will look at GC details for each run.

The code

I will focus on the core part of the code for simplicity.

public void test(long size, int batch) throws Exception {
    validate(size, batch);

    var total = 0;
    var temp = Files.createTempFile("benchmark", ".file");

    long t0 = System.nanoTime();

    try (var bw = new BufferedWriter(Files.newBufferedWriter(temp, StandardOpenOption.DELETE_ON_CLOSE), 8 * 1024)) {
        do {
            var lines = generate(batch);
            for (var l : lines) {
                bw.write(l);
                bw.newLine();
            }
            total += lines.size();
        } while (total < size);
    }

    long t1 = System.nanoTime();
    System.out.println("batch: " + batch + ", total: " + total + ", time: " + (t1 - t0) / 1_000000);
}

private void validate(long size, int batch) { /* Do validation */ }

private List<String> generate(int n) {
    var list = new ArrayList<String>();
    for (int i = 0; i < n; i++) {
        list.add("item" + UUID.randomUUID().toString()); // 40 bytes
    }
    return list;
}

JVM & machine specs

The application ran with the following specs:

openjdk version "17.0.5" 2022-10-18
OpenJDK Runtime Environment GraalVM CE 22.3.0 (build 17.0.5+8-jvmci-22.3-b08)

Chip: Apple M1 Pro
Total Number of Cores: 8 (6 performance and 2 efficiency)
Memory: 32 GB

First execution (Big batch size)

Total size: 30000000
Batch size: 100000
Iterations: 30000000/100000=300

$ java -Xmx3g -verbose:gc Benchmark.java 30000000 100000
[0.054s][info][gc] Using G1
[0.445s][info][gc] GC(0) Pause Young (Normal) (G1 Evacuation Pause) 25M->7M(516M) 2.760ms
[0.595s][info][gc] GC(1) Pause Young (Normal) (G1 Evacuation Pause) 37M->14M(516M) 3.502ms
[0.704s][info][gc] GC(2) Pause Young (Normal) (G1 Evacuation Pause) 70M->16M(516M) 2.144ms
[1.164s][info][gc] GC(3) Pause Young (Normal) (G1 Evacuation Pause) 314M->11M(516M) 0.734ms
[1.633s][info][gc] GC(4) Pause Young (Normal) (G1 Evacuation Pause) 315M->15M(516M) 1.584ms
[2.087s][info][gc] GC(5) Pause Young (Normal) (G1 Evacuation Pause) 315M->11M(516M) 0.661ms
[2.540s][info][gc] GC(6) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.442ms
[2.998s][info][gc] GC(7) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.612ms
[3.457s][info][gc] GC(8) Pause Young (Normal) (G1 Evacuation Pause) 315M->15M(516M) 1.281ms
[3.928s][info][gc] GC(9) Pause Young (Normal) (G1 Evacuation Pause) 315M->11M(516M) 0.663ms
[4.388s][info][gc] GC(10) Pause Young (Normal) (G1 Evacuation Pause) 315M->15M(516M) 1.311ms
[4.844s][info][gc] GC(11) Pause Young (Normal) (G1 Evacuation Pause) 315M->11M(516M) 0.675ms
[5.301s][info][gc] GC(12) Pause Young (Normal) (G1 Evacuation Pause) 315M->15M(516M) 1.333ms
[5.759s][info][gc] GC(13) Pause Young (Normal) (G1 Evacuation Pause) 315M->11M(516M) 0.642ms
[6.216s][info][gc] GC(14) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.346ms
[6.671s][info][gc] GC(15) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.723ms
[7.133s][info][gc] GC(16) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.533ms
[7.594s][info][gc] GC(17) Pause Young (Normal) (G1 Evacuation Pause) 314M->11M(516M) 0.609ms
[8.054s][info][gc] GC(18) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.364ms
[8.512s][info][gc] GC(19) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.747ms
[8.975s][info][gc] GC(20) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.403ms
[9.439s][info][gc] GC(21) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.686ms
[9.897s][info][gc] GC(22) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.318ms
[10.355s][info][gc] GC(23) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.754ms
[10.811s][info][gc] GC(24) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.298ms
[11.269s][info][gc] GC(25) Pause Young (Normal) (G1 Evacuation Pause) 314M->11M(516M) 0.634ms
[11.723s][info][gc] GC(26) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.300ms
[12.178s][info][gc] GC(27) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.597ms
[12.637s][info][gc] GC(28) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.331ms
[13.091s][info][gc] GC(29) Pause Young (Normal) (G1 Evacuation Pause) 316M->11M(516M) 0.636ms
[13.553s][info][gc] GC(30) Pause Young (Normal) (G1 Evacuation Pause) 315M->16M(516M) 1.376ms
batch: 100000, total: 30000000, time: 13195

The whole execution took 13195ms, while the sum of all GC pauses took 36.999ms

First execution (Small batch size)

Total size: 30000000
Batch size: 1000
Iterations: 30000000/1000=30000

$ java -Xmx3g -verbose:gc Benchmark.java 30000000 1000
[0.004s][info][gc] Using G1
[0.251s][info][gc] GC(0) Pause Young (Normal) (G1 Evacuation Pause) 25M->7M(516M) 2.864ms
[0.410s][info][gc] GC(1) Pause Young (Normal) (G1 Evacuation Pause) 37M->8M(516M) 2.859ms
[0.546s][info][gc] GC(2) Pause Young (Normal) (G1 Evacuation Pause) 88M->8M(516M) 7.700ms
[0.994s][info][gc] GC(3) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.932ms
[1.446s][info][gc] GC(4) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.987ms
[1.906s][info][gc] GC(5) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 1.007ms
[2.350s][info][gc] GC(6) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 1.145ms
[2.795s][info][gc] GC(7) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.878ms
[3.236s][info][gc] GC(8) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.911ms
[3.678s][info][gc] GC(9) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.920ms
[4.124s][info][gc] GC(10) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.937ms
[4.573s][info][gc] GC(11) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.904ms
[5.021s][info][gc] GC(12) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.941ms
[5.464s][info][gc] GC(13) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.990ms
[5.909s][info][gc] GC(14) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.981ms
[6.353s][info][gc] GC(15) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 1.009ms
[6.798s][info][gc] GC(16) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 1.251ms
[7.247s][info][gc] GC(17) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.508ms
[7.693s][info][gc] GC(18) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.601ms
[8.142s][info][gc] GC(19) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.539ms
[8.591s][info][gc] GC(20) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.490ms
[9.044s][info][gc] GC(21) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.609ms
[9.489s][info][gc] GC(22) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.575ms
[9.938s][info][gc] GC(23) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.525ms
[10.388s][info][gc] GC(24) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.614ms
[10.835s][info][gc] GC(25) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.501ms
[11.284s][info][gc] GC(26) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.628ms
[11.736s][info][gc] GC(27) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.522ms
[12.189s][info][gc] GC(28) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.488ms
[12.635s][info][gc] GC(29) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.559ms
[13.078s][info][gc] GC(30) Pause Young (Normal) (G1 Evacuation Pause) 312M->8M(516M) 0.550ms
batch: 1000, total: 30000000, time: 12876

The whole execution took 12876ms, while the sum of all GC pauses took 34.925ms

Comparing numbers

Surprisingly, the execution using a bigger batch size took longer than the one using a smaller batch size: 13195ms vs 12876ms. If we have a look at the sum of GC pauses we can observe that the execution using a bigger batch size took longer too: 36.999ms vs 34.925ms.

Heap details

We might be wondering what’s the difference in heap terms. To answer that question we can re-run the executions enabling JFR recording. To be able to register TLAB events we'll extend the default profile for JFR recording resulting in the following diff:

diff -u $JAVA_HOME/lib/jfr/default.jfc tlab_profile.jfc
@@ -640,12 +640,12 @@
     </event>

     <event name="jdk.ObjectAllocationInNewTLAB">
-      <setting name="enabled" control="gc-enabled-high">false</setting>
+      <setting name="enabled" control="gc-enabled-high">true</setting>
       <setting name="stackTrace">true</setting>
     </event>

     <event name="jdk.ObjectAllocationOutsideTLAB">
-      <setting name="enabled" control="gc-enabled-high">false</setting>
+      <setting name="enabled" control="gc-enabled-high">true</setting>
       <setting name="stackTrace">true</setting>
     </event>

$ java -Xmx3g -XX:StartFlightRecording=filename=flight1.jfr,settings=tlab_profile.jfc Benchmark.java 30000000 100000

$ java -Xmx3g -XX:StartFlightRecording=filename=flight2.jfr,settings=tlab_profile.jfc Benchmark.java 30000000 1000

After getting the two recordings we observe that the main difference lies on allocations outside of TLAB. These allocations are more expensive than allocations in TLAB and can make a difference adding a considerable penalty to our execution time. In this case we observe a difference in Total GC Time of 53,1ms - 30,5ms = 22,6ms.

The issue here is we’re creating big objects (array of strings) because of the batch size. So once the batch is bigger than what can be allocated inside a TLAB we start seeing allocations outside of TLAB which will add pressure to the heap making our execution slower.

References

Apache Curator Use Case

Juan Olivares — Tue, 16 May 2023 06:05:45 +0000

The purpose of this document is to explain how Apache Curator can be utilized for distributed leader election. Its application is aimed at addressing the need for high availability in the processing of a large dataset.

It will be useful to anyone who’s into trying to understand how to implement a solution for distributed leader election.

The problem

Currently, we have a dataset divided into partitions, and it is being processed by a single node. To enhance the system's availability, we intend to introduce a dynamic group of nodes responsible for processing the dataset. Each node will have the capability to exclusively process any given partition. In other words, only one node will be assigned to process a specific partition at any given time.

In order to handle increased workload, we have the flexibility to add or remove nodes from the pool as needed. This means that when nodes are added or removed, the partition workers must be redistributed accordingly. However, this redistribution can result in an imbalance of workload. To address this, we introduce the concept of a preferred node for each partition. This means that each partition will run on its preferred node whenever that node is available, helping to maintain a more balanced workload distribution.

Each node will maintain a full set of partition workers, which can exist in two states: active or latent. Only active workers perform actual work, and among the nodes, only one copy of each worker can be active. When a new node is added, it will initialize the entire partition worker set in the latent state. If any of these partition workers have the new node as their preferred node, they will eventually become active on that node while becoming latent on their former node. When a node is removed, all the active partition workers it was hosting will transition to an active state on other nodes.

The solution

To achieve this we’ll be using Apache Curator. For the active/latent status we’ll leverage the LeaderSelector recipe which will allow us to have exclusive access to a partition. An active partition worker will be the leader process, while latent partition workers will be the processes waiting to become leaders.

public void start() {
    this.selector = new LeaderSelector(client, "/partitions/" + partition.getID(),
            new LeaderSelectorListenerAdapter() {
                @Override
                public void takeLeadership(CuratorFramework client) throws Exception {
                    try {
                        LOGGER.info("Node {} taking leadership of partition {}", node.getID(), partition.getID());
                        job.star(); // blocking call
                        LOGGER.info("Node {} gracefully finishing leadership of partition {}", node.getID(),
                                partition.getID());
                    } catch (Exception e) {
                        LOGGER.error(
                                "Error running worker process in node " + node.getID() + " as leader of partition "
                                        + partition.getID(), e);
                        throw e;
                    } finally {
                        relinquish();
                    }
                }
            });

    client.getCuratorListenable().addListener((c, event) -> {
        if (event.getType() == CuratorEventType.CLOSING) {
            relinquish();
        }
    });

    selector.autoRequeue();
    selector.setId(node.getID());
    selector.start();

    LOGGER.info("Node {} started worker for partition {}", node.getID(), partition.getID());
}

We’ll have a scheduled process periodically checking for every partition worker if the current node is the preferred node for the partition. In case of not being the preferred one the worker will relinquish its leadership in that node. For this part we’ll make use of LeaderSelector participants to know what nodes are alive.

public boolean isLeaderNotPreferred() {
    return isLeader() && isNotPreferred();
}

public boolean isLeader() {
    return selector.hasLeadership();
}

private boolean isNotPreferred() {
    return partition.getPreferredNodeID().map(preferredNodeID -> {
        return !preferredNodeID.equals(this.node.getID()) && isNodeAlive(preferredNodeID);
    }).orElse(false);
}

private boolean isNodeAlive(String nodeID) {
    return getAliveNodeIDs().contains(nodeID);
}

public Set<String> getAliveNodeIDs() {
    try {
        return selector.getParticipants().stream()//
                .map(Participant::getId)//
                .collect(Collectors.toSet());
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

This design ensures high availability. The system is able to respond to node failures, as well as, scale up/down if required by adding/removing nodes. The upper limit is the number of partitions. OTOH it requires some manual tweaking of the partition's preferred node to balance load among nodes.

Oplog Update Version ($v) Digest

Juan Olivares — Fri, 05 May 2023 09:24:21 +0000

Oplog Update Version ($v) Digest

This document describes how the MongoDB oplog update version ($v) works. It will be useful to anyone who’s into trying to understand the internal parts of the MongoDB oplog format.

The MongoDB oplog is an internal part of the MongoDB replication system. It is also used for other purposes, such as creating incremental backups or replaying changes to data in order to perform point-in-time recovery.

What role does the $v field play into the oplog entries

This field was first introduced in Mongo 3.6, and it’s placed under the o field. It represents the version of the oplog format used for updates. Initially the only value of this field was 1, and nothing else changed. It kept defining the operations in terms of $set/$unset.

{
    "op": "u",
    "ns": "db.collection",
    "o": {
        "$v": 1,
        "$set": {
            "foo": "bar"
        }
    },
    "o2": {
        "_id": "xxxxxxxxxxxxxxxxxxxxxxxx"
    },
    "v": 2
}

In Mongo 4.6, the $v field started using the value 2. It resulted in a change in the format used to represent the operations. A new diff field was added. This field is a recursive structure made of the following optional fields:

u: Top level update (old $set)
d: Top level unset (old $unset)
i: Nested field update. (old $set of the nested field)
s-fields (fields starting by s) they can either be:
- array operations; or
- nested field operations

{
    "op": "u",
    "ns": "db.collection",
    "o": {
        "$v": 2,
        "diff": {
            "u": {
                "foo": "bar"
            }
        }
    },
    "o2": {
        "_id": "xxxxxxxxxxxxxxxxxxxxxxxx"
    },
    "v": 2
}

Examples of $v:2 operations to $v:1 operations

// u
{ "diff": { "u": { "foo": "2", "bar": "xx" } } }
{ "$set": { "foo": "2", "bar": "xx" } }


// d
{ "diff": { "d": { "foo": false } } }
{ "$unset": { "foo": true } }


// i
{ "diff": {"i": { "foo": { "bar": 2 } } } }
{ "$set": { "foo.bar": 2 } }


// nested field s+i
{ "diff": { "sa": { "i": { "foo": 3, "bar": 1 } } } }
{ "$set": { "a.foo": 3, "a.bar": 1 } }


// nested field s+d
{ "diff": { "sa": { "d": { "foo": false } } } }
{ "$unset": { "a.foo": true } }


// array field u
{ "diff": { "sfoo": { "a": true, "u0": 2 } } }
{ "$set": { "foo.0": 2 } }


// array field d
{ "diff": { "sfoo": { "a": true, "u0": null } } }
{ "$unset": { "foo.0": true } }

References

https://jira.mongodb.org/browse/SERVER-32240
https://github.com/meteor/meteor/blob/master/packages/mongo/oplog_v2_converter.js

MongoDB Sharding Digest

Juan Olivares — Tue, 28 Mar 2023 08:25:51 +0000

This document describes how sharding internally works in MongoDB. It will be useful to anyone who’s into trying to understand the internal parts of the MongoDB sharding.

This same information can be found in the MongoDB official documentation arranged in a different way. I just crafted it in a way that helped me into trying to understand how MongoDB sharding internally works.

How is data sharded

MongoDB shards data at the collection level, distributing the collection data across the shards in the cluster.

MongoDB uses the shard key to distribute the collection's documents across shards. The shard key consists of a field or multiple fields in the documents.

MongoDB partitions sharded data into chunks. Each chunk has an inclusive lower and exclusive upper range based on the shard key.

Sharding Strategies

MongoDB supports two sharding strategies:

Hashed sharding: it involves computing a hash of the shard key field's value. Each chunk is then assigned a range based on the hashed shard key values.
Range sharding: it involves dividing data into ranges based on the shard key values. Each chunk is then assigned a range based on the shard key values.

How is data evenly distributed

The MongoDB balancer is a background process that monitors the amount of data on each shard for each sharded collection. When the amount of data for a sharded collection on a given shard reaches specific migration thresholds, the balancer attempts to automatically migrate data between shards and reach an even amount of data per shard.

The server schedules an auto-split if:

It detected that the chunk exceeds a threshold based on the maximum chunk size.
There is not already a split in progress for the chunk.
Every time an update or insert gets routed to a chunk, the server tracks the bytes written to the chunk in memory through the collection's ChunkManager.

Note that because the balancer keeps distribution status per collection it optimizes for an even distribution per collection rather than for the entire cluster.

How are requests routed to the correct shard

The Config servers store the metadata for a sharded cluster. The metadata reflects state and organization for all data and components within the sharded cluster. The metadata includes the list of chunks on every shard and the ranges that define the chunks.

The config database contains the collections that contain the sharded cluster metadata. MongoDB writes data to the config database when the metadata changes, such as after a chunk migration or a chunk split.

The config.chunks collection stores a document for each chunk in the cluster.
The config.collections collection stores a document for each sharded collection in the cluster.
The config.databases collection stores a document for each database in the cluster. Each database in a sharded cluster has a primary shard that holds all the un-sharded collections for that database. Each database has its own primary shard.
The config.shards collection represents each shard in the cluster in a separate document,

The mongos instances cache this data and use it to route read and write operations to the correct shards. mongos updates the cache when there are metadata changes for the cluster, such as adding a shard. Shards also read chunk metadata from the config servers.
The cache may or may not be up-to-date with its corresponding source.

The routing table cache is “lazy.” It does not refresh from its source unless necessary.

Operations that change a collection’s routing information (for example, a moveChunk command that updates a chunk’s location) will mark the local node’s routing table cache as “stale” for affected shards. Subsequent attempts to access routing information for affected shards will block on a routing table cache refresh.