{'fraction_non_alphanumeric': 0.03495805033959249, 'fraction_numerical': 0.008015381542149421, 'mean_word_length': 4.634355655599324, 'pattern_counts': {'":': 0, '<': 0, '<?xml version=': 0, '>': 0, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
Design and Experimental Evaluation of Algorithms for Optimizing the Throughput of Dispersed Computing

With growing deployment of Internet of Things (IoT) and machine learning (ML) applications, which need to leverage computation on edge and cloud resources, it is important to develop algorithms and tools to place these distributed computations to optimize their performance. We address the problem of optimally placing computations (described as directed acyclic graphs (DAGs)) on a set of machines to maximize the steady-state throughput for pipelined inputs. Traditionally, such optimization has focused on a different metric, minimizing single-shot makespan, and a well-known algorithm is the Heterogeneous Earliest Finish Time (HEFT) algorithm. Maximizing throughput however, is more suitable for many real-time, edge, cloud and IoT applications, we present a different scheduling algorithm, namely Throughput HEFT (TPHEFT). Further, we present two throughput-oriented enhancements which can be applied to any baseline schedule, that we refer to as"node splitting"(SPLIT) and"task duplication"(DUP). In order to implement and evaluate these algorithms, we built new subsystems and plugins for an open-source dispersed computing framework called Jupiter. Experiments with varying DAG structures indicate that: 1) TPHEFT can significantly improve throughput performance compared to HEFT (up to 2.3 times in our experiments), with greater gains when there is less degree of parallelism in the DAG, 2) Node splitting can potentially improve performance over a baseline schedule, with greater gains when there's an imbalanced allocation of computation or inter-task communication, and 3) Task duplication generally gives improvements only when running upon a baseline that places communication over slow links. To our knowledge, this is the first study to present a systematic experimental implementation and exploration of throughput-enhancing techniques for dispersed computing on real testbeds.


I. INTRODUCTION
Dispersed computing, where complex applications that process data from multiple sources are distributed across networked computers, is a promising paradigm for real-world applications [1], [2], [3], [4]. Like real-time video processing and cloud computing, there's often a massive amount of data continuously and steadily fed into the system from multiple sources. In today's big data applications, the amount of compute, storage and network resources required to finish a job in a reasonable amount of time far exceeds those of a single computer and therefore a cluster of compute nodes is needed. However, when computation is not centralized on one machine, various challenges emerge, such as scheduling [5], coordination, fault tolerance [6] etc. Besides, machines are usually commodity hardware, and have heterogeneous performance when processing different types of jobs [7], [8]. Intermachine communication is also heterogeneous due to nonlinear relationship between transfer time and file size [9]. Distributed applications are often described by directed acyclic graphs (DAG). Each application DAG consists of multiple tasks with precedence requirements. Each input instance needs to go through the exact same process flow of the DAG, and generates an output. Traditionally, in grid computing, the objective is often minimizing the single-shot makespan, such as the famous HEFT schedule [10]. However, for dispersed computing involving processing of streaming data sources such as in IoT applications, the computation can be pipelined and thus a more useful parameter is throughput, defined as the number of instances processed per unit time. In this work, therefore, instead of minimizing makespan for a single data input, we consider maximizing the steady-state throughput for pipelined execution.
Google invented Kubernetes [11] (k8s) to manage container orchestration on large scale clusters. However, k8s doesn't have a scheduler that implements a globally optimized algorithm; nor does it have a comprehensive orchestration system to process pipelined DAG applications.
Keeping in mind these challenges, we propose three throughput optimized scheduling algorithms, namely "Throughput HEFT" (TPHEFT), "Node Splitting" (SPLIT) and "Task Duplication" (DUP), and implement a real-world system (JupiterTP) based on k8s and our previous Jupiter system [9] to perform end-to-end pipelined computation on a distributed cluster. Most prior work on throughput has focused on algorithms and simulations. To our knowledge, this is the first work to not only propose algorithms, but also implement a full-stack distributed system for throughput optimization, which can handle arbitrary DAGs, and can be applied over any network of computers -public or private cloud, IoT systems, edge networks etc. The system itself is much more than just schedulers and scheduling algorithms, the profilers, execution engines and orchestration engines together make the above possible. Our system also provides unified interfaces to include other scheduling algorithms of similar settings, so that they can go beyond simulations. What's more, we systematically studied the performance of the whole system using real clusters, and also some real world applications, which is much more convincing than just simulated results. Our system has been made publicly available as open-source software.
Though less studied, there has also been prior work on improving throughput. [17] proposed greedy and linear algorithms, and conducted simulation on SimGrid framework; [18] proposed real systems, but for only 2 devices -mobile and cloud; there are also other real systems designed for special DAGs, such as linear DAGs in [19], and the map-reduce framework [20]. Further work on throughput maximization such as [21] [22] [23] [24] are theory or simulation based. We considered implementing their algorithms as baselines, but some assumptions in those papers are idealized don't hold in the real system, thus cannot be implemented easily and uniformly. Our work is different from previous ones, in a sense that we have novel algorithms and end-to-end full-stack systems for general DAGs.
Some pipelined processing systems (such as k8s itself) use dynamic scheduling, where the scheduler, based on the current status of the system, decides where to put the next task. This is best applied when we don't have much knowledge (such as future workload), or the system has a high churn rate or unexpected behaviors. However, we're aimed at optimizing expected overall system performance, not for the current task. In our model, tasks are templates scheduled on nodes, they start their routines when an input instance arrives; and cluster nodes and links are working steadily with small variations. Given such knowledge, we can do a one-shot, globally optimized scheduling, which can be reapplied when the DAG or cluster status changes. Once we decide where to run a task, we run all input instances of this task on the same node until the next round of scheduling.

III. SYSTEM MODEL AND KEY CONCEPTS
In this paper, we use the terms 'node', 'processor' and 'proc' interchangeably, since we target single-core nodes. Tasks can't run in parallel on single-core nodes, same for file transfer on links. We optimize for single-core nodes in theory, but though not as precise, our algorithms can also improve throughput for multi-core nodes.

A. Application DAG and compute cluster
In the general form of applications, a DAG consists of a set of tasks, represented by vertexes; and inter-task file transfers, represented by directed weighted edges.
A cluster is a collection of resources, and our main interest is computation and communication resources. Among these two, the former is represented by a set of nodes, consisting of commodity hardware, which can be highly heterogeneous, with nodes perform differently under different types of tasks. Some previous work used a simple model, where i.e., assigning a single compute factor to each node. However, as a counter-example, some nodes have powerful CPU while others perform well for I/O. If a task consumes a lot of CPU, it will perform better on the first type of nodes; if another task is I/O heavy, it'll perform better on the second type. Thus we can't simply assign a compute factor to each node. Instead we use a data-driven approach to profile the expected execution time of each task on each node, and describe execution as a matrix.
On the other hand, communication is represented by a set of links connecting nodes. All nodes in the cluster are reachable to each other. In a regular commercial cluster, we don't know or have control over the link topology or the routing tables, thus we use an abstraction of links, namely "virtual links", and assume there is a pair of links between each pair of nodes. In reality, several virtual links can be overlapped as one actual (physical) link. We showed in later experiments how accurate our model is, and how to adjust parameters when the model is inaccurate. Similar to computation, most prior work states In our work, inspired by the findings in [9], instead of assigning a fixed bandwidth to each link, we used a quadratic model where a, b, c are parameters to describe a link, calculated by quadratic regression of existing data of file sizes and transfer times.

B. Example process to reach steady-state
Single-shot makespan and steady-state throughput are different concepts and don't have a direct relationship. Steadystate means the system is working periodically, repeating the same process for each period. Thus minimizing makespan is not the same as maximizing throughput. In each schedule's steady-state, there's always a bottleneck resource which defines the overall system throughput. So we need to find ways to reduce the current bottleneck.
During an actual process, there's usually a start-up stage, where the system is partly saturated, and after some time, when the first input instance reaches output, the system will reach a steady-state. Fig 1 is a simple example to illustrate how a system can reach steady-state, and how to identify the bottleneck. Imagine a linear cluster that consists of three worker nodes connected by two links; and a DAG that consists of four tasks with a diamond shape. Given a schedule, T0 (task0) is on on node1; T1 and T2 on node2; T3 on node3. Assume the four tasks take 3, 2, 2, 5 seconds to complete (on their scheduled nodes respectively). Also assume the file transfer between T0 and T1, T0 and T2, T1 and T3, T2 and T3 takes 2, 1, 3, 2 seconds respectively (on their scheduled links). Neither node nor link can process more than one job at the same time.
A detailed description of what each stage looks like is provided in table I. We use "Tx-Ny" to denote input instance y for task(x), use "Fmn-Ny" to denote the file transfer from task(m) to task(n) for input instance y. We assume continued input instances N1, N2, N3... are generated fast enough. In table I, during the first 15 seconds, parts of the system are idle with no fixed routine. After 20 seconds, when the first output (which belongs to input1) was generated, the system reaches steady-state, and generates an output every 5 seconds. The rate at which the outputs are generated (throughput) depends only on the system bottleneck, which is the critical resource working non-stop. Other resources can take a break in each time period. In this example, both node3 (handles one task in each period) and link23 (handles two file transfers in each period) are the system bottlenecks. Based on the above analysis, we define a term called "schedule time" of a resource, which is the actual working time of a resource in one period, excluding idle times. The maximum schedule time of a system is the maximum value of the schedule times of all resources. For the bottleneck resource, its schedule time equals actual working time. Thus finding the system bottleneck is finding the resource with largest schedule time. Note that in general, we only consider nodes and links. There are other resources and possible bottlenecks, such as multi-socket overhead during file transfer startup, we will summarize them in the experiment part and give an overview of how much they cost. But based on our experiment results, only considering these two resources is a useful model under most circumstances.
Assume we have a schedule where node P has scheduled n tasks, which takes time t1, t2, ..., tn respectively, thus the schedule time of node P is i=n i=1 (t i ). Assume file transfers on link L take time f1, f2, ..., fm respectively, then the schedule time of this link is i=m i=1 (f i ). The relationship between achievable steady-state throughput and schedule times of nodes and links is as follows IV. ALGORITHMS All of our algorithms are designed to maximize the steadystate throughput, under an environment with abundant resources. We start from HEFT, analyze the bottlenecks, then propose TPHEFT, which is an independent algorithm that aims to minimize bottlenecks; then propose SPLIT and DUP, which can be applied to any existing algorithms to improve their throughput.

A. Throughput HEFT (Algorithm Block 1)
In single-shot case, one key trick to minimize makespan is to reduce inter-task file transfer times. It is considered that on the same node, inter-task file transfer time is always zero, given that inter-process communications are much faster than network transfer. So HEFT tends to schedule tasks close to each other if they have dependencies. As an extreme case, for a singly linked-list DAG, HEFT will schedule all tasks on the same node. The only way HEFT can schedule tasks on different nodes is if some tasks can run in parallel. For a pipelined case, parent and child tasks can be processed at the same time using different input instances. In this case, HEFT clearly creates idle resources thus decrease throughput.
Thus we propose TPHEFT, whose overall goal is to minimize the maximum system schedule time across all resources, so that the bottleneck is minimized. There has been prior work to improve HEFT to boost throughput [17]. Our algorithm is different in terms of scheduling order, and also in each iteration we only minimize the schedule time over effected resources, which is more precise and can have better results. Similar to HEFT, we define task upward rank recursively as rank(n) = comp(n) + max k∈nchild (comm(nk) + rank(k)) (5) Where comp(n) is average computation cost of task(n) over all nodes, comm(nk) is average communication cost of file transfer from task(n) to task(k) over all links. Upward rank is a more precise topological ranking in weighted graphs. Scheduling in this order can also make sure that all parent tasks are already scheduled before scheduling a child task. Note that upward rank does not guarantee time dependency because it's not the precise value of task time on the scheduled node, so HEFT has another step to make sure the earliest start time of a child is later than the latest finish times of all parents.
In TPHEFT, we drop the time dependency concern, and only use upward rank, simply because in pipelined execution, a child task can be executed before the parent task is finished, they're just at different time stages. For example, the child may be executing task(c)-input1 while the parent is executing task(p)-input2. In our implementation, we use run time queues and file system monitoring process to handle this.
Following upward ranks, we tentatively schedule the current task on a node. This will cause one incurred node usage, and several incurred link usages. We add these new costs to these resources' original costs, and calculate the resulted max schedule time over effected resources. This value means, if current task is scheduled on this node, the max schedule time of effected resources. Note that this is not necessarily the current system max schedule time. We pick such a node to minimize this value, and schedule current task on this node.

B. Node Splitting (Algorithm Block 2)
We propose SPLIT, an algorithm that aims to improve the throughput of an existing schedule by making portion-based replicas of all tasks on a node. Consider HEFT, TPHEFT, or any other schedule, the critical resources can be significantly more time consuming than others, especially in HEFT where tasks are more centralized. In TPHEFT, although we already tried to balance them, we're bounded by the condition "one task can only be scheduled on one node". If we just happen to have a task which is very time consuming, there's nothing TPHEFT can do. Thus we propose SPLIT, where we reduce the bottleneck on nodes or links by splitting a node (meaning all tasks scheduled on this node) into several replicas, each with an assigned portion of original workload. There are several challenges arising from this idea, in terms of algorithm, system implementation and experiments. Below are some key remarks.
• We can't require the application programmer to write their tasks in a multi-thread way, and processing one input instance of a task on multiple nodes for DAGs is Algorithm 1 Throughput HEFT Input: 1: Application DAG, sorted tasks 2: Cluster (undirected weighted graph) 3: Network profile (list of (a, b, c)) 4: Execution matrix, exec[i][j] is time of task i on node j Output: Mapping from tasks to nodes, one task to one node 5: map from resources to their schedule times, update when a task is scheduled 7: assign entryT ask first, update resources 8: for all task ∈ tasks do 9: tmp ← {} map from a node to resulted max schedule time over *effected* resources if task is scheduled on this node 10: pT asks, pN odes ← parent tasks, nodes of task 11: for all proc ∈ nodes do 12: nodeT ime ← exec[task][proc] + resT ime[proc] 13: linkT ime ← copy existing link costs in resT ime 14: for all pT ask ∈ pT asks do 15: link = pN ode-proc 16: lT ime = time of file pT ask-task on link 17: linkT ime[link] += lT ime 18: maxLinkT ime = max linkT ime.values() 19: maxT ime = max(nodeT ime, maxLinkT ime) 20: tmp.add(proc, maxT ime) 21: assign task to key (proc) in tmp with smallest value 22: update effected resource usages very hard. In SPLIT, we still process each input instance on one node, but will dynamically distribute different instances of the same task to different nodes based on their portions, using both random number and hashcodebucket methods.
• If bottleneck is a link, we can do SPLIT on source or destination node. This will reduce the inter-task file transfer load between the two nodes, thus reducing the link bottleneck. So we only talk about node splitting. • When deciding where to split the bottleneck resource to, our number one priority is to NOT result in a bigger bottleneck, because when we incur extra load somewhere else, the new resource could become the new bottleneck. Our number two priority is to reduce the load on current node as much as possible. • When we split nodes, should we split it into two, three, or four? As mentioned before, improving anywhere other than the bottleneck is useless, and if at some point of splitting, this resource is not the bottleneck anymore, then further splitting it is wasteful. • SPLIT can always result in better throughput if handled properly, because we can control the portion of tasks to ship. Even shipping away a tiny portion will reduce the bottleneck.
We summarize the algorithm as two phases, the node and portion selection phase, and the do split phase. In case of link bottleneck, just split source or destination node. In phase one, first find the bottleneck node (btnk). Then iterate over all candidate nodes (We consider candidate nodes as idle nodes, simply because we have abundant resources. But this can be easily extended to any node with a small usage), for each candidate node, tentatively split 100% btnk workload onto this node. Like in TPHEFT, this will cause one extra node usage and several extra link usages. Get the max schedule time over effected resources, mark it as the max schedule time if btnk is scheduled on this candidate. After checking all candidates, find the one with smallest max schedule time as the objective node. After deciding on this candidate, we choose a portion of btnk to ship to the candidate, such that the schedule time of btnk node and effected resources become the same. Note that the portion means the workload percentage of each task relative to btnk, not original tasks. Fig 2 gives an example that btnk has already been split and carries different portions of different tasks. Splitting 40% of the node to a new node will result in a schedule in the fig. This way, we split each node into two for every iteration, but can keep splitting if the new nodes are still bottlenecks, which is more agile and simple than calculating "splitting btnk into how many candidate nodes and what portions". Note that task portions and node portions are not much related, because a task is on several nodes, a node has several tasks. We only keep track of the portion (relative to original task) of each task on each node, as in the previous SPLIT output example. In phase two, split btnk node into two nodes and update resource usages based on portion. Fig 3 describes an example case for SPLIT. Effected resources include: btnk node, candidate node, links between {all parent and child nodes of btnk node} and btnk node, links between {all parent and child nodes of btnk node} and candidate node. These resource usages need to be updated based on node splitting portion. Here parent nodes of nodeB are defined as nodes where any scheduled task is a parent of any scheduled task on nodeB. Cluster nodes are undirected graph and have no parent-child relationship originally, but after scheduling tasks on them, such relationships appear. Obviously, unlike in tasks, the patentchild relationships of nodes is not a DAG, a node can be a parent of itself, or nodes can form a cycle of parent-child relationships, or a node can have no parent or child nodes. These corner cases need to be carefully managed.
Although SPLIT generated more tasks and parent-child relations, we do NOT consider this as a change of the original graph, instead we consider the extra tasks as replicas, such as taskA-1, taskA-2, etc, which is different from task duplication in the next section.
Lastly, there's a very important runtime problem the probability based load balancing algorithm didn't address. When splitting a task which has multiple parents (such as task3 in fig 1), all the parent output files associated with the same input instance have to be sent to the same replica of child, otherwise the child task cannot start. For example, for the earlier diamond DAG (fig 1) with 4 tasks, task3 has parents task1 and task2. If task3 has two replicas, task3-1, task3-2, then output files output-task1-input1 and output-task2-input1 must both be sent to either task3-1 or task3-2. Later, output-task1-input2 and output-task2-input2 have to follow the same rules, but they don't have to be sent to the same task3 replica with output-taskX-input1 files. For simplicity, in the real system, tasks are not aware of their peers (thus no distributed consensus needed). But if they only choose the child replica based on probability, then there's no guarantee that all parent output files associated with the same input instance will be sent to the same place. This only happens when a multi-parent child task has more than one replica. The number of parents and the number of replicas for each parent doesn't matter, nor is this a problem if the child only has one parent task.
There are multiple solutions to this problem. For simplicity we chose the following. For single-parent tasks, we do the above probability based random selection; for multi-parent tasks, we extract the input task ID (tagged by the system), generate a hashcode using a self-defined hash function, then choose a child replica (rep) using An example hash function is simply atoi(input ID), which can result in a round-robin like distribution. Because same

Algorithm 2 Node Splitting
Input: 1: Same as previous + an existing schedule Output: map from nodes to scheduled tasks and portions 2: 3: resT ime ← mapping from resources to their schedule times, initially given by input mapping 4: while availableN odes = empty do 5: btnk ← get bottleneck 6: if bottleneck is link then 7: btnk ← src or dst node, choose a batter one 8: tasks ← tasks on btnk to be split 9: pT asks, pN odes ← parent tasks and nodes 10: cT asks, cN odes ← child tasks and nodes 11: tmp ← {} mapping from proc to resulted new max schedule time over effected resources if btnk is split on proc 100% 12: for all proc ∈ availableNodes do 13: nT ime ← task∈tasks exec[task][proc] + 0 14: lT imes ← existing link usages in resT ime 15: for all task ∈ tasks do 16: for all pt ∈ pT asks do 17: pN ode ← node of parent task pt 18: lk = pN ode-proc parent link 19: lT imes[lk] += time of file pt-task on lk 20: for all ct ∈ cT asks do update resource usages objects will always produce the same hashcodes, we can be sure any groups of intermediate files will not be scattered to different child replicas. As for hash collision, it doesn't hurt the correctness of the system, but we should try to distribute inputs as evenly as possible in order not to have a new bottleneck.

C. Task Duplication (Algorithm Block 3)
DUP is inspired by and adapted from theory-focused work [15]. It is designed to bypass slow links by rerouting the parent-child file transfer paths. As the name suggests, we're duplicating tasks, unlike in SPLIT, where we split nodes (all tasks on a node). We can see later that DUP would change the DAG itself as we modify the parent-child relations.
The idea behind duplication comes from the scenario described in fig 4. Under a certain schedule, if a link is very slow due to network issues, we want to find a way to bypass the link by relocating the tasks whose file transfers go though the link. In fig 4, we want to bypass the slow red link. There's just one task that uses this link, namely src. In order to bypass the link, we need to relocate src. However, we shouldn't directly Fig. 4: Example of duplication for a single task move it to a new node, because part of it doesn't go through the slow link (path {p, src, c1}). Our solution is to create a new task, which is same as src, namely dup, and place it on a selected node. With this, we redirected the traffic {p, src, c2} to {p, dup, c2}, but kept the traffic of {p, src, c1} that goes through the task but doesn't go through the bottleneck link.
Above is just a simple example of DUP where we used three parallel links to replace a slow link. In real-world cases, there are lots of other problems to consider.
• In DUP, using fig 4 as an example, we removed the parent-child relationships between src and c2, and added the relationship between p1 and dup, p2 and dup, dup and c2. Each round of duplication will change the DAG structure. • Unlike SPLIT, which in theory can always improve system performance if handled properly, duplication does not always improve performance. There are two conditions, first the bottleneck has to be a link, second the max schedule time on the new path has to be smaller than the bottleneck. • TPHEFT and SPLIT are very general algorithms that can be applied anywhere, whereas DUP has several limitations. In DUP, we can only bypass slow links. But the getBottleneck() function doesn't know whether a link bottleneck is due to slow link or large file transfer. Applying DUP to the latter case can make things worse in runtime. • Since DUP can change the parent-child relations, it can leave a trail of zombie tasks -the ones that are disconnected from the exit task, thus have no contribution to final output. We can set up a garbage collector to remove these tasks and their schedule to free up the nodes, and they should not be counted as bottlenecks. This algorithm also consists of two phases, the find best node phase, which returns the current best available node; and the duplicate phase, which updates the nodes and links usages and the DAG. We use the same greedy method, at each iteration, choose an idle node which will have the best result and duplicate required tasks on it, until there are no resources, or duplication decreases throughput, or bottleneck is not a link anymore, as shown in the algorithm block.
In the find best node phase, tasks to be duplicated are the ones on src node who transfer files to tasks on dst node. After tentatively choosing a candidate, the effected resources of duplication include, candidate node, link from candidate node to dst node, links from parent nodes of src node to candidate node. The resulted max schedule time across those resources have to be smaller than the bottleneck value.

V. SYSTEM IMPLEMENTATION
The original Jupiter is discussed in [25]. In this paper, we built three new schedulers, execution engines (called CIRCE) and orchestration engines.
Jupiter ( fig 5) is a containerized distributed system built on k8s, with network profilers, execution profilers, schedulers, execution and orchestration engines. Each profiler is a pod running on a k8s node, that collects computation and communication info and reports to the home node, where the scheduler pod is running. Then the scheduler calculates a mapping and sends it to CIRCE home pod via rest API. CIRCE home spawns a pod for each task (in SPLIT, a pod for each task replica), and ships the pod to the designated node. Pods can directly communicate with each other via clusterIP service. Each CIRCE pod has an input queue and output queue, whenever one or a group of (depending on task requirement) files is received in the input queue, it is used as input for that pod's task, and generates one or several files to the output queue. The output queue, upon receiving a file, decides which child pod to transfer the file to, either to a fixed child, or by random number generator or hashcode-bucket in case of SPLIT. Since we consider only single-core nodes, one node (not pod) can only be processing one instance at a time. At the node level, it looks like a FIFO queue. In pipelined computing, a CIRCE pod can be receiving multiple files from multiple parents, and it has to wait for all parent output files of one particular instance to arrive before execution. But during this time period, it could receive parent files from different input instances. The system adds labels to intermediate files to indicate their associated input instances. When a file arrives, CIRCE will first cache its location and input instance. When the bucket of an input instance reaches designated size, those files are used as inputs for current task.
A pod doesn't know the existence of parents or peers, it just monitors its input queues; but it knows everything about its children. An example of a pod's ENVs in SPLIT, which indicate two child pods, each with two replicas task0-1/0.15 : task0-2/0.85 : task1-1/0.5 : task1-2/0.5 In non hashnode-bucket case, this pod generates two random numbers under 1000. For the first number, if it's 0-149, transfer file to task0-1, else to task0-2. Similar for second number. In non-split cases, since there are no replicas, each replica has 100% probability. Such a design greatly simplifies the system, with no consensus needed.

A. Experiment Setup
We used a public cloud called Digital Ocean (DO) to conduct experiments. We designed 9 DAGs, 5 small dummy, 3 large dummy and 1 real application, with different shapes, sizes and parameters. For each DAG, we conduct 7 experiments, and measure the time of executing a few hundred instances in each experiment to calculate steady-state throughput. We first do HEFT, then TPHEFT to show improved throughput. Then we do HEFT+SPLIT and TPHEFT+SPLIT, to show that SPLIT works well on both baseline schedules, but has a better performance when used on an imbalanced schedule (HEFT). We also show that TPHEFT works best when the degree of parallelism inside the DAG (measure of how many tasks can be executed at the same time for the same input) is low.
To further show the impact of computation and communication parameters within the same DAG for SPLIT, we took the diamond shape DAG1, and did two sets of experiments. In the first set, we scaled the computation in each task by a factor of 1, 5, 10, 20, and measured the throughput improvement of HEFT+SPLIT over HEFT; in the second set, we scaled the communication in each transfer to a factor of 1, 5, 10, 20 and measured the improvement of MANUAL+SPLIT over MANUAL. Here the MANUAL is a schedule that places each task on a different node, thus forcibly creating several link bottlenecks to split.
We then show the improvement of DUP. This technique was invented to bypass slow links, so to show significant improvement, we simulate slow links by scheduling most tasks in US and one or a few in Europe (called MANUAL in table III). The slow links will be bypassed by duplicating European tasks back to US. We established DO clusters that span across geographical data centers. Note that DUP can be applied to any schedule, but using a schedule where a slow link is a clear bottleneck is easy to show improvement. Additionally we also showed TPHEFT+DUP, but we don't expect it to give much improvement, because in an already balanced schedule such as TPHEFT, it's not so likely that links will be big bottlenecks. Our way of simulating slow links has a feature -when main data center is in the US, slow links always come in pairs because both source and destination are in US. This sometimes decreased performance.
We considered but did not implement the combination of DUP and SPLIT, because: a) it is challenging from a system perspective to combine the two implementations and b) it is unclear that this combination would yield clear benefits. We leave open to future work to investigate this combination .
We use both dummy and real application DAGs for our experiments. The dummy application is called "count", where each task does math calculations for a designated loop range. Such task mainly consumes CPU, and the execution time is easy to control by controlling the loop range. The real application we used is machine learning based object detection.
In below experiments, we show the number of input instances processed per 1000 seconds under steady-state. This is calculated by measuring the time of executing hundreds of instances. The percentage showed relative throughput increase before rounding, and the actual values are after rounding.

B. Experiment Results
a) Small dummy DAGs on small cluster, DAG1-5: This part is done on DO, 1 master node, in SFO data center, with 4GB RAM and 2 cores; 15 worker nodes, with 1 GB RAM and 1 core, 8 in SFO, 3 in NYC, 3 in Germany. We used 5 random DAGs of different shapes and sizes (shown in fig 6). DAG1 and DAG2 are of same shape, but T1 in DAG2 is much heavier  than T1 in DAG1 while other parameters are the same, thus creating balanced and imbalanced version of the same DAG. Note that this part only refers to DAG1 to DAG5. It is not meaningful to compare results across DAG1-5 and DAG6-8, because they're done on different clusters. In table II, we see that TPHEFT (compared with HEFT) works best for DAGs with low degrees of parallelism. Linear DAGs (DAG5) have no parallelism, thus most improvement. DAG3 has high parallelism compared with the other four, thus a low improvement. This is expected from our previous analysis.
Also, SPLIT works best for imbalanced schedules. We see from table II that when used upon HEFT, SPLIT gives higher throughput improvement than when used upon TPHEFT, because the latter is a more balanced schedule. Another case we mentioned earlier is when one or a few tasks are significantly heavier than others. Compare DAG1 and DAG2, we see that for TPHEFT baseline, DAG2's SPLIT showed more improvement because it's imbalanced compared with DAG1. It's not the same case for HEFT+SPLIT between DAG1 and DAG2, i.e. though DAG2 is more imbalanced, the SPLIT on HEFT and DAG2 was not so significant compared with SPLIT on HEFT and DAG1. This is because the imbalance of task T1 is outweighed by the imbalance of HEFT.
For scaling computation and communication, results are shown in fig 11 and fig 12. As we see, with bigger scale factor, SPLIT gives better performance. This is due to higher degree of imbalance. For communication, SPLIT doesn't perform as well as for computation, because of the limitations of our model, but it still gives increasing performance.
Looking at DUP (table III), we see for most DAGs it resulted in good improvement compared with the manual schedule where cross continental links are big bottlenecks, except in DAG2, where task1 is significantly heavier than other tasks, and the bottleneck is not on links, in which case DUP can't improve overall system performance. This further shows where DUP really fits. Also, we see for TPHEFT, DUP has no improvement. Because we used public clouds, the real links are always good, besides TPHEFT will always try to avoid link bottltnecks anyway, thus it's hard to create or simulate a case where TPHEFT+DUP is any good.
As we mentioned earlier, there are multiple factors that might make our system model different from the real world case, including virtual link model; multi-socket communication cost; etc. Although we don't know exactly how much of an impact each one has, we can compare the compile time throughput with runtime ones. Our results of these particular experiment show the value ranges from 1.4 to 3.  It is worth noting that in DAG8, although TPHEFT improved a lot on HEFT, SPLIT decreased throughput when applied on TPHEFT. We analyzed the data. Our conclusion is, for SPLIT, when computation and communication costs are close, the actual system bottleneck is a link, but since we used virtual links, one actual link may be split into several virtual links, thus none of them was counted as bottleneck. Instead we end up splitting nodes, which are not actual bottlenecks. With additional overhead, the more we split, the lower the throughput. For link bottlenecks, different clusters require different models, unless we know the link topology, we don't know what's the best model for a particular cluster. But as a c) Machine learning based object detection application: To show an example of real-world application with our system, we composed an object detection application (fig 8), which is a good example of dispersed computing. The input images are taken by different cameras from different angles, and we want detect objects in the combined image, which gives an overview of the scene. The app tasks two images (left and right image, with overlapping parts) as inputs, pre-processes each image (such as resize), then sends these two processed images to the stitching task, where we find the overlapping part to stitch the two processed images together. After this, we use a pre-trained model to locate humans and cars in the stitched image. In the end we group all the results as one matrix in a csv file as output. Throughput results are shown in the two tables in the "IMGP" row. They're basically consistent with results in the dummy DAG, given the specific shape and parameters of the IMGP application.

VII. CONCLUSION AND FUTURE WORK
In this paper, we proposed three throughput optimized scheduling algorithms for dispersed computing, and implemented a real-world distributed system to support them. We further conducted experiments using different DAGs, both dummy and real, to show the throughput improvement under different circumstances. TPHEFT and SPLIT are common algorithms while DUP is specially designed for slow links. In the future, we can extend our algorithms to non-idle nodes, multi-core CPUs, different models for link topology, and different applications from compute oriented ones. Further, we plan to work on dynamic scheduling with machine learning based approaches.