Anne Benoit

Matthieu Gallet

Bruno Gaujal
email: bruno.gaujal@inria.fr

Yves Robert

Computing the throughput of probabilistic and replicated streaming applications *

Keywords: scheduling, probabilistic streaming applications, replication, throughput, timed Petri nets

In this paper, we investigate how to compute the throughput of probabilistic and replicated streaming applications. We are given (i) a streaming application whose dependence graph is a linear chain; (ii) a one-to-many mapping of the application onto a fully heterogeneous target platform, where a processor is assigned at most one application stage, but where a stage can be replicated onto a set of processors; and (iii) a set of random variables modeling the computation and communication times in the mapping. We show how to compute the throughput of the application, i.e., the rate at which data sets can be processed, under two execution models, the Strict model where the actions of each processor are sequentialized, and the Overlap model where a processor can compute and communicate in parallel. The problem is easy when application stages are not replicated, i.e., assigned to a single processor: in that case the throughput is dictated by the critical hardware resource. However, when stages are replicated, i.e., assigned to several processors, the problem becomes surprisingly complicated: even in the deterministic case, the optimal throughput may be lower than the smallest internal resource throughput. The first contribution of the paper is to provide a general method to compute the throughput when mapping parameters are constant or follow I.I.D. exponential laws. The second contribution is to provide bounds for the throughput when stage parameters (computation and communication times) form associated random sequences, and are N.B.U.E. (New Better than Used in Expectation) variables: the throughput is bounded from below by the exponential case and bounded from above by the deterministic case. An extensive set of simulation allows us to assess the quality of the model, and to observe the actual behavior of several distributions.

Introduction

In this paper, we deal with streaming applications, or workflows, whose dependence graph is a linear chain composed of several stages. Such applications operate on a collection of data sets that are executed in a pipeline fashion [START_REF] Spencer | Executing multiple pipelined data analysis operations in the grid[END_REF][START_REF] Subhlok | Optimal mapping of sequences of data parallel tasks[END_REF][START_REF] Vydyanathan | Toward optimizing latency under throughput constraints for application workflows on clusters[END_REF]. They are a popular programming paradigm for streaming applications like video and audio encoding and decoding, DSP applications, etc [START_REF]DataCutter Project: Middleware for Filtering Large Archival Scientific Datasets in a Grid Environment[END_REF][START_REF] Taura | A heuristic algorithm for mapping communicating tasks on heterogeneous resources[END_REF][START_REF] Wu | Supporting distributed application workflows in heterogeneous computing environments[END_REF]. Each data set is input to the linear chain and traverses it until its processing is complete. While the first data sets are still being processed by the last stages of the pipeline, the following ones have started their execution. In steady state, a new data set enters the system every P time-units, and several data sets are processed concurrently within the system. A key criterion to optimize is the period, or equivalently its inverse, the throughput. The period P is defined as the time interval between the completion of two consecutive data sets. The system can process data sets at a rate ρ = 1/P, where ρ is the throughput.

The application is executed on a fully heterogeneous platform, whose processors have different speeds, and whose interconnection links have different bandwidths. We assume that the mappping of the application onto the platform is given, and that this mapping is one-to-many. In other words, when mapping application stages onto processors, we enforce the rule that any given processor will execute at most one stage. However, a given stage may well be executed by several processors. Indeed, if the computations of a given stage are independent from one data set to another, then two consecutive computations (for different data sets) of the same stage can be mapped onto distinct processors. Such a stage is said to be replicated, using the terminology of Subhlok and Vondran [START_REF] Subhlok | Optimal mapping of sequences of data parallel tasks[END_REF][START_REF] Subhlok | Optimal latency-throughput tradeoffs for data parallel pipelines[END_REF] and of the DataCutter team [START_REF] Beynon | Optimizing execution of component-based applications using group instances[END_REF][START_REF] Spencer | Executing multiple pipelined data analysis operations in the grid[END_REF][START_REF] Vydyanathan | A duplication based algorithm for optimizing latency under throughput constraints for streaming workflows[END_REF]. This also corresponds to the dealable stages of Cole [START_REF] Cole | Bringing Skeletons out of the Closet: A Pragmatic Manifesto for Skeletal Parallel Programming[END_REF]. Finally, we consider two execution models, the Strict model where the actions of each processor are sequentialized, and the Overlap model where a processor can process a data set while it is simultaneously sending the previous data set to its successor and receiving the next data set.

Unsurprisingly, the problem of finding a mapping with optimal throughput of tasks on processors is NPcomplete, even in the deterministic case and without any communication cost [START_REF] Benoit | Computing the throughput of probabilistic and replicated streaming applications[END_REF]. However, even determining the throughput may be difficult. In the deterministic case, and without replication, the throughput of a given mapping is easily seen to be dictated by the critical hardware resource: the period is the largest cycle-time of any resource, be it a processor or communication link. However, when stages are replicated, the problem becomes unexpectedly complicated: even in the deterministic case, the optimal throughput may be lower than the smallest internal resource throughput. In this paper, we present a model based on timed event graphs to determine this throughput.

We also introduce randomness in the execution of the application onto the platform. Consider the computations performed by a given processor on different data sets: we assume that the execution times of the computations are random variables that obey arbitrary I.I.D. (Independent and Identically-Distributed) probability laws. Similarly, we assume that the execution times of all the communications taking place on a given interconnection link are random variables that obey arbitrary I.I.D. probability laws. Note that the I.I.D. hypothesis apply to all events (computations or communications) that occur on the same hardware resource (either a processor or a communication link), and does not restrict the heterogeneity of the application/platform mapping. In other words, processors may well have different speeds, links may well have different bandwidths, stages may well have very different computation and data volumes; furthermore, the distribution law may well vary from one computation to another, or from one communication to another. One model also allows for correlation between communication and processing times of a given data-set, on the different stages.

The main contribution is to provide a general method (although of exponential cost) to compute the throughput when mapping parameters follow I.I.D. exponential laws. This general method is based upon the detailed analysis of the timed Petri nets deduced from the application mapping for each execution model, Strict and Overlap. It turns out that the Petri nets exhibit a regular structure in the Overlap model, thereby enabling to reduce the cost and provide a polynomial algorithm. The second main contribution is to provide bounds for the throughput when stage parameters are associated (computation and communication times form associated random sequences, [START_REF] Esary | Association of random variables, with applications[END_REF]) and N.B.U.E. (New Better than Used in Expectation) variables: the throughput is bounded from below by the exponential case and bounded from above by the deterministic case. Note that the associated case is more general than the I.I.D. case, and also it is more realistic. Imagine for instance that the speed of the processors are constant, and that the size of the data is not altered while it is processed. Then, the time required to process a data is directly related to the processor on which it is executed, and these processing times cannot be assumed to be independent.

This paper unifies the results presented in [START_REF] Benoit | Computing the throughput of probabilistic and replicated streaming applications[END_REF][START_REF] Benoit | Computing the throughput of replicated workflows on heterogeneous platforms[END_REF] and adds several new results. In particular, we deal with the more realistic and more general case of associated variables, while [START_REF] Benoit | Computing the throughput of replicated workflows on heterogeneous platforms[END_REF] only deals with deterministic variables and [START_REF] Benoit | Computing the throughput of probabilistic and replicated streaming applications[END_REF] only considers I.I.D. variables. A detailed discussion on the differences between the three cases is given in Section 2.4.

The paper is organized as follows. First in Section 2, we formally describe the framework and the optimization problems, and we introduce the random variables that are used for the probabilistic study. Then we briefly introduce in Section 3 timed event graphs which are used to solve the deterministic case. Using these event graphs, we solve the deterministic (or static) case in Section 4. We explain how to compute the throughput when communication and computation times follow I.I.D. exponential laws (Section 5). We give a general method which turns out to be of exponential complexity in the general case, but we provide a polynomial algorithm for the Overlap model. Then in Section 6, we deal with arbitrary I.I.D. and N.B.U.E. laws, and we establish the above-mentioned bounds on the throughput. Some experimental results are given in Section 7, both to assess the quality of our model and to observe the behavior of more complex random laws. Finally, we present some conclusions and directions for future work in Section 8.

Models

In this section, we first describe the workflow application, the target platform, and the communication models that we consider (Section 2.1). The replication model is presented in Section 2.2. We formally define the throughput in Section 2.3. Finally, we give a detailed presentation of the random variables that model processor speeds and link bandwidths (Section 2.4).

Application, platform and communication models

We deal with streaming applications, or workflows, whose dependence graph is a linear chain composed of N stages, called T i (1 ≤ i ≤ N ). Each stage T i has a size w i , expressed in floating point operations (flop), and needs an input file F i-1 of size δ i-1 , expressed in bytes. Finally, T i produces an output file F i of size δ i , which is the input file of stage T i+1 . Note that T 1 produces the initial data and does not receive any input file, while T N gathers the final data.

The workflow is executed on a fully heterogeneous platform with M processors. The speed of processor P p (1 ≤ p ≤ M ) is denoted as s p (in flop per seconds). We assume bidirectional links link p,q : P p → P q between any processor pair P p and P q , with bandwidth b p,q bytes per second. These links are not necessarily physical, they can be logical. For instance, we can have a physical star-shaped platform, where all processors are linked to each other through a central switch. The time needed to transfer a file F i from P p to P q is δi bp,q , while the time needed to process T i on P p is wi sp . An example of linear chain application and fully connected target platform is provided in Figure 1.

We consider two different realistic common models for communications. The Overlap model allows to overlap communications and computations: a processor can simultaneously receive values for the next data set, compute result for the current data set, and send output data for the previous data set. Requiring multithreaded programs and full-duplex network interfaces, this model allows for a better use of computational resources. On the contrary, in the Strict model, there is no overlap of communications by computations: a processor can either receive a given set of data, compute its result or send this result. This is the typical execution of a single-threaded program, with one-port serialized communications. Although leading to a less efficient use of physical resources, this model allows for simpler programs and hardware.

Replication model

When mapping application stages onto processors, we enforce the rule that any given processor will execute at most one stage. But instead of considering one-to-one mappings [START_REF] Benoit | Mapping pipeline skeletons onto heterogeneous platforms[END_REF], we allow stage replication, and rather consider one-to-many mappings, in which each stage can be processed by several processors. This is possible when the computations of a given stage are independent from one data set to another. In this case, two consecutive computations (different data sets) for the same stage can be mapped onto distinct processors. Such a stage is said to be replicated [START_REF] Beynon | Optimizing execution of component-based applications using group instances[END_REF][START_REF] Spencer | Executing multiple pipelined data analysis operations in the grid[END_REF][START_REF] Subhlok | Optimal mapping of sequences of data parallel tasks[END_REF][START_REF] Subhlok | Optimal latency-throughput tradeoffs for data parallel pipelines[END_REF][START_REF] Vydyanathan | A duplication based algorithm for optimizing latency under throughput constraints for streaming workflows[END_REF] or dealable [START_REF] Cole | Bringing Skeletons out of the Closet: A Pragmatic Manifesto for Skeletal Parallel Programming[END_REF].

Note that the computations of a replicated stage can be fully sequential for a given data set, what matters is that they do not depend upon results from previous data sets, hence the possibility to process different data sets in different locations. The following scheme illustrates the replication of a stage T i onto three . . . T i-1 T i on P 1 : data sets 1, 4, 7, . . . --T i on P 2 : data sets 2, 5, 8, . . . --T i on P 3 : data sets 3, 6, 9, . . .

P 2 P 6 F 2 F 1 T 2 F 3 T 1 T 4 T 3
T i+1 . . . (1)
For 1 ≤ i ≤ N , let R i denote the number of processors participating to the processing of T i . For 1 ≤ p ≤ M , if P p participates to the work of T i , then we write p ∈ Team i and define R ′ p =R i . As outlined in the scheme, the processors allocated to a replicated stage execute successive data sets in a round-robin fashion. This may lead to a load imbalance: more data sets could be allocated to faster processors. But this would imply out-of-order execution and would require a complicated data management if, say, a replicated stage is followed by a non-replicated one in the application pipeline. In particular, large buffers would be required to ensure the in-order execution of the non-replicated stage. This explains why round-robin execution has been enforced in all the papers referenced above, and we enforce this rule too.

Because of the round-robin rule, the execution of a replicated stage is slowed down by the slowest processor involved in the round-robin. Let P slow be the slowest processor involved in the replication of T i . Then, if p ∈ Team i , P p processes one data set every R i data sets at the speed dictated by P slow , and thus its computation time (per data set) is C comp (p) = wi Ri×s slow . Note that this implies that if processors of different speeds are processing a same stage, some of them will remain partly idle during the execution.

Throughput definition

The throughput ρ is defined as the average number of data sets which can be processed within one time unit, and the period P is the inverse of the throughput and corresponds to the time-interval that separates two consecutive data sets entering the system. We can derive a lower bound for the period as follows. Let If no stage is replicated, then the throughput is simply determined by the critical resource (maximum cycle-time): ρ = 1/M ct . However, when stages are replicated, the previous result is no longer true, and more sophisticated techniques are required. In the following, we investigate how to compute the throughput when computation and communication times are constant or subject to random variations.

Random computation and communication times

We assume that at most one stage is processed by each processor. Since the mapping is given, the value of computation (or communication) times used by each resource fully defines the behavior of the system.

We distinguish three cases, the deterministic case where computation and communication times are constant, independent of the data set, the I.I.D. case where computation and communication times form random I.I.D. sequences, and the associated case where they form associated random sequences.

Deterministic case

In this case, we assume that the computation and communication times are constant and do not depend on the data set. Hence, we denote the computation time of stage T i on processor P p by c p = w i /s p . Similarly, the communication time of the file F i sent by P p to P q is given by d p,q = δ i /b p,q .

Stochastic cases

Here, we consider that the time to execute a stage, and the time to transfer data, depend on the data set. The n-th instance of a task may not be of the same size as the previous one, or the speed of the processor may depend on time (for example in the case where the processor is not dedicated to the streaming application but shares its processing capacity and communication bandwidth with other exogenous applications).

In such cases, it is often convenient to model these values as random variables. We denote by w i (n) the size of the n-th instance of the data at stage T i , s p (n) is the speed of processor p when processing for the n-th time, δ i (n) is the size of the n-th transfer file at stage T i , and b p,q (n) is the bandwidth available between processor p and q during the n-th transfer.

Once these variables are defined, it is possible to define X p (n), the random variable giving the computation time of the n-th data set, processed by P p , where p ∈ Team i (recall that each processor deals with only one stage). We have X p (n) = w i (n)/s p (n) for all n. Similarly, let Y p,q (n) be the random variable giving the communication time of the n-th file of type F i transferred from P p to P q , where p ∈ Team i and q ∈ Team i+1 . We have Y p,q (n) = δ i (n)/b p,q (n) for all n.

Two kinds of stochastic assumptions are used in the following.

• The first one assumes that the random sequences {X p (n)} n∈N and {Y p,q (n)} n∈N are I.I.D. (independent and identically distributed) and mutually independent sequences. This case is denoted as the I.I.D. case in the following. • The second case rather makes assumptions on the parameters defining the system, namely δ i (n), w i (n), s p (n) and b p,q (n). We assume that the processes {s p (n)} 1≤p≤M,n∈N , {b p,q (n)} 1≤p,q≤M,n∈N , and for all i, {δ i (n) n∈N } and {w i (n) n∈N } are I.I.D. and mutually independent. However, for any fixed n ∈ N, the set of variables S = {δ i (n)} 1≤i≤N ∪ {w i (n)} 1≤i≤N are assumed to be associated (i.e., for all U, V ∈ S, Pr (U < u, V < v) ≥ Pr (U < u) Pr (V < v), see [START_REF] Esary | Association of random variables, with applications[END_REF]). This case is called the associated case in the following. The second case is more general than the first one since one can choose δ i (n) and w i (n) to be constants so that {X p (n)} n∈N and {Y p,q (n)} n∈N become independent sequences. It should also be clear that the associated case is more realistic. Imagine that the speed of the processors as well as the communication bandwidths on all the links are constant (this is typically true when processors and resources are dedicated to this application and no other process or communication share the resources). Also assume that the size of the n-th data does not depend on the stage, that is δ i (n) = δ 1 (n) for all i. Then, the processing time on processor p, allocated to stage i, is X p (n) = δ i (n)/s p . The processing time of the same data on processor q at another stage (say j) of the pipeline is X q (n) = δ j (n)/s q . They are related by X p (n) = sq sp X q (n) so that they cannot be assumed to be independent. The I.I.D. assumption made in the first case only holds when the processing times of the same data over two different stages are not related in any manner.

The general relation that exists between the sequences {X p (n)} n∈N and {X q (n)} n∈N in this second case can be made more precise. Actually, it is rather direct to show that they are also associated, which implies that given any two processors p and q, the larger X p (n), the larger X q (n) is likely to be. This will be formally proved in Section 6.2.

We let (X, Y ) denote the mapping of an application (T 1 , . . . , T N ) on a platform (P 1 , . . . , P M ). In both the I.I.D. and associated cases, since all variables are identically distributed, the probability that the computation time of T i on P p is larger than x does not depend on n and is given by Pr (X p (1) > x), while its expected value is denoted by E[X p ]. So far, no assumption is made on the distributions of the random processing and communication times. However, some of our results are only valid for specific distributions of random variables. Below we recall the definition of these specific classes.

Exponential variables. An important class of random variables is the one with exponential distributions. The probability that an exponential random variable X with rate λ is larger than t is given by Pr (X > t) = e -λt .

New Better than Used in Expectation (N.B.U.E.) variables. A random variable X is said to have a N.B.U.E. distribution if, and only if, E[X -t|X > t] ≤ E[X], for all t > 0. In other words, the N.B.U.E. assumption for communication or computation times means that if a computation (or a communication) has already been processed for a duration t and it is not finished yet, then the remaining time is smaller than the processing time of a fresh operation. This assumption is often true since in most cases, a partial execution of a stage should not increase the remaining work, especially when the amount of computation and communication are bounded from above. Note that exponential variables have the N.B.U.E. property, with equality in that case (E[X -t|X > t] = E[X], for all t > 0). Also, note that there exist many statistical procedures to test if a random variable is N.B.U.E. [START_REF] Kumazawa | Tests for new better than used in expectation with randomly censored data[END_REF].

In the following, we always assume, by default, that we are in the I.I.D. case, rather than in the associated case, unless it is stated otherwise.

3 Modeling the problem with event graphs

Mappings with replication

In this section, we aim at modeling mappings with timed Petri nets (TPNs) as defined in [START_REF] Baccelli | Recursive Equations and Basic Properties of Timed Petri Nets[END_REF], in order to be able to compute the period of a given mapping. In the following only TPNs with the event graph property (each place has exactly one input and one output transition) will be considered (see [START_REF] Baccelli | Synchronization and Linearity[END_REF]). We consider mappings where some stages may be replicated, as defined in Section 2.2: a stage can be processed by one or more processors. As already stated, two rules are enforced to simplify the model: a processor can process at most one stage, and if several processors are involved in the computation of one stage, they are served in a round-robin fashion. In all our Petri net models, the use of a physical resource during a time t (i.e., the computation of a stage or the transmission of a file from a processor to another one) is represented by a transition with a firing time t, and dependences are represented using places. Now, let us focus on the path followed in the pipeline by a single input data set, for a mapping with several stages replicated on different processors. Consider Example A described in Figure 1: the first data set enters the system and proceeds through processors P 1 , P 2 , P 4 and P 7 . The second data set is first processed by processor P 1 , then by processor P 3 (even if P 2 is available), by processor P 5 and finally by processor P 7 . There are six different paths followed by the data sets, and then data set i takes the same path as data set i -6. We have the following easy result: Proposition 1. Consider a pipeline of N stages T 1 , . . . , T N , such that stage T i is mapped onto m i distinct processors. Then the number of paths followed by the input data in the whole system is equal to m = lcm (m 1 , . . . , m N ).

Proof. Let m be the number of paths P j followed by the input data. Assume that stage T i is processed by processors P i,1 , . . . , P i,mi . By definition, all paths are distinct. Moreover, the round-robin order is respected: path P j is made of processors (P 1,j mod m1 , . . . , P i,j mod mi , . . . , P N,j mod m N ). The first path P 1 is made of (P 1,1 , P 2,1 , . . . , P N,1 ). By definition, m is the smallest positive integer, such that the (m + 1)-th used path is identical to the first one: ∀i ∈ {1, . . . , N }, m mod m i = 0. Indeed, m is the smallest positive integer, which is divisible by each m i , i.e., m = lcm (m 1 , . . . , m N ).

The TPN model described below is the same flavor as what has been done to model jobshops with static schedules using TPNs [START_REF] Hillion | Performance evaluation of job shop systems using timed event graphs[END_REF]. Here, however, replication imposes that each path followed by the input data must be fully developed in the TPN: if P 1 appears in several distinct paths, as in Figure 1, there are several transitions corresponding to P 1 . Furthermore, we have to add dependences between all the transitions corresponding to the same physical resource to avoid the simultaneous use of the same resource by different input data. These dependences differ according to the model used for communications and computations.

Overlap model

We first focus on the Overlap model: any processor can receive a file and send another one while computing. All paths followed by the input data in the whole system have to appear in the TPN. We use the notations of Proposition 1.

Let m denote the number of paths in the mapping. Then the i-th input data follows the (i mod m)th path, and we have a rectangular TPN, with m rows of 2N -1 transitions, due to the N transitions representing the use of processors and the N -1 transitions representing the use of communication links. The i-th transition of the j-th row is named T j i . The time required to fire a transition T j 2i (corresponding to the processing of stage T i on processor P u ) is set to wi su , and the one required by a transition T j 2i+1 (corresponding the transmission of file F i from P u to P v ) is set to δi bu,v . Then we add places between these transitions to model the following set of constraints: 1. The file F i cannot be sent before the computation of T i : a place is added from T j 2i to T j 2i+1 on each row. Similarly, the stage T i+1 cannot be processed before the end of the communication of F i : a place is added from T j 2i+1 to T j 2(i+1) on each row j. 2. When a processor appears in several rows, the round-robin distribution imposes dependencies between these rows. Assume that processor P i appears on rows j 1 , j 2 , . . . , j k . Then we add a place from T j l 2i to T j l+1 2i with 1 ≤ l ≤ k -1, and a place from T j k 2i to T j1 2i . 3. The one-port model and the round-robin distribution of communications also impose dependences between rows. Assume that processor P i appears on rows j 1 , j 2 , . . . , j k . Then we add a place from T j l 2i+1 to T j l+1

2i+1 with 1 ≤ l ≤ k -1, and a place from T j k 2i+1 to T j1 2i+1 to ensure that P i does not send two files simultaneously, if P i does not compute the last stage. 4. In the same way, we add a place from T j l 2i-1 to T j l+1

2i-1 with 1 ≤ l ≤ k -1, and a place from

T j k 2i-1 to T j1
2i-1 to ensure that P i does not receive two files simultaneously, if P i does not compute the first stage.

Finally, any resource before its first use is ready to compute or communicate, only waiting for the input file. Indeed, a token is put in every place going from a transition T j k i to a transition T j1 i , as defined in the previous lines. The complete TPN of Example A for the Overlap model is given in Figure 2.

Strict model

In the Strict model, any processor can either send a file, receive another one, or perform a computation while these operations were happening concurrently in the Overlap model. Hence, we require a processor to successively receive the data corresponding to an input file F i , compute the stage T i+1 and send the file F i+1 before receiving the next data set of F i . Paths followed by the input data are obviously the same as in Subsection 3.2, and the structure of the TPN remains the same (m rows of 2N -1 transitions).

The first set of constraints is also identical to that of the Overlap model, since we still have dependences between communications and computations, as in Figure 2. However, the other dependences are replaced by those imposed by the round-robin order of the Strict model. Indeed, when a processor appears in several rows, the round-robin order imposes dependences between these rows. Assume that processor P i appears on rows j 1 , j 2 , . . . , j k . Then we add a place from T j l 2i+1 to T

j l+1 2i-1 with 1 ≤ l ≤ k -1,

T j1

2i-1 . These places ensure the respect of the model: the next reception cannot start before the completion of the current sequence reception-computation-sending.

Any physical resource can immediately start its first communication, since it is initially only waiting for the input file. Thus a token is put in every place from a transition T j k i to a transition T j1 i , as defined in the previous lines. The complete TPN of Example A for the Strict model is given in Figure 3.

Note that the automatic construction of the TPN in both cases has been implemented. The time needed to construct the Petri net is linear in their size: O (m × N ).

Computing throughputs in the deterministic case

For TPNs with the event graph property the computation of the throughput is possible through the computation of critical cycles, using the (max, +) algebra [START_REF] Baccelli | Synchronization and Linearity[END_REF]. For any cycle C in the TPN, let L(C) be its length (number of transitions) and t(C) be the total number of tokens in places traversed by C. Then a critical cycle achieves the largest ratio max Ccycle L(C) t(C) , and this ratio is the period P of the system: indeed, after a transitive period, every transition of the TPN is fired exactly once during a period of length P [START_REF] Baccelli | Synchronization and Linearity[END_REF].

Critical cycles can be computed with softwares like ERS [START_REF] Jean-Marie | ERS: a tool set for performance evaluation of discrete event systems[END_REF] or GreatSPN [START_REF] Chiola | GreatSPN: Graphical Editor and Analyzer for Timed and Stochastic Petri Nets[END_REF] with a complexity O m 3 N 3 . By definition of the TPN, the firing of any transition of the last column corresponds to the completion of the last stage, i.e., to the completion of an instance of the workflow. Moreover, we know that all the m transitions (if m is still the number of rows of the TPN) of this last column are fired in a round-robin order. In our case, m data sets are completed during any period P: the obtained throughput ρ is m P .

Overlap model

The TPN associated to the Overlap model has a regular structure, which facilitates the determination of critical cycles. In the complete TPN, places are linked to transitions either in the same row and oriented forward, or in the same column. Hence, any cycle only contains transitions belonging the same "column": we can split the complete TPN into 2N -1 smaller TPNs, each sub-TPN representing either a communication or a computation. However, the size of each sub-TPN (the restriction of the TPN to a single column) is not necessarily polynomial in the size of the instance, due to the possibly large number of rows, equal to m = lcm (m 1 , . . . , m N ).

It turns out that a polynomial algorithm exists to find the weight L(C)/t(C) of a critical cycle: only a fraction of each sub-TPN is required to compute this weight, without computing the cycle itself.

Theorem 1. Consider a pipeline of N stages T 1 , . . . , T N , such that stage T i is mapped onto m i distinct processors. Then the average throughput of this system can be computed in time

O N -1 i=1 (m i m i+1 ) 3 .
The proof of this theorem is available in [START_REF] Benoit | Computing the throughput of replicated workflows on heterogeneous platforms[END_REF]. In Example A (see Figure 1), a critical resource is the output port of P 1 , whose cycle-time is equal to the period, 189. However it is possible to exhibit cases without critical resource.

Strict model

Cycles in the TPN associated to the Strict model are more complex and less regular, since corresponding TPNs have backward edges. The intuition behind these backward edges is that a processor P u cannot compute an instance of S i before having completely sent the result F i of the previous instance of S i to the next processor P v . Thus, P u can be slowed by P v . As for the Overlap model, there exist mappings for which all resources have idle times during a complete period. With the Strict model, this is the case for Example A, the critical resource is P 2 , which has a cycle-time M ct = 215.8, strictly smaller than the period P = 230.7.

Computing the throughput with exponential times

In this section, we consider a system with stochastic I.I.D. communication and computing times with exponential distributions. In the corresponding Petri net, all transitions (modeling processing times or modeling communication times) have exponential firing times. The probability of the firing time t i of a transition is given by Pr (t i > x) = 1 -e -λix . The firing rate λ i corresponds either to the processing rate of one processor or the communication rate over one link.

The study of the exponential case is motivated by two facts. First, one can get explicit formulas in this case. Second (as we will see in Section 6), exponential laws are extreme cases among all N.B.U.E. variables.

In the rest of this section, we first explain the general method which allows us to compute the throughput for both the Overlap and the Strict models. However, this general method has a high complexity. In the Overlap case, we provide a simpler method, building upon the relative simplicity of the timed Petri net (Section 5.2). Finally in Section 5.3, we derive a polynomial algorithm for the Overlap case when we further assume that the communication network is homogeneous. Model the system as a timed Petri net. The transformation of the initial system into a timed Petri net is fully described in Section 3. Recall from Section 3 that it consists in R = lcm 1≤i≤N (R i ) rows and 2N -1 columns, and examples for both models are depicted in Figure 2 and Figure 3. This step is done in time O (RN ), and the expectation of the delay between two successive firings of any transition gives the throughput of the system.

General method to compute the throughput

Transformation of the timed Petri net into a Markov chain. To compute the expectation of the delay between two successive firings of any transition, we transform the above timed Petri net into a Markov chain (Z 1 , Z 2 , . . .). To each possible marking of the timed Petri net, we associate a state x i . There are (2N + 3(N -1))R places, and each place contains either zero or one token. Thus, there are at most 2 (2N +3(N -1))R possible different markings, leading to the same number of states in the Markov chain.

Due to the exponential size of the number of states of the Markov chain, we only consider the part of the timed Petri net corresponding to communications in examples. This part is shown in Figure 4.

On Example A, places are named (P 1 , P 2 , P 3 , P 4 , P 5 , P 6 , P 7 , P 8 , P 9 , P 10 , P 11 , P 12 ), while transitions are named (a, b, c, d, e, f ). Thus, a state is defined by a 12-tuple, each number equal to either 0 or 1 being the number of tokens in the place. In the Markov chain, moving from a state to another corresponds to the firing of a transition of the timed Petri net. Thus, arrows in the graphical representation of the Markov chain are labeled with the names of the transitions. The complete list of possible states and the corresponding transitions are given in Figure 5. If in state x i , transition T j can be fired leading to state x k , then the transition rate of the corresponding arrow is set to λ j .

Computation of the throughput. Using this new representation, we are able to compute the throughput. The throughput is the number of completed last stage T N per time unit. In terms of Petri nets, this is also the expected number of firings per time unit of the transitions in the last column. Thus, in terms of Markov chains, the throughput is given by the probability of being in one of the states enabling these transitions. By construction of the Markov chain, all of its states are positive recurrent. Thus, it admits a stationary distribution, giving the probability of each state. This stationary distribution can be computed in polynomial time in the size of the Markov chain by solving a linear system [START_REF] Häggström | Finite Markov Chains and Algorithmic Applications[END_REF]. The sum of the probability of the valid states returns the throughput. f c b (0, 1, 0, 0, 1, 0, 0, 1, 0, 1, 0, 1) (0, 0, 1, 0, 1, 0, 1, 0, 0, 1, 1, 0) (0, 1, 0, 0, 1, 0, 1, 0, 1, 0, 1, 0) (1, 0, 0, 0, 1, 0, 1, 0, 0, 1, 0, 1) (0, 1, 0, 1, 0, 0, 0, 1, 1, 0, 1, 0)

(1, 0, 0, 1, 0, 0, 0, 1, 0, 1, 0, 1) (0, 0, 1, 1, 0, 0, 0, 1, 0, 1, 1, 0)

(1, 0, 0, 1, 0, 0, 1, 0, 1, 0, 1, 0) (0, 0, 1, 0, 0, 1, 0, 1, 0, 1, 0, 1) (0, 1, 0, 0, 0, 1, 0, 1, 1, 0, 0, 1) (0, 0, 1, 0, 0, 1, 1, 0, 1, 0, 1, 0)

(1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0,

Overlap model

We now focus on the Overlap model. As in the deterministic case, constraints applying to our system form a very regular timed Petri net which is feed forward (dependencies only from column C i to column C i+1 , for 1 ≤ i ≤ 2N -2), giving an easier problem than the Strict model. Proof. First, let us give the overall structure of the proof:

1. split the timed Petri net into columns C i , with 1 ≤ i ≤ 2N -1; 2. separately consider each column C i ; 3. separately consider each connected component D j of C i ; 4. note that each component D j is made of many copies of the same pattern P j , of size u j × v j ; 5. transform P j into a Markov chain M j ; 6. determine a stationary measure of M j , using a combinatorial trick based on Young diagrams [START_REF] Knuth | The Art of Computer Programming[END_REF]; 7. compute the throughput of P j in isolation (called hereinafter inner throughput of component D j ); 8. combine the inner throughputs of all components to get the global throughput of the system. To decrease the overall complexity, we use the same idea as in Section 4.1: thanks to the regularity of the global timed Petri net, we split it into a polynomial number of columns, and we compute the throughput of each column independently.

Let us focus on a single column. We have two cases to consider: (i) the column corresponds to the computation of a single processor (columns C 2i-1 , for 1 ≤ i ≤ N ); (ii) the column corresponds to communications between two sets of processors (columns C 2i , for 1 ≤ i ≤ N -1).

In case (i), cycles do not interfere: any cycle involves a single processor, and any processor belongs to exactly one cycle. Thus, the inner throughput is easily computed, this is the expectation of the number of firings per time unit. The processing time X p (n) being exponential, this is equal to the rate λ p of X p .

On the contrary, case (ii) is more complex and requires a more detailed study. Let us consider the i-th communication (it corresponds to column C 2i ): it involves R i senders and R i+1 receivers. We know from its structure that the timed Petri net is made of g = gcd(R i , R i+1 ) connected components. Let u j be equal to R i /g and v j be equal to R i+1 /g. Then each connected component D j in this column is made of c = R lcm(Ri,Ri+1) copies of a pattern P j of size u j × v j . Since these components are independent, we can compute the throughput of each of them independently. In the case of Example B presented in Figure 6, we More precisely, we focus on the second communication (see Figure 6, involving 21 senders and 27 receivers). In this case, we have g = 3 connected components, made of 55 copies of pattern P j of size u j × v j = 9 × 7.

G ′ v = 9 3 connected components 27 5 R 1 R 2 R 4 R 3 T 1 T 2 T 4 T 3 11 21
Each pattern is a timed Petri net P j with a very regular structure, which can be represented as a rectangle of size (u j , v j ), also denoted (u, v) to ease notations, as shown in Figure 6. As said before, determining the throughput of P is equivalent to determining a stationary measure of a Markov chain. We know that the stationary measure of a Markov chain with t states can be computed in time O t 3 [START_REF] Häggström | Finite Markov Chains and Algorithmic Applications[END_REF]. Thus, we need to determine the number of states of the transformation of P j into a Markov chain. Let M j be this Markov chain.

The number of states of M j is by definition the number of possible markings, and we can directly determine it. A valid marking of P j is represented in Figure 7. The regularity of the structure imposes some constraints to valid markings: a transition can be fired for the k-th time if, and only if, all the transitions above it or on its left have been fired k times. In other terms, if a processor sends a file to q receivers P 1 , . . . , P q , it can send the k-th instance of the file to P i if and only if it has sent the k first instances of the file to P 1 , . . . , P i-1 .

In our rectangular representation of the timed Petri net, the borderline between transitions that have been fired k + 1 times and those that have been fired k times is the union of two Young diagrams, as displayed on Figure 7. Since there is only a single token in each column and in each row, we cannot have three simultaneous Young diagrams.

Let us compute the number of states of the Markov chain M j . As said in the previous paragraph, the (0, v) borderline can be seen as two Young diagrams, or two paths. The first one is from coordinates (i, 0) to (0, j), and the second one goes from (u, j) to (i, v) (see Figure 8). If i and j are given, then there are α i,j = i + j i possible paths from (i, 0) to (0, j), where n k is equal to

(u, v) u -i v -j (v, 0) i (0, 0) j
n! k!(n-k)! .
Similarly, there are α u-1-i,v-1-j possible paths from (u, j) to (i, v). Thus, if i and j are given, then there are α i,j ×α u-1-i,v-1-j possible markings. If i and j are not given anymore, then the total number S(u, v) of valid markings can be easily determined:

S(u, v) = u-1 i=0 v-1 j=0 α i,j α u-1-i,v-1-j = u-1 i=0 v-1 j=0 i + j i u + v -2 -i -j u -1 -i = u + v -1 u -1 v = (u + v -1)! (u -1)!v! v .
Thus, the Markov chain of a single connected component has exactly S(u, v) = (u+v-1)! (u-1)!v! v states, and its inner throughput can be computed in time S(u, v) 3 .

Let us now come to the computation of the global throughput of the system. Actually, the throughput is given by the following iteration. The throughput of one strongly connected component is the minimum of its inner throughput and the throughput of all its input components, so once all inner throughputs are known, the computation of the throughput is linear in the number of components.

In column C 2i , we have g = gcd(R i , R i+1 ) connected components so that the total computation time to obtain their throughput is equal to gS(u, v). Since we have S(gu, gv) ≥ gS(u, v), u = R i /g and v = R i+1 /g, the total computation time to determine the throughput of C 2i is less than S(R i , R i+1 ).

Finally, the total computation time of the throughput is equal to

N -1 i=1 S(R i , R i+1 ) 3
, leading to our result of a throughput that can be computed in time O (N exp(max 1≤i≤N (R i ))).

Overlap model, homogeneous communication network

In the case where all the communication times in one column are all I.I.D., with the same rate in component D j , denoted λ j , then the inner throughput of each strongly connected component (i.e., the throughput of the component if isolated from the global timed Petri net) can be computed explicitly with a very simple formula. This reduces the overall computation of the throughput to a simple computation of minimums over the strongly connected components, which can be done in polynomial time.

Theorem 4. Let us consider the system (X, Y ) formed by the mapping of an application onto a platform, following the Overlap communication model with a homogeneous communication network. Then the throughput can be computed in polynomial time.

The global throughput can then be computed in polynomial time from all inner throughputs, and it is equal to ρ = Dj ∈C 2N -1 min D j ′ ≺Dj ρ j ′ , where D j ′ ≺ D j means that there exists a path from component

D j ′ to component D j , or D j ′ = D j . Because of the structure of the timed Petri net, if D j ′ is in column C i ′ and D j is in column C i , then i ′ < i or j ′ = j.
The computation can thus be done column by column. For any component in the first column, its throughput must be equal to its inner throughput ρ j . The computation for column i only depends on results from column i -1 by construction of the Petri net. Moreover, the total number of components is polynomial in the number of processors M . We obtain therefore a polynomial complexity (2N -1 columns in the timed Petri net, and a polynomial number of components).

Proof. Platforms with the Overlap model and a homogeneous communication network are special cases of the Overlap model. Thus, the proof of Theorem 3 remains true, and we focus again on the Markov chain M j , obtained from a pattern of component D j . If D j corresponds to a processor, the formula given previously applies, and its inner throughput is ρ j = λ j .

Next, we focus on a strongly connected component corresponding to a communication. We already know that the throughput is given by an invariant measure of M j . Graphically, the set of reachable states from a given state is easy to define: any of the top-left corners in the line can be "inverted" into a bottom-right corner to obtain a new valid state. In terms of Petri nets, this corresponds to the firing of one fireable transition. On Figure 7, there are 4 fireable transitions, corresponding to 4 top-left corners on the bold line in the equivalent Figure 8. Moreover, this reasoning can be inverted: any bottom-right corner in Figure 7 can be inverted, giving a possible previous marking leading to the current marking. Since we have as many top-left corners as bottom-right ones on Young diagrams, any state of M j has the same number of incoming states as outgoing ones. Moreover, since the communication network is homogeneous, all transitions have the same firing rate. These two conditions imply that the invariant measure of the Markov chain M j is uniform [START_REF] Häggström | Finite Markov Chains and Algorithmic Applications[END_REF]: if S is the number of states of M j , then its invariant measure is ( 1 S , . . . , 1 S ). Lastly, let us compute the number of states of M j allowing a given transition to be fired. Due to symmetry, all transitions have the same firing rate and we can only consider the top-right transition T 0 of the net. By using the bijection with Young diagrams, the number of markings such that T 0 is fireable is exactly the number S ′ (u, v) of possible paths starting from this top-right corner. The quantity S ′ (u, v) is computed in the same way as S(u, v) (see proof of Theorem 3):

S ′ (u, v) = u-2 i=0 v-2 j=0 α i,j = u + v -2 u -1 = S(u, v) v + u -1 .
Finally, we know the rate of the states leading to a given transition, and the number of states leading to it. Between two successive firings of the same transitions, uv communications are made (remind that gcd(u, v) = 1). Thus, the inner throughput is equal to vuλj (v+u-1) . As in the previous case, the throughput of a component can be computed in an iterative way. The throughput of one component is equal to the minimum of its inner throughput and the throughput of all its incoming components. This allows one to compute the throughput of all components starting from the first column and ending in the last one. Now, the global throughout is the rate at which tasks exit the system. This is equal to the sum of the throughputs of all components in the last column 2N -1. The throughputs of the last components are equal to the minimum of the inner throughputs of all components on paths from the first column to the last.

Computing ρ column by column renders the computation of this formula polynomial in the number of tasks and processors. [START_REF] Beynon | Optimizing execution of component-based applications using group instances[END_REF] Comparison results in the I.I.D. and associated cases

In the previous section, we have shown how to compute the throughput when all communication and processing times are exponential variables (in polynomial time for the homogeneous Overlap case).

In general, it is well known that the computation of the throughput is hard for stochastic communication and processing times, even for very simple cases [START_REF] Kamburowski | Bounding the distribution of project duration in pert networks[END_REF]. However, the fact that in our case, the throughput is an increasing and convex function of communication and processing times implies that one can use stochastic comparisons to construct bounds on the throughput whenever communication and processing times are I.I.D. or associated variables, and have the N.B.U.E. property. Moreover, the lower and upper bounds are obtained by the deterministic and exponential cases respectively. This section is dedicated to the study of several comparisons between systems with different processing and communication times. To do so, we first need to introduce a way to define orders on random sequences. We use classical definitions. Definition 1. Let {V (n)} n∈N and {W (n)} n∈N be two real random variable sequences:

• V is smaller than W for the strong order (hereinafter denoted

V ≤ st W ) if for all increasing function f , E[f (V (1), V (2), • • • )] ≤ E[f (W (1), W (2), • • • )]. • V is smaller than W for the increasing convex order (hereinafter denoted V ≤ icx W ) if for all increasing convex function g, we have E[g(V (1), V (2), • • • )] ≤ E[g(W (1), W (2), • • • )].
We also consider another order for multi-dimensional variables which is weaker than the strong order but that will be enough to compare the expected throughputs. Definition 2. Let {V (n)} n∈N and {W (n)} n∈N be two real random sequences. V is smaller than W for the lower orthant order (denoted V ≤ lo W ) if for any infinite sequence x 1 , x 2 , . . .

P (V (1) ≤ x 1 , V (2) ≤ x 2 , • • • ) ≥ P (W (1) ≤ x 1 , W (2) ≤ x 2 , • • • ).

The I.I.D. case

In the following, we consider a system that is either Strict or Overlap and whose processing times and communication times are I.I.D.. Theorem 5. Consider two systems (X (1) , Y (1) ) and (X (2) , Y (2) ). If we have for all n, ∀1 ≤ p ≤ M , X

p (n) ≤ st X (2) p (n) and ∀1 ≤ p, q ≤ M, Y (1)
p,q (n) ≤ st Y (2) (1)
p,q (n), then ρ (1) ≥ ρ (2) .

b (n) = D b (n -1) ⊗ A b (n), where the matrices A b (n) are such that A b (n) ij = k T b k (n)
if a path connects transtions T p and T q with one token in the first place of the path and no token in any other place. Now, the firing times of the transitions T b k (n) are either communication times or processing times so that there exists p, q (only depending on k, in a bijective way) such that

T b k (n) = X (b) p (n) or T b k (n) = Y (b) p,q (n). Therefore, T 1 k (n) and T 2 k (n) are I.I.D. sequences such that T 1 k (n) ≤ st T 2 k (n)
for all n and k, so that the same holds for the sequence of matrices A b (n). Now, the (max,plus) matrix product and the sum are increasing functions. This implies that D1 (n) ≤ st D 2 (n).

Experimental results

To assess the quality of our model, and to observe the actual behavior of several distributions, we have conducted an extensive set of simulations. The event graph model presented in Section 3 is simulated using the toolbox ERS [START_REF] Jean-Marie | ERS: a tool set for performance evaluation of discrete event systems[END_REF]. Two tools were used, the first one, scscyc, is able to determine the cycle time in the deterministic case, while the second one, eg_sim, determines the cycle time in both the deterministic and exponential cases. The throughput of this second method relies on the number of simulated events. The whole system is simulated using the Simgrid framework [START_REF] Casanova | SimGrid: a Generic Framework for Large-Scale Distributed Experiments[END_REF]. Since Simgrid aims at being a realistic simulation, any communication cannot use more than 92% of the theoretical bandwidth [START_REF] Velho | Accuracy study and improvement of network simulation in the simgrid framework[END_REF]. Thus, all link bandwidths in the Simgrid platform are divided by 0.92 to achieve the same theoretical bandwidth as in the simulations with ERS. This system does not directly provide the throughput of the system: we define its throughput as the number of processed instances divided by the total completion time. Unless stated otherwise, simulations only use the Overlap model, since it is a fundamental assumption of most theorems. The whole code is available at http://graal.ens-lyon.fr/~mgallet/downloads/SimuStochaGrid.tar.gz.

Examples without any critical resources

In Section 4, we have shown examples of mappings without any critical resource, i.e., whose period is larger than any resource cycle-time, for both Overlap and Strict models. We have conducted extended experiments to assess whether such situations are very common or not. Several sets of applications and platforms were considered, with between 2 and 20 stages and between 7 and 30 processors. All relevant parameters (processor speeds, link bandwidths, number of processors computing the same stage) were randomly chosen uniformly within the ranges indicated in Table 1. Finally, each experiment was run for both models. We compared the inverse of the critical resource cycle-time and the actual throughput of the whole platform. A grand total of 5, 152 different experiments were run.

Number of processed data sets

As said in the previous paragraphs, the throughput is dependent of the number of processed data sets (Simgrid simulations) or of the number of simulated events (eg_sim simulations). Thus, we need to carefully assess this influence.

Figure 9 presents the throughput of a simulated system (7 stages, replicated 1, 3, 4, 5, 6, 7 and 1 times) using different numbers of processed data sets (resp. simulated events) and the theoretical throughput in the constant case. As we can see, the difference between exponential and constant cases is very small, and all measures tend to the same throughput. While the limit is reached in almost all cases as soon as 10,000 tasks (or events) are simulated, the Simgrid simulation with exponential times is a bit less stable but with 50,000 data sets, the difference with theoretical throughput is less than 1%.

Evaluation of the standard deviation in the exponential case

In this paper, we present several methods to compute the expectation of the throughput when communication and computation times are given by exponential random laws. However, we do not have any other information on this distribution. Thus, we used the same system to process 500 sets of 10, 50, 100, 500, 1,000, 5,000 or 10,000 data sets. This system is identical to the one used in Figure 9. Figure 10 displays the minimum, the maximum and the average throughput reached by the system after having processed 10, 50, 100, 500, 1,000, 5,000 or 10,000 data sets. Moreover, the standard deviation is also shown. Both the Simgrid and the eg_sim methods were used to determine the throughput. As can be observed, if the dispersion of Simgrid simulations is larger than the one of eg_sim simulations, it remains very small and the standard deviation is around 2% with 5,000 data sets and around 1% with 10,000 data sets. Thus, as expected from the small difference between constant and exponential cases, the variation is small: there is no need to use a lot of data sets to alleviate the effect of the random law.

Fidelity of the event graph model

Our event graph model clearly shows that the throughput can be determined by considering each stage and each communication between two successive stages as independent from other stages (resp. other communications).

We consider a system based on several instances of a pattern composed of two stages linked by a single costly communication. Figure 11 represents the throughput of systems with different number of stages (using a pattern with 5 senders and 7 receivers, and 10, 000 data sets are used in these simulations). As expected, the throughput does not vary with the number of stages. This is due to the absence of backward dependencies as explained in the event graph model: as soon as a data set is processed and pushed to the next step in the linear chain, it does not influence the previous stages anymore.

Thus, we consider a single communication to assess the difference between the Simgrid simulation and the different theoretical results. Since the complexity of the throughput evaluation stands in communications, we consider a single communication between two negligible computations. Both stages have replication factors comprised between 2 and 9. In another simulation set, we still consider a single communication with negligible computations, but we assume a heterogeneous network. Thus, average communication time through a given link is randomly fixed between 100 and 1000, and all links have different mean communication times. As before, we evaluate the throughput with replication factors comprised between 2 and 9. Figure 13 shows the throughput reached by Simgrid and eg_sim simulations in both constant and exponential cases, and the theoretical throughput of the constant case, computed using the event graph model with scscyc. All throughputs are normalized to the throughput returned by Simgrid in the constant case.

We observe that the throughput of Simgrid and scscyc are identical, and all values are very close to the constant case; the difference is less than 2%. Unlike the homogeneous network case, there is almost no difference between the exponential and the constant case: due to the round-robin distribution, a single link limits all communications, and the global behavior tends to the behavior of one communication through a single link.

Comparison between the exponential and deterministic cases

Figure 14 presents the difference between the expected throughput when using exponential random times (as reached by Simgrid simulations and as returned by Theorem 4) and the throughput returned by Simgrid simulations using constant times. If u is the number of senders and v is the number of receivers in the single involved communication, then the ratio between both cases is equal to 1 ≥ max (u,v) u+v-1 > 1 2 . Again, this figure clearly shows the correlation between the throughput achieved by Simgrid simulations and the predicted throughput. As predicted by Theorem 7, their throughput is comprised between the throughput reached by constant times and the one reached by exponential times. In this experiment, Gauss X means a normal distribution with variance √ X, and Beta X means a beta distribution of shape X. The mean value is the same for all distributions.

Comparison of several random laws

Figure 16 presents the throughput reached by Gamma distributions, which are non-N.B.U.E. random laws. Even if Theorem 8 does not apply here, their throughputs are still between the throughput reached by constant times and the one reached by exponential times, being very close to constant ones.

Running time of simulations

All tools (scscyc, eg_sim and Simgrid simulator) are coded in C and are quite fast: less than one second is sufficient to generate all tasks and run all tools with 100 data sets (for Simgrid simulator) and 100 events (for eg_sim). Event with 100,000 events and tasks, the whole simulation takes around three minutes.

Conclusion

In this paper, we have investigated how to compute the throughput achieved by a given one-to-many mapping of a streaming application onto a target heterogeneous platform. The major novelty is the introduction of I.I.D. and associated variables to model computation and communication times. Our model is based on event graphs and Markov chains, and allows us to compute the expected throughput when variables are constant or follow exponential random laws. In the general case of arbitrary I.I.D. or associated random variables having the N.B.U.E. property, we have established bounds, and the lower and upper bounds are obtained by the deterministic and exponential cases respectively. Both bounds can be computed in polynomial time under the Overlap model with a homogeneous communication network. We have conducted an extensive set of simulations, hence assessing the quality of our model, and observing the behavior of more complex random laws. It turns out that even the throughput obtained with non-N.B.U.E. laws is between the deterministic and exponential throughput.

All the results of this paper are established in the context of a given mapping. In previous work [START_REF] Benoit | Computing the throughput of probabilistic and replicated streaming applications[END_REF], we have shown that computing the best mapping, i.e., the mapping that maximizes the throughput, is an NPcomplete problem, even in the simpler deterministic case with no communication costs. Now that we have new methods to evaluate the throughput of a given mapping in a probabilistic setting, we will devote future work to designing polynomial time heuristics for determining a mapping which achieves a good throughput in such a setting. Thanks to the methodology introduced in this paper, we will be able to compute the throughput of various heuristics, and to compare their efficiency. This would be a first and important step for mapping and scheduling streaming applications on large-scale platforms, whose load and performance parameters are subject to dynamic variations.

Figure 1 :

 1 Figure 1: Example A: Four-stage pipeline, seven-processor computing platform, mapping with replication.

  C exec (p) be the cycle-time of processor P p . If we enforce the Overlap model and constant communication and computation times, then C exec (p) is equal to the maximum of its reception time C in (p), its computation time C comp (p), and its transmission time C out (p): C exec (p) = max {C in (p), C comp (p), C out (p)} . If we enforce the Strict model, then C exec (p) is equal to the sum of the three operations: C exec (p) = C in (p) + C comp (p) + C out (p). Note that in both models, the maximum cycle-time, M ct = max 1≤p≤M C exec (p), is a lower bound for the period.

Figure 3 :

 3 Figure 3: Complete TPN of Example A for the Strict model.

Theorem 2 .Figure 4 :

 24 Figure 4: Example A: Part of the timed Petri net corresponding to communication F 2 .

Figure 5 :

 5 Figure 5: List of all possible states of the Markov chain corresponding to the reduced timed Petri net of Example A.

Theorem 3 .

 3 Let us consider the system (X, Y ) formed by the mapping of an application onto a platform, following the Overlap communication model. Then, the throughput can be computed in time O N exp( max 1≤i≤N (R i )) .

Figure 6 :

 6 Figure 6: Example B, with stages replicated on 5, 21, 27 and 11 processors, and structure of the timed Petri net corresponding to the second communication.

Fired k + 1

 1 times Fired k -1 times Fired k times

Figure 7 :

 7 Figure 7: Valid marking of P j , the reduced timed Petri net of the second communication of Example B.

Figure 8 :

 8 Figure 8: Reachable states from a given position.

Proof.

  Consider the Petri nets modeling both systems. They only differ by the firing times of the transitions. Then for b = 1, 2, let D b k (n) be the time when transition T k ends its n-th firing. The Petri net being an event graph (all places have a single input transition and all places have a single output transition), the variables D b k (n) satisfy a (max,plus) linear equation 1 : D

Figure 10 :

 10 Figure 10: Minimum, maximum and average throughput across 500 simulation runs.

Figure 11 :Figure 12 :

 1112 Figure 11: Variation of the throughput with the number of stages.

Figure 12 Figure 13 :

 1213 Figure 13: Throughput of constant and exponential cases and reached by Simgrid simulations. All throughputs are normalized to the constant throughput.

Figure 14 :

 14 Figure 14: Comparison between constant and exponential laws.

Figure 15 :Figure 16 :

 1516 Figure 15: Comparison of several N.B.U.E. laws.

Figure 15

 15 Figure15presents the throughput reached by several other N.B.U.E. random laws. As predicted by Theorem 7, their throughput is comprised between the throughput reached by constant times and the one reached by exponential times. In this experiment, Gauss X means a normal distribution with variance √ X, and Beta X means a beta distribution of shape X. The mean value is the same for all distributions.Figure16presents the throughput reached by Gamma distributions, which are non-N.B.U.E. random laws. Even if Theorem 8 does not apply here, their throughputs are still between the throughput reached by constant times and the one reached by exponential times, being very close to constant ones.

  and a place from T j k 2i+1 to Figure 2: Complete TPN of Example A for the Overlap model.

	P 1 T 1	P 1 → P 1 F 1	P 2 T 2	F 2 P 2 → P 3	P 4 T 3	F 3 P 4 → P 6	P 7 T 4
	P 1	P 1 → P 2	P 3	P 3 → P 4	P 5	P 5 → P 6	P 7
	P 1	P 1 → P 1	P 2	P 2 → P 5	P 6	P 6 → P 6	P 7
	P 1 T 1 P 1	P 1 → P 1 F 1 P 1 → P 2	P 2 T 2 P 3	F 2 P 2 → P 3 P 3 → P 3	P 4 T 3 P 4	F 3 P 4 → P 6 P 4 → P 6	P T P 7
	P 1 P 1	P 1 → P 2 P 1 → P 1	P 3 P 2	P 3 → P 4 P 2 → P 4	P 5 P 5	P 5 → P 6 P 5 → P 6	P P 7
	P 1 P 1	P 1 → P 1 P 1 → P 2	P 2 P 3	P 2 → P 5 P 3 → P 5	P 6 P 6	P 6 → P 6 P 6 → P 6	P P 7
	P 1	P 1 → P 2	P 3	P 3 → P 3	P 4	P 4 → P 6	P
	P 1	P 1 → P 1	P 2	P 2 → P 4	P 5	P 5 → P 6	P
	P 1	P 1 → P 2	P 3	P 3 → P 5	P 6	P 6 → P 6	P

Table 1 :

 1 Table1shows that the cases without critical resources are very rare, and the difference remains below 9%. In fact no such case was actually found with the Overlap model! Numbers of experiments without critical resource.

	Size (stages, processors) Comp. times Comm. times #exp without critical
		(seconds)	(seconds)	resource / total
	With overlap:
	(10, 20) and (10, 30)	5 . . . 15	5 . . . 15	0 / 220
	(10, 20) and (10, 30)	10 . . . 1000	10 . . . 1000	0 / 220
	(20, 30)	5 . . . 15	5 . . . 15	0 / 68
	(20, 30)	10 . . . 1000	10 . . . 1000	0 / 68
	(2, 7) and (3, 7)	1	5 . . . 10	0 / 1000
	(2, 7) and (3, 7)	1	10 . . . 50	0 / 1000
	Without overlap:
	(10, 20) and (10, 30)	5 . . . 15	5 . . . 15	14 / 220
	(10, 20) and (10, 30)	10 . . . 1000	10 . . . 1000	0 / 220
	(20, 30)	5 . . . 15	5 . . . 15	5 / 68
	(20, 30)	10 . . . 1000	10 . . . 1000	0 / 68
	(2, 7) and (3, 7)	1	5 . . . 10	10 / 1000
	(2, 7) and (3, 7)	1	10 . . . 50	0 / 1000
		20

The product ⊗ is defined as:(V ⊗ M ) k = max i (V i + M ik ).

Anne Benoit and Yves Robert are with This work was supported in part by the ANR StochaGrid and RESCUE projects, and by the Inria ALEAE project.

Finally, the throughput ρ (b) is the limit of n/E[D b (n)] when n goes to infinity, so that ρ (1) ≥ ρ (2) , which concludes the proof. Theorem 6. Let us consider two systems with I.I.D. communication and processing times (X (1) , Y (1) ) and (X (2) , Y (2) ). If we have for all n, ∀ 1 ≤ p ≤ M , X

p,q (n), then ρ (1) ≥ ρ (2) . Proof. The proof is similar to the previous one, using the fact that D b k (n) is also a convex function (a composition of maximum and sums) of the communication and processing times.

Theorem 7. Let us consider any I.I.D. system (X (1) , Y (1) ), and two new systems (X (2) , Y (2) ) and (X (3) , Y (3) ) such that:

p (n) has an exponential distribution, and E[X

p,q (n) has an exponential distribution, and

p (n) is deterministic and for all integers n, X

p,q (n) is deterministic and for all n, Y

Then we have ρ (3) ≥ ρ (1) and if

p,q (n) are N.B.U.E., then ρ (1) ≥ ρ (2) . Proof. A direct consequence of the N.B.U.E. assumption is that if V is N.B.U.E. and W is exponential with the same mean as V , then V ≤ icx W (see [START_REF] Gaujal | Introduction to Scheduling, chapter Comparisons of stochastic taskresource systems[END_REF], for example). It is also direct to show that if U is deterministic and U = E[V ], then U ≤ icx V . Therefore, a direct application of Theorem 6 shows that ρ (3) ≥ ρ (1) ≥ ρ (2) .

In particular, Theorem 7 implies that in the Overlap case with a homogeneous communication network, as soon as communication times and processing times are N.B.U.E., then the throughput ρ can be bounded explicitly. It is comprised between the throughput of the system in which all random processing times are replaced by their mean values where the inner throughput of processing components are the same as in the exponential case while the throughput of communication components is replaced by uiviλi max(ui,vi) = min(u i , v i )λ i ), and the throughput of the system in which all random processing times are replaced by exponential variables with the same mean value.

The associated case

In this section, we study the associated case. We recall that in this case, we assume that the speed of the processes and the bandwith of the communication links are I.I.D. sequences while the sizes of one data-set and the sizes of its transfer files, on all stages, are associated.

The first following lemma establishes that under such assumptions, the communication and processing times of one data-set are also associated. Lemma 1. Consider any two variables U (n) and V (n) belonging to the set of the processing and communication times of the n-th data set, namely {X p (n), Y p,q (n)} 1≤p,q≤M . Then they are associated.

Proof. The proof is carried for the computation times only since the proof with communication times as well is exactly the same. Since n is fixed throughout, it can be skipped in the notations. Let us set X p = δ i /s p and X q = δ j /s q . Therefore, using the densities f p and f q of the variables s p and s q respectively, Pr (X p < t, X q < u) = Pr (δ i /s p < t, δ j /s q < u) ,

= x y

Pr (δ i < tx, δ j < uy|s p = x, s q = y) f p (x)f q (y)dxdy ≥ Pr (δ i < ts p ) Pr (δ j < ts q ) , by association of δ i and δ j and by independence of s p and s q w.r.t. the other variables.

Let us denote by {X p (n) * } 1≤p≤M,n∈N (resp. {Y q,p (n) * } 1≤p,q≤M,n∈N ) a set of processing times (resp. communication times) with the same distribution as {X p (n)} 1≤p≤M,n∈N (resp. {Y q,p (n)} 1≤p,q≤M,n∈N ), but mutually independent.

Lemma 2. Under the foregoing notations,

Proof. The proof comes from a direct generalisation of Lemma 1 from two variables to all of them. Theorem 8. Let us consider a system (X (1) , Y (1) ), such that {X

p,q (n)} 1≤p,q≤M are associated. Let us also consider two new systems (X (2) , Y (2) ) and (X (3) , Y (3) ) such that:

p,q (n)} 1≤p,q≤M are I.I.D. with the same distribution as in (X (1) , Y (1) );

p,q (n) and X

(3) p (n) are deterministic (i.e., they do not depend on n) and satisfy X

(3)

Then we have ρ (3) ≥ ρ (1) ≥ ρ (2) .

Proof. The proof follows the same ideas as the proof of Theorem 5. The first inequality comes from Jensen inequality that also holds for associated variables. Since the throughput is the limit of an increasing convex function of the variables Y (1) p,q (n) and X

(1) p (n), then its expectation satisfies the first inequality. As for the second inequality, it uses directly Lemma 10.3 from [START_REF] Gaujal | Introduction to Scheduling, chapter Comparisons of stochastic taskresource systems[END_REF], to get a comparison for the firing times of all the transitions in the Petri net model for both systems:

The throughput being the limit of n/E[D b (n)] when n goes to infinity, one gets directly ρ (1) ≥ ρ (2) .

When the task sizes (resp. bandwidth requirements) have the N.B.U.E. property while the processor speeds and the communication speeds can be arbitrary (but uniformly bounded away from 0, by, say, b), with a density f sp , then the processing times as well as the communication times become DFR (Decreasing Failure Rate), which implies the N.B.U.E. property. Indeed, by definition of N.B.U.E. variables, s Pr (δ i > t + s|δ i > t) ds ≤ s Pr (δ i > s) ds.

Let us consider the processing time of one task (the case of communication times is similar). Since n is fixed, one can remove the dependence on n for simplicity: X p = δ i /s p and let us compute

Pr (X p > s) ds.

By combining Theorem 7 and Theorem 8, one gets the following result, which is the final result of this section: if all processors are independent of each other and the data-set sizes are associated and have the N.B.U.E. property, then, the throughput of the system is bounded from below by a version of the system where all the processing times and the communication times are replaced by I.I.D. exponential variables with the same means and is bounded from above by another version of the system where all processing times and communication times are replaced by constants (equal to the mean). This justifies the focus on the constant and exponential cases, as extreme cases.