Avoiding Spark OOMEs

Spark can process more data than it can fit into memory. So why does it sometimes fail with OutOfMemoryExceptions when joining unskewed data sets?

An interesting way to counter OOMEs in a large join is here [SO] where rows are given a random integer seed that is used in addition to the usual condition. In theory, this breaks down the data into more manageable chunks.

Another standard exercise is to repartition the data. But this causes a shuffle and it may actually be the repartition itself that causes of an OOME.

In practice, I've found persisting the data frame to disk and reading it back yields better results. The number of partitions being written is rarely the number that is read back. That is, you get a more natural partition for free (or almost free. Obviously, some time is taken in writing to disk). And there is no repartition that could throw an OOME.

This question came up on Discord where somebody is trying to crossJoin a huge amount of data. I suggested a solution that uses mapPartitions. The nice thing about this method is that your code is passed a lazy data structure. As long as you don't try to call something like toList on it, it will pull data into memory as needed and garbage collect it after it's written out.

By using a lazy Iterator, Spark can write far more memory than it has to disk. As Spark consumes from the Iterator, it measures its memory. When it starts looking a bit full, it flushes to disk. Here is the memory usage of this code that uses mapPartitions to write to /tmp/results_parquet a data set that is much larger than the JVMs heap:

Spark with 0.5gb heap writing 1.3gb files

If we run:

watch "du -sh /tmp/results_parquet"

we can see that upon each GC, more is written to disk.

The result is a huge dataframe that could not fit into memory can now be joined with another.

As an aside: Uber has been doing some work on dealing with OOMEs in Spark. See their article here. TL;DR; they're proposing that in the event of an OOME, Spark adapts and increases the memory to CPU ratio by asking come cores to step down before it re-attempts the failed stage. Ergo, each compute unit has more memory than before. 

Unsafe Fences

The mysterious sun.misc.Unsafe class in JDK 1.8 has some methods that have been recently added. These are loadFence(), storeFence() and fullFence(). The documentation in Unsafe.java is sparse telling us that they "ensure lack of re-ordering of" load, stores and loads/stores before the fence "with loads or stores after the fence."

These load-, store- and full-fence native methods are defined in unsafe.cpp in the OpenJDK source code just defer to OrderAccess::acquire(), ::release() and ::fence() respectively.

OrderAccess::release() doesn't do a great deal:

inline void OrderAccess::release() {
  // Avoid hitting the same cache-line from
  // different threads.
  volatile jint local_dummy = 0;
}

It is inlined so it's as if it were copy-and-pasted into all parts of the code that call it and when I put something similar into a (non-inlined) piece of C++ and disassemble it on an x86_64 with objdump -d, it looks like this:

00000000000003dd <_Z19OrderAccess_releasev>:

 3dd: 55                   push   %rbp

 3de: 48 89 e5             mov    %rsp,%rbp

 3e1: c7 45 fc 00 00 00 00 movl   $0x0,-0x4(%rbp)

 3e8: 5d                   pop    %rbp

 3e9: c3                   retq   

The highlighted line is what sets 0 to something in the stack (all other lines are setting up and tearing down the stack frame). So, I can only conclude that the volatile in the release() method is there to stop the compiler re-ordering things as nothing special is happening at the hardware level. This appears to be what the OpenJDK source code suggests:

"Note: as of 6973570, we have replaced the originally static "dummy" field (see above) by a volatile store to the stack. All of the versions of the compilers that we currently use (SunStudio, gcc and VC++) respect the semantics of volatile here. If you build HotSpot using other compilers, you may need to verify that no compiler reordering occurs across the sequence point represented by the volatile access."

The only place in the JDK's source that I can see a call to loadFence() is in StampedLock. What this call to loadFence() appears to be avoiding is a write to a volatile variable to ensure a LoadLoad memory barrier (that is, the data loaded in the second read is at least as new as the data loaded in the first). Why this LoadLoad is necessary is explained at this link here.

The LoadLoad is desired after reading a non-volatile variable but before reading a volatile variable. But no memory barrier is otherwise issued in a Normal Store/Volatile Load combination. So, one way to provide it is to write to a volatile variable. Unfortunately, this is expensive.

[Aside: the semantics of primitive memory fences operate between instructions. It does not make sense to associate a fence with a single instruction.]

An acquire fence is a LoadLoad and a LoadStore (see here for a good explanation). From OpenJDK's orderAccess.hpp, the comments tell us what this acquire does:

"Execution by a processor of acquire makes the effect of all memory accesses issued by it subsequent to the acquire visible to all processors *after* the acquire completes.  The effect of prior memory accesses issued by it *may* be made visible *after* the acquire.  I.e., prior memory accesses may float below the acquire, but subsequent ones may not float above it."

So, acquire issues the LoadLoad we are looking for (plus LoadStore which appears is not needed in this case) and avoids an expensive write to a volatile.

A note on x86 architecture

LoadLoad, LoadStore and StoreStore seem to be no-ops on x86. This is what Doug Lea's JMM Cookbook says and looking at the OpenJDK code, acquire() and release() do nothing but add 0 to registers (they do, however, use volatile so the compiler should not do any re-ordering). So, calls to these methods should have no effect on this architecture even if they're needed on others.

The method fence() does, however, have an effect on x86 as uses the lock assembly instruction.

Empirically, this seems to be the case. Using JMH, I put a simple class together that benchmarks the times for these three calls. Each method is called and nothing more:

Benchmark                               Mode  Samples  Score  Score error  Units
c.p.m.m.FenceBenchmarks.acquireFence    avgt        5  0.338        0.021  ns/op
c.p.m.m.FenceBenchmarks.fence           avgt        5  6.343        0.090  ns/op
c.p.m.m.FenceBenchmarks.releaseFence    avgt        5  0.340        0.013  ns/op

The results show fence()  to be an order of magnitude slower than the other two methods.

Heartbleed and the JVM

The Heartbleed bug that affected OpenSSL earlier this year was a failure to do a proper bounds check. It allowed the client to define how much memory it was assigned in the server. The server then might have only filled some of this memory but all of its contents were returned to the client. The problem in a language like C is that this memory might be dirty. That is, it may contain other peoples' discarded data.

The patch was to check the bounds so no superfluous data was returned to the client. But why was the data not clean in the first place?

Basically, it's an optimization. Clearing the content of memory is expensive.

Java is much more secure than C. If you grab some off-heap memory, it cleans it before giving it to you. If you call ByteBuffer.allocateDirect(int) you are given an instance of a packge protected class, DirectByteBuffer. In its constructor you see:

            base = unsafe.allocateMemory(size);

where unsafe is an object of class sun.misc.Unsafe. The method allocateMemory takes us to unsafe.cpp in OpenJDK where we see:

  void* x = os::malloc(sz, mtInternal);

  if (x == NULL) {

    THROW_0(vmSymbols::java_lang_OutOfMemoryError());

  }

  //Copy::fill_to_words((HeapWord*)x, sz / HeapWordSize);

  return addr_to_java(x);

This doesn't clean the memory (that's done later). But as an interesting aside, set -XX:+PrintMalloc as a command line parameter and you'll see the JVM output lines like:

os::malloc 536870912 bytes --> 0x00007f1ce7fff028

Anyway, the cleaning of the memory comes here, a few lines later in DirectByteBuffer:

        unsafe.setMemory(base, size, (byte) 0);

which is a call to:

UNSAFE_ENTRY(void, Unsafe_SetMemory(JNIEnv *env, jobject unsafe, jlong addr, jlong size, jbyte value))

  UnsafeWrapper("Unsafe_SetMemory");

  size_t sz = (size_t)size;

  if (sz != (julong)size || size < 0) {

    THROW(vmSymbols::java_lang_IllegalArgumentException());

  }

  char* p = (char*) addr_from_java(addr);

  Copy::fill_to_memory_atomic(p, sz, value);

UNSAFE_END

Despite the mention of filling atomically, this call to the Copy class looks no more complicated than something like this:

    for (uintptr_t off = 0; off < size; off += sizeof(jlong)) {

      *(jlong*)(dst + off) = fill;

    }

The consequence of this security is a loss of performance:

"It's a well known fact that direct bytebuffer allocation is much slower than non direct byte buffers.

"If you think about it when you allocate a non direct byte buffer then it basically just needs to dereference a pointer to some memory on the Java heap, which is very quick. But for a direct buffer, it has to malloc real memory from the OS, and do a bunch of other house keeping.

"So yes, if you're using direct byte buffers it pays to re-use them."

https://community.jboss.org/thread/82231

What do you mean by memory?

If you set the JVM memory arguments Xmx and Xms, what are you actually doing? You're setting the heap size right? So, if I run 4 java processes using 5gb each on a machine with, say, 8gb, I've maxed the memory, right? Well, not necessarily.

This was the output from top for a Linux box running a dozen or so JVMs with Xmx4g:

top - 17:31:05 up 187 days, 12:24,  6 users,  load average: 1.75, 1.20, 0.88 
Tasks: 293 total,   1 running, 292 sleeping,   0 stopped,   0 zombie 
Cpu(s):  5.6%us,  0.5%sy,  0.0%ni, 93.7%id,  0.0%wa,  0.0%hi,  0.2%si,  0.0%st 
Mem:     48274M total,    29096M used,    19178M free,      439M buffers 
Swap:     8191M total,      990M used,     7201M free,     4889M cached 

  PID USER      PR  NI  VIRT  RES  SHR S %CPU %MEM    TIME+  COMMAND
28448 mds       17   0 5325m 4.8g 6600 S    1 10.1 502:39.47 java
29368 mds       17   0 4718m 1.8g 5668 S    0  3.8  74:00.60 java
16996 mds       17   0 4711m 1.7g  15m S   10  3.5   1:40.33 java
15366 mds       17   0 4661m 1.5g  15m S    0  3.3   0:59.42 java
15365 mds       17   0 4654m 1.5g  15m S    0  3.2   0:54.01 java
16997 mds       17   0 4711m 1.5g  15m S    0  3.1   1:01.36 java
.
. 

The RESident space is for all but one less than 4gb. The VIRTual memory is over 4gb. Which do we trust?

First, what is RESident and VIRTual memory?

"A virtual memory scheme splits memory used by each program into small, fixed-size units called pages... At any one time, only some of the pages of a program need to be resident in physical memory page frames; these pages for the so-called resident set. Copies of the unused pages of a program are maintained in the swap area - a reserved area of disk space used to supplement the computer's RAM - and loaded into physical memory only as required." [1]

But this machine is barely using swap space. So, the missing memory is not there.

"When your JVM starts it reserves your maximum heap size as one continuous block of virtual memory on startup. However, only the pages actually used become resident pages (actual main memory). When you set the minimum size, it doesn't force the JVM to use that much memory if it doesn't need it." [2]

"Address space (ie. virtual memory in the process list) doesn't cost anything; it's not real. What's real is the RSS (RES) column, which is resident memory." [3]

So, we're setting something that is "not real". What's the point? Well, we're setting logical memory limitations, not physical. When a garbage collection is started and the size of the different areas of the garbage collector (eden, old etc) is a function of this parameter.

This is all became relevant for us this week because one of our boxes went wild and starting killing Java processes. Note: it is not necessarily the offender who is punished. The Linux OOM Killer [4] calculates who is the best process to kill using an algorithm that takes into account a number of variables (you can see who is most likely to die as the score is shown in the file /proc/PID/oom_score).

[1] The Linux Programming Interface, Kerrisk
[2] StackOverflow.
[3] ServerFault.
[4] Blog on the Linux OOM Killer.

The Case of the Missing Memory

If you start Java with -Xmx512m, you'd expect the max heap size to be 512 megs, right? So, when you connect via JConsole (or JMX) why is the heap size smaller? My JConsole shows a committed and maximum size of the "Heap Memory Usage" as about 491MiB.

I've been fiddling with the OpenJDK source and been putting log statements into it to see how the GC is behaving. So, for instance, I changed:

hotspot/src/share/vm/gc_implementation/parallelScavenge/parallelScavengeHeap.cpp

added:

#include <stdio.h>

and liberally sprinkled the code with statements like:


  printf("PH: ParallelScavengeHeap::initialize: size %d: \n", _reserved.byte_size());

And sure enough, when I run my application, I see the JVM spit out:

PH: ParallelScavengeHeap::initialize: size 536870912: 

So, it's using the collector that is the "default on certain machine types" [1] with the full 512MiB allocated to it. This collector is "targeted towards large young generation heaps and to scale with more CPUs. It works very well with large young generation heap sizes that are in gigabytes, like 12GB to 80GB or more, and scales very well with increase in CPUs, 8 CPUs or more" [2] (I have 16 CPUs and 64GBs, although obviously have not set my heap that high).

Now, when I connect to the JVM and look at the memory usage, I see statements like:

PH: EdenMutableSpacePool::get_memory_usage: size 134217728: 

PH: SurvivorMutableSpacePool::get_memory_usage: size 22347776: 

PH: PSGenerationPool::get_memory_usage: size 357957632: 

since I have changed these C++ classes that can all be found in

hotspot/src/share/vm/services/psMemoryPool.cpp

(Note that these three classes are all the subtypes of CollectedMemoryPool for the PS collector.)

This memory usage totals the 491MiB usage we saw in JConsole but is still exactly 22 347 776 bytes short of the 512MiB we defined at start up time.

The "missing" memory is of course because there are two survivor spaces [3] but since only one has live objects at any point, we only count its size.

[1] Dr Richard Warburton's blog.
[2] Oracle's Improving Java Application Performance and Scalability by Reducing Garbage Collection Times and Sizing Memory Using JDK 1.4.1.
[3] Oracle's Virtual Machine Garbage Collection Tuning.