Overview
Large-scale data analytics, scientific simulation, and deep learning codes in HPC perform massive computations on data greatly exceeding the bounds of main memory. These out-of-core algorithms suffer from severe data movement penalties, programming complexity, and limited code reuse. To solve this, HPC sites have steadily increased DRAM capacity. However, this is not sustainable due to financial and environmental costs. A more elegant, low-cost, and portable solution is to expand memory to distributed multi-tiered storage. In this work, we propose MegaMmap: a software distributed shared memory (DSM) that enlarges effective memory capacity through intelligent tiered DRAM and storage management. MegaMmap provides workload-aware data organization, eviction, and prefetching policies to reduce DRAM consumption while ensuring speedy access to critical data. A variety of memory coherence optimizations are provided through an intuitive hinting system. Evaluations show that various workloads can be executed with a fraction of the DRAM while offering competitive performance.
System Description
-
Distributed Memory and Storage Hierarchy (DMSH)
The DMSH represents the set of storage devices in a distributed system. Storage devices are grouped into tiers, and then into layers. A tier is a set of devices with similar architecture. For example, the set of all RAM devices in the entire distributed system would be in the RAM tier. Layers group devices based on their performance characteristics. For example, RAM-1600Mhz and RAM-3200MHz would be in two separate layers in the RAM tier.
-
The API
To give applications the illusion of a large memory, MegaMmap implements a shared memory vector API, providing implementations of several functions and operators including array index, memory copy, acquiring current size, appending, resizing, and destroying the data container. Processes connect to the shared vector using a semantic, user-defined key common to all processes. The vector API is highly versatile and allows more complex distributed data structures, such as matrices, logs, and multi-dimensional arrays, to be developed using simple offset calculations and appends. Through C++ templating, MegaMmap can theoretically store any type of data -- including complex C++ classes, so long as a serialization method is provided. Unlike standard C++ vectors, MegaMmap shared vectors are not destroyed in the destructor -- users must explicitly destroy them. This is to avoid the race condition where processes finish at separate times. Shared vectors can be either volatile or nonvolatile, where volatile vectors store temporary data that is difficult to fit in memory and nonvolatile vectors represent data that should eventually be persisted to storage for future use.
-
System Monitor
In the background, HCompress will gather statistics about the tiers in the DMSH. It will periodically determine a device's availability and current capacity using built-in tools like du and iostat.
-
Ares
Ares is a compression service that unifies the interfaces of multiple different compression libraries: bzip2, zlib, huffman, brotli, bsc, lzma, lz4, lzo, pithy, snappy, and quicklz. When using the compression service, the user must specify which compression library they want to use, the data they want to compress, and the size of that data. Ares will allocate a buffer for storing the compressed data and then compress the input data by calling the chosen library from the Compression Library Pool. If the compression library ends up increasing the size of the data, the uncompressed data will be used instead. The buffer holding the data to return will be prepended with metadata including the uncompressed size of the data and whether or not the data is compressed. We do not store the compression library used in the metadata. We will assume the user remembers which library was used to compress the data when they go to decompress it.
-
HCDP
The HCDP decides where to put data in the DMSH and which compression libraries to use when compressing the data. This computation is based off of each device's capacity and performance as well as each compression library's performance and estimated compression ratio. After finding the least costly data placement, we actually place the data there. When placing data in the DMSH, we use HDF5 filters to divide the dataset into 1MB chunks. The filter will compress each chunk using Ares.
The data placement is determined as follows:
- \(S\) is the size of the data to be stored in DMSH
- \(i\) is the tier
- \(j\) is the compression library
- \(dmsh(S, i, j)\) is the time it takes to store S bytes in the DMSH starting from tier i using compression j
- \(tier(S, i, j)\) is the time it takes to store S bytes in tier i using compression j
- \(C_i\) is the capacity of tier i
- \(R_i\) is the remaining capacity of tier i
- \(B_i\) is the bandwidth of tier i
- \(t_c(S, j)\) is the time it takes to compress S bytes of data using compression j
- \(r(j)\) is the compression ratio for compression j
- \(t_d(S, j)\) is the time it takes to decompress S bytes of data using compression j
- \(S_j\) is the compressed size of S bytes using compression j
- \(w_1, w_2, w_3\) are user-defined weights
$$ tier(S, i, j) = w_1 t_c(S, j) + S_j/B_i - w_2(S_j/B_i)\frac{r(j)-1}{r(j)} + w_3 t_d(S_j, j) $$
$$ \begin{equation*} dmsh(S, i, j) = Min \begin{cases} tier(S, i, j) & S_j \lt R_i \\ tier(S, i, j) + dmsh(S_j - C_i, i+1, j) & R_i \lt S_j \leq C_i \\ tier(R_i, i, j) + dmsh(S_j - R_i, i+1, j) & C_i \lt S_j \\ dmsh(S, i, j+1) & \\ dmsh(S, i+1, j) & \\ \end{cases} \end{equation*} $$
-
Compression Cost Predictor
Every time a compression or decompression is made using the HCDP, we record the compression time, compression ratio, and the decompression time. Periodically, for each compression library, we will run a least squares regression using DLIB in order to estimate compression time, compression ratio, and decompression time in the HCDP.
The models we have for these are as follows:
\(c_1, c_2, c_3\) are parameters we are trying to estimate $$ t_c(S, j) = c_1 * S $$ $$ r(j) = c_2 $$ $$ t_d(S, j) = c_3 * S $$