{'fraction_non_alphanumeric': 0.042751123654228076, 'fraction_numerical': 0.012882826382356015, 'mean_word_length': 4.071428571428571, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
Using Straggler Replication to Reduce Latency in Large-scale Parallel Computing (Extended Version)

Users expect fast and fluid response from today's cloud infrastructure. Large-scale computing frameworks such as MapReduce divide jobs into many parallel tasks and execute them on different machines to enable faster processing. But the tasks on the slowest machines (straggling tasks) become the bottleneck in the completion of the job. One way to combat the variability in machine response time, is to add replicas of straggling tasks and wait for one copy to finish. In this paper we analyze how task replication strategies can be used to reduce latency, and their impact on the cost of computing resources. We use extreme value theory to show that the tail of the execution time distribution is the key factor in characterizing the trade-off between latency and computing cost. From this trade-off we can determine which task replication strategies reduce latency, without a large increase in computing cost. We also propose a heuristic algorithm to search for the best replication strategies when it is difficult to fit a simple distribution to model the empirical behavior of task execution time, and use the proposed analysis techniques. Evaluation of the heuristic policies on Google Trace data shows a significant latency reduction compared to the replication strategy used in MapReduce.


Introduction
Applications such as Google search, Dropbox, Netflix need to perform enormous amounts of computing, or data processing on the cloud. Recently, cloud computing is also being offered as a service by Amazon S3, Microsoft Azure etc. where users can rent machines by the hour to run their computing jobs. The large-scale sharing of computing resources makes cloud computing flexible and scalable.
Cloud computing frameworks such as MapReduce [8] and Hadoop [18] employ massive parallelization to reduce latency. Large jobs are divided into hundreds of tasks that can be executed parallely on different machines. An important class of parallel execution is "embarrasingly parallel" computation [22], which requires little or no effort in dividing the computation into independent parallel tasks. Several algorithms used in optimization and machine learning, for example the Alternating Direction Method of Multipliers (ADMM) [4], and Markov Chain Monte-Carlo (MCMC) [14], fall into this class and can be parallelized easily.
The execution time of a task on a machine is subject to stochastic variations due to co-hosting, virtualization and other hardware and network variations [11]. Thus, the key challenge in executing a job with a large number of tasks is the latency in waiting for the slowest tasks, or the "stragglers" to finish. As pointed out in [11, Table 1], the latency of executing many parallel tasks could be significantly larger (140 ms) than the median latency of a single task (1 ms).
In this work we analyze how replication of straggling tasks can be used to reduce latency. In particular, we develop a mathematical framework and provide insights into how task replication affects the trade-off between latency and computing cost.

Related prior work
The idea of replicating tasks in parallel computing has been recognized by system designers [5,10], and first adopted at a large scale via the "backup tasks" in MapReduce [8]. A line of systems work [2,3,15,23] further developed this idea to handle various performance variability issues in data centers.
While task replication has been studied in systems literature and also adopted in practice, there is not much work on careful mathematical analysis of replication strategies. In [21] replication strategies are analyzed, mainly for the single task case. In this paper we consider task replication in a computing job consisting of a large number of tasks, which corresponds closely to today's largescale cloud computing frameworks. We note that using replication or redundancy to reduce latency has also attracted attention in other contexts such as cloud storage and networking applications [12,13,17,19].

Our contributions
To the best of our knowledge, we establish the first formal analysis of task replication in jobs consisting of a large number of tasks, for the system model and relevant performance measures proposed. We analyze the trade-off between the latency and the cost of computing resources, and provide insights into design of task replication strategies.
In particular, we study a class of scheduling policies called single-fork policies and characterize how latency and resource usage depend on three parameters: when we replicate tasks, how many replicas are launched, and whether the original replicas are killed or not. We show that the tail of the execution time distribution (heavy, light or exponential tail) is the key factor that determines the choice of the task replication policy. In particular for heavy tail distributions e.g. Pareto, we identify scenarios where the latency and computing cost can be reduced simultaneously. We also propose a heuristic algorithm to find the best single-fork policy when it is hard to use the proposed analysis techniques for the empirical distribution of task execution time.

Organization
The rest of the paper is organized as follows. In Section 2 we formulate the problem, define the latency and cost metrics and introduce notation used in the paper. In Section 3 we provide a summary of the analysis of single-fork task replication policies. Then in Section 4 we describe a heuristic algorithm to find a good scheduling policy for execution time distributions for which we cannot find an analytical distribution that fits well. Proofs of the analysis are given in Section 5. In Section 6 we conclude with a discussion of the implications and future perspectives. Finally, results from order statistics and extreme value theory used in this work are given in the Appendix.

Problem Formulation
We now describe the system model, and propose the performance metrics used to evaluate a task replication strategy.

Notation
First, we define some notation used in this paper. Lower-case letters (e.g., x) denote a particular value of the corresponding random variable, which is denoted in upper-case letters (e.g., X). We denote the cumulative distribution function (CDF) of X by F X (x). Its complement, the tail distribution is denoted byF which may be more convenient to use than the c.d.f. sometimes. We denote the upper end point For i.i.d. random variables X 1 , X 2 , · · · , X n , we define X j:n as the j-th order statistic, i.e., the j-th smallest of the n random variables.

System Model
Consider a job consisting of n parallel tasks, where n is large. Analysis of real-world trace data shows that it is common for a job to contain hundreds or even thousands of tasks [16]. We assume the executing time of each task on a computing node is independent and identically distributed (i.i.d.) according to F X , where F X is the cumulative distribution function (CDF) of random variable X. The distribution F X accounts for the variability in the machine response due to various factors such as congestion, queuing, virtualization etc. We consider that there is an unlimited pool of machines such that each new task (or new replica) is assigned to a new machine. Hence, the execution time of a task can be assumed to be i.i.d. across machines.
Since we assume an unlimited pool of machines, the delay due to queueing of tasks at machines is small and can be subsumed as a constant additive term in the execution time X. Also, we do not consider the dependence of the task execution time on the size of the task itself. But it can be accounted for similarly by adding a constant delay (fixed across tasks of the same job) to the expected completion time of the job defined in Section 2.4.

Scheduling Policy
A scheduling policy or scheduler assigns tasks to different machines, possibly at different time instants. We assume that the scheduler receives instantaneous feedback notifying it when a machine finishes its assigned task. But there is no intermediate feedback indicating the status of processing of a task. When the scheduler receives notification that at least one replica of each of the n tasks has finished, it kills all the residual running replicas. It can also use the feedback to decide when to selectively launch or kill replicas in order to reduce the overall job completion time.
The times when the scheduler launches or kills replicas could be pre-determined (static policy) or be dependent on the feedback about the execution of other tasks in the job (dynamic policy). We focus our attention on a set of dynamic policies called single-fork policies, defined as follows. Definition 1 (Single-fork Scheduling policy). A single-fork scheduling policy π SF (p, r, l) launches a single replica of all n tasks at time 0. It waits until (1 − p)n tasks finish and then for each of the pn straggling tasks, chooses one of the following two actions: • replicate without relaunching (l = 0): launch r new replicas; • replicate with relaunching (l = 1): kill the original copy and launch r + 1 new replicas.
When the earliest replica of a task is executed, all the other replicas are terminated.
We use l to denote the number of original replicas of each task remaining after the forking point. Hence l = 0 when the original replica is killed and restarted, and l = 1 otherwise. Note that in for both the relaunching (l = 0) and no relaunching (l = 1) cases there are a total of r + 1 replicas running after the forking point. The effect of r and l on the replication of straggling tasks is illustrated in Fig. 1.
For simplicity of notation we assume that p is such that pn is an integer. We note that p = 0 corresponds to running n tasks in parallel and waiting for all to finish, which is the baseline case without any replication or relaunching.
Remark 1 (Backup tasks in MapReduce). The idea of 'backup' tasks used in Google's MapReduce [8], is a special case of the single-fork policy. Following our notation, it corresponds to r = 1 and l = 1. The value of p is tuned dynamically and hence not specified in [8].
Although we focus on single-fork policies in this paper, the analysis can be generalized to multi-fork policies, where new replicas of straggling tasks are launched at multiple times during the execution of the job. Forking multiple times can give a better latency-cost trade-off, but could be undesirable in practice due to additional delay and complexity in obtaining new and killing existing replicas.

Performance metrics
Our objective is to find the best single-fork policy π SF (p, r, l) for a given task execution time distribution F X . We now define the performance metrics of latency and resource usage that are used to evaluate different scheduling policies.
Definition 2 (Expected Latency). The expected latency E [T ] is the expected value of T , the time taken for at least one replica of each of the n tasks to finish. It can be expressed as, where T i is the minimum of the finish times of the replicas of task i, which depends on the scheduling policy used. Figure 2: Illustration of T and C for a job with two tasks, and two replicas of each task. The latency T = max(8, 10) = 10, and the computing cost is C = (8 + 6 + 10 + 5)/2 = 14.5.
Definition 3 (Expected Cost). The expected computing cost E [C] is the sum of the running times of all machines, normalized by n, the number of tasks in the job. The running time is the time from when the task is launched on a machine, until it finishes, or is killed by the scheduler.
For a user of a cloud computing service such as the Amazon Web Service (AWS), which charges the user by time and number of machines used, the money paid by the user to rent the machines is proportional to E [C].
Example 1. Suppose the scheduler launches r replicas of each of the n tasks at times t i,j for j = 1, 2, . . . r. Then the latency T i of the i th task is given by where X i,j are i.i.d. draws from the execution time distribution F X . The computing cost C can be expressed as where |x| + = max(0, x). Fig. 2 illustrates the execution of a job with two tasks, and evaluation of the corresponding latency T and cost C. Given two tasks, we launch two replicas of task 1 t 1,1 = 0 and t 1,2 = 2, and two replicas of task 2 at t 2,1 = 0 and t 2,2 = 5. The task execution times are X 1,1 = 8, X 1,2 = 7, X 2,1 = 11, and X 2,2 = 5. Machine M 1 finishes the task first at time t = 8, T 1 = 8 and the second replica running on M 2 is terminated before it finishes executing. Similarly, machine M 4 finishes task 2 at time T 2 = 10, and the replica running on M 3 is terminated. Thus the latency of the job is T = max {T 1 , T 2 } = 10. The cost is the sum of all running times normalized by n, i.e., C = (8 + 6 + 10 + 5)/2 = 14.5.
In this work we analyze the trade-off between E [T ] and E [C] for the single-fork policy and provide insights into design of scheduling policies that achieve a good trade-off.

Single-fork policy analysis
In this section we evaluate the performance metrics E [T ] and E [C] for a given single fork policy π SF (p, r, l). The main insight we get from this analysis is that the tail behavior (heavy, light or exponential) of F X is the key factor in characterizing the latency-cost trade-off. We demonstrate this for two canonical distributions: Pareto and Shifted Exponential. All proofs are deferred to Section 5.

Performance characterization
For a job with a large number of tasks n, the expected latency and cost can be expressed in terms of the single-fork policy parameters p, r and l as given by Theorem 1 below.
Theorem 1 (Single-Fork Latency and Cost). For a computing job with n tasks, and task execution time distribution F X , the latency and cost metrics as n → ∞ are where F Y is given by Lemma 2 below, and it is the distribution of the residual time after forking when the earliest replica of a straggling task finishes. The term E [Y pn:pn ] is the expected maximum of pn i.i.d. random variables drawn from F Y . Its behavior for n → ∞ is given by Lemma 3 below.
Lemma 2 (Residual Straggler Execution Time). As n → ∞, the tail distributionF Y of the residual execution time (after the forking point) of each of the pn straggling tasks is Lemma 3. As n → ∞, the asymptotic behavior of E [Y pn:pn ] is given by where DA (·) is the domain of attraction of F Y which can be determined using Lemma 9 and Theorem 12. The termsã pn andb pn are the normalizing constants of F Y as given by Theorem 11, γ EM is the Euler-Mascheroni constant, and Γ(·) is the Gamma function, i.e., The domain of attraction of a distribution depends on its tail behavior (exponential, heavy or light). For example, exponentially decaying distributions belong to DA (Λ) while heavy tailed distributions belong to DA (Φ ξ ). We now give a sketch of the proof of Theorem 1. A detailed proof can be found in Section 5.
Proof sketch of Theorem 1 . The expected latency of a single fork policy π SF (p, r, l; n) can be decomposed into two parts: where, T (1) is the time to execute the first (1 − p)n tasks and T (2) is the time to execute the rest of the pn tasks with replication. We evaluate each of these parts separately.
It is straightforward to see that T (1) is the ((1−p)n) th order statistic of n i.i.d. random variables with distribution F X . Thus its expected value is where (11) follows from the Central Value Theorem (Theorem 10) which states that the ((1−p)n) th order statistic concentrates sharply around F −1 X (1 − p) as n → ∞. After n(1−p) tasks finish, the scheduler adds redundancy by launching replicas of the straggling tasks and waits for one replica to finish. We denote the residual execution time distribution of the straggling tasks by F Y . It depends of F X and the parameters r, p and l of the scheduling policy as given by Lemma 2. For example, for r = 2 and l = 0, the tail of distributionF Y =F 2 X , which is the minimum of two i.i.d. random variables with distribution F X . The proof of Lemma 2 is given in Section 5.
The second part of the latency, T (2) is the maximum of the times until each of the pn straggling tasks finish. Thus, its expected value E T (2) = E [Y pn:pn ]. The behavior of the maximum order statistic of a large number of random variables is given by the Extreme Value Theorem Theorem 11. We can use it to show that E T (2) is given by Lemma 3.
Thus we can evaluate the expected latency E [T ] = E T (1) + E T (2) . Similarly, the expected cost E [C] can be evaluated by decomposing it into two parts: before and after the replication of straggling tasks. The details can be found in the proof in Section 5.

Examples of the Effect of Tail Behavior
We now demonstrate how the the tail of the distribution F X is a major factor in determining the trade-off between E [T ] and E [C], and hence the choice of the best single-fork policy. We consider two canonical execution time distributions, the Pareto distribution (heavy tailed) and the Shifted Exponential distribution (exponential tail) and evaluate the latency-cost trade-off in Theorem 1 for them. One key insight from this analysis is that in certain regimes, it is possible to reduce latency while simultaneously reducing cost.

Pareto execution time
The cumulative distribution function of the Pareto distribution Pareto (α, x m ) is Pareto distribution is a heavy-tail distribution, with a polynomially decaying tail. It has been observed to fit task execution time distributions in data centers [11,16].
The values ofã pn and E [Y ] depend on the relaunching parameter l, and are given as follows.
The termã pn is the solution to and E [Y ] is evaluated numerically as discussed in the proof.
The proof is given in Section 5. From this theorem we can infer that the latency E [T ] grows polynomially o(n 1 /α(r+1) ) with n. This can be seen directly from (15) for the case of relaunching (l = 0). For the no relaunching case we know thatã pn grows with n. Thus for large enough n, it should be greater than x m p −1/α . Hence from (17), The o(n 1 /α(r+1) ) growth follows from this. Fig. 3 compares the latency obtained from simulation and analytical calculations for Pareto distribution, indicating latency obtained from analytical calculation is very close to the actual performance for n ≥ 100, especially for the case with relaunching (l = 0). In Fig. 4 we plot the expected latency and cost as p varies, for different values of r and l. The black dot is the baseline  case (p = 0), where no replication is used and we simply wait for the original copies of all n tasks to finish. The baseline case is also equivalent to the policies with r = 0, l = 1 and any p.
In Fig. 4 we observe that a small amount of replication (small p and r) can reduce latency significantly in comparison with the baseline case. But as p increases further, the latency may increase (as observed for r = 0) because of the second term in (2). For a given r, relaunching leads to lower latency when p satisfies the condition in Lemma 5 below.
Lemma 5. For given r, relaunching (l = 0) gives lower latency E [T ] than no relaunching (l = 1) when p satisfies, where α is the shape parameter of the Pareto distribution, as given in (12).
Intuition suggests that replicating earlier (larger p) and more (higher r) will increase the cost E [C]. But Fig. 4 shows that this is not necessarily true. Since we kill all the machines running a task when one of its replicas finish, there is in fact a saving in the computing cost! However this benefit diminishes as p and r increase above a certain threshold. Lemma 6. For a given r and relaunching (l = 0), the range of p for which the single-fork policy is sub-optimal is given by where a policy π SF (p, r, l) is said to be sub-optimal if there exists another policy π SF (p , r, l) which gives lower E [T ] and E [C] than π SF (p, r, l).
For example for the r = 1 and relaunch (l = 0) case, we can solve (20) to show that all policies with p < p * 1 ≈ 0.05 are sub-optimal, where p * 1 is marked in Fig. 5. Similarly, for cases r = 0 and r = 2, the sub-optimal ranges [0, p * 0 ] and [0, p * 2 ] are shown respectively in Fig. 5. We conjecture that the convex hull of the curves for different r and l gives the optimal latencycost trade-off. Points on the hull, which lie between some two curves can be achieved by time-sharing between the corresponding two policies.

Shifted Exponential execution time
We now analyze the latency-cost trade-off when the task execution time distribution F X is a Shifted Exponential Distribution SExp (∆, λ). Unlike the Pareto distribution which is heavy-tailed, shifted exponential has an exponentially decaying tail. Its cumulative distribution function is given by The special case ∆ = 0 corresponds to the pure exponential distribution, which is popular in queueing theory and scheduling due to its memoryless property. But it may not be suitable for modeling the execution time of a task, as a task seldom finishes instantaneously, and usually it is lower bounded by a constant delay due to machine start-up or task initialization. Hence we add the constant delay ∆ and model the execution time using the SExp (∆, λ).
Theorem 7. For a computing job with n tasks, if the execution time distribution of each task is SExp (∆, λ), then as n → ∞, the latency and cost metrics are Similar to Fig. 3, Fig. 6 compares the latency obtained from simulation and analytical calculations for the Shifted Exponential distribution, which again demonstrates the effectiveness of the asymptotic theory.
We can draw the following observations from Theorem 7. Given r and l, replicating earlier (larger p) gives an Θ(ln p) decrease in latency, and a linear increase the cost. This is also illustrated in Fig. 7 for execution time distribution SExp (1, 1) and n = 400. Fig. 8 illustrates the latency-cost trade-off. Unlike Fig. 5 there is no range of p for which both latency and cost decrease (or increase) simultaneously.
For the special case of ∆ = 0 by Theorem 7, the cost E [C] = 1 λ , which is independent of p and r. But latency always reduces with r and p. This suggests that we can achieve arbitrarily low latency without any increase in cost. Since this is not observed in practice, we can conclude that the pure exponential distribution is not a useful model for the task execution time.
For a given p and r, relaunching always gives larger latency than no relaunching. But the cost may increase or decrease depending on the values of λ and ∆. Lemma 8 below gives the set of the parameters for which latency and cost are strictly larger with relaunching.

Heuristic Algorithm
In certain practical systems may be difficult to fit a well-known distribution such as Pareto or shifted exponential to the empirical behavior of the task execution time. This makes the analysis of the latency-cost trade-off using the framework presented in Section 3 hard. In this section we present algorithms that use traces of task execution times to estimate the latency and cost metrics, and use these estimates to search for the best single-fork policy π SF (p, r, l). We also evaluate the algorithms on Google trace data [1].

Estimation of Latency and Cost Metrics
In Algorithm 1 we propose an algorithm to estimate E [T ] and E [C] based on random sampling of distributions F X and F Y . The estimates are denoted byT andC respectively.
We first construct empirical CDFsF X andF Y from experimental traces of start and finish times of tasks on a large cluster of machines. The traces need to sufficient to be capture the tail behavior of the distribution, which plays a key role in characterizing the latency-cost trade-off, as seen in Section 3. To estimate the terms of E [T ] and E [C] in (5) and (6) involving F X , we draw n i.i.d. samples fromF X and find the maximum and mean of the smallest n(1 − p) samples. Similarly, to estimate the terms in (5) and (6) involving Y , we draw pn samples fromF Y and find the maximum and mean of the samples. The above steps are repeated m times andT andC are set to means of the corresponding estimates from each step.
While it is possible to estimateT andC directly from traces of the execution time, we take the two-step approach of first finding the empirical CDFsF X andF Y and then estimatingT andC. This is because unlike samples of the task execution time X, the samples of the residual straggler execution time Y (Lemma 2) cannot be directly obtained from the traces of task execution time. Also, this two-step approach allows system designers to improve each of two estimates separately. For example, the CDFsF X andF Y can be smoothed using bootstrapping methods [9].
By the Central Limit Theorem we can show that the standard deviation of the error in estimating , which is slower. Thus, the estimate ofC is more robust than that ofT .

Heuristic Search for the Best Policy
We now present a heuristic algorithm that uses traces of execution time to search for the best single-fork policy. This policy can be used to perform task replication in future jobs that have similar task execution time statistics. The best single-fork policy is said to be the policy that minimizes the objective function J defined as, The parameter µ is the priority given to the minimizing the cost. Since we know only estimatesT andC of the expected latency and cost from Algorithm 1, we use the estimated objective function in (24). Algorithm 2 gives the pseudo-code of the search for the best single-fork parameters p, r and l. For a given p, we first find the optimal r and l, and then perform gradient descent on p. This is repeated for k iterations. To optimize r and l we keep increasing r by 1 until the objective function decreases. For each r, we set l to the value (0 or 1) which gives a smaller J. The termsT (π) and C(π) are the latency and cost estimates for the policy π = π SF (p, r, l) found using Algorithm 1.
The policy found by Algorithm 2 may not be the true optimal single-fork policy due to the following error factors: 1. Error in the estimates of E [T ] and E [C] from Algorithm 1.
2. The gradient descent in p could be slow and may not converge to the optimum in k iterations.
Also, note that E [T ] and E [C] in (5) and (6) are convex in r, but not in p and l. Thus, the algorithm is not guaranteed to converge to the optimal single-fork policy.

Demonstration using Google Traces
We now demonstrate the results of running Algorithm 2 on Google Trace data [1]. We use the traces to estimate the distribution F X and then run the heuristic algorithm on it to find the best single-fork policy. The Google Trace data gives timestamps of events such as SCHEDULE, EVICT, FINISH, FAIL, KILL etc. for each of the tasks of computing jobs that are run on Google's machine clusters. We consider the difference between the SCHEDULE and FINISH timestamps as the task execution time, and construct the empirical distributionF X . Note that because we consider only the SCHEDULE and FINISH task, we are not accounting for the computing resources consumed due to failure or eviction of tasks.
We consider two large Google cluster jobs with n = 1017 and n = 488 tasks respectively. Their normalized histograms are shown in Fig. 9 and Fig. 10. Fig. 9 shows heavy-tail behavior of task execution time, whereas Fig. 10 has bimodal behavior with a very small percentage of tasks finishing in around 1400 seconds.
Then we run Algorithm 2 on the empiricalF X found using the histograms. We use ∆ p = 0.002,   Fig. 11 and Fig. 12. We also plot the estimated latency-cost trade-off for r = 1 and l = 1 as p varies from 0 to 1. These are the parameters of the back-up tasks option in MapReduce as described in Remark 1.
We observe in Fig. 11 that the heuristic algorithm finds policies with r > 1 that give lower latency for the same cost, than the policies with r = 1 and l = 1. We also observe that the latency reduction is more when µ is smaller, that is, the priority given to minimizing the cost is lower. But in Fig. 12, the reduction in latency by adding additional replicas r > 1 is very small as compared to the r = 1, l = 1 case because it has a lighter tail. In both Fig. 11 and Fig. 12 we observe that adding redundancy, that is r ≥ 1 signficantly reduces latency for a small cost, in comparison with the baseline case (p = 0), which is also equivalent to r = 0, l = 1 with any p.

Proofs of Single-fork Analysis
In this section we give proofs of the results in Section 3.

Latency and Cost Analysis for general F X
Proof of Theorem 1. The expected latency E [T ] can be divided into two parts: before and after replication.
= E X (1−p)n:n + E max j=1,2,...,pn The time before forking T (1) is the time until (1 − p)n of the n tasks launched at time 0 finish. Thus, its expected value E T (1) is the expectation of the (1 − p)n th order statistic X (1−p)n:n of n i.i.d. random variables with distribution F X . By the Central Value Theorem stated as Theorem 10, for n → ∞, this term goes to inverse CDF value F −1 X (1 − p). At this forking point, the scheduler introduces replicas of the pn straggling tasks. The distribution F Y of the residual execution time (minimum over the r + 1 replicas) of each straggling task is given by Lemma 2. Thus the term E T (2) in (25)  number of tasks n.
The cost before forking E C (1) consists of two parts: the cost for the (1 − p)n tasks that finish first, and the cost for the pn straggling tasks. The first term in (27) is the sum of the expected values of the smallest (1 − p)n execution times. Using Theorem 10, we can show that the i th term in the summation goes to F −1 X (i/n) as n → ∞. Expressing the sum as an integral over h = i/n we get the first term in (29). The second term in (27), is the normalized running time of the pn straggling tasks before forking. Substituting E T (1) from (25) and simplifying, we get (29).
The cost after forking, E C (2) is the normalized sum of the runtimes of the r + 1 replicas of each of the pn straggling tasks. By Lemma 2, the residual execution time of the j th straggling task is Y j ∼ F Y . Since the scheduler kills all replicas as soon as one replica finishes, the expected runtime for the j th straggling task is (r + 1)E [Y j ]. Thus, the cost in (30) is the sum of (r + 1)E [Y j ] over the pn tasks, normalized by n. Since Y j are i.i.d, we can reduce this to (31).
Proof of Lemma 2. First consider the case where we relaunch (l = 0) the original copy, and add r replicas for each of the pn straggling tasks. Thus, there are r + 1 identical replicas of each task after forking. The residual execution time distribution F Y (after time T (1) when the replicas are added) of each task is the minimum of r + 1 i.i.d. random variables with distribution F X . Hence, Pr(Y > y) = Pr(min(X 1 , X 2 , . . . X r+1 ) > y), For the case without relaunching (l = 1), there is 1 original replica and r new replicas of each of the straggling tasks. Thus, the tail distributionF Y (y) = 1 − F Y (y) is given by As the number of tasks n → ∞ by Theorem 10 we know that T (1) → F −1 X (1 − p). Hence we have, To prove Lemma 3, we characterize the expected maximum of a large number of random variables using Theorem 11. First, we state Lemma 9 which implies that the domain of attraction (see Theorem 12) of F Y is same as that of F X .
Lemma 9 (Domain of attraction for F Y ). Given a single fork policy π SF (p, r, l; n) with 0 < p < 1, The proof follows from Lemma 2 and Theorem 12 and is omitted here.
Proof of Lemma 3. We can use Lemma 9 to find the domain of attraction of F Y . Then from (55) we have where E [G(y)] can be found using Theorem 11 and Lemma 13.

Latency and Cost Analysis for Pareto F X
We prove Theorem 4, which evaluates the latency E [T ] and computing cost E [C] metrics when the task execution time distribution F X is the Pareto, as defined in (12).
Case 1: Relaunching (l = 0) In the single-fork policy with relaunching (l = 0), the scheduler waits for (1 − p)n tasks to finish and then relaunches each of the pn straggler tasks on a new machine.
From (59) in Theorem 11 we can evaluateã pn as follows And E [Y ] of (41) can be evaluated as Case 2: No Relaunching (l = 1) In the single-fork policy with no relaunching (l = 1), the scheduler keeps the original copy, and adds r additional replicas for each straggling task. Using Lemma 2 we can show that which simplifies to (17). The expected value of Y can be found by numerically integratingF Y (y) in (43) over its support.
Proof of Lemma 5. For given r, the expected latency is lower with relaunching when p satisfies, Proof of Lemma 6. Given r and relaunching (l = 0), the single-fork policy is sub-optimal in both Substituting E [T ] and E [C] from Theorem 4 and simplifying, we get (20).

Latency and Cost Analysis for Shifted Exponential F X
Now we prove Theorem 7, which gives the latency-cost trade-off when the distribution of the execution time X is a shifted exponential given by (21).
Proof of Theorem 7.
= ∆ + 1 λ (p ln p + (1 − p)) + p∆ − p λ ln p, To find E [Y ],ã pn andb pn we consider the cases of relaunching (l = 0) and no relaunching (l = 1) separately. Case 1: Relaunching (l = 0) ∼ SExp (∆, (r + 1)λ) (51) Based on Theorem 12, for η(y) = 1/((r + 1)λ) we have By Theorem 11 and Theorem 12, the maximum of shifted exponential belongs to the Gumbel family withã Case 2: No Relaunching (l = 1) In the case of no relaunching, Note that the first term does not include ∆ because for large n the original task would have run for at least ∆ seconds. Thus the tail distribution of Y is given bȳ The expected value E [Y ] is the integration ofF Y (y) over its support. Proof of Lemma 8. For the shifted exponential distribution it is clear that no relaunching always gives a lower latency. We now find conditions on λ and ∆ for which no relaunching gives lower cost. Define β = λ∆, then ⇔ n + pn λ∆ + r 1 − e −λ∆ > n + pnλ(r + 2)∆, And note that the function g(β) = βr + β − r + re −β is monotonically increasing since g (β) > 0 for any r ∈ Z + . 6 Concluding Remarks

Major Implications
In this paper we show that replication of the slowest tasks in a job (the stragglers) is a powerful way to reduce latency in large-scale parallel computing. We characterize the trade-off between latency and computing cost for a set of replication strategies called single-fork policies. The policy used in practical systems such as MapReduce belongs to this set of single-fork policies. A non-intuitive insight we get from this analysis is that in certain scenarios it is possible to reduce latency and cost simultaneously. We also propose a heuristic algorithm to find the scheduling policy which achieves a good trade-off between latency and cost. Experiments on Google Traces data show that policies found by the heuristic algorithm can give a better latency-cost trade-off than the back-up tasks option used in MapReduce.

Future Perspectives
Although we focus on single-fork policies in this work, the analysis can be extended to multi-fork policies where the scheduler can add or kill replicas at multiple instances during the execution. We also plan to consider queueing of tasks at machines and analyze the percentile latency in addition to the expected value considered here. Another future direction is to analyze the performance of approximate computing where only a subset of the tasks of a job are required to be completed, for example information retrieval and machine learning applications.
In this work, we assume prior knowledge of the execution time distribution F X when designing the task replication strategy. An interesting research direction is to develop an adaptive scheduling algorithm that simultaneously estimates the distribution and schedules task replication. This shares some similarity to the celebrated multi-arm bandit problem, with an exploration-exploitation trade-off between estimating the distribution F X , and using the current estimate to design a task replication policy.
Our analytical framework can be used in other applications where the response time of the components is stochastic. For example, in crowdsourcing, each worker can take a variable amount of time to complete a task. Then the overall latency can be modeled in the same way as our work, with the cost function being equal to the number of workers [20].
In this section we introduce some definitions and results in order statistics that play an important role in the analysis of scheduling policies in Section 3. In particular, we describe how order statistics in different regimes have different concentration behavior in Appendices A and B.

A Central order statistics
For an order statistic X k:n , we called it a central order statistic if k ≈ np for some p ∈ (0, 1). In this case, X k:n is asymptotically normal, concentrated around the p-th quantile of X, as indicated by the following result called the Central Value Theorem (Theorem 10.3 in [6]).
∼ F , if 0 < p < 1 and 0 < f (x p ) < ∞, where x p = F −1 (p), then for k = k(n) such that k = np + o ( √ n), where f (·) is the p.d.f. corresponds to F and P → denotes convergence in probability as n → ∞.

B Extreme order statistics
Extreme value theory (EVT) is an asymptotic theory of extremes, i.e., minima and maxima. It shows that if a distribution belongs to one of three families of distributions Theorem 12), then its maxima can be well characterized asymptotically as given by Theorem 11, which is also referred to as the Fisher-Tippett-Gnedenko Theorem (Theorem 1.1.3 in [7]).
∼ F , if there exist sequences of constants a n > 0 and b n ∈ R such that P [(X n:n − b n )/a n ≤ x] → G(x) (55) as n → ∞ and G(·) is a non-degenerate distribution. The extreme value distribution G(x) and the values of a n and b n depend on the domain of attraction (and hence the tail behavior) of F X given by Theorem 12.
Intuitively, F ∈ DA (Λ) corresponds to the case thatF has an exponentially decaying tail, F ∈ DA (Φ ξ ) corresponds to the case thatF has heavy tail (such as polynomially decaying), and F ∈ DA (Ψ ξ ) corresponds to the case thatF has a short tail with finite upper bound.
It is worth noting that the distribution function for −X may be in a different domain of attraction from that of X.