Scaling Your Java EE Applications -- Part 2

Lessons learned from failed projects

Use share nothing clustering architectures

Figure 3: share nothing cluster

The most scalable architectures are sharing-nothing clusters. In such clusters, each node has the same functionalities, and knows nothing about the existence of other nodes. The load balancer will decide how to dispatch the requests to the back end server instances. It is rare that a load balancer will become a bottleneck, because a load balancer only need to do some very simple jobs, such as distributing requests, health checking, and session sticking. By adding more nodes, the computing ability of the cluster is increasing almost lineally, if the back end database or other information systems are powerful enough.

However, almost all popular Java EE vendors implement HTTPSession failover in their cluster products to ensure that all client requests can be processed properly without losing any session state in case of failure of some server nodes. This will break the share nothing principle. The same session data objects will be shared among two or more nodes in order to implement the failover functionality. In my previously-referenced article, I recommended not to use the session failover, unless it is strictly required. The session failover functionality cannot avoid errors completely when failures happen, as my article mentioned, but it will damage the performance and scalability.

Use scalable session replication mechanisms
If you are required to use session failover to give a more friendly experience to users, it is important to choose scalable replication products or mechanisms. Different vendors implement variable replication solutions – some use database persistence, some use centralized state server, while others use memory replications among nodes. The most scalable one is paired node replication, which is implemented by most popular vendors, such as BEA Weblogic, JBoss and IBM Websphere. Sun implemented the paired node replication in Glassfish V2 and above. The least scalable solution is using database as session persistence storage. One of projects tested in our laboratory using database persistence as session replication found that no more than three to four nodes would kill the database if the session objects were updated frequently.

Use collocated deployment instead of distributed one
Java EE technology, especially EJB, was born for distributed computing. Decoupled business functionality and reused remote components make multi-tier applications popular. But for scalability, reducing the distributed layers maybe a good choice.

A government project tested in our laboratory compared two different deployments in the same number of server instances – a distributed structure and a collocated one, show as the following: figures:

Figure 4: distributed structure

Figure 5: collocated structure

The result was that the collocated structure scaled much better than the distributed one! Imagine in one method in your application you may be invoking a couple of EJBs. If you load balance on every one of those EJBs, you're going to end up with instances of your application spread out across the multiple server instances. As a result, you're going to have a lot of server-to-server cross talk that's unnecessary. And more, if your method is under a transaction, your transaction boundary will include many server instances which will impact performance heavily.

Shared resources and services
In Java EE clustering systems servicing concurrent requests, scalable performance is limited by the number of operations which must utilize any shared resource that does not exhibit linear scalability. Database servers, JNDI trees, LDAP Servers, and external file systems can be shared by the nodes in the cluster.

Although not recommended by the Java EE specification, the external I/O operations are used for various purposes. For example, some applications tested in our laboratory used file systems to save uploading files by users, or create dynamic configuration XML files. Within a cluster, the application server nodes need a way of replicating these files across to other instances. But this was proven unscalable. With more nodes added, copying files among server instances will occupy all the network bandwidth and consuming considerable CPU resources. To work in a cluster, the solution is to use the database in place of external files, or choose SAN as the central deposits for files. An alternative way is to use an effective distributed file systems, such as Hadoop DFS (http://wiki.apache.org/hadoop/).

Shared services are common in a clustering environment. Those services are not deployed on every node of the cluster. Instead, they are installed on a dedicated server node, such as distributed logging service or timing service. A distributed lock manager (DLM) provides the applications on the server nodes in a cluster with a means to synchronize their accesses to shared services. A lock manager must operate correctly regardless of network latencies and system failures. For example, an ERP system tested in our laboratory has met such problems. They have written their own DLM system, but finally found their lock system would hold for locks forever, in case of failure of nodes which hold locks.

Distributed cache

Almost every Java EE project I experienced used object caching to improve performance, and all popular application servers provide extra degrees of caching to enable faster applications. But some caches are typically designed for a standalone environment, and can only work within one JVM instance. We need cache because some objects are so heavy that creating a new one will cost a great deal. So we maintain an object pool to reuse the object instances without further creation. We gain performance only if the maintenance of the cache is cheaper than objects creation. In a clustered environment, each JVM instance will maintain its own copy of the cache, which should be synchronized with others to provide inconsistent state in all server instances. Sometimes this kind of sync will bring worse performance than no caching at all. A scalable distributed cache system is very important for the scalability of the whole cluster.

Many open sourced Java products related to distributed cache are very popular. Followings are ones tested in our laboratory:

• 1 project tested on JBoss Cache
• 3 projects tested on Terracotta
• 9 projects tested memcached

Test result showed that Terracotta scaled well on 10 nodes and got a very high performance among no more than 5 nodes, but memcached scaled extremely well on more than 20 server instances.

Memcached

Memcached is a high-performance, distributed memory object caching system, generic in nature, but intended for use in speeding up dynamic web applications by alleviating database load. Memcached's magic lies in its two-stage hash approach. It behaves as though it were a giant hash table, looking up key = value pairs. Give it a key, and set or get some arbitrary data. When doing a memcached lookup, first the client hashes the key against the whole list of servers. Once it has chosen a server, the client then sends its request, and the server does an internal hash key lookup for the actual item data. The big benefit, when dealing with giant systems, is memcached's ability to massively scale out. Since the client does one layer of hashing, it becomes entirely trivial to add dozens of nodes to the cluster. There's no interconnect to overload, or multicast protocol to implode.

Memcached actually is not a Java product, but it has a Java client API. That means if you want to port your Java EE applications to use memcached, an amount of modification is unavoidable to use the API to get and put the values to/from the cache. Using memcached is very simple, but there are still some things you should try to avoid with respect to scalability and performance.

• Don't cache objects who are updated more frequently than read. Memcached is used to reduce your database reads instead of writes. You have to see what the read/write ratio per object will be. If the number of ratio is high, then caching makes a lot of sense.
• Try to avoid paging in the nodes running memcached. Swapping may be a disaster for memcached.
• Try to avoid row based cache. Instead, cache complex object will make memcached more efficient.
• Choose the right hashing algorithms. With default algorithms, adding or removing servers will invalidate all of your cache. Since the list of servers to hash against has changed, most of your keys will likely hash to different servers. Use consistent hashing algorithms (http://weblogs.java.net/blog/tomwhite/archive/2007/11/consistent_hash.html) to add and remove servers, will keep most of your cached objects valid.

Terracotta

Terracotta (http://www.terracottatech.com/) is an enterprise-class, open-source, JVM-level clustering solution. JVM-level clustering simplifies enterprise Java by enabling applications to be deployed on multiple JVMs, yet interact with each other as if they were running on the same JVM. Terracotta extends the Java Memory Model of a single JVM to include a cluster of virtual machines such that threads on one virtual machine can interact with threads on another virtual machine as if they were all on the same virtual machine with an unlimited amount of heap.

Figure 6: Terracotta JVM clustering

The programming model of applications clustered using Terracotta is the same or similar to that of an application written for a single application. There is no developer API specific to Terracotta. Terracotta uses bytecode manipulation—a technique used by many Aspect-Oriented software development frameworks such as AspectJ and AspectWerkz—to inject clustered meaning to existing Java language features.

Among the Terracotta server and client JVM instances, there I presume that must be some interconnect or multicast protocol to communicate among them. That may be the reason that the Terracotta failed to scale very well on more than 20 nodes for the projects tested in my laboratory. (Note: The test result is only reflected on the projects in our laboratory; your results may vary.)

Paralleled processing

As I mentioned earlier, a single thread of tasks will be the scalability bottleneck of the system. However, some single-thread jobs, such as processing or generating huge data sets, not only need to be paralleled to multithreaded or multiprocessed, but also have the requirement of scaling to multiple server nodes. For example, a Java EE project tested in our laboratory has a scenario to analyze the URL access patterns in their website from the logging files. The logging files would be accumulated more than 120 GB every week. Analyzing these logging files required four hours when using a single thread of Java application. This customer resolved this problem by using Hadoop Map-Reduce to scale it horizontally. Now, this URL access patterns analyzing program can not only be run in a multiprocessed mode, but also parallelized to more than 10 nodes, and it only spends 7 minutes to finish all the work.

There are a lot of frameworks and tools to help Java EE developers to scale their applications horizontally. In addition to the Hadoop, a lot of Java implementation of MPI can also help to parallel single thread of tasks to scale to many nodes.

MapReduce

MapReduce is a distributed programming model intended for processing massive amounts of data in large clusters, developed by Jeffrey Dean and Sanjay Ghemawat at Google. MapReduce is implemented as two functions – Map, which applies a function to all the members of a collection and returns a list of results based on that processing, and Reduce, which collates and resolves the results from two or more Maps executed in parallel by multiple threads, processors, or stand-alone systems. Both Map() and Reduce() may run in parallel, though not necessarily in the same system at the same time.

Hadoop is an open-source, end-to-end, general-purpose Java implementation of MapReduce. It is a Lucene-derived framework for deploying distributed applications running on large clusters of commodity computers, and being built and used by a community of contributors from all over the world. Yahoo's Search Webmap, Amazon EC2/S3 services and Sun's grid engine can all be run on Hadoop.

To simply put, by using “Hadoop Map-Reduce”, the “URL access patterns analyzing” program first divides the logging files into 128MB splits and the framework assigns one split to each Map function. The Map function will analyze the split assigned to it and generate a temporary result. All the temporary results from Map functions are sorted and assign to each Reduce function. The Reduce function will combine all the temporary results to get the final result. Those Map and reduce functions can be run paralleled on all the nodes comprising the cluster, and controlled by the Hadoop framework.

MapReduce is useful in a wide range of applications, including distributed grep, distributed sort, web link-graph reversal, term-vector per host, web access log stats, inverted index construction, document clustering, machine learning, statistical machine translation and other areas.

MPI

MPI is a language-independent communications protocol used to program parallel computers . Many Java versions of the MPI standard have been developed. mpiJava and MPJ are typical. mpiJava is implemented as a thin Java native interface (JNI) wrapper on top of a native MPI library. MPJ is a 100 percent Java implementation of the MPI standard. Both mpiJava and MPJ adhere closely to MPI’s Fortran and C oriented API; for example, each provides a Comm class with the same method names and arguments as MPI’s message passing subroutines.

CCJ is a Java library of MPI-like collective communication operations. CCJ provides the barrier, broadcast, scatter, gather, all-gather, reduce, and all-reduce operations (but not point-to-point communication operations like send, receive, and send-receive). Rather than implementing its own low-level communication protocol, CCJ is implemented on top of Java remote method invocation (RMI). This allows CCJ to transfer arbitrary serializable objects, not just primitive data types as in MPI. Furthermore, CCJ can scatter and gather portions of arbitrary collection objects among a group of parallel processes, as long as the collections implement CCJ’s DividableDataObject interface.

Using unorthodox approach to achieve high scalability

There are a lot of books and articles to teach us how to design the systems in an Object Oriented way to get a flexible structure, how to make your services transparent to the clients to get a maintainable systems, and how to design your database schema with normalized model to make your data more integrated. But sometimes, in order to get high scalability, unorthodox ways may be chosen.

Google designs its highly scalable distributed file system(GFS) without implementation of the POSIX API, and GFS is not transparent to the users at all. To use the GFS, you must use the GFS API library. Google also designs a highly scalable distributed database system (Bigtable) without complying with the ANSI SQL standard. The whole concept and structure are totally different with traditional relational database. But the most important thing is both GFS and Bigtable are extremely scalable to successfully meet Google's storage needs and are widely deployed within Google as the storage platform.

Traditionally, we scale databases by using bigger, faster and more expensive server machines, or we may use enterprise cluster database such as Oracle RAC to scale our database to more server nodes. But a social networking website tested in our laboratory chooses the unorthodox way. This web based social networking application allows users to create personal profiles, blogs in the website and share photos and music with friends. This application is Java EE based, and built upon Tomcat and MySQL. Unlike other applications tested in our laboratory, it only wants to be tested on more than 20 inexpensive PC servers. Their data model is shown below:

Figure 7: Users data partitions

The special part of this lies in that different users' data (such as profile and blog) may be stored on different database server instances. For example, user 00001 is stored on server A, while user 20001 is stored on server C. A meta table in a dedicated database server describes the partition rules. When the Java EE application deployed in Tomcat wants to access (or update) the user information, it will first connect to the “meta table” to find the database node to serve the user, then connect to that server to operate the query or update actions.

The partition of user data and two steps access approach have following advantages:

• Scalable write bandwidth: In this kind of applications, blogging, ranking and BBS will make the write bandwidth the major bottleneck for social networking websites. Distributed Cache system will help little on the performance for the database write actions. With data partitions, you can write in parallel which increases your write throughput. To support more registered users just add more database nodes, modify the meta table to reflect the new servers.
• High availability: If one database server goes down, only some users are impacted, while others still operate.

The disadvantages are:

• Since database nodes can be added very dynamically, it is hard for Java EE applications in Tomcat to use the database connection pools.
• Since user data access involve two steps actions, it is hard to use ORMapping tools for developers.
• To create a complex search or joining data, you usually must pull together lots of different data from different database servers.

“We know well about these disadvantages, and have prepared for it”, the architect said about their system. “We even prepared for the extreme situation that the meta table server becomes the bottleneck. Then we will depart from the meta table, and create a top level meta table to point to many second level meta table server instances.”

References

1. Scalability definition in wikipedia: https://en.wikipedia.org/wiki/Main_Page/Scalability
2. Javadoc of atomic APIs: http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/atomic/package-summary.html
3. Alan Kaminsky. Parallel Java: A unified API for shared memory and cluster parallel programming in 100% Java: http://www.cs.rit.edu/~ark/20070326/pj.pdf
4. OMP-an OpenMP-like interface for Java: http://portal.acm.org/citation.cfm?id=337466
5. Google MapReduce white paper: http://labs.google.com/papers/mapreduce-osdi04.pdf
6. Google Bigtable white paper: http://labs.google.com/papers/bigtable-osdi06.pdf
7. Hadoop MapReduce tutorial: https://hadoop.apache.org/core/docs/r0.17.0/mapred_tutorial.html
8. Memcached FAQ: http://www.socialtext.net/memcached/index.cgi?faq
9. Terracotta: http://www.terracotta.org/

About the Author

Wang Yu presently works for ISVE group of Sun Microsystems as a Java technology engineer and technology architecture consultant. His duties include supporting local ISVs, evangelizing and consulting on important Java technologies such as Java EE, EJB, JSP/Servlet, JMS, Web services technologies. He can be reached at [email protected].