Bogdan Nicolae
email: bogdan.nicolae@ie.ibm.com

Leveraging naturally distributed data redundancy to reduce collective I/O replication overhead

Keywords: data resilience, high availability, data replication, deduplication, collective I/O scalability, redundancy management

Dumping large amounts of related data simultaneously to local storage devices instead of a parallel file system is a frequent I/O pattern of HPC applications running at large scale. Since local storage resources are prone to failures and have limited potential to serve multiple requests in parallel, techniques such as replication are often used to enable resilience and high availability. However, replication introduces overhead, both in terms of network traffic necessary to distribute replicas, as well as extra storage space requirements. To reduce this overhead, state-of-art techniques often apply redundancy elimination (e.g. compression or deduplication) before replication, ignoring the natural redundancy that is already present. By contrast, this paper proposes a novel scheme that treats redundancy elimination and replication as a single co-optimized phase: remotely duplicated data is detected and directly leveraged to maintain a desired replication factor by keeping only as many replicas as needed and adding more if necessary. In this context, we introduce a series of high performance algorithms specifically designed to operate under tight and controllable constrains at large scale. We present how this idea can be leveraged in practice and demonstrate its viability for two real-life HPC applications.

I. INTRODUCTION

Scientific and data-intensive computing have matured over the last couple of years in all fields of science and industry. Their rapid increase in complexity and scale has prompted ongoing efforts dedicated to reach exascale infrastructure capability by the end of the decade. However, advances in this context are not homogeneous: I/O capabilities in terms of networking and storage are lagging behind computational power and are often considered a major limitation that that persists even at petascale [START_REF] Dongarra | The international exascale software project roadmap[END_REF].

A particularly difficult challenge in this context are collective I/O access patterns where all processes simultaneously dump large amounts of related data simultaneously to persistent storage (which we henceforth refer to as collective dump). This pattern is often exhibited by large-scale, bulk-synchronous applications in a variety of circumstances, e.g., when they use checkpoint-restart fault tolerance techniques to save intermediate computational states at regular time intervals [START_REF] Elnozahy | A survey of rollback-recovery protocols in message-passing systems[END_REF] or when intermediate, globally synchronized results are needed during the lifetime of the computation (e.g. to understand how a simulation progresses during key phases). Under such circumstances, a decoupled storage system (e.g. a parallel file system such as GPFS [START_REF] Schmuck | GPFS: A shared-disk file system for large computing clusters[END_REF]) does not provide sufficient I/O bandwidth to handle the explosion of data sizes: for example, Jones et al. [START_REF] Jones | Application Monitoring and Checkpointing in HPC : Looking Towards Exascale Systems[END_REF] predict dump times in the order of several hours.

In order to overcome the I/O bandwidth limitation, one potential solution is to equip the compute nodes with local storage (i.e., HDDs, SSDs, NVMs, etc.). Using this approach, a large part of the data can be dumped locally, which completely avoids the need to consume and compete for the I/O bandwidth of a decoupled storage system. However, this is not without drawbacks: the local storage devices are prone to failures and as such the data they hold is volatile. Furthermore, the availability of the data may also suffer under concurrency due to the limited I/O bandwidth of the local storage devices and/or network links.

Partner replication is a technique often used to mitigate the limitations of using local storage devices: instead of storing only one local copy of the dataset, a predefined number of extra copies are send remotely to the local storage devices of other compute nodes. Using this approach, resilience and high availability of the data can be achieved in a scalable fashion by leveraging the network bandwidth allocated to the compute nodes for communication, which is often orders of magnitude higher than the I/O bandwidth of a decoupled storage system.

However, with increasing scale, partner replication quickly hits on an important limitation: due to an increasing failure rate and an increasing number of processes potentially interested in a dataset, it is necessary to increase the replication factor in order to guarantee the same level of resilience and/or availability. As a consequence, the processes need to send more data to each other: this increases the network bandwidth contention because of larger data transfers, as well as the space utilization and I/O pressure on the local storage devices because more data is received from other processes. Ultimately, both aspects introduce an overhead that not only negatively impacts performance, but also increases operational costs (e.g. need to buy larger local storage devices).

Thus, it is important to be able to achieve a high replication factor with minimal overhead. A common strategy in this context is to apply some form of redundancy elimination (i.e., compression or deduplication) before the replication, under the assumption that it leads to a significant reduction of replication overhead, which improves overall performance and reduces resource utilization. How-ever, although straightforward, this two-phase approach is not optimal: first, an effort is made to eliminate data redundancy, only to reintroduce it later through replication.

In this paper, it is precisely this co-optimization aspect that we explore. Inspired by several studies that confirm high data redundancy for HPC workloads (such as Meister et al. [START_REF] Meister | A Study on Data Deduplication in HPC Storage Systems[END_REF] and our own previous work [START_REF] Nicolae | Towards Scalable Checkpoint Restart: A Collective Inline Memory Contents Deduplication Proposal[END_REF]), we propose to identify any data redundancy that already exists across distributed processes and group together duplicated data into natural replicas. Using this approach, redundancy elimination and partner replication are only selectively needed when a data piece is duplicated by more and, respectively, by less than the desired replication factor. We summarize our contributions as follows:

• We present a series of design principles that facilitate efficient deduplication of distributed chunks, eliminating those that are remotely duplicated beyond a fixed replication factor and evenly distributing the partner replication workload for the remaining chunks among the processes to achieve load balancing. Furthermore, all processes closely coordinate to help each other out and minimize the overhead of network traffic using single-sided communication. (Section III-B)

• We show how to materialize these design principles in practice through a series of algorithmic descriptions that are applied to implement an I/O library that exposes a dedicated collective I/O write primitive at application level. This library is then integrated with the AC-FTE [START_REF]Ac-fte[END_REF] fault tolerance runtime, which leverages the collective I/O write capabilities of the library in the context of checkpoint-restart. (Sections III-C and IV)

• We evaluate our approach in a series of experiments conducted on the Shamrock testbed, using two representative real-life HPC applications that exhibit a high degree of redundancy in the context of checkpointrestart. Our experiments demonstrate a large reduction of performance overhead and resource utilization compared to techniques that are not aware of naturally distributed duplicates. (Section V)

II. RELATED WORK

Replication is a widely used technique to improve resilience and high availability in parallel file systems and other special purpose storage services [START_REF] Schmuck | GPFS: A shared-disk file system for large computing clusters[END_REF], [START_REF] Shvachko | The Hadoop distributed file system[END_REF], [START_REF] Nicolae | BlobSeer: Next-generation data management for large scale infrastructures[END_REF], [START_REF] Nicolae | BlobCR: Virtual disk based checkpointrestart for HPC applications on IaaS clouds[END_REF].

However, due to limited scalability of remote I/O accesses, local storage devices saw an increasing adoption. At first, they were exploited as an intermediate write cache layer that is used to flush the application data asynchronously to remote storage systems in the background. This was for example introduced in the context of multilevel checkpointing [START_REF] Moody | Design, modeling, and evaluation of a scalable checkpointing system[END_REF], [START_REF] Dong | Hybrid checkpointing using emerging nonvolatile memories for future exascale systems[END_REF], node-level aggregation of I/O from multiple cores [START_REF] Dorier | Damaris: How to Efficiently Leverage Multicore Parallelism to Achieve Scalable, Jitter-free I/O[END_REF], or I/O forwarding [START_REF] Ilsche | Optimizing i/o forwarding techniques for extreme-scale event tracing[END_REF]. To avoid or at least limit the need to make use of a remote storage system, several proposals aim to directly make local storage resilient either through point-to-point replication [START_REF] Zhu | Shifted declustering: A placement-ideal layout scheme for multi-way replication storage architecture[END_REF] or erasure codes [START_REF] Bautista-Gomez | FTI: High Performance Fault Tolerance Interface for Hybrid Systems[END_REF].

Reducing the amount of replicated data is possible either using compression [START_REF] Ibtesham | On the viability of checkpoint compression for extreme scale fault tolerance[END_REF], [START_REF] Lakshminarasimhan | Isabela for effective in situ compression of scientific data[END_REF] or deduplication. The latter broadly falls into two categories: static and content-defined. Static approaches split the input data into small, fixedsized chunks that are then compared to each other, either directly or by using fingerprints. Since comparing only hash values increases speed at the expense of false positives (due to potential collisions), some approaches [START_REF] Alvarez | Netapp deduplication for fas and v-series deployment and implementation guide[END_REF] even combine hash comparisons with direct comparisons in order to be able to leverage computationally cheap hash functions. Content defined approaches [START_REF] Muthitacharoen | A low-bandwidth network file system[END_REF] on the other hand use a variable chunk size calculated using a sliding window over the data that hashes the window content at each step using Rabin's fingerprinting method [START_REF] Rabin | Fingerprinting by random polynomials[END_REF]. This approach was used in several storage systems [START_REF] Zhu | Avoiding the disk bottleneck in the data domain deduplication file system[END_REF], [START_REF] Dubnicki | Hydrastor: a scalable secondary storage[END_REF].

Different studies of block replication and erasure codes [START_REF] Reed | Polynomial codes over certain finite fields[END_REF] were performed before in the context of RAID systems [START_REF] Brown | Towards availability benchmarks: a case study of software raid systems[END_REF], whose implementation is at hardware level, as well as for distributed data storage [START_REF] Weatherspoon | Erasure coding vs. replication: A quantitative comparison[END_REF] implemented at the software level. Several works such as DiskReduce [START_REF] Fan | Diskreduce: Raid for data-intensive scalable computing[END_REF] and Zhe Zhang et al. [START_REF] Zhang | Does erasure coding have a role to play in my data center?[END_REF] study the feasibility of replacing three-way replication with erasure codes in cloud storage and large data centers. Such techniques can complement our approach in the sense that data not duplicated to a sufficient degree can be made resilient through erasure codes as an alternative to replication.

Our own previous work [START_REF] Nicolae | Towards Scalable Checkpoint Restart: A Collective Inline Memory Contents Deduplication Proposal[END_REF] focuses on the benefits of eliminating duplicates at global level, before datasets are collectively written to persistent storage. Although closely related, the goal of our previous work is to minimize redundancy, which is precisely what we want to avoid in this work: it focuses on how to efficiently apply interprocess deduplication techniques in the context of partner replication in order to leverage naturally available distributed duplicated data pieces as natural replicas without compromising the desired resilience and high availability level, while at the same time decreasing performance overhead and resource utilization. To our best knowledge, we are the first to explore the benefits of deduplication under such circumstances.

III. SYSTEM DESIGN

A. Assumptions

We target applications that are composed of a set of tightly coupled processes (also referred to as ranks) that need to simultaneously write a local dataset (potentially related to the other datsets) during runtime. A typical example of this I/O access pattern is exhibited by checkpointrestart: at regular intervals, all processes save checkpointing information that can be later used to restart in case of failures. While we mainly use this scenario in Section V to demonstrate the benefits of our approach experimentally, our proposal is general enough to address other related scenarios as well, e.g. dumping of visualization output during a numerical simulation. To this end, we define a collective I/O primitive that acts as a synchronization point and is used by all processes to specify the local dataset and initiate the parallel write: DUMP OUTPUT(buf f er, K)

For simplicity, we assume the local dataset buf f er resides in the memory of each process and needs to be written on the local storage device of the node that hosts the process. Note that buf f er does not necessarily need to be a contiguous region. Since the local storage device and the node as a whole is prone to failures, we also assume that the data needs to be replicated to at least K -1 other remote nodes and stored on their local storage devices as well. For the rest of this paper, we refer to K as the replication factor and to the K -1 remote nodes as the replication partners of the initial node. Although a common scenario, it is not required for all processes to write the same amount of data. Our goal is to maximize the performance of the DUMP OUTPUT primitive while minimizing its resource utilization (i.e. storage space on the local devices and network traffic due to communication with replication partners).

B. Design overview

Our proposal relies on four key design principles, which are visually illustrated using an example in Figure 1.

Identify natural redundancy through collective interprocess deduplication: The central idea of of our approach is to identify the data pieces that are already duplicated between multiple processes hosted on different nodes, such that it is possible to replicate only the data pieces that are not already naturally duplicated at least as much as the desired replication factor.

To this end, we split the local dataset into small fixed sized chunks and compute a hash value (called fingerprint) for each chunk that "uniquely" represents it (the term unique is abused here, because hash collisions are theoretically possible but negligible in practice [START_REF] Nicolae | Towards Scalable Checkpoint Restart: A Collective Inline Memory Contents Deduplication Proposal[END_REF]). By using fingerprints, the complexity of the problem is greatly reduced, as comparisons and exchanges between partners involve only a small fraction of the original size of the local dataset. Based on this observation, we introduce a twophase deduplication strategy: in the first phase, each process identifies the duplicate chunks of its own dataset and keeps only one copy, which results in a set of locally unique fingerprints. In the second phase, the processes identify the frequency of each fingerprint (i.e., number of processes where it shows up). Depending on the frequency of each fingerprint, duplicates of the corresponding chunk are either added or removed to match the desired replication factor.

How to identify the frequency of fingerprints in the second phase is a non-trivial issue: at large scale, the number of fingerprints can quickly explode, making an exact solution expensive to compute. Thus, we aim to enforce an upper bound on the computational complexity of the problem while accepting a non-optimal solution. To this end, we relax the problem in the sense that we select only a maximum of F fingerprints for which we count the frequency, while considering the rest of them unique even if they are not. Although this relaxation does not affect correctness, the quality of the deduplication is highly dependent on which F fingerprints are chosen. Obviously, selecting the most frequent F fingerprints would maximize the deduplication potential, however, it is not possible to rank the fingerprints apriori without computing an exact solution.

To deal with this dilemma, we propose an efficient (logarithmic in the number of processes) reduction-based algorithm that performs both the selection and the frequency counting in a hierarchic bottom-up fashion. More specifically, it is based on a merge step that given two sets of fingerprints and the frequency of their appearance, outputs the F most frequent fingerprints of the union (the frequency of a fingerprint in the union is defined as the sum of its frequencies in the two sets). This merge step is performed in parallel starting from the initial fingerprints of pairs of processes until a single set of F fingerprints remains. Besides counting the frequency, the merge step also associates at most K processes for each fingerprint (which we refer to as designated ranks). Thus, the end result is a set of F fingerprints, each of which is mapped to its frequency and set of designated ranks. Once this global view is obtained, it is broadcast to all processes.

At this point, each process can consult the global view in order to check whether there is any fingerprint it holds for which it is not among the K designated ranks. If that is the case, then it means there are other K processes designated to store the chunk corresponding to the fingerprint, so it can be safely discarded as the desired replication factor was reached. Otherwise, for each of the remaining fingerprints (regardless whether it is in the global view or not), it needs to store the corresponding chunk locally. If the fingerprint of the chunk is in the global view and the number of designated ranks D is less than K, the chunk needs to be replicated to K -D remote partners. This can happen in parallel: each of the D designated ranks will be assigned to a subset of the K -D partners (distributed in a roundrobin fashion) and will send the chunk to all members of the subset. Otherwise, if the fingerprint of the chunk is not in the global view, each process needs to select K -1 partners and send them copies of the chunk. All chunks received by a process from its partners are saved locally together with the other chunks. Finally, once all processes have finished receiving and saving the chunks, the collective dump completes.

Load balancing by means of uniform rank assignment: How the designated ranks for each fingerprint are assigned in the reduction process is important. To better illustrate this point, consider the extreme case when all processes need to store the same local dataset that is not possible to deduplicate locally (i.e. all local chunks are Fig. 1. Our approach in a nutshell: Example with three processes that call the DUMP OUTPUT primitive with a replication factor K = 3. Fig. 2. Naïve partner selection (left) vs. load aware partner selection (right) for a replication factor of three: first two processes send 100 chunks (illustrated by arrows), the rest send 10 chunks (not illustrated). The value of each node is the total number of received chunks. Using a rank shuffling of (1,3,4,2,5,6) spreads the load more evenly: the maximum number of received chunks is lowered from 200 (left) to 110 (right). unique). In this case, only K copies of the dataset needs to be stored overall. However, if we use a naive solution that designates the same ranks for each chunk, we end up with K processes that are fully loaded and a large number of remaining processes that are idle and simply need to wait until all K designated processes have finished.

Thus, a much better idea is to try to assign the ranks in such way that the overall load is evenly distributed, which speeds up the most loaded (and thus slowest) process, effectively leading to a better overall performance. To this end, we propose a load-balancing algorithm that is embedded into the merge phase. More specifically, for each process we count the number of fingerprints it was designated for. Whenever we need to merge two fingerprints, if the combined list of ranks is larger than K, we truncate it in such way that the most loaded ranks are eliminated first, effectively shifting the assignment in favor of the lesser loaded processes. Thus, as the reduction progresses, the rank assignment for a particular fingerprint is constantly changing in order to reflect a better global load balancing.

Reduce unavoidable imbalance using load-aware partner selection: Even in the ideal case when the load corresponding to the F fingerprints is evenly spread, the remaining chunks that were not identified as duplicates may not be themselves evenly spread, which can lead to an overall imbalance. However, the negative consequences of this "unavoidable" imbalance can be limited by choosing the partners in an optimal configuration. To illustrate this point, consider six processes that need to replicate their chunks to two partners (K = 3). Let's assume all processes have 20 chunks in common. In this case, after applying the collective deduplication strategy mentioned above using a uniform rank assignment, each process needs to store and replicate 10 chunks to its partners. However, assuming the first two processes also have 90 unique chunks each, using a naive partner selection strategy that simply chooses ranks (i + 1)..(i + K -1) (mod N ) as partners for rank i results in a high imbalance, as shown in Figure 2.

To address this issue, we propose a load-aware partner selection strategy that is centered around the idea of shuffling the ranks in such way that we interleave the ranks that need to send a high number of chunks with the ranks that need to send a smaller number of chunks. Once the ranks are shuffled, applying the naive strategy will result in a much better load balancing, as shown in Figure 2.

A key requirement in this context is that all ranks agree on the same shuffling. To this end, we gather from each rank information about the load: how many chunks need to be stored locally and how many chunks need to be sent to each partner. Once each rank is aware of the load of every other rank (e.g., by using an all-gather collective), we calculate an interleaving that is uniquely shared by all ranks and achieves our goal of load-balancing of receive size. We detail this process in Section III-C. Note that more elaborate schemes are possible (e.g. that take into account topology or rack-awareness), however this is outside the scope of this work.

Low-overhead exchanges using single sided communication planning: Once each node has identified its partners, the exchanges between the processes can begin. However, a straightforward solution where each process tells its partners how much data it wants to send and then starts streaming the chunks suffers from significant overhead on the receiving side: the chunks need to be collected from multiple parallel streams and buffered before being written to local storage.

To address this issue, we propose a different highperformance communication model that relies on singlesided operations and thus can take advantage of technologies such as RDMA. The key difficulty in this context is how to expose a designated memory region to each partner in a consistent fashion, such that they can independently send their chunks directly at the right location to avoid extra buffering overhead. This is a non-trivial issue, because standardized APIs for single-sided operations (e.g. MPI) use the concept of window, which implies that a single memory region is exposed by one process to all other processes. Thus, it is insufficient for one process to know how much data it needs to send to its partners, because it would not know at what offset in each of the destination window to put it.

However, in our context we can take advantage of the load information that was gathered during the partner selection phase: since there is a unique shuffling, rank i (in the shuffled order) knows how many chunks the other ranks need to send to its partners. Thus, it is possible to calculate an offset for each of the partners of rank i in such way that the other ranks that share the same partners can implicitly agree without extra communication. We detail in Section III-C how this can be efficiently calculated. Furthermore, since each rank knows how many chunks it needs to receive from all other ranks, it can open a window of the right size from the beginning, avoiding any waste. This is an important issue, because applications typically occupy a large part of the memory by the time they call DUMP OUTPUT.

C. Algorithms

In this section, we show how to materialize the design principles presented in Section III-B through a series of algorithmic descriptions.

Algorithm 1 provides an overview of the process. In a first phase, we compute the hash values corresponding to locally unique chunks into the LHashes set. Starting from LHashes, we apply the collective parallel reduction strategy described in Section III-B in order to obtain the most globally frequent fingerprints and their designated ranks. The result is stored in the GHashes set. Note that the reduction can be efficiently parallelized using an optimized ALLREDUCE collective primitive (e.g., as implemented by MPI). The load balancing happens during the merge step (denoted HMERGE), as described in Section III-B.

By convention, Load[0] denotes the number of chunks that need to be stored locally, while Load[1..(K -1)] denotes the number of chunks that need to be sent to partner 1..(K -1). In the next step, we compute Load. This happens in two steps: first, for each fingerprint h i that is part of GHashes and for which the current rank M is among the list of designated ranks R i , it is necessary to compute the number of partners P to which the corresponding chunk needs to be sent to. If D = |R i | is equal or larger than K, then P = 0 (because there are Algorithm 1 Overview of our approach 1: procedure DUMP OUTPUT(buf, K)

2:

LHashes ← LOCAL DEDUP(buf, K) SendLoad ← ALLGATHER(Load)

11:

Shuf f le ← RANK SHUFFE(SendLoad)

12:

Of f sets ← CALC OFF(Shuf f le, SendLoad)

13:

for all i is my partner do write both designated and received chunks to local storage 17: end procedure enough replicas already). Otherwise, the number can be calculated by applying the round-robin allocation of the K -D replicas, as mentioned in Section III-B. Once P is calculated, Load[0..P ] is incremented. Since M is only one source for the remaining h i that are not part of GHashes, in the second step, Load[0..(K -1)] is incremented for each such h i .

Once Load is calculated, information about the load is gathered from all ranks and disseminated to everybody. Thus, each process has its own global view of the send load, which is held by SendLoad. Armed with this knowledge, the rank shuffling phase can begin, which is detailed in Algorithm 2: in a first step, we sort the ranks in descending order of the total send size. Then, we repeatedly pair a rank that has the most amount of chunks to send (head) with K -1 ranks that have the least amount of chunks to send (tail) until all ranks were processed. The result of the rank pairing is a new permutation Shuf f le, which is then used to calculate the Of f sets corresponding to the partner windows. This process is detailed in Algorithm 3. Finally, in a last phase, each process opens a window of the appropriate size and puts the chunks that need to be replicated remotely into the partner windows at the appropriate offset. Once the chunk exchange is complete, both the chunks for which the process was designated and the chunks received from its partners are committed to the local storage.

Algorithm 3 zooms on the offset calculation that enables efficient data transfers through single-sided communication.

The key idea here is to leverage the global knowledge about the load of each process in order to allocate a static welldefined region for each sender. More specifically, rank i uses offset 0 for its partner i + 1, offset j for its partner return Shuf f le 13: end function i + 2 (where j is the send size from i + 1 to i + 2), offset l + m for its partner i + 3 (where l is the send size from i + 1 to i + 3 and m is the send size from i + 2 to i + 3), etc.

Algorithm 3 Compute the offsets for partner windows

1: function CALC OFF(Shuf f le, SendLoad) 2: Of f [0..(K -1)] ← 0 3:
for all 1 <= i < K do 4: return Of f 10: end function

P ← Shuf f le[(i + 1) mod N ] 5: for all i + 1 <= j < K do 6: Of f [j] ← Of f [j] + SendLoad[P ][j -i]

IV. IMPLEMENTATION DETAILS

We implemented a prototype of our approach as a library that exposes the DUMP OUTPUT primitive to the application. We make use of the Boost C++ collection of libraries for standard data structures and algorithms. For the implementation of the collectives (i.e. ALLREDUCE and ALLGATHER, used in Section III-C), we rely on the Boost API that builds on top of MPI to include support for advanced features, such as automatic serialization of data structures. With respect to the hash function used to calculate the hash values that represent the chunks, we decided to use SHA1, a crypto-grade hash function specifically designed to minimize the chance of collisions. However, our approach fully supports other hash functions if a better trade-off between performance and collision chance is desired (e.g. in some scenarios occasional collisions are acceptable).

For the purpose of this work, we integrated our library with AC-FTE [START_REF]Ac-fte[END_REF], a run-time environment specifically designed to provide scalable checkpoint-restart functionality for tightly coupled applications, both transparently and at application level. In this context, we use the transparent mode to capture all memory pages that were allocated by the application during its runtime and then pass them to the DUMP OUTPUT primitive when a checkpoint is desired. The mechanism by which the memory pages are captured is built on top of jemalloc [START_REF] Evans | A scalable concurrent malloc(3) implementation for freebsd[END_REF], a scalable high performance malloc implementation designed to efficiently support concurrent allocations.

Note that since we rely on AC-FTE as a demonstrator, we match chunks with memory pages. Thus, the chunk size matches the system memory pages size (i.e. 4 KB by default). How to select an optimal chunk size to strike a good trade-off between deduplication overhead and amount of identified chunk redundancy is an interesting topic in itself but outside the scope of this work. Our library can be easily adapted to work with arbitrarily large chunk sizes or to integrate with other collective I/O standardization efforts, such as MPI-IO [START_REF] Snir | MPI-The Complete Reference[END_REF].

V. EVALUATION

A. Experimental Setup

Our experiments were performed on the Shamrock testbed of the Exascale Systems group of IBM Research in Dublin. For the purpose of this work, we used a reservation of 34 nodes interconnected with Gigabit Ethernet, each of which features an Intel Xeon X5670 CPU (6 cores, 12 hardware threads), HDD local storage of 1 TB and 128 GB of RAM. With respect to the software configuration, each node runs RedHat 6.3 Linux distribution, while the MPI environment is MPICH2 1.4.1. The deduplication was performed using fixed-sized chunks of 4 KB (corresponding to the size of the memory page), which were hashed using the SHA1 function of OpenSSL 1.0.1. All internal data structures for the collective deduplication and the MPI wrappers are based on Boost 1.53.

B. Methodology

We compare three approaches throughout our evaluation.

a) Full replication: In this setting, each process dumps all chunks of the dataset to local storage and replicates them to its partners before returning control to the application. The partner selection is based on a simple strategy that sends each chunk stored by rank i to ranks i + 1, i + 2, ... (i + j) mod N , where 0 < j < K, depending on how many duplicates of the chunk were identified. Furthermore, it makes use of the single sided communication planning strategy introduced in Section III-C. For the rest of this paper, we refer to this setting as no-dedup.

b) Replication of locally deduplicated data: In this setting, each process eliminates the duplicate chunks of its dataset before storing them to local storage and replicating them to its partners. Other than this, it is identical to the previous setting: it uses a simple replication strategy and makes use of our single sided communication planning strategy. For the rest of this paper, we refer to this setting as local-dedup. c) Replication using our collective approach: This setting implements our approach. It performs all stages introduced in Section III-C: in addition to the local deduplication, our collective scheme identifies the duplicate chunks across all processes and replicates only those that have not already reached the desired replication factor. The partner selection strategy performs a rank shuffling to improve the load balancing. and applies the single sided communication planning. For the rest of this paper, we refer to this setting as coll-dedup.

We use two real-life HPC applications to motivate the our approach. The collective I/O scenario we consider is checkpointing at regular intervals as part of a checkpointrestart fault tolerance strategy. In both cases, the local HDD acts as the local storage and the dataset to be dumped to the local HDD and replicated to the partners represents a full checkpoint of the content of all dynamic memory allocated by the process at the time when the checkpoint was taken. This corresponds to a system-level approach as performed by a transparent checkpointing library. We detail the two applications below.

1) HPCCG: is a simple conjugate gradient benchmark code for a 3D chimney domain on an arbitrary number of processes that generates a 27-point finite difference matrix with a user-prescribed sub-block size on each process. It is part of the Mantevo [START_REF]The mantevo project[END_REF] set of mini-apps, a set of small, self-contained programs that embody essential performance characteristics of key HPC applications. HPCCG was specifically designed to enable weak scalability benchmarking. This makes it an ideal candidate for the purpose of our work, because it is easy to control the amount of data per process. We fix the sub-block size per process to 150x150x150, which corresponds to approx. 1.5 GB per process. The total number of iterations is 127, with a checkpoint scheduled to be taken at iteration 100.

2) CM1: is a three-dimensional, non-hydrostatic, nonlinear, time-dependent numerical model suitable for idealized studies of atmospheric phenomena. This application is used to study small-scale processes that occur in the atmosphere of the Earth, such as hurricanes. It is representative of a distributed stencil computation that studies how a phenomenon evolves in time by splitting a large spatial domain into sub-domains that are assigned to distributed tightly-coupled processes. For the purpose of this work, we have chosen as input data a 3D hurricane that is a version of the Bryan and Rotunno simulations [START_REF] Bryan | The maximum intensity of tropical cyclones in axisymmetric numerical model simulations[END_REF]. We use a weak scalability setting for which the size of the subdomain solved by each process remains constant at 200x200. This corresponds to a memory usage of around 800 MB, out of which approx. 500 MB is constantly changed. The duration of the simulation is 70 time-steps, with a checkpoint taken each 30 time-steps.

C. Results using a variable number of processes

In this section we evaluate the performance our our approach for a variable number of processes (up to 408, corresponding to our maximum reservation of 34 nodes). We solve a weak scalability scenario that keeps the size of problem for each process constant, as detailed above. We aim to study two aspects: how effective the deduplication is at reducing the total amount of data that needs to be stored, as well as the impact this reduction has on speeding up the DUMP OUTPUT primitive.

Figure 3(a) shows the results for the total size of unique content identified by each of the three approaches in four configurations: HPCCG-196, CM1-256, HPCCG-408, CM1-408. Since, no-dedup does not attempt to identify any duplication, its total size of unique content coincides with the sum of the data sizes of all processes and will be used as a baseline. Comparatively, local-dedup identifies a large amount of data duplication for both applications: overall, the amount of unique content identified at the extreme of 408 is reduced to 33% for HPCCG and 30% for CM1. However, going even further, coll-dedup manages a reduction down to as little as 6% for HPCCG and 5% for CM1. These results have a significant impact on both the local HDD space needed to save the chunks as well as the network traffic generated by the replication, because both are proportional to the amount of unique content. Thus, under favorable circumstances, we conclude that even lowcapacity local storage is feasible to use or perhaps even RAM itself. Furthermore, less network traffic has the benefit of lower energy consumption and frees up bandwidth for other purposes (e.g., application communication).

Since the process of identifying duplicated chunks across processes incurs and an increasing overhead at increasing scale, we illustrate in Figure 3(b) and Figure 3(c) this overhead for an increasing number of processes. The threshold F is fixed at 2 17 . Since local deduplication is not affected by scale, it is used as a baseline for comparison. What is interesting to note is the relatively small overhead when increasing the number of replicas: even if the list of designated ranks grows for each fingerprint, the difference between the three coll-dedup curves is small. This shows that the parallel reduction can efficiently handle an increasing replication factor. Also interesting to note is the difference between HPCCG and CM1: the relative overheads are smaller for HPCCG compared to CM1. Further analysis revealed that this effect happens because the parallel reduction for HPCCG resulted in more fingerprints with less designated ranks when compared with CM1.

Overall, the overhead of hash calculation and parallel reduction pays off for the lesser amount of stored and replicated chunks, as can be observed from the performance improvements obtained for DUMP OUTPUT in Table I. We have chosen to focus on the weak scalability of each of the applications for a replication factor of three. The baseline in this context represents the time it takes the application to run to completion without any call to DUMP OUTPUT. For HPCCG, at the extreme of 408 processes, coll-dedup is 2.8x faster than local-dedup and 9.8x faster than no-dedup. Similarly, for CM1, coll-dedup is 2.5x faster than local-dedup and 7.4x faster than no-dedup. Thus, we conclude that our approach exhibits a good weak scalability and significantly improves the performance of DUMP OUTPUT, in addition to the benefits related to storage space and network traffic reduction.

D. Results using a variable replication factor

In this section, we focus on the scalability of our approach for an increasing replication factor. For all our experiments, we fix the number of processes to the maximum of 408 (corresponding to 34 nodes, the reservation size).

We depict the increase in execution time with respect to the baseline for all replication factors between one and six in Figure 4(a) and Figure 5(a). The baseline in this context means the completion time to run the application without checkpointing (i.e., without using DUMP OUTPUT). Thus, the increase in execution time is the difference between the completion time of the approaches we compare and the completion time of the baseline.

As expected, the scalability of no-dedup is poor when the replication factor increases: both in the case of HPCCG and CM1, the increase in execution time is 3x and, respectively, 5x higher for a replication factor of six compared with a replication factor of one. On the other hand, coll-dedup exhibits excellent scalability: for both HPCCG and CM1, the performance cost of increasing the replication factor is minimal, leading to a situation where a replication factor of six with coll-dedup is faster than a minimalist replication scenario (i.e., replication factor of two), both compared with local-dedup and no-dedup. Performancewise, in the case of HPCCG, coll-dedup is 2x faster than local-dedup and 6x faster than no-dedup for a replication factor of six. A similar trend is observable for CM1 as well: for a replication factor of six, coll-dedup is more than 8x faster than no-dedup, as well as 2.3x faster than local-dedup.

Since there is a strong correlation between the performance overhead and the communication overhead, we depict in Figure 4(b) and Figure 5(b) the average and maximal amount of data that each process has to send to its partners. In the case of HPCCG, no-dedup exhibits an overlap between the average and the maximum, which means every process needs to write and replicate the same amount of data. Comparatively, local-dedup exhibits a small gap between maximum and the average, which slowly grows with increasing replication factor. Since coll-dedup starts with a larger gap between average and maximum already at a replication factor of two, the growth of the gap is more pronounced. This insight reveals a potential limitation caused by load imbalance: for example, when using a replication factor of six, coll-dedup sends on the average 5x less data to its partners compared with local-dedup. However, the actual speedup is only 2x. In the case of CM1, we observe a growing gap between the maximum and the average send size for all three approaches. However, in this case, the maximum of coll-dedup is well below the average of local-dedup, enabling a higher speedup than in the case of HPCCG. Based on these observations, we conclude that the load imbalance resulting from collective deduplication can be significant in practice and needs to be addressed efficiently in order to fully exploit the reduction of communication overhead in order to improve the performance.

E. Impact of rank shuffling

In the previous section, we noted the importance of load balancing. While it is impossible to avoid the load imbalance caused by the different amounts of data that is unique to each process (and consequently the imbalance in the amount of data that each process has to send to its partners), we argue that it is at least possible to mitigate its negative effects by means of rank shuffling, as detailed Section III-B.

To quantify the benefits of rank shuffling, in this section we analyze the maximal receive size for an increasing replication factor both with and without rank shuffling active. We focus on coll-dedup in both cases. The case when our approach does not use rank shuffling is denoted coll-no-shuffle, while the case it does is denoted coll-shuffle.

The results for both HPCCG and CM1 are depicted in Figure 4(c) and, respectively, Figure 5(c). As can be observed, for a replication factor of two, there is no difference between coll-shuffle and coll-noshuffle. However, with increasing replication factor, the gap between the two approaches becomes clearly visible. In the case of HPCCG, there is an 8% reduction in maximal receive size, while in the case of CM1, the reduction is much larger and almost reaches 30%. In both cases, the difference between the two approaches remains roughly constant. Note that it is not necessary to depict the average receive size, because it matches the average send size discussed in Section V-D.

Thanks to this reduction of maximal receive size through better selection of the replication partners, coll-shuffle manages a more even distribution of the overall communication overhead when compared to coll-no-shuffle. Furthermore, very important to note is also the fact that the receive size directly corresponds to the additional amount of data that each process has to store locally. Thus, the benefits of better load balancing of communication overhead directly translate into a better load balancing of write overhead to local storage devices.

VI. CONCLUSIONS

Partner replication to local storage is a crucial technique to enable resilience and high availability at large scale for massive datasets that result from collective I/O writes under concurrency. However, the explosion of data sizes, scale and the need for everincreasing replication factors poses difficult challenges in terms of I/O performance, scalability and resource utilization (bandwidth and storage space). This paper introduced a novel approach that identifies and leverages naturally distributed data redundancy to minimize the amount of data that needs to be replicated, while at the same time eliminates data pieces that are duplicated more than the desired level. To this end, we introduced a low-overhead collective deduplication technique that is enhanced with even rank distribution, load-aware partner selection and single-sided communication planning.

We illustrated the benefits of our approach for two real-life applications in experiments that involve dozens of nodes and hundreds of cores. Performance-wise, our approach is 2.5x-2.8x faster compared with replication of locally deduplicated data and 7.4x-9.8x faster compared to full replication. Furthermore, as the replication factor increases, our approach can conserve more storage space and bandwidth (up to several orders of magnitude!) when compared to the other two approaches. Also important to note is the impact of the partner selection strategy: compared to a naive solution, our load-aware strategy managed to obtain a much better load balancing, with maximal receive size per process reduced by up to 30%.

Encouraged by these results, we plan to broaden the scope of our work in future efforts. One interesting direction is to combine our approach with other redundancy mechanisms, in particular erasure codes, which would act as a replacement for replication. Furthermore, we proposed a load-aware partner selection strategy that is based on the amount of send data only. What would be interesting to explore in this context are other partner selection criteria, such as rack-awareness or topology.

3 : 7 :

 37 GHashes ← ALLREDUCE(HMERGE, LHashes) ⊲ M ← my rank 4: for all h i ∈ GHashes where M ∈ R i do 5: increment Load based on R i for all h i ∈ LHashes and h i / ∈ GHashes do 8: increment all Load[0..K -1]

Algorithm 2 2 : 3 :sort RankIndex according to SendLoad 4 : 5 :i < N do 6 : 8 : 9 :

 22345689 Load aware partner selection based on rank shuffling to balance receive size 1: function RANK SHUFFLE(SendLoad) RankIndex ← [0..(N -1)] ⊲ N = no. of ranks head ← tail ← i ← 0 while Shuf f le[i++] ← RankIndex[head++] while j < K and head < tail do Shuf f le[i++] ← RankIndex[tail--]

Fig. 3 .

 3 Fig. 3. Effectiveness of deduplication: size of unique content and scalability of collective hash value reduction overhead.

  Increase in execution time: lower is better. Baseline (no checkpointing) is 279s. -max no-dedup-avg local-dedup-max local-dedup-avg coll-dedup-max coll-dedup-avg (b) Amount of replicated data per process: lower is better.

Fig. 4 .

 4 Fig. 4. HPCCG: scalability results for increasing number of replicas. Problem size is 408 processes (corresponding to 34 nodes).

  Impact of rank shuffling: lower maximal receive size indicates better load balancing.

Fig. 5 .

 5 Fig. 5. CM1: scalability results for increasing number of replicas. Problem size is 408 processes (corresponding to 34 nodes).


Olivier Garet
email: olivier.garet@univ-lorraine.fr

AND Régine Marchand
email: regine.marchand@univ-lorraine.fr

Marie Théret
email: marie.theret@univ-paris-diderot.fr

CONTINUITY OF THE ASYMPTOTIC SHAPE OF THE SUPERCRITICAL CONTACT PROCESS

Keywords: 2000 Mathematics Subject Classification. 60K35, 82B43 contact process, shape theorem, continuity

We prove the continuity of the shape governing the asymptotic growth of the supercritical contact process in Z d , with respect to the infection parameter. The proof is valid in any dimension d ≥ 1.

Introduction

The contact process is a famous interacting particle system modelling the spread of an infection on the sites of Z d . The evolution depends on a fixed parameter λ ∈ (0, +∞) and is as follows: at each moment, an infected site becomes healthy at rate 1 while a healthy site becomes infected at a rate equal to λ times the number of its infected neighbors. There exists a critical value λ c (Z d ) such that the infection, starting from the origin, infinitely expands with positive probability if and only if λ > λ c (Z d ).

Durrett and Griffeath [START_REF] Durrett | Contact processes in several dimensions[END_REF] proved that when the contact process on Z d starting from the origin survives, the set of sites occupied before time t satisfies an asymptotic shape theorem, as in first-passage percolation. In [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF], two of us extended this result to the case of the contact process in a random environment. The shape theorem can be stated as follows: provided that λ > λ c (Z d ), there exists a norm µ λ on R d such that the set H t of points already infected before time t satisfies:

P λ ∃T > 0 : t ≥ T =⇒ (1 -ε)tS(λ) ⊂ Ht ⊂ (1 + ε)tS(λ) = 1,
where Ht = H t + [0, 1] d , S(λ) is the unit ball for µ λ and P λ is the law of the contact process with parameter λ, starting from the origin and conditioned to survive. The growth of the contact process is thus asymptotically linear in time, and governed by the shape S(λ).

The aim of this note is to prove the continuity of the map λ → S(λ). More precisely, we prove the following result: Theorem 1. For every λ > λ c (Z d ), lim

λ ′ →λ sup x∈S d-1 |µ λ ′ (x) -µ λ (x)| = 0, where S d-1 = x = (x i ) 1≤i≤d ∈ R d : x 1 = d i=1 |x i | = 1 .
It is then easy to deduce the following continuity for the asymptotic shape. Denote by d H the Hausdorff distance between non-empty compact sets in R d . For every λ > λ c (Z d ), lim

λ ′ →λ d H (S(λ ′ ), S(λ)) = 0.
Continuity properties for asymptotic shapes in random growth models have already been investigated. In first passage percolation, perhaps the most famous random growth model, Cox and Kesten [START_REF] Cox | The time constant of first-passage percolation on the square lattice[END_REF][START_REF] Cox | On the continuity of the time constant of first-passage percolation[END_REF][START_REF] Kesten | Aspects of first passage percolation[END_REF] proved that the time constant is continuous with respect to the distribution of the passage-time of an edge. In a forthcoming paper, Garet, Marchand, Procaccia and Théret [START_REF] Garet | Continuity of the time and isoperimetric constants in supercritical percolation[END_REF] extend their result to the case of possibly infinite passage times by renormalization techniques.

In these two cases, thanks to a good subadditivity property, the quantity whose continuity is studied appears as an infimum of a decreasing sequence of continuous functions, which gives quite easily one half of the continuity.

Because of the possibility of extinction of the contact process, the subadditivity properties are not so obvious and we thus use the essential hitting time presented in Garet-Marchand [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF].

Note that the one-dimensional case is simpler because the growth of the supercritical contact process in dimension 1 is characterized by the right-edge velocity: its continuity is proved in Liggett [START_REF] Thomas | Interacting particle systems, volume 276 of Grundlehren der Mathematischen Wissenschaften[END_REF], Theorem 3.36. See also Durrett [3] for an analogous result about 2D oriented percolation.

In Section 2, we introduce the notation, build contact processes with distinct infection parameters on the same space thanks to the Harris construction and recall the definition and properties of the essential hitting time introduced in [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF]. Section 3 is devoted to the proof of the left-continuity, while in Section 4 we prove the rightcontinuity.

Notation and known results

We work on the grid Z d , with d ≥ 1, and we put an edge between any pair of sites at distance 1 for . 1 . We denote by E d the set of these edges.

To define the contact process, we use the Harris construction [START_REF] Harris | Additive set-valued Markov processes and graphical methods[END_REF]. It allows to couple contact processes starting from distinct initial configurations and distinct parameters λ ∈ (0, λ max ], where λ max > 0 is fixed and finite, by building them from a single collection of Poisson measures on R + .

Construction of the Poisson measures.

We endow R + with the Borel σalgebra B(R + ), and we denote by M the set of locally finite counting measures m = +∞ i=0 δ ti . We endow this set with the σ-algebra M generated by the maps m → m(B), where B describes the set of Borel sets in R + .

As the continuity is a local property, it will be sufficient in the sequel to build a coupling for contact processes with parameters in (0, λ max ], for a fixed and well chosen λ max > 0. We define the measurable space (Ω, F ) by setting

Ω = M E d × M Z d × ([0, λ max ] N ) E d and F = M ⊗E d ⊗ M ⊗Z d ⊗ ([0, λ max ] ⊗N ) ⊗E d .
On this space, we consider the probability measure defined by

P = P ⊗E d λmax ⊗ P ⊗Z d 1 ⊗ (U ([0, λ max ]) ⊗N ) ⊗E d ,
m e λ = m λ (ω e , (U e i ) i≥1 ) = +∞ i=1 1 1 {U e i ≤ λ λmax } δ S e i .
Under P, the random variable m λ is a Poisson point process with parameter λ. We then define, for λ ≤ λ max , the application

Ψ λ : Ω -→ M E d × M Z d ((ω e ) e∈E d , (ω z ) z∈Z d , (U i e ) e∈E d ,i≥1 ) -→ ((m λ (ω e , (U e i ) i≥1 )) e∈E d , (ω z ) z∈Z d ).
The law of Ψ λ under P is then

P λ = P ⊗E d λ ⊗ P ⊗Z d 1 .
We thus recover infection processes, indexed by E d , with parameter λ and recovering processes, indexed by Z d , with parameter 1. Note that the Poisson measures for recoverings, (ω z ) z∈Z d , do not depend on λ. The following lemma will be useful to compare the evolution of two contact processes with different parameters. Lemma 2. Let t > 0 and let S be a finite subset of

E d . Assume 0 < λ ′ ≤ λ ≤ λ max and note Idem(S, t, λ, λ ′ ) = ∩ e∈S m e λ [0,t] = m e λ ′ [0,t] . For each ε > 0, there exists δ = δ(S, t, ε) > 0 such that ∀λ, λ ′ ∈ (0, λ max ] |λ ′ -λ| ≤ δ ⇒ P(Idem(S, t, λ, λ ′ )) ≥ 1 -ε. Proof. Let λ, λ ′ ∈ (0, λ max ],
and assume without loss of generality that λ ≤ λ ′ .

For each e ∈ E d and t > 0, set

D e t = +∞ i=1 1 1 { λ ′ λmax <U e i ≤ λ λmax } 1 1 {S e i ≤t} , then E(D e t ) = λ -λ ′ λ max E(ω e ([0, t])) = λ -λ ′ λ max λ max t = (λ ′ -λ)t. Now, P(Idem(S, t, λ, λ ′ ) c ) ≤ e∈S P(m e λ [0,t] = m e λ ′ [0,t] ) ≤ e∈S P(D e t ≥ 1) ≤ e∈S E(D e t ) ≤ |S|t(λ ′ -λ),
so we can take δ = 1/(t|S|ε).

Graphical construction of the contact process.

This construction is exposed in all details in Harris [START_REF] Harris | Additive set-valued Markov processes and graphical methods[END_REF]; we just give here an informal description. Suppose that λ ∈ (0,

λ max ] is fixed. Let ω = ((ω e ) e∈E d , (ω z ) z∈Z d , (U i e ) e∈E d ,i≥1 ) ∈ Ω. Above each site z ∈ Z d ,
we draw a time line R + , and we put a cross at the times given by ω z , corresponding to potential recoverings at site z. Above each edge e ∈ E d , we draw at the times given by m λ ((ω e ) e∈E d , (U i e ) e∈E d ,i≥1 ) an horizontal segment between the extremities of the edge, corresponding to a potential infection through edge e (remember we fix the infection rate λ).

An open path follows the time lines above sites -but crossing a cross is forbidden -and uses horizontal segments to jump from a time line to a neighboring time line: in this description, the evolution of the contact process looks like a percolation process, oriented in time but not in space. For x, y ∈ Z d and t ≥ 0, we say that ξ λ,x t (y) = 1 if and only if there exists an open path from (x, 0) to (y, t), then we define:

ξ λ,x t = {y ∈ Z d : ξ λ,x t (y) = 1}, ∀A ∈ P(Z d ) ξ λ,A t = x∈A ξ λ,x t . (1)
For instance, we obtain

(A ⊂ B, λ ≤ λ ′ ) ⇒ (∀t ≥ 0 ξ λ,A t ⊂ ξ λ ′ ,B t ).
Harris proved that under P, or under P λ , the process (ξ λ,A t ) t≥0 is the contact process with infection rate λ, starting from initial configuration A.

Translations.

For t ≥ 0, we define the translation operator θ t on a locally finite counting measure m = +∞ i=1 δ ti on R + by setting

θ t m = +∞ i=1 1 1 {ti≥t} δ ti-t .
The translation θ t induces an operator on Ω, still denoted by θ t :

for every ω = ((ω e ) e∈E d , (ω z ) z∈Z d , (U i e ) e∈E d ,i≥1 ) ∈ Ω, we set θ t (ω) = ((θ t ω e ) e∈E d , (θ t ω z ) z∈Z d , (U i+ωe([0,t]) e ) e∈E d ,i≥1 ).
Since the Poisson point processes are translation invariant and ω e ([0, t]) is independent from the (U i e )'s, P and P λ are invariant under θ t . There is also a natural action of Z d on Ω, which preserves P and P λ , and which consists in changing the observer's point of view: for x ∈ Z d , we define the translation operator T x by setting:

∀ω ∈ Ω T x (ω) = ((ω x+e ) e∈E d , (ω x+z ) z∈Z d , (U i x+e ) e∈E d ;i≥1
), where x + e the edge e translated by vector x.

Notation and classical estimates for the contact process. For a set

A ⊂ Z d , we define the life time τ A λ of the process starting from A by

τ A λ = inf{t ≥ 0 : ξ λ,A t = ∅}. If y ∈ Z d , we write τ y λ instead of τ {y} λ
and we simply write τ λ for τ 0 λ . We also note

{τ λ = +∞} = {0 λ ↔ ∞}.
The critical parameter for the contact process in Z d is then

λ c (Z d ) = inf{λ > 0 : P λ (τ λ = +∞) > 0} ∈ (0, +∞).
Define, for λ > λ c (Z d ), the following conditional probability

P λ (.) = P λ (.|τ λ = +∞) = P( . ∩ {0 λ ↔ ∞}) P(0 λ ↔ ∞)
.

For A ⊂ Z d and x ∈ Z d , we also define the first infection time

t A λ (x) of site x from set A by t A λ (x) = inf{t ≥ 0 : x ∈ ξ λ,A t }.
If y ∈ Z d , we write t y λ (x) instead of t {y} λ (x) and we simply write t λ (x) for t 0 λ (x). The set of points infected before time t is then

H λ t = {x ∈ Z d : t λ (x) ≤ t} and Hλ t = H t + [0, 1] d .
The following estimates are classical for the supercritical contact process; they are stated here with some extra uniformity in the parameter λ (this uniformity is mainly obtained by stochastic comparison): Proposition 3 (Proposition 5 in Garet-Marchand [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF]). Let λ min , λ max with λ c (Z d ) < λ min ≤ λ max . There exist A, B, C, c, ρ > 0 such that for every λ ∈ [λ min , λ max ], for every x ∈ Z d , for every t ≥ 0,

P(τ λ = +∞) ≥ ρ, ( 2
)
P(H λ t ⊂ [-Ct, Ct] d ) ≤ A exp(-Bt), (3)
P(t < τ λ < +∞) ≤ A exp(-Bt), (4)
P t λ (x) ≥ x c + t, τ λ = +∞ ≤ A exp(-Bt). (5)
2.5. Essential hitting times and shape theorem. We now recall the definition of the essential hitting time σ λ (x). It was introduced in [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF] to prove an asymptotic shape result for the supercritical contact process in random environment. See also Garet-Marchand [START_REF] Garet | Large deviations for the contact process in random environment[END_REF] and Garet-Gouéré-Marchand [START_REF] Garet | The number of open paths in oriented percolation[END_REF] for further uses. The essential hitting time σ λ (x) is a time when the site x is infected from the origin 0 and also has an infinite life time. It is defined through a family of stopping times as follows: we set u 0 (x) = v 0 (x) = 0 and we define recursively two increasing sequences of stopping times (u n (x)) n≥0 and (v n (x)) n≥0 with

u 0 (x) = v 0 (x) ≤ u 1 (x) ≤ v 1 (x) ≤ u 2 (x) . . . : • Assume that v k (x) is defined. We set u k+1 (x) = inf{t ≥ v k (x) : x ∈ ξ λ,0 t }. • Assume that u k (x) is defined, with k ≥ 1. We set v k (x) = u k (x)+τ x λ •θ u k (x) . We then set (6) K λ (x) = min{n ≥ 0 : v n (x) = +∞ or u n+1 (x) = +∞}.
This quantity represents the number of steps before we stop: either we stop because we have just found an infinite v n (x), which corresponds to a time u n (x) when x is occupied and has infinite progeny, or we stop because we have just found an infinite u n+1 (x), which says that after v n (x), site x is never infected anymore. When the contact process survives, the second case does almost surely not occur.

In [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF], using (4) and ( 5), it is proved that K λ (x) is almost surely finite, which allows to define the essential hitting time σ λ (x) by setting

σ λ (x) = u K λ (x) .
It is of course larger than the hitting time t λ (x) and can been seen as a regeneration time. At the same time, we define the operator Θλ

x on Ω by:

Θλ x = T x • θ σ λ (x) if σ λ (x) < +∞, T x otherwise.
The advantage of the essential hitting time σ λ (x) is that it enjoys, unlike t λ (x), some good invariance and integrability properties when conditioned to survive. We now recall the main results of [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF] we will need here. In the following, we fix λ min , λ max > 0 such that λ c (Z d ) < λ min ≤ λ max .

Proposition 4 (Garet-Marchand [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF], Theorems 1 and 3, Corollary 21, Theorem 22 and Lemma 29).

• For each λ > λ c (Z d ), for every x ∈ Z d , [START_REF] Garet | Large deviations for the contact process in random environment[END_REF] the probability measure P λ is invariant under the map Θλ x . • There exist constants (C p ) p≥1 such that for every λ ∈ [λ min , λ max ], for every

x ∈ Z d , for every p ≥ 1,

(8) E λ [σ λ (x) p ] ≤ C p (1 + x ) p .
• For each λ > λ c (Z d ), for every x ∈ Z d , there exists a deterministic µ λ (x) such that

(9) lim n→+∞ t λ (nx) n = lim n→+∞ σ λ (nx) n = µ λ (x).
The convergence holds P λ almost surely, and also in L 1 (P λ ). • The function x → µ λ (x) can be extended to a norm on R d . Let [START_REF] Kesten | Aspects of first passage percolation[END_REF] S(λ) = {x ∈ R d : µ λ (x) ≤ 1}.

• For every ε > 0, P λ -a.s., for every t large enough,

(11) (1 -ε)S(λ) ⊂ Hλ t t ⊂ (1 + ε)S(λ).
It has also been noted in [START_REF] Garet | Asymptotic shape for the contact process in random environment[END_REF] (see proof of Lemma 25 there) that there exists M 1 > 0 such that, for each λ ∈ [λ min , λ max ] and each x ∈ Z d \{0}, the sequence (E λ σ λ (nx) + M 1 ) n≥1 is subadditive. Thus, with (9), we can represent µ λ (x) as the following infimum:

(12) ∀λ ∈ [λ min , λ max ] ∀x ∈ Z d µ λ (x) = inf n≥1 M 1 + E λ (σ λ (nx)) n .
As a corollary of (9), we obtain the following monotonicity property:

Corollary 5. For each x ∈ Z d , λ → µ λ (x) is non-increasing on (λ c (Z d ), +∞). Proof. Suppose λ c (Z d ) < λ ′ < λ < +∞. Choose λ min , λ max with λ c (Z d ) < λ min < λ ′ < λ ≤ λ max .
Use the construction of Subsection 2.2 to build the two contact processes with respective parameters λ and λ ′ . On the event {0 λ ′ ↔ ∞}, which has positive probability, we have that for each n ≥ 1, t λ (nx) n ≤ t λ ′ (nx) n . Letting n go to infinity, we get µ λ ′ (x) ≤ µ λ (x).

Left-Continuity

We prove here the left-continuity of µ λ . More precisely, we prove that for each λ 0 > λ c (Z d ), for every ε > 0, there exists δ > 0 such that

(13) ∀λ ∈ [λ 0 -δ, λ 0 ] ∀x ∈ S d-1 |µ λ0 (x) -µ λ (x)| ≤ ε.
When proving continuity theorems for the time constant in first passage percolation (see Cox and Kesten [1,[START_REF] Cox | On the continuity of the time constant of first-passage percolation[END_REF][START_REF] Kesten | Aspects of first passage percolation[END_REF]), the left-continuity is usually considered as the easy part, due to the fact that the time constant is an infimum. In the case of the contact process, there are extra difficulties, because contact processes with different intensities can not be coupled in such a way that they die simultaneously.

Lemma 6. Let λ > λ c (Z d ). For each x ∈ Z d , lim λ ′ →λ - E λ ′ (σ λ ′ (x)) ≤ E λ (σ λ (x)). Proof. Fix λ > λ c (Z d ). Choose λ min such that λ c (Z d ) < λ min < λ and set λ max = λ. Fix x ∈ Z d . Use the construction of Subsection 2.2.
In this proof, for λ ′ ∈ [λ min , λ], we note σ λ ′ = σ λ ′ (x). Suppose that for every λ ′ ∈ [λ min , λ], we have managed to construct a "good" event G(λ ′ ) such that

σ λ ′ = σ λ on G(λ ′ ). Then, for λ ′ ∈ [λ min , λ], E σ λ ′ , 0 λ ′ ↔ ∞ = E σ λ ′ , 0 λ ′ ↔ ∞, G(λ ′ ) + E σ λ ′ , 0 λ ′ ↔ ∞, G(λ ′ ) c ≤ E σ λ , 0 λ ′ ↔ ∞ + E σ 2 λ ′ , 0 λ ′ ↔ ∞ P(0 λ ′ ↔ ∞, G(λ ′ ) c ).
Now, using the fact that {0

λ ′ ↔ +∞} ⊂ {0
λ ↔ +∞} and the control (8) on the moments of σ λ ′ , we get

E λ ′ (σ λ ′ ) = E σ λ ′ , 0 λ ′ ↔ ∞ P(0 λ ′ ↔ ∞) ≤ P(0 λ ↔ +∞) P(0 λ ′ ↔ +∞) E λ (σ λ ) + E λ ′ (σ 2 λ ′ ) P(0 λ ↔ +∞) P(0 λ ′ ↔ +∞) P λ (G(λ ′ ) c ) ≤ P(0 λ ↔ +∞) P(0 λ ′ ↔ +∞) E λ (σ λ ) + C 2 (1 + x 2 ) P λ (G(λ ′ ) c ) . ( 14
)
Thus if we prove that P(0 λ ↔+∞)

P(0 λ ′ ↔+∞)
and P λ (G(λ ′ )) are close to 1 when λ ′ is close to λ, we can complete the proof.

We now build the "good" event G(λ ′ ) such that σ λ ′ = σ λ on G(λ ′ ) and such that P λ (G(λ ′ )) goes to 1 as λ ′ goes to λ. Since σ λ is P λ -a.s. finite and H λ σ λ is P λ -a.s. a finite set, we can first choose M > 0 such that (15)

P λ (A M ) ≥ 1 - ε 3 , where A M = {H λ σ λ ⊂ [-M, M ] d } ∩ {σ λ ≤ M }.
Then, estimates (3) and (4) let us choose L > 0 such that for each λ ′ ∈ [λ min , λ] (16)

P λ (B L (λ ′ )) ≥ 1 - ε 3 , where B L (λ ′ ) = {H λ ′ L ⊂ [-CL, CL] d } ∩ {L < τ λ ′ < ∞} c . Set S = [-(M + CL), (M + CL)] d ∩ Z d and t = M + L. With Lemma 2, we can choose δ > 0 such that (17) ∀λ ′ ∈ [λ -δ, λ] P λ (Idem(S, t, λ, λ ′ )) ≥ 1 -ε/3.
Finally, we consider, for every

λ ′ ∈ [λ -δ, λ], the event G(λ ′ ) = A M ∩ Θ-1 x,λ (B L (λ ′ )) ∩ Idem(S, t, λ, λ ′
). The choices (15), ( 16) and (17) we respectively made for M, L and δ, and the invariance property [START_REF] Garet | Large deviations for the contact process in random environment[END_REF] ensure that

P λ (G(λ ′ )) ≥ 1 -ε.
The event A M says that, apart from the fact that (x, σ λ ) λ ↔ ∞, the time σ λ is determinated by the configuration of the Poisson processes in the space-time box

[-M, M ] d × [0, M ]. The event Θ-1
x,λ (B L (λ ′ )) says that, if (x, σ λ ) has a progeny for parameter λ ′ still alive at time σ λ +L, then (x, σ λ ) λ ↔ ∞. The event Idem(S, t, λ, λ ′ ) says that the infection at rate λ ′ in the box S×[0, t] behaves exactly like the infection at rate λ in the same box. Now, on the event G(λ ′ )∩{0 λ ↔ ∞}, the point (x, σ λ ) has a progeny for parameter λ that is still alive at time σ λ + L. But the infections with rate λ and λ ′ coincide in the box S × [0, t], so the point (x, σ λ ) has a progeny for parameter λ ′ that is also still alive at times σ λ + L. Then (x, σ λ )

λ ′
↔ ∞, and it is now easy to see that

σ λ ′ = σ λ . Note also that G(λ ′ ) ∩ {τ λ = +∞} ⊂ {τ λ ′ = +∞}. This gives P(0 λ ↔ +∞) P(0 λ ′ ↔ +∞) = 1 P λ (0 λ ′ ↔ +∞) ≤ 1 P λ (G(λ ′ )) ≤ 1 1 -ε
and, coming back to (14), we see that

∀λ ′ ∈ [λ -δ, λ] E λ ′ (σ λ ′ ) ≤ 1 1 -ε E λ (σ λ ) + C 2 (1 + x 2 ) √ ε .
This completes the proof.

Lemma 7. For each x ∈ Z d , λ → µ λ (x) is left-continuous on (λ c (Z d ), +∞).
Proof. Fix x ∈ Z d . Since, from Corollary 5, the application λ → µ λ (x) is nonincreasing on (λ c (Z d ), +∞), we can define L = lim λ ′ →λ -µ λ ′ (x). Obviously L ≥ µ λ (x) and we must prove L ≤ µ λ (x). Note λ n = λ -1/n. Using the representation (12) of µ λ (x) as an infimum, we have

L = inf n≥1 µ λn (x) = inf n≥1 inf k≥1 E λn (σ λn (kx)) + M 1 k = inf k≥1 inf n≥1 E λn (σ λn (kx)) + M 1 k = inf k≥1 M 1 k + inf n≥1 E λn (σ λn (kx)) k By Lemma 6, for each k, inf n≥1 E λn (σ λn (kx)) ≤ E λ (σ λ (kx)), so L ≤ inf k≥1 M 1 k + E λ (σ λ (kx)) k = µ λ (x),
which completes the proof.

The difference between (13) and Lemma 7 is the uniformity of the control. For all λ > 0, since µ λ is a norm and by symmetry of the model, we have for all x, y ∈ R d ,

µ λ (x) -µ λ (y) ≤ µ λ (x -y) ≤ x -y 1 µ λ (e 1 ) ,
where e 1 = (1, 0, . . . , 0). We obtain that |µ λ (x) -µ λ (y)| ≤ x -y 1 µ λ (e 1 ). Fix λ 0 ∈ (λ c , +∞) and ε > 0. By Lemma 7 we know that lim λ→λ - 0 µ λ (e 1 ) = µ λ0 (e 1 ), thus there exists δ > 0 such that for all λ ∈ [λ 0 -δ, λ 0 ], for all x, y ∈ R d , we have |µ λ (x) -µ λ (y)| ≤ 2 x -y 1 µ λ0 (e 1 ). We obtain the existence of η > 0 such that for all x, y ∈ R d satisfying x -y 1 ≤ η, we have sup

λ∈[λ0-δ,λ0] {|µ λ (x) -µ λ (y)|} ≤ ε .
There exists a finite set of points y 1 , . . . , y m in R d such that

S d-1 ⊂ m i=1 {x ∈ R d : x -y i 1 ≤ η} , thus for all λ ∈ [λ 0 -δ, λ 0 ] we obtain sup x∈S d-1 |µ λ (x) -µ λ0 (x)| ≤ 2ε + max i=1,...,m |µ λ (y i ) -µ λ0 (y i )| .
By homogeneity of µ λ , the result of Lemma 7 also holds for all y ∈ R d , in particular for y i , i ∈ {1, . . . , m}. This concludes the proof of (13).

We can notice that the previous argument also applies to the study of the rightcontinuity of µ λ . However, as we will see in the next section, we do not need it since we perform directly the study of the right-continuity of µ λ uniformly in all directions.

Right-continuity

We prove here the right-continuity of µ λ . More precisely, we prove that for each λ 0 > λ c (Z d ), for every ε > 0, there exists δ > 0 such that

(18) ∀λ ∈ [λ 0 , λ 0 + δ] ∀x ∈ S d-1 |µ λ0 (x) -µ λ (x)| ≤ ε.
As we will see, the right-continuity of the asymptotic shape of the contact process can be obtained by a slight modification of a part of the proof of the large deviations inequality for the contact process established by Garet and Marchand in [START_REF] Garet | Large deviations for the contact process in random environment[END_REF].

Let λ 0 > λ c (Z d ) be fixed. Fix λ min , λ max with λ c (Z d ) < λ min ≤ λ 0 < λ max . Let α, ε > 0 and L, N be positive integers. Consider λ ≥ λ 0 and close to λ 0 . We define the following event, relative to the space-time box

B N = B N (0, 0) = ([-N, N ] d ∩ Z d ) × [0, 2N ]: A α,L,N,ε λ,λ0 = ∀(x 0 , t 0 ) ∈ B N ξ x0,λ αLN -t0 • θ t0 ⊂ x 0 + (1 + ε)(αLN -t 0 )S(λ 0 ) ∩ ∀(x 0 , t 0 ) ∈ B N ∪ 0≤s≤αLN -t0 ξ x0,λ s • θ t0 ⊂] -LN, LN [ d .
Consider first A α,L,N,ε λ0,λ0 . The first part of the event ensures that the descendants, at time αLN , of any point (x 0 , t 0 ) in the box B N are included in x 0 +(1+ε)(αLN )S(λ): it is a sharp control, requiring the asymptotic shape Theorem for parameter λ 0 . The second part ensures that the descendants, at all times in [0, αLN ], of the whole box B N are included in ] -LN, LN [ d : the bound is rough, only based on the (at most) linear growth of the process with parameter λ 0 . Thus, the "good growth" event A α,L,N,ε λ0,λ0 is typical, and it has been proved that Lemma 8 ([7]). Fix λ 0 > λ c (Z d ). There exists α = α(λ 0 ) ∈ (0, 1) such that for every ε ∈ (0, 1), every L 0 > 0, there exists an integer L > L 0 such that lim

N →+∞ P(A α,L,N,ε λ0,λ0 ) = 1.
Garet and Marchand used Lemma 8 to prove the upper large deviations for the contact process: for every λ 0 ∈ [λ min , λ max ], provided that α = α(λ 0 ) is fixed as in Lemma 8, then for L greater than some L 0 = L > L 0 (ε, λ 0 ), they prove that there exists p 1 = p 1 (λ 0 , ε, L) > 0 such that P(A α,L,N,ε/3 λ0,λ0

) > p 1 =⇒ ∃A, B ∀t > 0 P(ξ 0,λ0 t ⊂ (1 + ε)tS(λ 0 )) ≤ A exp(-Bt).

The idea of the proof is as follows: a too fast infection from (0, 0) to Z d × {n} uses a too fast path, along which we should find a number of order θn of "bad growth" events, i.e. translated versions of (A α,L,N,ε/3 λ0,λ0

) c . The proof ends with a Peierls argument: the event (A α,L,N,ε/3 λ0,λ0

) c is local, thus its translated events are only locally dependent. If their probability is small enough, we can control the probability that there exists a path from (0, 0) to Z d × {n} with at least θn "bad growth" events.

Let's come back to the right-continuity. Fix λ 0 > λ c (Z d ) and ε > 0. Take α given by Lemma 8, L large enough, and p 1 (λ 0 , ε, L) > 0 as before. A look at the proof of the Peierl argument in [START_REF] Garet | Large deviations for the contact process in random environment[END_REF] should convince the reader that for each λ ≥ λ 0 , we have ) > p 1 .

Then, it follows that ∀λ ∈ [λ 0 , λ 1 ] ∃A, B ∀t > 0 P(ξ 0,λ t ⊂ (1 + ε)tS(λ 0 )) ≤ A exp(-Bt). Now, we can deduce from (2) and (3) that ∀λ ∈ [λ 0 , λ 1 ] ∃A, B ∀t > 0 P λ (H 0,λ t ⊂ (1 + ε)tS(λ 0 )) ≤ A exp(-Bt).

A detailed proof is provided in [START_REF] Garet | Large deviations for the contact process in random environment[END_REF] when deducing (62) from (61). Fix λ ∈ [λ 0 , λ 1 ], η > 0 and, using the asymptotic shape result [START_REF] Thomas | Interacting particle systems, volume 276 of Grundlehren der Mathematischen Wissenschaften[END_REF], choose t large enough to have A exp(-Bt) < 1/2 and P λ ((1 -η)tS(λ) ⊂ H 0,λ t ) < 1/2. This implies that the event {(1 -η)tS(λ) ⊂ H 0,λ t ⊂ (1 + ε)tS(λ 0 )} has positive probability; particularly, (1 -η)S(λ) ⊂ (1 + ε)S(λ 0 ), and, letting η tend to 0, we have This completes the proof of (18).

  where, for every λ ∈ R + , P λ is the law of a Poisson point process on R + with intensity λ and U ([a, b]) is the uniform law on the compact set[a, b].Fix an edge e. Denoting by (S e i ) i≥1 the atoms of ω e , we build the classical coupling between the Poisson measures of the infection processes with different parameters λ ∈ (0, λ max ]. Define

)

  > p 1 =⇒ ∃A, B ∀t > 0 P(ξ 0,λ t ⊂ (1 + ε)tS(λ 0 )) ≤ A exp(-Bt). Remember that the event A α,L,N,ε/3 λ,λ0is local. Thus, applying Lemma 2 with S = [-N, N ] d ∩ Z d and t = αLN , we obtain the existence of λ 1 ∈ (λ 0 , λ max ] such that ∀λ ∈ [λ 0 , λ 1 ] P(A α,L,N,ε/3 λ,λ0

  ∀λ ∈ [λ 0 , λ 1 ] S(λ) ⊂ (1 + ε)S(λ 0 ), or equivalently ∀λ ∈ [λ 0 , λ 1 ] ∀x ∈ R d µ λ0 (x) ≤ (1 + ε)µ λ (x).