Performance Tuning

Transparent Huge Pages (THP) is the Linux Memory Management feature which aims to improve the application performance by using the larger memory pages. In most of the cases it works fine but for databases and in-memory data grids it usually causes a significant performance drop. Since it’s enabled on most of the Linux distributions, we do recommend disabling it when you run Hazelcast.

Use the following command to check if it’s enabled:

Or an alternative command if you run RHEL:

To disable it permanently, please see the corresponding documentation for the Linux distribution that you use. Here is an example of the instructions for RHEL: https://access.redhat.com/solutions/46111.

Swapping behavior can be configured by setting the kernel parameter (/proc/sys/vm/swappiness) and can be turned off completely by executing swapoff -a as the root user in Linux systems. We highly recommend turning off the swapping on the machines that run Hazelcast. When your operating systems start swapping, garbage collection activities take much longer due to the low speed of disc access.

The Linux kernel parameter, vm.swappiness, is a value from 0-100 that controls the swapping of application data from physical memory to virtual memory on disk. To prevent Linux kernel to start swapping memory to disk way too early, we need to set the default of 60 to value between 0 and 10. The higher the parameter value, the more aggressively inactive processes are swapped out from physical memory. The lower the value, the less they are swapped, forcing filesystem buffers to be emptied. In case swapping needs to be kept enabled, we recommend setting the value between 0 and 10 to prevent the Linux kernel to start swapping memory to disk way too early.

Provisioning a dedicated physical network interface controller (NIC) for Hazelcast members ensures smooth flow of data, including business data and cluster health checks, across servers. Sharing network interfaces between a Hazelcast member and another application could result in choking the port, thus causing unpredictable cluster behavior.

TCP uses a congestion window to determine how many packets it can send at one time; the larger the congestion window, the higher the throughput. The maximum congestion window is related to the amount of buffer space that the kernel allocates for each socket. For each socket, there is a default value for the buffer size, which you can change by using a system library call just before opening the socket. You can adjust the buffer sizes for both the receiving and sending sides of a socket.

To achieve maximum throughput, it is critical to use the optimal TCP socket buffer sizes for the links you are using to transmit data. If the buffers are too small, the TCP congestion window will never open up fully, therefore throttling the sender. If the buffers are too large, the sender can overrun the receiver such that the sending host is faster than the receiving host, which causes the receiver to drop packets and the TCP congestion window to shut down.

Hazelcast, by default, configures I/O buffers to 128KB; you can change these using the following properties:

Typically, you can determing the throughput by the following formulae:

To increase the TCP maximum buffer size in Linux, use the following properties:

To increase TCP auto-tuning by Linux, see the following properties:

Keeping track of garbage collection (GC) statistics is vital to optimum performance, especially if you run the JVM with large heap sizes. Tuning the garbage collector for your use case is often a critical performance practice prior to deployment. Likewise, knowing what baseline GC behavior looks like and monitoring for behavior outside of normal tolerances will keep you aware of potential memory leaks and other pathological memory usage.

The best way to minimize the performance impact of GC is to keep heap usage small. Maintaining a small heap saves countless hours of GC tuning and provides improved stability and predictability across your entire application. Even if your application uses very large amounts of data, you can still keep your heap small by using Hazelcast’s High-Density Memory Store. Some common off-the-shelf GC tuning parameters for Hotspot and OpenJDK are as follows:

To enable GC logging, use the following JVM arguments for Java 8:

The above arguments only work for Java 8. For Java 9+, you can use the following arguments:

Azul Systems, the industry’s only company exclusively focused on Java and the Java Virtual Machine (JVM), builds fully supported, certified standards-compliant Java runtime solutions that help enabling real-time business. Zing is a JVM designed for enterprise Java applications and workloads that require any combination of low latency, high transaction rates, large working memory, and/or consistent response times. Zulu and Zulu Enterprise are Azul’s certified, freely available open source builds of OpenJDK with a variety of flexible support options, available in configurations for the enterprise as well as custom and embedded systems. Azul Zing is certified and supported in Hazelcast Enterprise. When deployed with Zing, Hazelcast gains performance, capacity, and operational efficiency within the same infrastructure. Additionally, you can directly use Hazelcast with Zulu without making any changes to your code.

For queries on fields with ranges, you can use an ordered index. Hazelcast, by default, caches the deserialized form of the object under query in the memory when inserted into an index. This removes the overhead of object deserialization per query, at the cost of increased heap usage. See the Indexing Ranged Queries section.

Composite indexes are built on top of multiple map entry attributes; thus, increase the performance of complex queries significantly when used correctly. See the Composite Indexes section

Setting the hazelcast.query.predicate.parallel.evaluation property to true can speed up queries when using slow predicates or when there are huge amount of entries per member.

If you’re using queries heavily, you can benefit from increasing query thread pools. See the Configuring the Query Thread Pool section.

Setting the queried entries' in-memory format to OBJECT forces the objects to be always kept in object format, resulting in faster access for queries, but also in higher heap usage. It will also incur an object serialization step on every remote get operation. See the Setting In-Memory Format section.

The Portable interface allows individual fields to be accessed without the overhead of deserialization or reflection and supports query and indexing support without full-object deserialization. See the related Hazelcast Blog and the Portable Serialization section.

See also the Serialization Configuration Wrap-Up section for details.

An executor queue may be configured to have a specific number of threads dedicated to executing enqueued tasks. Set the number of threads (pool-size property in the executor service configuration) appropriate to the number of cores available for execution. Too few threads will reduce parallelism, leaving cores idle, while too many threads will cause context switching overhead. See the Configuring Executor Service section.

An executor queue may be configured to have a maximum number of tasks (queue-capacity property in the executor service configuration). Setting a bound on the number of enqueued tasks will put explicit back pressure on enqueuing clients by throwing an exception when the queue is full. This will avoid the overhead of enqueuing a task only for it to be canceled because its execution takes too long. It will also allow enqueuing clients to take corrective action rather than blindly filling up work queues with tasks faster than they can be executed. See the Configuring Executor Service section.

Any time spent blocking or waiting in a running task is thread execution time wasted while other tasks wait in the queue. Tasks should be written such that they perform no potentially blocking operations (e.g., network or disk I/O) in their run() or call() methods.

By default, tasks may be executed on any member. Ideally, however, tasks should be executed on the same machine that contains the data the task requires to avoid the overhead of moving remote data to the local execution context. Hazelcast executor service provides a number of mechanisms for optimizing locality of reference.

If you find that your work queues consistently reach their maximum and you have already optimized the number of threads and locality of reference, and removed any unnecessary blocking operations in your tasks, you may first try to scale up the hardware of the overburdened members by adding cores and, if necessary, more memory.

When you have reached diminishing returns on scaling up (such that the cost of upgrading a machine outweighs the benefits of the upgrade), you can scale out by adding more members to your cluster. The distributed nature of Hazelcast is perfectly suited to scaling out, and you may find in many cases that it is as easy as just configuring and deploying additional virtual or physical hardware.

In addition to the regular distributed executor service, Hazelcast also offers durable and scheduled executor services. Note that when a member failure occurs, durable and scheduled executor services come with "at least once execution of a task" guarantee, while the regular distributed executor service has none. See the Durable and Scheduled executor services.

Each member-specific executor will have its own private work-queue. Once a job is placed on that queue, it will not be taken by another member. This may lead to a condition where one member has a lot of unprocessed work while another is idle. This could be the result of an application call such as the following:

This could also be the result of an imbalance caused by the application, such as in the following scenario: all products by a particular manufacturer are kept in one partition. When a new, very popular product gets released by that manufacturer, the resulting load puts a huge pressure on that single partition while others remain idle.

This can lead to OutOfMemoryError because the number of queued tasks can grow without bounds. This can be solved by setting the queue-capacity property in the executor service configuration. If a new task is submitted while the queue is full, the call will not block, but will immediately throw a RejectedExecutionException that the application must handle.

There is currently no load balancing available for tasks that can run on any member. If load balancing is needed, it may be done by creating an executor service proxy that wraps the one returned by Hazelcast. Using the members from the ClusterService or member information from SPI:MembershipAwareService, it could route "free" tasks to a specific member based on load.

An executor service must be shut down with care because it will shut down all corresponding executors in every member and subsequent calls to proxy will result in a RejectedExecutionException. When the executor is destroyed and later a HazelcastInstance.getExecutorService is done with the ID of the destroyed executor, a new executor will be created as if the old one never existed.

When a task fails with an exception (or an error), this exception will not be logged by Hazelcast by default. This comports with the behavior of Java’s thread pool executor service, but it can make debugging difficult. There are, however, some easy remedies: either add a try/catch in your runnable and log the exception, or wrap the runnable/callable in a proxy that does the logging; the last option keeps your code a bit cleaner.

Hazelcast clients use an internal executor service (different from the distributed executor service) to perform some of its internal operations. By default, the thread pool for that executor service is configured to be the number of cores on the client machine times five; e.g., on a 4-core client machine, the internal executor service will have 20 threads. In some cases, increasing that thread pool size may increase performance.

Hazelcast allows you to update the whole or a part of map or cache entries in an efficient and a lock-free way using entry processors.

By default the entry processor executes on a partition thread. A partition thread is responsible for handling one or more partitions. The design of entry processor assumes users have fast user code execution of the process() method. In the pathological case where the code is very heavy and executes in multi-milliseconds, this may create a bottleneck.

We have a slow user code detector which can be used to log a warning controlled by the following system properties:

The defaults catch extremely slow operations but you should set this much lower, say to 1ms, at development time to catch entry processors that could be problematic in production. These are good candidates for our optimizations.

We have two optimizations:

These are enabled very simply by implementing these interfaces in your entry processor. These optimizations apply to the following map methods only:

See the Entry Processors section.