DAOS as HPC Storage: a View From Numerical Weather Prediction

With prospects of increasing simulation resolution and diversity of parameters modelled and output, it is expected shortly the volume of forecast data generated will reach approximately 180 TiB per time-critical window, and up to 700 TiB per time-critical window in the near future.

Therefore, the pressure on the HPC storage system will increase significantly, requiring higher throughput to maintain the time-critical and operational aspects of the service. Exploration and adoption of novel storage technologies will be key for a satisfactory implementation of the upcoming resolution increases.

A pair of C functions have been developed to perform writing and reading of weather fields to and from a DAOS object store, using the DAOS C API. They have been developed based on the design of FDB5, the domain-specific object store already employed within the ECMWF, so that the same type of operations as in operational workflows is carried out. The diagram in Fig. 1 shows the different DAOS concepts and APIs involved in the field I/O functions.

At the top level a DAOS Key-Value object, in its own container, acts as a main index allowing location of the data belonging to a same model run or forecast. That index maps the most-significant part of the key identifying a field (e.g. “‘class’: ‘od’, ‘date’: ‘20201224’”) to another DAOS container, at the lower layer. In that container is another Key-Value object that acts as a forecast index, enabling location of the data of the fields comprising a forecast. That index maps the least-significant part of the field key to a further DAOS container and a DAOS Array in that container where the data of the indexed weather field is stored.

How the developed functions perform field writes and reads using these concepts and objects is described, in a very simplified way, in Algorithms 1 and 2, respectively.

When the field write function is called, with the binary field and its key as parameters, the most-significant part of the key is retrieved from the main Key-Value. If it exists, the indexed references to the forecast containers are retrieved and the containers are opened. If they do not exist, creation of a new pair of forecast index and store containers is attempted, with container IDs computed as md5 sums of the most-significant part of the key so that any concurrent processes attempting creation of the same pair of containers will avoid creation of inaccessible containers. Immediately after creation, the forecast containers are opened and the ID of the forecast store container is stored in a special entry in a newly created Key-Value in the forecast index container. Next, a reference to the forecast index container is indexed in the main Key-Value, using the most-significant part of the field key as key. Then, the binary field is written into an Array in the forecast store container, with a new object ID, and a reference to it is indexed in the forecast index Key-Value, using the least-significant part of the field key as key.

Note that when a field is written with a key that already exists in the overall store, a new Array object is created and indexed, and the previously existing one is de-referenced. No read-modify-write is performed upon re-write, and the functions do not delete de-referenced objects by design.

When the field read function is called, with the key of a field the parameter, the most-significant part of the key is looked up in the main Key-Value. If it exists, the indexed reference to the forecast containers are retrieved and the containers are opened. Next, the least-significant part of the key is retrieved from in the forecast index Key-Value. If that is present, the indexed references to the forecast store container and Array are retrieved. Finally, the forecast store container is opened and the Array is read.

For the NWP applications we are investigating, the separation of the different Key-Values and Arrays in different containers is motivated by the need to avoid contention for the same container or indexing Key-Value object during intensive I/O workloads. This separation also allows the implementation of the different parts of the object store with different storage back-ends or on different systems.

IOR is a community-developed I/O benchmark which relies on MPI to run and coordinate parallel processes performing I/O operations against a storage system. It includes back-ends to operate with various popular storage systems, including DAOS.

IOR has been employed in this analysis with the intent to measure the throughput an application could achieve if it were programmed to operate in a traditional parallel, bulk synchronous I/O manner. That is, running a number of parallel client processes which use the high-level DAOS Array API to write or read data from a DAOS object store, all of them starting each I/O operation simultaneously, and waiting for each other to finish.

For the tests in this analysis, the IOR benchmark has been run in segments mode. IOR is invoked with a set of parameter values which instruct each client process to perform a single I/O operation, transferring its full data size. This is with the intent to assess the performance of a hypothetical optimised parallel application which is designed to minimise the number of I/O operations interacting with the storage. Processes in such applications issue a single transfer operation comprising all the data parts they manage, in contrast to an equivalent, non-optimised application where processes issue a transfer or even an open and a close operation for each data part.

Unless the storage is not optimised to handle large transfers or objects, this benchmark mode should give an idea of what is the maximum, ideal throughput the storage can deliver. This mode has been implemented by setting both the -b and -t IOR parameters to the size of each data part managed by each process, and -s to the number of data parts managed by each process. The -i parameter has been set to 1, and the -F flag (file per process) has been enabled. Each process performs the following during the benchmark execution: a) initial barrier, b) pre-I/O barrier, c) object create/open of size t * s bytes, d) transfer (write or read) of t * s bytes, e) object close, f) post-I/O barrier, g) post-I/O processing and logging, h) final barrier.

To gain insight on how DAOS performs in the previously outlined NWP I/O server use case, we use our custom field I/O benchmark, which mimics operational NWP I/O workflows. Here, parallel processes perform a sequence of field I/O operations using the functions previously described in the Weather field I/O section, with no synchronisation. Pool and container connections in a process are cached once initialised.

In contrast to the IOR benchmark in segments mode, the processes in this benchmark perform multiple I/O operations of smaller sizes, one for each data part. Each field I/O operation involves an Array open, transfer and close, and usually involves a few operations with Key-Value objects. We also do not enforce I/O synchronisation, as in IOR.

Fig. 2 shows a schematic representation of the processes and sequence of field I/O operations involved in an example benchmark run.

The field I/O benchmark has been configured in three different modes for this evaluation:

• •

  full: the fully-functional field I/O functions are used as described in Weather field I/O.

• •

  no containers: the use of container layers in the field I/O functions is disabled with the intent to analyse any potential issues in the use of container layers. The functions create and operate with all DAOS Key-Values and Array objects in the main container.

• •

  no index: the use of indexing Key-Value objects and container layers in the field I/O functions are disabled with the intent to compare with results from other benchmark modes and assess the overhead of field indexing. To preserve the ability to locate previously written fields, the field I/O functions map the field identifiers to DAOS Array object IDs by calculating a md5 sum of the identifiers. The arrays are stored and read directly from the main container.

By default, in modes with indexing enabled, each parallel process writes and reads fields indexed in its own forecast index Key-Value and therefore there should be little contention for the different objects within DAOS. The benchmark, however, can be configured to have a single shared forecast index Key-Value among all processes, inducing high contention on that Key-Value object.

The Field I/O benchmark is maintained and documented in a publicly accessible open-source Git repository[15].

The IOR and field I/O benchmarks, in each of their modes, can be further adjusted and employed as part of larger workflows to generate specific combinations of I/O workload of interest, hereafter referred to as access patterns. Two access patterns used in this analysis are described next.

• •

  A (unique writes then unique reads): This access pattern has two phases. Each client process performs a number of writes to new objects in the storage (single transfer to single object for IOR in segments mode). Once all writer processes have finished, a second phase begins, where another process set of the same size and distribution is executed performing an equivalent number of reads from the objects generated in the write phase. The access pattern ends when all second-phase processes terminate.

  This access pattern is designed to provide a situation where there is no contention for same fields and when there is only either write or read workloads in operation at any one time, analogous to a single large-scale application utilising the object store, or to systems where the predominate I/O patterns are only one I/O phase (writing or reading).

  In this access pattern, when run with the IOR benchmark, there is no contention in Array writes or reads, as each process accesses its own Array independently from others. When run with the field I/O benchmark in any of its modes with a forecast index Key-Value per process, there is no contention in any of the forecast index or store objects, as each process writes or reads its own Key-Value and Arrays sequentially.

• •

  B (repeated writes while repeated reads): This access pattern starts with a setup phase to populate the storage with data, where half of the client processes (and thereby half the client nodes) perform a single write to a new object. Once all writer processes have terminated, the main phase begins. Half the client processes perform a number of re-writes to designated objects; simultaneously, the other half of the client processes perform the same number of reads from their designated objects. The access pattern ends when all main-phase processes terminate.

  This access pattern has only been implemented with the field I/O benchmark as IOR does not allow coordination between a set of writers and a set of readers. It aims to mimic situations where one or more large-scale applications issue simultaneous writes and reads of the same objects or fields. This is the equivalent of the I/O behaviour of actual NWP and product generation workloads, or to systems with mixed applications of varying I/O workloads.

  In this access pattern, when run with Field I/O in full or no containers modes with a forecast index Key-Value per process, there is no contention in Array writes or reads, but there is some contention in each forecast index Key-Value between reader and writer processes on the same object. When run in no index mode, the same degree of contention occurs at the Array level.

The access patterns presented can be configured not only to run with the different benchmarks and modes, but also to use a range of parameter values. The following list summarises these parameters and the values tested in this paper.

• •

  number of DAOS server nodes: 1, 2, 4, 8, 10, 12, 14, 16

• •

  number of client nodes: 0.5x, 1x, 2x and 4x the amount of server nodes

• •

  number of processes per client node: 1, 3, 4, 6, 8, 9, 12, 18, 24, 36, 48, 72, 96

• •

  number of iterations or data parts per proc.: 100, 2000

• •

  object class: S1 (no striping), S2 (striping across two targets) and SX (striping across all targets)

• •

  object size: 1, 5, 10, 20 and 50 MiB

To quantify, compare, and discuss the performance delivered by the storage, a set of metric and throughput definitions are introduced here and applied in the Results section. During execution, the IOR and Field I/O benchmarks report timestamps for various events. Each timestamp is reported together with an identifier of the client node, process and iteration or I/O they belong to, as well as the event name. We record timestamps for: execution start, I/O start, object open start, object open end, data write/read start, data write/read end, object close start, object close end, I/O end, and execution end.

In Field I/O, I/O start is recorded immediately before calling the field write or read functions. In IOR it is equivalent to object open start.

The following derived metrics are calculated where relevant from the timestamps.

• •

  single iteration parallel I/O wall-clock time: calculated as the maximum I/O end minus the minimum I/O start for a given I/O iteration across all parallel processes. This metric is only valid in benchmarks where I/O is synchronised.

• •

  total parallel I/O wall-clock time: calculated as the maximum I/O end of the last I/O iteration minus the minimum I/O start of the first I/O iteration across all parallel processes. This metric is calculated separately for the write and read phases of the described access patterns, and is valid for synchronised and non-synchronised benchmarks. In the example in Fig. 2, the metric would be calculated as $t_{f}(IO8)-t_{0}(IO1)$.

With these derived metrics, the throughput is calculated in two different ways, outlined below. Information such as I/O size and number of parallel processes is available at analysis time and is used as input for the calculation of these throughputs.

• •

  synchronous bandwidth: the sum of the sizes of the I/O operations for an iteration across all parallel processes, divided by the single iteration parallel I/O wall-clock time. The averaged result across all iterations is the synchronous bandwidth, only applicable to synchronous benchmarks (i.e. IOR).

  $$\begin{split}sync.\ bw.=\frac{1}{n}*\sum_{i=1}^{n}\frac{\sum_{j=1}^{m}size(IO_{ij})}{max_{j}\{t_{f}(IO_{ij})\}-min_{j}\{t_{0}(IO_{ij})\}}\\ n=\#iterations,\ m=\#processes\end{split}$$

  IOR internally calculates the synchronous bandwidth following this definition.

• •

  global timing bandwidth: the sum of the I/O sizes across all I/O operations in the workload and divided by the total parallel I/O wall-clock time.

  $$\begin{split}\ \ global\ timing\ bw.=\frac{\sum_{i=1}^{n}size(IO_{i})}{t_{f}(IO_{n})-t_{0}(IO_{1})}\\ n=\#I/O\ operations\end{split}$$

  It is calculated separately for the write and read phases of the described access patterns. The value of this bandwidth is inversely proportional to global parallel I/O wall-clock time. If a benchmark performs any work between I/O iterations this will impact the global timing bandwidth measure as it will increase the total parallel I/O wall-clock time.

When running patterns A or B with Field I/O, the distribution of I/O operations should be analysed and taken into account at benchmark configuration time to ensure the resulting distribution is representative of the desired access pattern. To characterise the overall distribution of I/Os, the following offsets can be measured for a benchmark run:

• •

  initial I/O offset ($off_{0}$): measured separately for the write and read phases of an access pattern, calculated as the last first-iteration I/O start time minus the first I/O start time across all processes, as shown in Fig. 2.

• •

  final I/O offset ($off_{f}$): measured separately for the write and read phases of an access pattern, calculated as the last I/O end time minus the first last-iteration I/O end time.

• •

  phase start offset ($po_{0}$): calculated as the absolute value of the difference between the first I/O start time of the write and read phases, only applicable to access pattern B.

• •

  phase end offset ($po_{f}$): calculated as the absolute value of the difference between the last I/O end time of the write and read phases, only applicable to access pattern B.

For the most accurate results, we should ensure $off_{0}$ and $po_{0}$ are not large relative to the total parallel I/O wall-clock time by adjusting access pattern configuration.

$off_{f}$ may account for a large portion of the total parallel I/O wall-clock time as, for example, the storage may process operations from the various parallel processes in an unbalanced manner. Similarly, $po_{f}$ may account for a large portion of the total duration of I/O (write and read) of a pattern B run, for example, as the storage may process write and read operations in an unbalanced manner.

Outside these offset time spans, there are as many in-flight I/O operations as parallel processes at any point in time during the benchmark run.

The benchmarks we present have been conducted using the NEXTGenIO system[16], which is a research HPC system composed of dual-socket nodes with Intel Xeon Cascade Lake processors. Each socket has six 256 GiB first-generation Intel’s Optane DCPMMs configured in AppDirect interleaved mode. Each processor is connected with its own network adapter to a low-latency OmniPath fabric. Each of these adapters has a maximum bandwidth of 12.5 GiB/s. The fabric is configured in dual-rail mode, that is, two separate OmniPath switches interconnect first-socket adapters and second-socket adapters separately, respectively.

The HPC system nodes use CentOS7 as the operating system, with DAOS v2.0.1. For the majority of the tests, two DAOS engines have been deployed in each node used as DAOS server, one on each socket using the associated fabric interface and interleaved SCM devices, with 12 targets per engine.

In this test set, the IOR benchmark in segments mode has been run (access pattern A) with the intent to analyse the maximum write and read performance a client application can achieve.

In our initial experiment we deployed DAOS on a single server node, with one or two engines, each running on a socket of the node and using the corresponding fabric adapter (ib0 and ib1, respectively). The IOR client processes have been pinned to one or two sockets (or interfaces) in one or two client nodes, as detailed in Table I. The segment count has been set to 100.

A segment size of 1 MiB has been motivated by the current weather field size at the exemplar weather centre. This segment size results in objects of 100 MiB in size being written or read by IOR. Striping has been disabled (OC_S1) to avoid complexity in network behaviour for the initial tests. The number of processes per client node has been set to 24, 48, 72 and 96 after verifying with several other IOR tests that process counts in that range usually resulted in higher benchmark bandwidths. Each test has been repeated 9 times for the range of client processes, and the maximum synchronous bandwidth obtained among the 36 repetitions is reported in Table I for both the write and read phases.

With a single server engine using one interface and a set of client processes using a single client interface on the first socket of the corresponding nodes (first row of the table; 1 client node), the maximum write bandwidth reaches 3 GiB/s and, for read, 4.2 GiB/s. When using multiple client sockets and interfaces against a single server engine on the first socket and interface (last column of first row and second row), the maximum write bandwidth still saturates at 3 GiB/s whereas the maximum read bandwidth reaches up to 7.7 GiB/s.

For the test in the last row of the table, with two DAOS engines deployed in a server node, and using both interfaces on a single client node (last row; 1 client node), the maximum write bandwidth reaches up to 5.5 GiB/s. For read, the maximum bandwidth remains at 7.5 GiB/s on a single client node, whereas it reaches up to 9.5 GiB/s with two client nodes (a total of 4 network interfaces). This indicates that more client interfaces than server interfaces are necessary to saturate server interface bandwidth for read, which is likely to be the deployment configuration of DAOS for production systems.

To further analyse the bandwidth increase when more engines and interfaces are employed, the same access pattern with identical parameters has been run with different numbers of server and client nodes, using both processors and network interfaces on each node. Using as many client nodes as server nodes has been pursued for all tested configurations to effectively make use of the available bandwidth theoretically provided by the server interfaces, and, where possible, configurations with twice or four times as many client nodes have been tested to explore if, as suggested by results in Table I, these result in higher bandwidths. A few special configurations have also been tested, for instance with less client nodes than server nodes (e.g. 2 server nodes and 1 client node), or with slightly less than double the server nodes (e.g. 10 server nodes and 18 client nodes) to explore bandwidth behaviour in unbalanced configurations. Fig. 3 shows, for each combination of server and client node count, the mean synchronous bandwidth obtained across all repetitions for the best performing number of client processes per client node.

For a single dual-engine server, the bandwidth achieved per engine is of approximately 2.5 GiB/s for write and 5 GiB/s for read. As the number of servers increases, the bandwidths increase linearly at a rate of approximately 2.5 GiB/s write and 3.75 GiB/s read for every additional engine.

It is worth noting that configurations with twice as many client nodes generally deliver best performance, both for write and for read. Setups with a substantially higher ratio (e.g. four times as many) of client nodes do not show significant increases in performance, and configurations with a slightly lower ratio show a substantial decrease in performance.

Above 8 server nodes, the scaling rate seems to decrease slightly even if using twice as many client nodes, as seen in the bandwidth results for the setup with 10 server nodes and 20 client nodes.

The fact that the increase in read bandwidth per engine is not equal to the bandwidth achieved with a single engine is possibly due to the fact that multiple engines on both sockets in different nodes may contend to communicate through a single interface on one socket. Nonetheless, the linear increase in bandwidth with increasing engines is encouraging.

Due to the limitations encountered with the PSM2 fabric provider, we have not been able to benchmark DAOS with the highest-performance network functionality. However, it is possible to deploy DAOS with PSM2 using a single engine per server and only one socket used per client node. This restricts the usable network, processing, memory, and storage resources within nodes, but does allow comparison of TCP and PSM2 configurations of DAOS and an assessment of the impact of using a lower-performance network configuration.

We benchmarked a configuration using 4 DAOS server nodes, and up to 16 client nodes, which we tested with a number of different process counts (4, 8, 12, and 24 processes per client node). We used IOR in segments mode with the same configuration as used previously.

Fig. 4 compares both read and write performance when using the IOR benchmark with the different communication mechanisms. It is evident that PSM2 does provide higher performance than TCP, especially when scaling up the number of processes per client node. PSM2 demonstrates 10% to 25% higher bandwidth than TCP, and can also provide higher performance at lower node counts. However, it can be observed that the DAOS performance using PSM2 follows the same general patterns as with TCP communications and, whereas we expect the former would result in larger absolute bandwidths by up to a 25%, we consider their behaviour at scale would be similar.

We ran the Field I/O benchmark with the intent to analyse the order of magnitude and behaviour of bandwidths obtained with the field I/O functions rather than simple segment transfers; how the different field I/O modes scale; and how bandwidth behaves in access pattern B with an I/O workload similar to operational workloads.

For the rest of the tests in this paper, both sockets and network interfaces have been used on each of the nodes employed for the benchmark runs, with two DAOS engines deployed on each server node and client processes pinned to both sockets on each client node.