Didier El Baz
email: elbaz@laas.fr

Tung The

Nguyen
email: ttnguyen@laas.fr

A self-adaptive communication protocol with application to high performance peer to peer distributed computing#

Keywords: communication protocol, self-adaptive protocol, micro-protocols, high performance computing, peer to peer computing, nonlinear optimization, network flow problems

A self adaptive communication protocol is proposed for peer to peer distributed computing. This protocol can configure itself automatically in function of application requirements and topology changes by choosing the most appropriate communication mode between peers. The protocol was designed in order to be used in conjunction with a decentralized environment for high performance distributed computing. A first set of computational experiments is also presented and analyzed for an optimization application, i.e. nonlinear network flow problems.

I. INTRODUCTION

The peer to peer concept has known great developments with file sharing applications like Gnutella [START_REF]Gnutella Protocol Development[END_REF] or FreeNet [START_REF]The FreeNet Network Projet[END_REF]. Recent advances in microprocessors architectures and high bandwidth networks permit one to consider high performance peer to peer distributed computing as an economic and attractive solution. Nevertheless, the implementation of high performance computing methods on peer to peer networks gives rise to numerous challenges like scalability issues, machine heterogeneity and peers volatility. Existing protocols like TCP and UDP are designed to answer to application requirements like fault tolerance and exchanged messages order. Generally, they are not well suited to peer to peer networks.

In this paper, we propose a self adaptive communication protocol for high performance peer to peer distributed computing. The configurable protocol chooses the most appropriate communication mode between peers according to choices made at the application level, like the computation scheme, i.e. synchronous or asynchronous iterative method, or events that occur at the network level like topology changes. This protocol will be further used in a decentralized environment for high performance peer to peer computing [START_REF] Baz | CIP -Calcul intensif pair à pair[END_REF]. Section 2 deals with micro-protocols and protocol composition frameworks. In section 3, we propose some modifications to the Cactus framework and our self adaptive communication protocol. Section 4 deals with a first version of the decentralized environment for high performance peer to peer distributed computing. Finally, a first set of computational experiments for optimization problems, i.e. nonlinear network flow problems is presented and analyzed in Section 5.

II. MICRO-PROTOCOLS

Micro-protocols are an interesting approach to design self adaptive communication protocols. Micro-protocols were first introduced in x-kernel [START_REF] Hutchison | The x-kernel: An architecture for implementing network protocols[END_REF]. They have been widely used since in several systems. A micro-protocol implements merely a functionality of a given protocol. A protocol results from the composition of a given set of micro-protocols. This approach permits one to reuse the code, facilitate the design of new protocols and give the possibility to configure the protocol dynamically.

Several protocol composition frameworks have been proposed in order to deploy communication architectures. One can divide these frameworks according to 3 models: the hierarchical, nonhierarchical and hybrid models. In the hierarchical model, a stack of micro protocols composes a given protocol. This model can be found in the x-kernel [START_REF] Hutchison | The x-kernel: An architecture for implementing network protocols[END_REF] and APPIA [START_REF] Miranda | Appia: A flexible protocol kernel supporting multiple coordinated channels[END_REF] frameworks. In the nonhierarchical model, there is no particular order between micro-protocols; the Coyote [START_REF] Bhatti | Coyote: a system for constructing fine-grain configurable communication services[END_REF] and ADAPTIVE [START_REF] Schmidt | ADAPTIVE-A Dynamically Assembled Protocol Transformation, Integration and eValuation Environment[END_REF] frameworks correspond to this model. The hybrid model is a combination of the two previous models; micro-protocols are composed here hierarchically and nonhierarchically. One can find this last model in the XQoS [START_REF] Exposito | FPTP: the XQoS aware and fully programmable transport protocol[END_REF] and Cactus [START_REF] Hiltunen | The Cactus Approach to Building Configurable Middleware Services[END_REF] frameworks.

We have concentrated on the Cactus framework since this approach is flexible and efficient. We present the Cactus framework in the next subsection.

A. Cactus framework

Cactus is a system for constructing highly-configurable protocols for networked and distributed system [START_REF] Hiltunen | The Cactus Approach to Building Configurable Middleware Services[END_REF]. It has two grain levels. Individual protocols, termed composite protocols, are constructed from micro-protocols. Composite protocols are then layered on top of each other to create a protocol stack using an interface similar to the standard x-kernel API.

Cactus is an event-based framework. Events are used to signify state changes, such as arrival of messages from the network. Each micro-protocol is structured as a collection of event handlers, which are procedure-like segments of code and are bound to events. When an event occurs, all handlers bound to that event are executed. Events can be raised in different ways, explicitly by micro-protocols or implicitly by the runtime # The present work was funded by ANR-07-CIS7-011 grant system, with either blocking or non-blocking semantics, with a specific delay and a priority execution number. Arguments can be passed to handlers in two ways, statically when a handler is bound to an event and dynamically when an event is raised. The runtime system also provides operations for unbinding handlers, creating and deleting events, halting event execution, and canceling a delayed event. Handler execution is atomic with respect to concurrency, i.e., a handler is executed till completion before another handler is started unless it voluntarily yields the CPU.

The Cactus framework provides a message abstraction named dynamic messages, which is a generalization of traditional message headers. A dynamic message consists of a message body and an arbitrary set of named message attributes. Micro-protocols can add, read, and delete message attributes. When a message is passed to a lower-level protocol, a pack routine combines message attributes with the message body; while an analogous unpack routine extracts message attributes when a message is passed to a higher-level protocol. Cactus also supports shared data that can be accessed by all microprotocols configured in a composite protocol.

The CTP Configurable Transport Protocol is designed and implemented using the Cactus framework [START_REF] Wong | A configurable and extensible transport protocol[END_REF]. Fig. 1 shows the CTP implementation with events on the right side and micro-protocols on the left side. An arrow from a microprotocol to a given event indicates that the micro-protocol binds a handler to this event.

The CTP protocol includes a wide range of micro-protocols including a small set of basic micro-protocols like Transport Driver, Fixed Size or Resize and Checksum that are needed in every configuration and a set of micro-protocols implementing various transport properties like acknowledgements, i.e. PositiveAck, NegativeAck and DuplicateAck, retransmissions, i.e.

Retransmit, forward error correction, i.e. ForwardErrorCorrection, and congestion control, i.e. WindowedCongestionControl and TCPCongestionAvoidance. III. SELF-ADAPTIVE COMMUNICATION PROTOCOL In this Section, we propose the so-called P2PSAP Peer To Peer Self Adaptive communication Protocol for high performance peer to peer distributed computing. Fig. 2 shows the architecture of the protocol; this protocol has a Socket interface and two channels: a control channel and a data channel. We present now in detail those components.

A. Socket API

A main lack of Cactus CTP is that it has no application programming interface. Application has to use an interface as though it was just another composite protocol. So, we placed on the top of our protocol a socket-like API. Application can open and close connection, send and receive data. Furthermore, application can get session state and change session behavior or architecture through socket options, which are not available in Cactus. Session management commands like listen, open, close, setsockoption and getsockoption are directed to Control channel; while data exchange commands, i.e. send and receive commands are directed to Data channel.

B. Data channel

The data channel transfers data packets between peers; it is built by using the Cactus framework. To improve protocol performance and facilitate the reconfiguration, we have made some modifications to Cactus framework.

Firstly, Cactus doesn't allow concurrency handler execution. It means that a handler must wait for current executed handler completion before being executed. But nowadays, almost PCs have more than one core and concurrent handler execution is necessary to improve performance. So, we have modified Cactus to allow concurrency handler execution. Each thread has its own resources and its handler execution is independent of others.

Secondly, we have eliminated unnecessary messages copies between layers. In the Cactus framework, when a message is passed to upper or lower layers, Cactus runtime creates a new message and pass it to upper or lower layer. Hence, a significant number of CPU cycles and memories are consumed in multiple-layers systems. In our protocol, message copies occur between Socket API layer and Data channel, and within the Data channel. In order to eliminate message copies, we have modified the pack and unpack functions so that only a pointer to message is passed between layers. Therefore, no message copy is made within the stack.

Finally, Cactus provides operations for unbinding handler but it has no explicit operation for removing a micro-protocol. In order to facilitate protocol reconfiguration, we have added to Cactus API an operation for micro-protocol removing. In addition to the micro-protocol initializing function, each microprotocol must have a remove function, which unbinds all its handlers and releases its own resources. This function will be executed when the micro-protocol is removed.

After presenting our modifications to Cactus framework, we now consider in detail the data channel. The data channel has two levels: the physical layer and the transport layer; each layer corresponds to a Cactus composite protocol. We encompass the physical layer to support communications on different networks, i.e. Ethernet, InfiniBand and Myrinet. Each communication type is carried out by a composite protocol. The data channel can be triggered between the different types of networks; one composite protocol is then substituted to another. The transport layer is constituted by a composite protocol made of several micro-protocols, which is based on CTP. We have added to the existing micro-protocols a remove function corresponding to the modifications we have introduced to Cactus framework. In addition, we have designed some new micro-protocols that enable CTP to be used for sending and receiving messages in parallel computing applications.

Synchronization micro-protocols. CTP supports only asynchronous communication. But distributed application may have plural communication synchronization semantics. Hence, we have implemented 2 micro-protocols corresponding to 2 communication modes: synchronous and asynchronous. These micro-protocols introduce some new events, UserSend and UserReceive, which will be raised when send and receive socket commands are called by an application. In response to message sent from application, these micro-protocols may return the control to application immediately after message sent (asynchronous send) or wait for an acknowledgement indicating that message was received by receiver side application (synchronous send). Likely, in response to receive call from application, they may return the control to application immediately with or without message (asynchronous receive), or wait until message arrives (synchronous receive).

Buffer management micro-protocol. There are 2 buffers to manage: send buffer and receive buffer. Send buffer stores messages waiting to be sent or waiting for an acknowledgement from receiver application. Receive buffer stores messages which arrive from network and waits to be received by application. So, this micro-protocol must implement handlers for the UserSend and MsgFromNet events to catch messages from application and from network.

New congestion control micro-protocols. CTP actually has serveral micro-protocols implementing SCP congestion control and TCP-Tahoe congestion control. We have designed and used new micro-protocols implementing the TCP New-Reno congestion control [START_REF] Floyd | The New-Reno Modification to TCP's Fast Recovery Algorithm[END_REF] which is widely used and the H-TCP congestion control for high speed-latency network [START_REF] Leith | H-TCP protocol for high-speed long distance networks[END_REF].

At this level, data channel configuration is carried out by substituting or removing and adding micro-protocols. The behavior of the data channel is triggered by the control channel.

C. Control channel

The Control channel manages session opening and closure; it captures context information and (re)configures the data channel at opening or operation time; it adapts itself to these information and their changes; it is also responsible for coordination between peers during reconfiguration process. Note that we use the TCP/IP protocol to exchange those messages since Control messages cannot be lost.

Before describing the main components of the control channel, we present first a session life cycle, like in Fig. 3. Suppose process A wants to exchange data with process B, it opens a session through socket create and connect command. Then, a TCP connection is opened between 2 processes. Process B accepting connection must send its context information to process A. Process A, based on its context information and those of B, chooses the most appropriate configuration for data channel and send configuration command to process B. After that, the two processes carry out Fig. 3: Session life cycle the configuration of data channel. When the configuration is done, each process has to inform the other process and waits for the notification of other process. Data is exchanged only when both processes have finished data channel configuration. During the communication, a process can decide to change configuration of data channel due to context changes or user requirements, like process A in Fig. 3. Then, a similar procedure will be realized. When session is closed, the data channel is closed firstly; the control channel with TCP connection is closed later on.

We describe now the main components of the control channel.

1) Context monitor: the context monitor collects context data and their changes. Protocol adaptation is based on context acquisition, data aggregation and data interpretation. Context data can be requirements imposed by the user or the algorithm at the application level, i.e. asynchronous algorithms, synchronous algorithms or hybrid methods. Context data can also be related to peers location and machine loads. Context data are collected at specific times, periodically or by means of triggers. Data collected by the context monitor can be referenced by the controller.

2) Controller: the controller is the most important component in the control channel; it manages session opening and close through TCP connection opening and close; it also combines and analyzes context information provided by the context monitor so as to choose the configuration (at session opening) or to take reconfiguration decision (during session operation) for data channel. The choice of the most appropriate configuration is determined by a set of rules that are described by a specification language such as OWL, ECA, etc. These rules specify new configuration and actions needed to realize it. The (re)configuration command along with necessary information is sent to component Reconfiguration and to other communication end point.

3) Reconfiguration: reconfiguration actions are made by the reconfiguration component via the dedicated Cactus functions. Reconfiguration is made at the physical layer by substituting a composite protocol supporting communication on a network board to a composite protocol supporting communication on another type of network board or at the transport layer by removing and adding or substituting microprotocols.

4) Inter-peer coordination: the coordination component is responsible of context information exchanges and coordination of peers reconfiguration processes so as to insure proper working of the protocol.

D. Example of scenario

We present now a simple scenario for the P2PSAP protocol so as to illustrate its behavior.

We consider a high performance computing application, like for instance a large scale numerical simulation application or a complex optimization problem, solved on the network composed of two clusters shown in Fig. 4. The first cluster is composed of two similar machines: P1 and P2 that can be connected via Ethernet or InfiniBand. The second cluster is made of two similar machines: P3 and P4 connected only via Ethernet. The communication protocol between machines P3 and P4 is based on synchronous communication (since machines have similar characteristics and loads) via micro-protocols insuring e.g. reliability and order, i.e. ReliableFifo, and congestion control, i.e. TCPNewRenoCongestionControl. The communication protocol between machines P1 and P2 is based on synchronous communication for the same reasons via Infiniband since Infiniband is faster than Ethernet. Moreover, InfiniBand insures reliability and message order; as a consequence, the data channel needs only micro-protocols insuring buffer management (Buffer management), synchronous communication (Synchronous) and segment size management (Resize). Communications between machines of the different clusters are asynchronous; as a consequence, we need neither order, nor reliability micro-protocols in this case.

IV. ENVIRONMENT FOR HIGH PERFORMANCE PEER TO PEER DISTRIBUTED COMPUTING

In this section, we present a first prototype of an environment for high performance peer to peer distributed computing. For previous research work, see [START_REF] Baz | Some solutions for Peer to Peer Global Computing[END_REF]. Our environment uses the P2PSAP protocol for data exchanges between peers. Fig. 5 illustrates the architecture of the environment. In the future, we shall develop a decentralized environment based on P2PSAP.

We describe now the main components of the decentralized environment.

1) User daemon is the interaction interface between the application and the environment. It allows users to submit their tasks and retrieve final results.

2) Topology management organizes connected peers into clusters and maintains links between peers and between clusters. 3) Task management is responsible for task splitting into sub-tasks, sub-tasks distribution and results collection.

4) Task execution executes sub-tasks and exchanges intermediate results.

5) Load balancing estimates peer workload and migrates a part of work from overload peer to non-loaded peer.

6) Fault tolerance ensures the integrity of the calculation in case of peer failure. We intend to use the rollback recovery mechanism with checkpoint to recover state of failure peer on another peer.

7) Communication provides support for data exchange between peers using self-adaptive protocol.

V. EXPERIMENTATION

We present now a first set of computational experiments for nonlinear network flow problems.

A. Network flow problems

Network flow problems [START_REF] Baz | Asynchronous iterative algorithms with flexible communication for nonlinear network flow problems[END_REF] consist in computing the network flow from a source to a destination that minimizes the total traffic cost. The problems occur in many domains: electrical networks, gas or water distribution, financial models, communication and transportation networks. The solution of nonlinear network flow problems requires intensive computations, so a distributed or parallel solution of these problems is very attractive. . The problem is to minimize total cost subject to a conservation of flow constraint at each node:

1) Problem formulation Let

∑ ∈A j i ij ij f c ) , ( ) ( min , subject to ∑ ∑ ∈ ∈ ∈ ∀ = - A i m i mi A j i ij N i b f f ) , ( ) , ( . ,
We assume that problem (V.1) has a feasible solution. We consider the following standing assumptions on ij c .

Assumption V.1. ij c is strictly convex. Assumption V.2. ij c is lower semicontinuous. Assumption V.3. The conjugate convex function of , ij c defined by , } ) ( . { sup ) ( * ij ij ij ij ij ij f c f t t c - = is real valued, i.e., R t t c ij ij ij ∈ ∀ +∞ < < -∞ , ) ( * . 2) Dual problem A dual problem for (V.1) is given by n R p ) p ( q min ∈ , subject to no constraint on the vector } | { N i p p i ∈ =
, where q is the dual functional given by

∑ ∑ ∈ ∈ - - = A j i N i i i j i ij p b p p c p q ) , ( * . ) ( ) (
We refer to p as a price vector and its components as prices. The i-th price i p , is a Lagrange multiplier associated with the i-th conservation of flow constraint. The necessary and sufficient condition for optimality of a pair ) , ( p f is given as follows: a feasible flow vector

} ) , ( | { A j i f f ij ∈ =
is optimal for (V.1) and a price vector

} | { N i p p i ∈ =
is optimal for (V.2) if and only if for all A j i ∈ ) , ( ,

j i p p -is a sub-gradient of ij c at . ij f An equivalent condition is ), ( * * j i ij ij p p c f - ∇ = , ) , ( A j i ∈ ∀ where ) ( * x c ij ∇
denotes the gradient of ). ( * x c ij 3) Dual optimal solution set The optimal solution of the dual problem is never unique since adding the same constant to all coordinates of a price vector p leaves the dual cost unaffected. We can remove this degree of freedom by constraining the price of one node, say the destination node d, to be zero. Consider the set 0}.

| { = ∈ = d n p R p P
We concentrate on the reduced dual problem:

). We note that assumption V.4 is not very restrictive (see [START_REF] Baz | Asynchronous iterative algorithms with flexible communication for nonlinear network flow problems[END_REF]). In the sequel, ) ( p g will denote the gradient of the dual functional, the components ) ( p g i of ) ( p g are given by: .

, ) ( ) ( ) ( ) ( ) , ( ) , ( * * N i b p p c p p c p p q p g i i A j i A i m m mi j i ij i i ∈ - - ∇ - - ∇ = ∂ ∂ = ∑ ∑ ∈ ∈ 4

∈ = β α

(see [START_REF] Baz | Asynchronous iterative algorithms with flexible communication for nonlinear network flow problems[END_REF]).

(V.1) (V.2) (V.3)
The gradient type method consists in iterating on the i-th component of the vector of prices as follows:

} { , ' ,..., 2 , 1 , ,..., 1 ), , ( 1 1 1 q q n i p p g p p q i i q i q i ∈ = - = - - α where i i p p = 0 and '
q is the number of iterations such that ε ≤ ) ( ' q i i p g where i i P p = 0 and ε is the research accuracy.

B. Experiment platform

Experiments have been carried out on the LAASNETEXP experimental network (see [START_REF] Owezarski | LaasNetExp: a generic polymorphic platform for network emulation and experiments[END_REF]). The topology of the small network used for this first set of computational experiments is displayed in Fig. 6 where peers are connected via a Gigabit Ethernet network. Machines P1, P2, P3 and P4 are similar i.e.: Dual Core Xeon 3050, 2.13GHz with Linux Debian.

We have used the C implementation of Cactus 2.2 for micro-protocol composition over Linux UDP sockets.

C. Self-adaptation

In these experiments, the context data used for configuring the protocol are the computational schemes chosen by the user (synchronous iterative schemes, asynchronous iterative schemes or hybrid schemes, i.e. combination of synchronous and asynchronous iterative schemes) and topology parameters i.e. peer location which implies inter or intra cluster communication. Decision rules are summarized in Table 1. In the sequel, we explain those rules.

In clusters, machines are often identical; they are connected via a local network with small latency, high bandwidth and reliable data transfer. As a consequence, if computational loads are well balanced, then synchronous communication between peers can be very efficient. On the other hand, in inter-cluster data exchanges where there may be heterogeneity (processors, OS, bandwidth) and communications may have high latency and unreliability, synchronization may be an obstacle to efficiency due to waiting times and overheads. Thus, asynchronous communication seems more appropriate in this case.

Communication modes have to correspond to computational schemes which depend on algorithm nature and are generally chosen by users. In the case of synchronous schemes, the synchronous communication must be implemented in both intra-cluster and inter-cluster data exchanges. Thus, micro-protocols used for the data channel will be the Synchronous micro-protocol with some reliability and order micro-protocols. To explore the high-speed long distance networks, the data channel can use H-TCP congestion control micro-protocol for inter-cluster communication instead of TCP New-Reno congestion control micro-protocol which works well only in low latency network.

Likely, in the case of asynchronous schemes, the asynchronous communication must be implemented in both intra-cluster and inter-cluster data exchanges. We note that asynchronous schemes of computation are fault tolerant in some sense since they allow messages losses. However, messages losses may lead to some iterations increases. For this reason, in intra-cluster communication with low latency, it may be better to add some reliability micro protocols to the data channel along with the Asynchronous micro-protocol. While in inter-cluster communication with high latency and message losses recovery time may be comparable with updating time, thus those messages can become obsolete. Hence, reliability micro protocols are not needed in this case.

There are also some problems that can be solved by using all computational schemes. In this last case, the user can leave the protocol to freely choose communication mode. As a consequence, it can choose the most appropriate mode according to topology parameters, i.e., synchronous communication for intra-cluster data exchanges and asynchronous communication for inter-cluster data exchanges; this corresponds to the so-called Hybrid computational scheme.

D. Results

Computations have been carried out on 1, 2 and 4 machines. In the distributed case, i.e. for several machines, the original network flow problem is decomposed into several subnetworks. In the particular case of 2 machines, computations are made within the same cluster. We have carried out experiments with different computational schemes and communication scenarios i.e. synchronous, asynchronous and hybrid (synchronous / asynchronous). A first set of computational results is displayed in Table II for gradient type methods applied to gas distribution problems with 2000 nodes.

Fig. 2 :Fig. 1 :

 21 Fig. 2: P2PSAP Protocol architecture

Fig. 5 :Fig. 4 :

 54 Fig. 5: Environment prototype

  directed graph. N is referred to as the set of nodes, N N A × ⊆is referred to as the set of arcs, and the cardinal number of N is denoted by n.

4 .

 4 The reduced dual optimal solution set * P is nonempty and compact.

Fig. 6 :

 6 Fig. 6: Network used for computational tests

TABLE I :

 I COMMUNICATION MODE ADAPTATION RULES TABLE II.

				COMPUTATIONAL RESULTS
	Number of PCs	Method	P1	Number of iterations P2 P3	P4	Times (s)	Speedup	Efficiency
	1	-	399813	-	-	-	2135	-	-
	2	Syn	400694 400694	-	-	1481	1,44	0,72
	2	Asyn	385780 583735	-	-	1209	1,76	0,88
	4	Syn	402056 402056 402056 402056	1241	1,72	0,43
	4	Hybrid	449372 449372 398421 398421	935	2,28	0,57
	4	Asyn	419175 389144 464128 743636	656	3,25	0,81

B REFERENCES

Synchronous

Asynchronous Hybrid For the application and topology considered, we note that asynchronous schemes have performed better than the synchronous ones. Moreover, the efficiency of synchronous schemes deteriorates greatly when the number of processors increases, i.e. 0.72 with 2 machines and 0.44 with 4 machines. The efficiency of asynchronous distributed schemes decreases slowly with the number of processors, i.e. 0.88 with 2 machines and 0.81 with 4 machines. This is mainly due to waiting time due to synchronization and synchronization overhead. When using asynchronous schemes of computation, some processors may iterate faster than others; this is particularly the case when loads are unbalanced as for the application considered here. The use of hybrid schemes, i.e. synchronous communication between peers in the same cluster and asynchronous communication between peers in different clusters gave efficiency that is in between efficiencies of pure synchronous and asynchronous methods.

It follows from the computational experiments that the choice of communication mode has important consequences on the distributed method efficiency. The ability for the protocol P2PSAP to choose the best communication mode in function of network topology, context and computational scheme choices appears as a crucial property. We note also that the choice of communication mode has important consequences on the reliability of the distributed method and everlastingness of the high performance computing application. In this last context, we note that asynchronous communications are more appropriate in the case of communications between clusters.

VI. CONCLUSIONS

In this paper, we have proposed a self adaptive communication protocol for high performance peer to peer distributed computing. This protocol has been implemented on a small network for the solution of nonlinear optimization problems, i.e. network flow problems. A first set of computational experiments shows that the protocol permits one to obtain good efficiency when using asynchronous communications or a combination of synchronous and asynchronous communications.

We plan to study a specification language for controller decision rules description. We shall also concentrate on the design of a decentralized environment for high performance peer to peer distributed computing. This type of environment will permit one to use all the specificities offered by the peer to peer concept for high performance computing purpose. Self organization of peers for efficiency purpose or for insuring everlastingness of applications in hazardous situations or in the presence of faults will also be studied. Finally, we plan to consider other applications e.g. financial mathematics problems like American or European options problems. The different applications considered will permit us to validate our protocol and decentralized environment in different high performance computing contexts.


Abdelamine Boukedjar
email: aboukedj@laas.frmlalami@laas.frelbaz@laas.fr

Mohamed Esseghir Lalami

Didier El Baz

Didier El-Baz

Parallel Branch and Bound on a CPU-GPU System

come   L'archive ouverte pluridisciplinaire

Introduction

Graphics Processing Units (GPUs) are high-performance many-cores processors. Tools like Compute Unified Device Architecture (CUDA) or Open Computing Language (OpenCL) have been developed in order to use GPUs for general purpose computing; this has led to GPU computing and hybrid computing. CUDA-based NVIDIA GPUs are Single Instruction Multiple Thread (SIMT) architectures which is akin to Single Instruction Multiple Data (SIMD) architecture (see [START_REF]Cuda 2.0 programming guide[END_REF]).

In this paper, we concentrate on the implementation via CUDA of branch and bound algorithms on GPU for an important class of integer programming problems, i.e. Knapsack Problems (KP).

Knapsack problems occur in many domains like logistics, manufacturing, finance and telecommunications or as subproblems of hard problems in combinatorial optimization like multidimensional knapsack problems (see for example [START_REF] Boyer | Heuristics for the 0-1 multidimensional knapsack problem[END_REF], [START_REF] Kellerer | Knapsack Problems[END_REF] and [START_REF] Martello | Knapsack Problems -Algorithms and Computer Implementations[END_REF]). The knapsack problem is among the most studied discrete optimization problems; it is also one of the simplest prototypes of integer linear programming problems. Several parallel algorithms have been proposed for KP (e.g. see [START_REF] Baz | Load balancing methods and parallel dynamic programming algorithm using dominance technique applied to the 0-1 knapsack problem[END_REF], [START_REF] Gerash | A survey of parallel algorithms for one-dimensional integer knapsack problems[END_REF] and [START_REF] Lou | A parallel two-list algorithm for the knapsack problem[END_REF]). In particular, implementations on a SIMD machine have been performed on a 4K processor ICL DAP [START_REF] Kindervater | An introduction to parallelism in combinatorial optimization[END_REF], a 16K Connection Machine CM-2 (see [START_REF] Lin | Processor-efficient algorithms for the knapsack problem[END_REF] and [START_REF] Ulm | Dynamic programming implementations on SIMD machines -0/1 knapsack problem[END_REF]) and a 4K MasPar MP-1 machine (see [START_REF] Ulm | Dynamic programming implementations on SIMD machines -0/1 knapsack problem[END_REF]).

We are interested in the solution via GPU of difficult combinatorial optimization problems like problems of the knapsack family. We are presently developing a series of parallel codes that will be combined in order to provide efficient parallel hybrid methods. In [START_REF] Boyer | Solving knapsack problems on gpu[END_REF], we have presented an original implementation via CUDA of the dynamic programming method for KP on a CPU/GPU system with a single GPU. Experiments carried out on a CPU with 3 GHz Xeon Quadro Intel processor and GTX 260 GPU have shown substantial speedup. In [START_REF] Boyer | Dense dynamic programming on multi gpu[END_REF], Boyer et al. have proposed an implementation via CUDA of the dense dynamic programming method on multi GPU architectures. This solution is well suited to the case where CPUs are connected to several GPUs; it is also particularly efficient. We have also proposed parallel simplex methods that run on a GPU or several GPUs (see [START_REF] Lalami | Efficient implementation of the simplex method on a cpu-gpu system[END_REF] and [START_REF] Lalami | Multi gpu implementation of the simplex algorithm[END_REF]); these codes are particularly interesting when one wants to compute bounds of knapsack problems. We refer also to [START_REF] Van Luong | Large neighborhood local search optimization on graphics processing units[END_REF], for a study on local search methods and GPU computing for combinatorial optimization problems. The use of CPU-GPU systems for solving difficult combinatorial optimization problems is a great challenge so as to reduce drastically the time needed to solve problems and the memory occupancy or to obtain exact solutions yet unknown. One of the difficulties of branch and bound methods is that they often lead to an irregular data structure that is not well suited to GPU computing. In this paper, we propose an original hybrid implementation of the branch and bound method via CUDA. The reader is referred to [START_REF] Gendron | Parallel branch-and-bound algorithms: Survey and synthesis[END_REF] for a survey on parallel branch and bound algorithms.

The knapsack problem and the branch and bound method are presented in Section 2. Section 3 deals with the implementation via CUDA of the branch and bound method on the CPU-GPU system. Computational results are displayed and analyzed in Section 4. Section 5 deals with conclusions and future work.

Knapsack problem

Given a set of items i ∈ {1, ..., n}, with profit p i ∈ N * + and weight w i ∈ N * + and a knapsack with the capacity C ∈ N * + , the KP can be defined as the following integer programming problem:

(KP )              max n i=1 p i .x i , s.t. n i=1 w i .x i ≤ C, x i ∈ {0, 1}, i ∈ {1, ..., n}. (1)
To avoid any trivial solution, we assume that we have

     ∀i ∈ {1, ..., n}, w i ≤ C, n i=1 w i > C.
This insures that each item i fits into the knapsack and the overall weight sum of the items exceeds the knapsack capacity.

Branch and bound method

The branch and bound method is a general method that permits one to find optimal solution of various optimization problems like discrete and combinatorial optimization problems (see in particular [START_REF] Kellerer | Knapsack Problems[END_REF] and [START_REF] Martello | Knapsack Problems -Algorithms and Computer Implementations[END_REF]). It is based on enumeration of all candidate solutions and pruning of large subsets of candidate solutions via computation of lower and upper bounds of the criterion to optimize (see Figure 1). We concentrate here on breadth-first search strategy that is well suited to parallel implementation on GPU. We assume that items are sorted according to decreasing profit per weight ratio. At each step of the branch and bound method, the same branching and bounding tasks are carried out on a list of states called also nodes. Let k denote the index of the current item relative to a branching step. We denote by N k e the set of items in the knapsack at step k for a given node e. A node e is usually characterized by a tuple (w e , p e , X e , U e , L e ) where w e represents the weight of node e, p e represents the profit of the node, X e is the solution subvector associated with node e, U e and L e , respectively, are an upper bound and a lower bound of the node, respectively. We have:

w e = i∈N k e w i , p e = i∈N k e p i .
The so-called Dantzig bound (see [START_REF] Dantzig | Discrete variable extremum problems[END_REF]), derived from the solution of the continuous knapsack problem is a classical upper bound; it is given as follows:

U e = p e + se-1 i=k+1 p i + C. p se w se ,
where s e is the so-called slack variable such that se-1 i=k+1

w i ≤ C -w e < se i=k+1
w i ,

(we note that one can compute index s e for a node e and item k+1 via a greedy algorithm applied to a knapsack of capacity C -w e ) and C is the residual capacity given by

C = C -w e + se-1 i=k+1 w i .
Classically, one can obtain also a lower bound of the problem with a feasible solution of the knapsack problem with {0, 1} variables. For example, a lower bound L e , can be computed via a greedy algorithm after selection of all items before the slack variable, s e , since we have assumed that items are sorted by decreasing price per weight ratio.

For the sake of simplicity and efficiency, the following representation of a given node e will be used in the sequel: ( ŵe , pe , s e , U e , L e ), where So, we have:

U e = pe + C.
p se w se .

We note that bound computation is time consuming in the branch and bound method and obtaining this way the bounds U e and L e is more efficient.

Branching

If k < s e , then a node generates two sons at step k :

• a node with x k = 0, ŵe = ŵe -w k and pe = pe -p k ;

• a node with x k = 1 (in this case, no new node is created and the father is just kept in the list).

The case where k = s e yields only one son with x k = 0. This comes to keep the father in the list and set s e = s e + 1. Indeed, item k cannot be packed into the knapsack.

Bound computation

We note that the upper and lower bounds do not change in the case where x k = 1.

In the case where x k = 0, one has to compute the new slack variable. So, ŵe and pe are updated as presented in subsection 2.1. Consequently, the upper and lower bounds U e and L e , respectively, are updated.

Remark: Pruning subset of candidate solutions is then done via comparison of the best current lower bound with the upper bound of each node. We denote by L the best lower bound. If we have U e ≤ L, then node e is discarded.

3 Hybrid computing

CPU/GPU systems and CUDA

NVIDIA GPU cards and computing systems are highly parallel, multithreaded, many-core architectures. GPU cards are well known for image processing applications. NVIDIA has introduced in 2006 the Compute Unified Device Architecture (CUDA). CUDA is a software development kit that enables users to solve many complex computational problems on GPU cards. Our parallel dynamic programming code has been implemented via CUDA 3.2.

CUDA-based GPUs and computing systems are Single-Instruction, Multiple-Threads (SIMT) architectures, i.e. the same instruction is executed simultaneously on many data elements by different threads. GPUs are well suited to address problems that can be expressed as data-parallel computations since GPUs devote more transistors to data processing than to data caching and flow control. GPUs can nevertheless be used for task parallel applications with success.

As shown in Figure 2, a parallel code on GPU (the socalled device) is interleaved with a serial code executed on the CPU (the so-called host). At the top level, the threads are grouped into blocks. These blocks contain up to 1024 threads and are organized in a grid which is launched via a single CUDA program (the so-called kernel).

Figure 2. Thread and memory hierarchy in GPUs

When a kernel is launched, the blocks whithin the grid are assigned to idle groups of processors, the so-called multiprocessors. Threads in different blocks cannot communicate with each other explicitly. They can nevertheless share their results by means of a global memory.

The multiprocessor executes threads in groups of 32 parallel threads called warps. Threads composing a warp start together at the same program address, they are nevertheless free to branch and execute independently. However, a divergent branch may lead to a poor efficiency.

Threads have access to data from multiple memory spaces (see Figure 2). Each thread has its own register and private local memory. Each block has a shared memory with high bandwidth only visible to all threads of the block and which has the same lifetime as the block. Finally, all threads have access to a global memory. Furthermore, there are two other read-only memory spaces accessible by all threads which are cache memories:

• the constant memory, for constant data used by the process,

• the texture memory space, optimized for 2D spatial locality.

In order to have a maximum bandwidth for the global memory, memory accesses have to be coalesced. Indeed, the global memory access by all threads whithin a half-warp (a group of 16 threads) is done in one or two transactions if

• the size of the words accessed by the threads is 4, 8, or 16 bytes,

• all 16 words lie:

in the same 64-byte segment, for words of 4 bytes,

in the same 128-byte segment, for words of 8 bytes,

in the same 128-byte segment for the first 8 words and in the following 128-byte segment for the last 8 words, for words of 16 bytes;

• threads access the words in sequence (the kth thread in the half-warp accesses the kth word).

Otherwise, a separate memory transaction is issued for each thread. This degrades significantly the overall processing time.

Reference is made to [START_REF]Cuda 2.0 programming guide[END_REF] for further details on the NVIDIA GPUs and computing systems architecture and how to optimize the code.

In this paper, we study the parallel implementation of the branch and bound methods on a Tesla C2050 GPU.

Parallel algorithm

Bound computation is particularly time consuming in branch and bound algorithm. Nevertheless, this task can be efficiently parallelized on GPU. The main tasks of our parallel algorithm are presented below (see also Figure 3).

Task carried out on the CPU

• Perform branch and bound algorithm on CPU when the size of the list is small.

• Transfer the current list of nodes to the GPU.

• Launch branching phase on GPU.

• Launch bound computation phase on GPU.

• Get the sublist of created nodes.

• Discard the nonpromising nodes.

Task carried out on the GPU

• Perform branching (kernel 1).

Figure 3. Data exchange between CPU and GPU

• Perform bound computation (kernel 2).

• Find the best lower bound L.

In the sequel, we detail the different tasks carried out on the host and device, respectively. We begin by the tasks implemented on the device.

Computations on GPU

We note that each thread in the grid of blocks performs computation on only one node of the branch and bound list, so as to have coalesced memory access. The table of items that contains weight w and profit p is stored in the texture memory in order to reduce the memory latency. As a consequence, these variables will not be referred to as arguments in the different kernels presented below so as to simplify algorithmic notation.

Branching

The kernel 1 corresponds to the branching step carried out on the GPU. The e-th thread branches on the node e and creates a new node at the address e.

• In the case where k < s e , the e-th thread computes the value of ( ŵe , pe , s e ) of its son with x k = 0. We recall that its son with x k = 1 is already in the list stored in the CPU. We see that substantial speedup can be obtained by using the Tesla C2050 GPU.

The proposed parallel branch and bound algorithm permits one to reduce drastically the processing time. The more streaming processor cores of the Tesla C2050 GPU are made available for a given computation, the more threads are executed in parallel and better is the global performance.

In general, the speedup increases with the size of the problem. The speedup meets a level around 9.

The speedup depends greatly on the size and difficulty of the considered instance. In particular, the best speedups have been obtained for instances with great number of nodes. As a matter of fact, there are few nodes that are discarded in this case and the GPU occupancy is particularly important. For example, we have experienced speedup equal to 11 in some cases with 500 items.

We have also performed experiments for non correlated problems that turn out to be easy. In this last case, pruning is particularly important and thus sequential branch and bound is very efficient. Thus, implementation on a CPU-GPU system has given no speedup since all computation is performed on the CPU.

We consider the solution of hard problems of the knapsack family, like multidimensional knapsack problems or multiple knapsack problems, to become possible in reasonable time with the help of GPU architectures.

Conclusions and future work

In this paper, we have proposed an original parallel implementation via CUDA of the branch and bound method for knapsack problems on a CPU-GPU system with Tesla C2050 GPU. The proposed approach is very efficient. In particular, computational results have shown that difficult problems can be solved within small processing time. This work shows the relevance of using CPU/GPU systems for solving difficult combinatorial optimization problems.

Our approach seems to be robust since the results remain good when the size of the problem increases.

We believe that further speedup can be obtained on multi GPU clusters.

We are currently parallelizing a series of methods for integer programming problems like dynamic programming and Simplex methods. The combination of parallel methods will permit us to propose efficient hybrid computing methods for difficult integer programming problems like multidimensional knapsack problems, multiple knapsack problems and knapsack sharing problems.

Figure 1 .

 1 Figure 1. Data exchange between CPU and GPU

Table 1 . time on CPU and CPU-GPU system of the branch and bound algorithm

 1

	size n of the	time	on	time	on	speedup
	problem	CPU (s)		CPU-GPU(s)
	100	1.59		0.41		3.84
	200	4.85		0.91		5.33
	300	9.82		1.44		6.80
	400	10.94		1.27		8.61
	500	13.39		1.44		9.27

Table 1

 1 displays computational times of the branch and bound algorithm on the CPU and the CPU-GPU system.

Acknowledgment

Dr Didier El Baz would like to thank NVIDIA for his support through Academic Partnership.

variable of the son with x k = 0 is updated as follows:

and values of ŵe and pe are not modified. Moreover, node e with x k = 1 must be labelled as a nonpromising node in the CPU; this is done by assigning

Bound computation

The parallel bound computation procedure computes bounds of new nodes, this is made via kernel 2 which is a loop of n -h iterations; where h = max {k, s} and s is the first slack variable computed when the knapsack is empty. At each new iteration, a new item is considered and the current value of ŵj and pj and of the lower bound L j are updated. Moreover, it may happen that the upper bound U j is computed according to some condition that will be detailed in the sequel.

We note that the knapsack capacity C is stored in the constant memory of the GPU since it remains constant. This permits one to reduce memory latency.

When kernel 2 is launched, the j-th thread updates at iteration i, ( ŵj , pj , s j , L j ) if the item i has not been considered yet by this node i.e. if i ≥ s j .

The upper bound U j is computed and ŵj , pj and s j are updated as soon as the condition w+w i ≤ C is not satisfied.

We note that the lower bound L j is obtained only at the end of the procedure, i.e. at iteration n.

Finding the best lower bound

The best lower bound L is obtained in the GPU via a reduction method making use of the atomic instruction atomicM ax applied to the table of lower bounds (see [START_REF] Harris | Optimizing parallel reduction in cuda[END_REF]).

Branch and bound algorithm

If the size of the list is small, then it is not efficient to launch the branch and bound computation kernels on GPU since the GPU occupancy would be very small and computations on GPU would not cover communications between CPU and GPU. This is the reason why the branch and bound algorithm is implemented on CPU in this particular context. We note that for a given problem, the branch and bound computation phases can be carried out several times on the CPU according to the result of the pruning procedure. In this study, GPU kernels are launched only when the size of the list, denoted by q, is greater than 5000 nodes (see Figure 3). We have noticed that this condition ensures generally a 100 % occupancy for the GPU.

Pruning

This procedure starts after the list of nodes and the best lower bound L have been transfered from the GPU to the CPU. The size q of the list is updated by taking a value of q which is twice as much as its original value, then the list is treated from e = 1 to q. A node e is considered to be non promising (NP) if U e ≤ L otherwise it is promising (P) (see Figure 4).

Non promising states are replaced via an iterative procedure that starts from the beginning of the list. The different steps of the procedure that permits one to replace a non promising node with index l by a promising node are presented below.

• search a promising node j starting form the end of the list.

• copy data ( ŵj , pj , s j , U j ) of the promising node j in tuple ( ŵl , pl , s l , U l ).

• update the size of the list as follows :

Computational results

The CPU/GPU system considered in this study is a DELL Precision T7500 Westmere with processor Quad-Core Intel Xeon E5640 2.66 GHz and 12 GB of main memory and NVIDIA Tesla C2050 GPU. The Tesla C2050 GPU, which is based on the new-generation CUDA architecture codenamed Fermi, has 3 GB DDR5 of memory and 448 streaming processor cores (1.15 GHz) that deliver a peak performance of 515 Gigaflops in double precision floating point arithmetic. The interconnection between the host and device is done via a PCI-Express Gen2 interface. We have used CUDA 3.2 for the parallel code and gcc for the serial one.

We have carried out computational tests on randomly generated correlated problems; the problems are available at [9]. They present the following features:

• w i , i ∈ {1, ..., n}, is randomly drawn in [1, 100],

• p i = w i + 10, i ∈ {1, ..., n},

For each problem, the displayed results correspond to an average of ten instances. We have ovserved that the best number of threads per block is 192.