• Non ci sono risultati.

A Three-tier Active Replication Protocol for Large Scale Distributed Systems

N/A
N/A
Protected

Academic year: 2021

Condividi "A Three-tier Active Replication Protocol for Large Scale Distributed Systems"

Copied!
9
0
0

Testo completo

(1)

PAPER Special Issue on Dependable Computing

A Three-tier Active Replication Protocol for Large Scale Distributed Systems

Carlo MARCHETTI, Sara TUCCI PIERGIOVANNI, and Roberto BALDONI,Nonmembers

SUMMARY The deployment of server replicas of a service across an asynchronous distributed system (e.g., Internet) is a real practical challenge. This target cannot be indeed achieved by classical software replication techniques (e.g., passive and active replication) as these techniques usually rely on group communi- cation toolkits that require server replicas to run over a partially synchronous distributed system to solve the underlying agreement problem. This paper proposes a three-tier architecture for soft- ware replication that encapsulates the need of partial synchrony in a specific software component of a mid-tier to free replicas and clients from the need of underlying partial synchrony as- sumptions. Then we propose how to specialize the mid-tier in order to manage active replication of server replicas.

key words: high availability, fault tolerance, software replica- tion, three-tier architectures

1. Introduction

Active [24] and passive [9] software replication are well- known approaches used to keep strongly consistent the internal states of a set of replicas implementing a repli- cated services. Informally, strong consistency means that all replicas have to execute the same sequence of updates before failing. This can be obtained by com- bining (i) request ordering and (ii) atomic updates.

In order to maintain consistency despite replica failures, these approaches commonly let replicas em- ploy group communication primitives and services such as total order multicast, view synchronous multicast, group membership, etc [18]. For example, in active replication clients and deterministic replicas employ a total order multicast primitive to ensure consistency.

However such primitives to be implementable require replicas to be deployed in a partially synchronous dis- tributed system, as they rely on distributed agreement protocols.∗∗ A partially synchronous system is an asyn- chronous distributed system with some additional tim- ing assumptions on message transfer delays, processors’

Manuscript received March 31, 2003.

Manuscript revised July 7, 2003.

The authors are with the Faculty of Engineering of the University of Rome “La Sapienza” — Dipartimento di Infor- matica e Sistemistica, Via Salaria 113, 00198 Rome, Italy.

A preliminary version of this paper appears in the Pro- ceedings of the 2002 Pacific Rim Conference on Dependable Computing (PRDC 2002). This work is supported by a grant from the Italian Ministry of University and Research (MIUR) in the context of the MAIS project (Multiple- channel Adaptive Information Systems) and by a grant from UE in the context of the EUPubli.com project (IST project

#IST-2001-35217).

relative speeds etc. [14], [15]. As examples, the timed asynchronous model [16] and the asynchronous system model with unreliable failure detectors [11] can both be seen as partially synchronous distributed systems models.∗∗∗ Synchrony assumptions can be guaranteed (most of the time) only on small size distributed sys- tems deployed over either a LAN or a CAN (Controlled Area Network). This makes then difficult the deploy- ment of server replicas over large-scale (asynchronous) distributed systems, as service availability could be re- duced by asynchrony even in the absence of replica fail- ures [8].

In this paper we first introduce two architectural properties, namely Client and Server Asynchrony that, if satisfied by a replication technique, allow to deploy clients and server replicas within an asynchronous dis- tributed system. Then we propose, in the context of active replication, a three-tier architecture in which clients (the client-tier) interact with a middle tier (the mid-tier) that forwards client requests to server replicas (the end-tier). The mid-tier is responsible for maintain- ing consistency. To achieve this, the mid-tier embeds two basic components, namely the sequencer and the active replication handler. The first component is con- cerned with defining a total order on client requests, while the second masters the client/server interaction enforcing atomicity on the end-tier. This clear sepa- ration of concerns allows to isolate the need of partial synchrony within the sequencer component and to free clients and server replicas from satisfying any timing as- sumptions to ensure the liveness of the service. There- fore this architecture satisfies both Client and Server Asynchrony.

Finally we propose an implementation of the ac- tive replication handler, working in an asynchronous distributed system. The active replication handler re- ceives a request from a client and gets a sequence num-

∗∗It is well known that inasynchronous distributed sys- tems, due to FLP impossibility result [17], it is not possible to implement these primitives while ensuring both safety conditions and the deterministic termination of the agree- ment protocols that they embed. In other words, these primitives can block the system due to asynchrony and de- spite replication [8].

∗∗∗Even the weakest unreliable failure detector allowing to solve Consensus is not implementable in asynchronous distributed systems [11].

(2)

ber from the sequencer. Then it forwards the request along with its sequence number to all end-tier server replicas. The latter execute requests according to their sequence numbers, computing the result. The active replication handler waits for the first result from the end-tier and sends it back to client. Waiting only for the first result from the end-tier allows to cope with replica asynchrony, as during the computation all repli- cas but one can slow down without reducing the service availability. As mid-tier components are involved in ev- ery client-server interaction, they are replicated so that if an entity of a mid-tier component that was carrying out a client/server interaction crashes, another entity can conclude the job.

The remainder of this paper is organized as fol- lows: Section 2 introduces basic notions on software replication and the Client and Server Asynchrony prop- erties, illustrates active replication and defines correct- ness conditions; Section 3 introduces a three-tier archi- tecture that allows to satisfy these architectural prop- erties; Section 4 details a provable correct mid-tier pro- tocol whose correctness, due to lack of space, is shown in [4]. Finally, Sect. 5 concludes the paper.

2. Software Replication

The basic idea underlying software replication is to replicate a server on different sites of a communica- tion network so that the service’s clients can access dif- ferent replicas in order to increase the probability of obtaining a reply to their requests. When the server is stateless (i.e., a result to a request depends only on the same request message content) improving service availability boils down to address the issue of letting the client invoke all of the replicas before returning a “no-response” exception to the application. On the other hand, when dealing with servers maintaining an internal state (stateful servers), it rises the problem of maintaining consistency among the replicas’ state to let clients interact with the stateful replicated server as if it was a singleton, highly available, object.

In this section we first introduce two architectural properties for software replication and then we deal with the active replication technique, providing a set of correctness conditions that ensure liveness and for- malize replica consistency.

2.1 Two Desirable Properties for Software Replication To match the nature of current distributed systems (e.g., client/server architectures, n-tiers architectures), apart from enforcing consistency, clients and server replicas implementing a replication technique should also satisfy the following architectural properties:

Client (resp. Server) Asynchrony. Clients (resp.

server replicas) run in a distributed system without any timing assumption i.e., an asynchronous distributed

system.

These properties allow clients and server replicas of a stateful service to be deployed across a system like the Internet without reducing the availability of the service.

Replication techniques are usually implemented adopting two-tier architectures in which replicas run agreement protocols to enforce consistency [18]. As a consequence, two-tier architectures for software repli- cation do not allow to satisfy Server Asynchrony. Fur- thermore, some protocols (e.g., [22]) involve clients in taking decisions about consistency, which hinder satis- fying also Client Asynchrony (see also Sect. 2.4).

2.2 Active Replication

Active replication [20], [24] can be specified taking into account two types of participants, namely clients and a set of deterministic server replicas. Each client request is received by each available replica that independently executes the request and sends back the reply to the client. We abstract the computation performed by a generic replica upon the arrival of a request req using the method compute (req), which returns a result (res).

Assuming a crash failure model for processes, the client waits only for the first reply and filters out others.

2.3 Correctness of Active Replication

To enforce strong replica consistency, a correct imple- mentation of an actively replicated deterministic service should satisfy the following properties [13]:

Termination. If a client issues a request, unless it crashes, it eventually receives a reply.

Uniform Agreed Order. If a replica processes a re- quest req as the i-th request, then each replica that pro- cesses the i-th request, processes req as the i-th request.

Update Integrity.For each request req, every replica executes compute (req) at most once, and only if a client has issued req.

Response Integrity.If a client issues a request req and delivers a result res, then res has been computed by some replica which executed compute (req).

These properties guarantee consistency and live- ness of an actively replicated service. In particular Uniform Agreed Order implies that if replicas execute requests, such requests are executed in a total order. The Termination property ensures live client/server in- teractions. The Update Integrity property ensures that replicas compute replies only upon receiving each client request for the first time. Finally, the Response In- tegrity property guarantees that the replies returned to clients are actually generated by the service.

As server replicas are assumed to be deterministic, if they executeall requests in the same order then they will produce the same result for each request. This satisfieslin- earizability [19].

(3)

2.4 Implementing Active Replication

To enforce consistency of a service replicated using ac- tive replication, a total order multicast primitive is com- monly employed. Intuitively, such a primitive ensures that all replicas deliver all messages in the same order.

Therefore, in the case of active replication, satisfying Client and Server Asynchrony, turns out in checking if the implementation of total order multicast primitives satisfies these properties.

A large number of total order multicast and broad- cast algorithms has been proposed in the last twenty years (a comprehensive survey can be found in [13]).

Due to the objectives of our protocol, in this work we only consider algorithms designed for “asynchronous distributed systems”. A common characteristics of these algorithms is that they always ensure safety, while liveness is guaranteed only if the platform provides an amount of synchrony sufficient to take agreed decisions concerning the total order of messages in a safe man- ner. A classifications of these algorithms considers the entities involved in building the total order of messages i.e., the processes involved in agreement protocols that can block upon a lack of synchronicity. In particular, message ordering can be built by:

1.senders (e.g., [12], [20]): in this case senders agree on the order to submit requests to the replicas;

2.destinations (e.g., [6], [11]): in this case replicas agree on the order to process client requests;

3.external processes (e.g., [1], [7]): ordering may in- volve processes that are neither senders nor destina- tions.

It is easy to see that algorithms belonging to the first and second classes do not allow to meet Client and Server Asynchrony, respectively. In fact, to guaran- tee live agreement on a total order of messages, algo- rithms of the first class require synchrony among clients while algorithms of the second require synchrony among server replicas.

Concerning the third class, algorithms falling in this category are commonly designed assuming that the

“external process” is elected among clients or servers that run in a partially synchronous distributed system.

As example, in [7] processes elect a sequencer process that is in charge of defining total order. Election is based on a group membership service whose liveness can not be deterministically guaranteed when the un- derlying system is asynchronous [10]. In contrast, the algorithm proposed in [1] is not based on a membership service. The objective of this algorithm is to implement a transport-level total order multicast primitive capa- ble of scaling to several processes possibly exchanging large messages. To meet such requirements, the proto- col adopts an architecture similar to the one proposed in this paper i.e., with a centralized fault-tolerant se- quencer (called master in [1]) that distributes to pro-

cesses tokens that suffice (i) to label messages with se- quence numbers, and (ii) to let processes deliver mes- sages in a total order. This protocol differs from ours in that it aims to implement a transport-level proto- col i.e., it does not provide support for client/server interactions. Therefore, under a deployment perspec- tive, it could be regarded as a good candidate to im- plement our mid-tier replicated objects (i.e., the ARH replicas and the sequencer, see also Sect. 3). Unfor- tunately, the synchrony requirements of the protocol are too stringent for our purposes: it can indeed tol- erate only sequences of at most t + 1 failures, being t a protocol’s parameter (which is difficult to estimate for large networks, according to [1]), and being a fail- ure the crash of a process as well as the performance failure of a channel between a process and the master.

On the contrary, we let our mid-tier processes interact with the sequencer through asynchronous links, and we isolate all synchrony requirements within the sequencer (see Sect. 3.1).

From the above discussion it follows that, to the best of our knowledge, no existing total order algorithm allows to implement active replication while enforcing both Client and Server Asynchrony. In the remainder of this paper we describe an architecture and a replica- tion protocol that satisfy both these properties.

3. Three-Tier (3T) Active Replication

The basic idea underlying 3T active replication is to in- troduce a middle tier (mid-tier) between asynchronous clients (client-tier) and asynchronous server replicas (end-tier). The mid-tier is in charge of accepting client requests, enforcing a total order on them and forward- ing requests to the end-tier that returns a result to the mid-tier, which finally provides clients with results.

Furthermore, inside the mid-tier, we separate the prob- lem of agreeing on a total order of client requests (to satisfy the Uniform Agreed Order property) from the problem of letting all available server replicas execute all (ordered) requests. The first problem is addressed by the sequencer mid-tier component and the second by the active replication handler mid-tier component. To ensure the liveness (i.e., the Termination property) of client/server interactions in the presence of failures, the mid-tier must be fault tolerant i.e., both the sequencer and the active replication handler components must be replicated.

In the remainder of this section we introduce the system model and present our 3T architecture for active replication.

A channel’s performance failure occurs each time the channel delays some messages for an amount of time ex- ceeding a predefined maximum channel latency value.

(4)

3.1 System Model

The distributed system is composed by a set of pro- cesses that communicate by message passing.

Processes can fail by crashing. After a crash event a process stops executing any action. A process is cor- rect if it never fails, otherwise it is faulty.

Processes communicate through eventual quasi- reliable channels that are modelled by the following properties:

Channel Validity. If a process receives a message m, then m has been sent by some process.

Channel No Duplication. Messages are delivered to processes at most once.

Channel Termination. If a correct process sends a message m to a correct process, the latter eventually delivers m.

Due to the absence of timing assumptions, previ- ous properties actually state that processes run over an asynchronous distributed system.

Processes are of four disjoint types: a set {c1, . . . , cl} of client processes (client-tier), a set {h1, . . . , hn} of active replication handler (ARH) repli- cas (implementing the mid-tier ARH component), a set {r1, . . . , rm} of deterministic server replicas (end-tier) and a single correct process, belonging to the mid-tier, namely the Seq component. This component actually abstracts the fault-tolerant implementation of the se- quencer service specified in Sect. 4.3 and detailed in [4].

Finally, in order to ensure the termination of our protocol, we need to assume:

ARH Correctness. A majority of ARH replicas (i.e.,

n+12 ) is correct.

Replica Correctness. There is at least one correct end-tier replica.

3.2 The 3T Architecture

Figure 1 shows the components of the three-tier archi- tecture for active replication. In the following we intro- duce the interfaces and a short functional description of each component.

• Retransmission/Redirection message han- dler (RR). RR is a message handler embedded by each client to cope with ARH replica failures and with the asynchrony of communication chan- nels (i.e., to enforce the Termination property).

RR exposes to client applications the IssueReq (request) method that accepts a request as input parameter. This method is invoked by client ap- plications to submit requests, and blocks the client process until the result to the request has been re- ceived. Operationally, RR first labels each out- going client request with a unique request identi- fier (req id) and then sends the request to distinct

Fig. 1 A three-tier architecture for active replication.

ARH replicas until a reply is eventually received.

Only to mitigate network traffic, RR sets a local timeout to transmit the request to a different ARH replica if a reply is not received within the time- out, and it continues to issue the request towards distinct ARH replicas until a reply is eventually received.

• Sequencer (Seq). The sequencer is a fault- tolerant service that returns a unique and consec- utive sequence number for each distinct request.

This is the basic building block to get the Uniform Agreed Order property. In particular, Seq exposes two methods of the Seq interface:

– GetSeq (req id) that returns the sequence number associated to req id;

– GetReqID (seq) that returns either the re- quest identifier associated to seq (if any) or null (otherwise).

• Active Replication Handler (ARH). ARH is the core of the replication logic: by exploiting Seq, it orders all incoming client requests and en- sures that at least one copy of each client request is eventually delivered to every non-crashed end- tier replica. Once replicas return results, ARH returns them to clients. To achieve Termination despite failures, ARH is implemented by n repli- cas maintaining an internal state composed of a set of req id, request pairs.

In order to face ARH replica failures, ARH repli- cas keep their internal states consistent using the following primitives (described in Sect. 4.4):

– UpdateMaj (req id, request) that updates a majority of ARH replica states;

– FetchMaj() that reads a majority of replica states;

• Filtering and Ordering message handler

(5)

(FO). FO is a message handler placed in front of each end-tier replica (i) to ensure ordered ex- ecution of client requests according to the number assigned by Seq to each request and (ii) to avoid re- peated executions of the same client request (possi- bly sent multiple times by ARH). Previous features allow to enforce Uniform Agreed Order and Update Integrity on the end-tier replicas.

RR and FO message handlers are co-located with the client and with each end-tier server replica, respec- tively. Hence we assume that the crash of a client (resp.

server replica) implies the crash of its RR (resp. FO) and vice-versa.

4. The Mid-Tier Protocol

In this section we present the protocol run by the mid- tier to enforce the correctness properties introduced in Sect. 2.3. To this aim, we first describe an introductory example showing how a client/server interaction is car- ried out in the absence of failures. Then we detail the protocols of the components depicted in Fig. 1.

4.1 Introductory Example

Figure 2 shows a failure-free run of the protocol.

In this scenario, a client c1 invokes RR method Is- sueReq (request1) to issue request request1. This method first assigns a unique request identifier req id1

to request1 and then it sends the req id1, request1 pair to an ARH replica (e.g., h1). Upon receiving

Fig. 2 A failure-free run.

thereq id1, request1 pair, h1 first updates a major- ity of ARH replicas ({h1, h3}) invoking UpdateMaj (req id1, request1). Then h1 invokes the sequencer GetSeq (req id1) method to assign a unique sequence number (1 in the example) to the request. Hence h1 sends the pair 1, request1 to all end-tier repli- cas and starts waiting for the first result from an end- tier replica. The FO component of each non-crashing end-tier replica checks if the incoming request is the expected one i.e., if the request sequence number is 1. In the example, the FO components of r1 and r2

verify this condition, and invoke compute(request1) on its replica that produces a result result1. Then FO sends to h1 the pair 1, result1. Upon deliver- ing the first pair, h1 sends req id1, result1 back to c1. h1 discards following results produced for request1

(not shown for simplicity in Fig. 2). Then h1 serves request2 sent by c2: it updates a majority of ARH replicas ({h1, h2}), gets the sequence number 2 from the sequencer, sends2, request2 to all end-tier repli- cas and waits for the first reply from end-tier. Note that in this scenario r3 receives request 2, request2 before receiving1, request1. However, upon receiving

1, request1, the r3 FO component executes both the requests in the correct order returning to h1 both the results. This ensures that the state of r3 evolves con-

As shown in Sect. 4.4, updating a majority of ARH replicas upon the receipt of each client request allows any ARH replica to retrieve any client request despite ARH replicas and client failures, which is needed to enforce strong consistency of end-tier server replicas. Interested readers can refer to [4] for further details.

(6)

sistently with respect to the state of r1 and r2. When h1receives the first2, result2 pair, it sends the result back to the client.

4.2 Retransmission/Redirection

The RR pseudo-code is shown in Fig. 3. RR maintains a state variable #cl seq increased for each new client request to form a unique request identifier along with the client identifier cl id. To issue a request, a client invokes the IssueReq() method (line 4). In detail, the IssueReq() method first assigns to the ongoing request the unique request identifier req id = cl id, #cl seq

(line 8), and then enters a loop (line 9). Within the loop, the client (i) sends the request to an ARH replica (line 10) and (ii) sets a local timeout (line 11). Then, a result is returned if the client process receives a re- sult within the timeout period for the request (line 14).

Otherwise another ARH replica is selected (line 15) and the request is sent again towards such a replica (line 10).

4.3 Sequencer

The Seq component is a fault-tolerant implementation of a sequencer service. A sequencer service dynami- cally assigns unique and consecutive sequence numbers to unique request identifiers (defining a total order) and allows to query its internal state to inspect the total or- der. In particular, it receives assignment requests from other processes and returns an positive integer sequence number denoted #seq for each distinct request. A pro- cess issues an assignment request invoking the GetSeq (req id) method where req id is an unique request iden- tifier. To cope with assignment request retransmissions, the sequencer service maintains a state A composed by a set of assignments{a1, a2. . . ak−1, ak} where each as- signment a is a pair req id, #seq, in which a.req id is a request identifier and a.#seq is the sequence number returned by the sequencer service.

Formally, a correct sequencer service implementa- tion must satisfy the following properties:

Fig. 3 Pseudo-code of RR message handler (code executed by clients).

Assignment Validity. If a ∈ A then (i) there exists a process p that issued an assignment request GetSeq (req id) and (ii) req id = a.req id.

Assignment Response Validity. If a process p de- livers a reply #seq, then (i) p issued an assignment re- quest GetSeq (req id), and (ii) ∃a = req id, #seq ∈ A;

Bijection. ∀ai, aj ∈ A : ai.#seq = aj.#seq ⇔ ai.req id = aj.req id

Consecutiveness. ∀ai ∈ A : (ai.#seq ≥ 1) ∧ (ai.#seq > 1 ⇒ ∃aj: aj.#seq = ai.#seq − 1)

Sequencer Termination. If a process p issues a re- quest then, unless p crashes, it eventually delivers a reply.

A fault-tolerant sequencer implementation that satisfies the previous specification is described in [5].

Furthermore, a sequencer service receives from processes query requests through the GetReqID (#seq) method, which either returns req id if exists ai =req id, #seq ∈ A or null otherwise. We thus in- troduce the following property to capture the sequencer behavior upon the receipt of a query request:

Query Response Integrity. If a process p invokes GetReqID (#seq) then: if ∃req id, #seq ∈ A when the sequencer receives a query request then GetReqID (#seq) returns req id; else it returns null.

Interested readers can find in [4] a modified version of the original sequencer implementation that satisfies the complete specification introduced above.

Implementing a fault-tolerant replicated sequencer mandates the solution of agreement problems. As a consequence, sequencer replicas have to run within a partially synchronous distributed system, in order not to reduce service availability. Let us note that the sequencer service can be implemented by a set of server replicas exploiting group communication primi- tives (e.g., total order multicast) [21].

In [21] we show that the sequencer problem is equivalent to theUniform Consensus problem [11].

(7)

4.4 Active Replication Handler

The basic idea underlying the ARH protocol is to let each ARH replica enforce end-tier consistency indepen- dently from other ARH replicas i.e., each ARH replica forwards to the end-tier each client request at most once (actually, exactly once if the ARH replica is correct).

Figure 4 illustrates the pseudo-code of an ARH replica, based on the following data structures and com- munication primitives.

Data Structures

• the LocalState variable is a set of req id, request

pairs. LocalState is used by ARH replica to hold information on client requests despite client and ARH replica failures. It is read and written by the UpdateMaj and FetchMaj communication primitives (see below).

• the LastServedReq variable stores the sequence number associated by Seq to the last client request that the ARH replica has served;

• the #seq variable stores the sequence number as- sociated by Seq to a client request that ARH has received by a client.

Communication Primitives.

• UpdateMaj() accepts as input parameter a

req id, request pair p and multicasts p to all ARH replicas{h1, . . . , hn}. An ARH replica upon receiving such an update message, inserts p in its LocalState variable and then acknowledges the message receipt. The UpdateMaj() method re- turns the control to the invoker as soon as it has receivedn+12  acknowledgement messages;

• FetchMaj() does not take input parameters and returns the union of the set of req id, request

pairs contained in the LocalState variables of a strict majority of ARH replicas. Upon invocation, FetchMaj() multicasts a state request message to all ARH replicas{h1, . . . , hn}. An ARH replica, upon receiving a state request message responds with a state message containing the current value

Fig. 4 Pseudo-code of an ARH replicahi.

of LocalState. The FetchMaj() method waits until it has received n+12  state messages, com- putes the union and returns the latter to the in- voker;

Due to space limitations, an implementation of these primitives is given [4].

ARH Protocol. Each ARH replica runs the proto- col of Fig. 4. In particular, upon receiving a message

req id, request from a client c (line 4), an ARH replica hi(i) updates a majority of ARH replicas exploiting the UpdateMaj() primitive (line 5) and (ii) invokes the sequencer to assign a sequence number (stored in the

#seq variable) to the request (line 6). Hence if #seq >

LastServedReq+1 then some other ARH replica served other client requests with sequence number comprised in the interval [LastServedReq + 1, #seq − 1]. In this case, to preserve the protocol termination, hi sends such requests again to the end-tier. To this aim:

• it invokes the FetchMaj() primitive (line 8) to retrieve the union of a majority of local states;††

• it sends to server replicas all the requests with sequence number comprised [LastServedReq + 1, #seq − 1], along with their sequence numbers that are retrieved from the sequencer (lines 9-12);

Finally hisends to server replicas the current client request along with its sequence number (line 13), up- dates the LastServedReq variable (line 14) and then waits for the first result from a server replica (line 15).

Upon the receipt of the first result, hiforwards the re- sult to the client (line 16).

4.5 Filtering and Ordering

Figure 5 illustrates the FO message-handler pseudo-

In this way other ARH replicas are able to retrieveall client requests and to forward them to the end-tier. This is done to letall request messages reach the end-tier despite client and ARH replica failures.

††As each client request is stored in a majority of ARH local states (line 5), the execution of FetchMaj() at line 8 stores intoLocalState all the client requests served by some ARH replicas.

(8)

Fig. 5 Pseudo-code of FO message handler (code executed by replicas).

code. FO has an internal state composed (i) by the ExpectedSeq variable, storing the sequence number as- sociated with the next request to execute, and (ii) by the Executed array, storing in the j-th position the re- sult of the request with sequence number j.

For each receipt of a “TORequest” message from an ARH replica hi (line 3), FO waits until the re- quest sequence number seq is lower than or equals to ExpectedSeq (line 4). When such condition holds FO atomically executes statements (5–8). In particular, when a request has seq = ExpectedSeq (line 5), FO computes the result, stores the result in Executed and increases the ExpectedSeq variable. Finally, FO sends a “TOResult” message, containing the request sequence number and the request’s result, to hi.

4.6 Discussion

As neither server replicas nor clients run agreement pro- tocols, they can be deployed within an asynchronous distributed system without reducing the service avail- ability if some clients/replicas/channels abruptly slow down. Therefore, the proposed 3T architecture satis- fies both Client and Server Asynchrony properties, en- abling the deployment of clients and server over wide area networks. The need for partial synchrony to en- force strong consistency is confined as much as pos- sible in the proposed architecture. As shown, syn- chrony is actually needed only for implementing the fault-tolerant sequencer service. Another option (pur- sued in [21]) is to merge both ARH and sequencer func- tionality in a single mid-tier component. This approach eliminates the hop required by ARH replicas to invoke Seq. However, the need of partial synchrony assump- tion embraces the whole mid-tier. Let us finally note that the protocol has been presented omitting several possible optimization detailed in [4], [21].

5. Conclusions

Server replicas of a stateless service can be deployed across an asynchronous distributed system without re- ducing the service availability as long as a replica is available and connected to clients. This does not hold if replicas implement a stateful service following clas- sical replication techniques as they need high coverage of timing assumptions (i.e., synchrony) on the system

connecting replicas in order to guarantee the liveness of the sophisticated and costly agreement protocols used to enforce strong consistency. This actually limits the dissemination of replicated stateful services over asyn- chronous, large-scale systems as the Internet.

In this paper we introduced the Client and Server Asynchrony properties aiming to capture the possi- bility of deploying replicas of a service across asyn- chronous systems. Then we presented a 3T architecture for handling active software replication. This architec- ture meets Client and Server Asynchrony by confin- ing the need of partial synchrony assumptions within a sequencer service detached from clients and server replicas. Finally, we presented an implementation of a provable correct active replication protocol whose cor- rectness, due to lack of space, is shown in [4].

An optimized version of the protocol proposed in this paper has been implemented in the Interoperable Replication Logic (IRL) infrastructure, which adopts a three-tier architecture to enhance CORBA objects with high availability and fault tolerance through soft- ware replication in compliance to the Fault-Tolerant CORBA standard [23]. IRL performance shows that a three-tier approach to software replication is feasible.

Interested readers can refer to [2], [3], [21] for further details.

Acknowledgements

The authors would like to thank the anonymous re- viewers for their valuable comments, which allowed us to significantly enhance the quality of the presentation.

We are also grateful to Xavier D´efago, Roy Friedman, Ricardo Jim´enez-Peris, Dahlia Malkhi, Michel Raynal, Lu´is Rodrigues, and Andr´e Schiper for the positive im- pact on our work of the interesting discussion and com- ments.

References

[1] S. Armstrong, A. Freier, and K. Marzullo, Multicast trans- port protocol, RFC 1301, IETF, Feb. 1992.

[2] R. Baldoni and C. Marchetti, “Three-tier replication for FT-CORBA infrastructures,” Softw. Pract. Exp. Wiley In- terScience, vol.33, no.8, pp.767–797, 2003.

[3] R. Baldoni, C. Marchetti, and A. Termini, “Active software replication through a three-tier approach,” Proc. 22th IEEE International Symposium on Reliable Distributed Systems (SRDS02), pp.109–118, Osaka, Japan, Oct. 2002.

[4] R. Baldoni, C. Marchetti, and S. Tucci-Piergiovanni, “Ac- tive replication in asynchronous three-tier distributed sys- tem,” Technical Report 05-02, Dipartimento di Informat- ica e Sistemistica, Universit`a di Roma “La Sapienza,”

http://www.dis.uniroma1.it/∼irl, Feb. 2002.

[5] R. Baldoni, C. Marchetti, and S. Tucci-Piergiovanni, “A fault-tolerant sequencer for timed asynchronous systems,”

Proc. 8th Euro-Par Conference, pp.578–588, Paderborn, Germany, Aug. 2002.

[6] K. Birman and T. Joseph, “Reliable communication in the presence of failures,” ACM Trans. Comput. Syst., vol.5,

(9)

no.1, pp.47–76, Feb. 1987.

[7] K. Birman, A. Schiper, and P. Stephenson, “Lightweight causal and atomic group multicast,” ACM Trans. Comput.

Syst., vol.9, no.3, pp.272–314, Aug. 1991.

[8] K.P. Birman, “A review of experiences with reliable multi- cast,” Softw. Pract. Exp., vol.29, no.9, pp.741–774, 1999.

[9] N. Budhiraja, F.B. Schneider, S. Toueg, and K. Marzullo,

“The primary-backup approach,” in Distributed Systems, ed. S. Mullender, chapter 8, pp.199–216, Addison Wesley, 1993.

[10] T. Chandra, V. Hadzilacos, S. Toueg, and B. Charron-Bost,

“On the impossibility of group membership,” Proc. 15th ACM Symposium of Principles of Distributed Computing, pp. 322–330, 1996.

[11] T. Chandra and S. Toueg, “Unreliable failure detectors for reliable distributed systems,” J. ACM, vol.43, no.2, pp.225–

267, March 1996.

[12] F. Cristian, “Asynchronous atomic broadcast,” IBM Tech.

Discl. Bull., vol.33, no.9, pp.115–116, Feb. 1991, also, 1st IEEE Workshop on Management of Replicated Data, Hous- ton, TX, Nov. 1990.

[13] X. D´efago, “Agreement-Related Problems: From Semi- Passive Replication to Totally Ordered Broadcast,” Ph.D.

Thesis, ´Ecole Polytechnique F´ed´erale de Lausanne, Switzer- land, 2000, Ph.D. Thesis no.2229.

[14] D. Dolev, C. Dwork, and L. Stockmeyer, “On the minimal synchronism needed for distributed consensus,” J. ACM, vol.34, no.1, pp.77–97, Jan. 1987.

[15] C. Dwork, N.A. Lynch, and L. Stockmeyer, “Consensus in the presence of partial synchrony,” J. ACM, vol.35, no.2, pp.288–323, April 1988.

[16] C. Fetzer and F. Cristian, “The timed asynchronous dis- tributed system model,” IEEE Trans. Parallel Distrib.

Syst., vol.10, no.6, pp.642–657, 1999.

[17] M. Fischer, N. Lynch, and M. Paterson, “Impossibility of distributed consensus with one faulty process,” J. ACM, vol.32, no.2, pp.374–382, April 1985.

[18] R. Guerraoui and A. Schiper, “Software-based replication for fault tolerance,” Computer, vol.30, pp.68–74, April 1997.

[19] M. Herlihy and J. Wing, “Linearizability: A correctness condition for concurrent objects,” ACM Trans. Program.

Lang. Syst., vol.12, no.3, pp.463–492, 1990.

[20] L. Lamport, “Time, clocks and the ordering of events in a distributed system,” Commun. ACM, vol.21, no.7, pp.558–

565, 1978.

[21] C. Marchetti, “A Three-tier Architecture for Active Soft- ware Replication,” Ph.D. Thesis, Universit`a di Roma “La Sapienza,” Dipartimento di Informatica e Sistemistica, 2003.

[22] L. Moser, P.M. Melliar-Smith, D. Agarwal, R. Budhia, and C. Lingley-Papadopoulos, “Totem: A fault-tolerant multicast group communication system,” Commun. ACM, vol.39, no.4, pp.54–63, April 1996.

[23] Object Management Group (OMG), Framingham, MA, USA, Fault Tolerant CORBA Specification, V1.0, OMG Document ptc/2000-12-06 edition, April 2000, OMG Final Adopted Specification.

[24] F.B. Schneider, “Replication management using the state- machine approach,” in Distributed Systems, ed. S. Mullen- der, chapter 7, pp.169–198, Addison Wesley, 1993.

Carlo Marchetti earned a doctoral degree in computer engineering from the Faculty of Engineering of the University of Rome “La Sapienza” in 2003. His re- search interests include fault tolerance, software replication, large scale group communications, dependable distributed architectures, and (adaptive) middleware platforms. He also investigated problems related to cooperative and large scale in- formation systems, and to mobile net- works.

Sara Tucci Piergiovanni was born in Rome on 26 June 1975 and earned a laurea degree in computer engineer- ing from the Faculty of Engineering of the University of Rome “La Sapienza” in 2002. Her thesis won the 2002 AICA- Confidustria award for the best Italian thesis in ICT. Since November 2002, Sara Tucci Piergiovanni is Ph.D. student at the Department of Computer and Sys- tems Science of the University of Rome

“La Sapienza”. Her research interests include fault tolerance, software replication, publish and subscribe, and shared memory systems.

Roberto Baldoni is a faculty of the School of Engineering at the Universita di Roma a Sapienza, Italy. He was visit- ing researcher at INRIA in 1995 and vis- iting professor at Cornell University and EPFL in 1996 and 2003 respectively. He has been for several years now consultant to the European Commission in Informa- tion Technologies. He has been invited to serve in the PCs of many international symposiums among which DISC, ICDCS, SRDS, CoopIS, ISORC, DSN. He chaired the PC of the istributed algorithms track of the 19th IEEE International Conference on Distributed Computing Systems, the 6th International Workshop on Object-Oriented Real-Time Dependable Systems (WORDS- 01) and of the ACM workshop on Principles on Mobile Comput- ing. He has been also general chair of WORDS-03.

Riferimenti

Documenti correlati

Type 2 – the capacity of dump tiers is formed during working the regular stope at the face side with the preparation of additional capacity on the second tier (Fig.. Type 3 –

Then we propose, in the context of active replication, a three-tier architecture in which clients (the client-tier) interact with a middle tier (the mid-tier) that forwards

In this paper we first introduce two desirable ar- chitectural properties for software replication, namely Client/Server-Asynchrony and Client-Autonomy that, when satisfied by

The main idea underlying IRL design is to use three-tier replication in order (i) to allow independent stateful CORBA objects to be replicated and managed, (ii) to decouple

FO is a message handler placed in front of each end-tier replica (i) to avoid repeated execution of the same client request (possibly sent by ARH) and (ii) to ensure ordered

Then, by exploiting traces, we simulate simplified versions of three replication protocols (i.e., active, passive, and three-tier replication), and we show how the end-to-end latency

FO intercepts all incoming/outgoing messages from/to ARHs in order to ensure ordered request execution (operations are computed by replicas according to the request sequence

AnyDbms package and AnyDbProvider class will be used to give the user much more flexibility in the choice of the DBMS: thanks to Microsoft DbProviderFactory, a great example of