Toàn Nguyên
email: tnguyen@inrialpes.fr

A Distributed Workflow Platform for High-Performance Simulation

Keywords: workflows, fault-tolerance, resilience, simulation, distributed systems, high-performance computing I

This paper presents an approach to design, implement and deploy a simulation platform based on distributed workflows. It supports the smooth integration of existing software, e.g., Matlab, Scilab, Python, OpenFOAM, Paraview and user-defined programs. Additional features include the support for application-level fault-tolerance and exception-handling, i.e., resilience, and the orchestrated execution of distributed codes on remote high-performance clusters.

INTRODUCTION

Large-scale simulation applications are becoming standard in research laboratories and in the industry [START_REF] Nguyên | A Distributed Workflow Platform for Simulation[END_REF] [START_REF] Abbas | High Computing Power: A radical Change in Aircraft Design Process[END_REF]. Because they involve a large variety of existing software and terabytes of data, moving around calculations and data files is not a simple avenue. Further, software and data are often stored in proprietary locations and cannot be moved. Distributed computing infrastructures are therefore necessary [6][8].

This article explores the design, implementation and use of a distributed simulation platform. It is based on a workflow system and a wide-area distributed network. This infrastructure includes heterogeneous hardware and software components. Further, the application codes must interact in a timely, secure and effective manner. Additionally, because the coupling of remote hardware and software components is prone to run-time errors, sophisticated mechanisms are necessary to handle unexpected failures at the infrastructure and system levels [START_REF] Dongarra | The International Exascale Software Roadmap[END_REF]. This is also true for the coupled software that contribute to large simulation applications [START_REF] Joseph | A Strategic Agenda for European Leadership in Supercomputing: HPC 2020[END_REF]. Consequently, specific management software is required to handle unexpected application and software behavior [START_REF] Kandaswamy | Fault-tolerant and recovery of scientific workflows on computational grids[END_REF][11] [12][15]. This paper addresses these issues. Section II is an overview of related work. Section III is a general description of a sample application, infrastructure, systems and application software. Section IV addresses fault-tolerance and resiliency issues. Section V gives an overview of the implementation using the YAWL workflow management system [START_REF] Van Der Aalst | Modern Business Process Automation: YAWL and its support environment[END_REF]. Section VI is a conclusion.

II. RELATED WORK

Simulation is nowadays a prerequisite for product design and for scientific breakthrough in many application areas ranging from pharmacy, biology to climate modeling that also require extensive simulation testing. This requires often large-scale experiments, including the management of petabytes volumes of data and large multi-core supercomputers [START_REF] Simon | Future directions in High-Performance Computing 2009-2018[END_REF].

In such application environments, various teams usually collaborate on several projects or part of projects. Computerized tools are often shared and tightly or loosely coupled [START_REF] Bachmann | Advances in Generalization and Decoupling of Software Parts in a Scientific Simulation Workflow System[END_REF]. Some codes may be remotely located and nonmovable. This is supported by distributed code and data management facilities [START_REF] Sherp | Using UNICORE and WS-BPEL for Scientific Workfow execution in Grid Environments[END_REF]. And unfortunately, this is prone to a large variety of unexpected errors and breakdowns [START_REF] Ludäscher | Scientific Workflows: Business as usual ?[END_REF].

Most notably, data replication and redundant computations have been proposed to prevent from random hardware and communication failures [START_REF] Plankensteiner | A New Fault-Tolerant Heuristic for Scientific Workflows in Highly Distributed Environments based on Resubmission impact[END_REF], as well as failure prediction [START_REF] Lan | Adaptive Fault Management of Parallel Applications for High-Performance Computing[END_REF], sometimes applied to deadline-dependent scheduling [START_REF] Crawl | A Provenance-Based Fault Tolerance Mechanism for Scientific Workflows[END_REF].

System level fault-tolerance in specific programming environments are proposed, e.g., CIFTS [START_REF] Gupta | CIFTS: a Coordinated Infrastructure for Fault-Tolerant Systems[END_REF], FTI [START_REF] Bautista-Gomez | FTI: high-performance Fault Tolerance Interface for hybrid systems[END_REF]. Also, middleware usually support mechanisms to handle faulttolerance in distributed job execution, usually calling upon data replication and redundant code execution [START_REF] Kandaswamy | Fault-tolerant and recovery of scientific workflows on computational grids[END_REF][15] [START_REF] Abramson | Embedding Optimization in Computational Science Workflows[END_REF] [START_REF] Duan | DEE: a Distributed Fault Tolerant Workflow Enactment Engine for Grid Computing[END_REF].

Also, erratic application behavior needs to be supported [START_REF] Sindrilaru | Fault-Tolerance and Recovery in Grid Workflow Mangement Systems[END_REF]. This implies evolution of the simulation process in the event of such occurrences. Little has been done in this area [START_REF] Plankensteiner | Fault-tolerant Behavior in State-of-the-Art Grid Workflow Management Systems[END_REF] [START_REF] Hwang | Grid Workflow: A Flexible Failure Handling Framework for the Grid[END_REF]. The primary concerns of the designers, engineers and users have so far focused on efficiency and performance [START_REF]The Grid Workflow Forum[END_REF] [49] [START_REF] Moody | Design, Modeling and Evaluation of a Scalable Multi-level checkpointing System[END_REF]. Therefore, application unexpected behavior is usually handled by re-designing and re-programming pieces of code and adjusting parameter values and bounds. This usually requires the simulations to be stopped and restarted.

A dynamic approach is presented in the following sections. It support the evolution of the application behavior using the introduction of new exception handling rules at run-time by the users, based on occurring (and possibly unexpected) events and data values. The running workflows do not need to be suspended in this approach, as new rules can be added at run-time without stopping the executing workflows.

This allows on-the-fly management of unexpected events. This approach also allows a permanent evolution of the applications that supports their continuous adaptation to the occurrence of unforeseen situations [START_REF] Hwang | Grid Workflow: A Flexible Failure Handling Framework for the Grid[END_REF]. As new situations arise and data values appear, new rules can be added to the workflows that will permanently take them into account in the future. These evolutions are dynamically hooked onto the workflows without the need to stop the running applications. The overall application logics is therefore maintained unchanged. This guarantees a constant adaptation to new situations without the need to redesign the existing workflows. Further, because exception-handling codes are themselves defined by new ad-hoc workflows, the user interface remains unchanged [START_REF] Adams | The worklet custom service for YAWL, Installation and User Manual[END_REF].

III. TESTCASE APPLICATION

A. Example

This work is performed for the OMD2 project (Optimisation Multi-Discipline Distribuée, i.e., Distributed Multi-Discipline Optimization) supported by the French National Research Agency ANR.

An overview of two running testcases is presented here. It deals with the optimization of an auto air-conditioning system [START_REF] Gill | Practical Optimization[END_REF]. The goal of this particular testcase is to optimize the geometry of an air conditioner pipe in order to avoid air flow deviations in both pressure and speed concerning the pipe output (Figure 1). Several optimization methods are used, based on current research by public and industry laboratories. This example is provided by a car manufacturer and involves several industry partners, e.g., software vendors, and academic labs, e.g., optimization research teams (Figure 1).

The testcases are a dual faceted 2D and 3D example. Each facet involves different software for CAD modeling, e.g. CATIA and STAR-CCM+, numeric computations, e.g., Matlab and Scilab, and flow computation, e.g., OpenFOAM and visualization, e.g., ParaView (Figure 12, at the end of this paper).

The testcases are deployed on the YAWL workflow management system [START_REF] Van Der Aalst | Modern Business Process Automation: YAWL and its support environment[END_REF]. The goal is to distribute the testcases on various partners locations where the software are running (Figure 2). In order to support this distributed computing approach, an open source middleware is used [START_REF] Baude | Programming, composing, deploying for the grid[END_REF].

A first step is implemented using extensively the virtualization technologies (Figure 3), i.e., Oracle VM VirtualBox, formerly SUN's VirtualBox [START_REF]Oracle VM VirtualBox[END_REF]. This allows hands-on experiments connecting several virtual guest computers running heterogeneous software (Figure 10, at the end of this paper). These include Linux Fedora Core 12, Windows 7 and Windows XP running on a range of local workstations and laptops (Figure 11, at the end of this paper).

B. Application Workflow

In order to provide a simple and easy-to-use interface to the computing software, a workflow management system is used (Figure 2). It supports high-level graphic specification for application design, deployment, execution and monitoring. It also supports interactions among heterogeneous software components. Indeed, the 2D example testcase described in Section III.A involves several codes written in Matlab, OpenFOAM and displayed using ParaView (Figure 7). The 3D testcase involves CAD files generated using CATIA and STAR-CCM+, flow calculations using OpenFOAM, Python scripts and visualization with ParaView. Extensions allow also the use of the Scilab toolbox.

Because proprietary software are used, as well as opensource and in-house research codes, a secured network of connected computers is made available to the users, based on existing middleware (Figure 8). This network is deployed on the various partners locations throughout France. Web servers accessed through the SSH protocol are used for the proprietary software running on dedicated servers, e.g., CATIA v5 and STAR-CCM+.

An interesting feature of the YAWL workflow system is that composite workflows can be defined hierarchically [START_REF] Adams | Facilitating Flexibility and Dynamic Exception Handling in Workflows through Worklets[END_REF]. They can also invoke external software, i.e., pieces of code written in whatever language suits the users. They are called by custom YAWL services or local shell scripts. Remote Web services can also be called.

YAWL thus provides an abstraction layer that helps users design complex applications that may involve a large number of distributed components (Figure 6). Further, the workflow specifications involve possible alternative execution paths, as well as parallel branches, conditional branching and loops. Combined with the run-time addition of code with the corresponding dynamic selection procedures as well as new exception handling procedures (see Section IV), a very powerful environment is provided to the users.

IV. RESILIENCE

A. Rationale

Resilience is defined here as the ability of the applications to handle unexpected situations. Usually, hardware, communication and software failures are handled using fault-tolerance mechanisms [START_REF] Ramakrishna | VGrADS: Enabling e-Science workflows on grids and clouds with fault tolerance[END_REF]. This is the case for communication software and for middleware that take into account possible computer and network breakdowns at runtime. These mechanisms use for example data and packet replication and redundant code execution to cope with these situations [START_REF] Russel | Workflow Control Flow Patterns, A Revised View[END_REF].

However, when unexpected situations occur at run-time, very few options are usually offered to the application users: ignore them or abort the execution, reporting the errors and analyze them, to later modify and restart the applications.

B. Exception Handling

Another alternative is proposed here. It is based on the dynamic selection and exception handling mechanism featured by YAWL [START_REF] Adams | Facilitating Flexibility and Dynamic Exception Handling in Workflows through Worklets[END_REF].

It provides the users with the ability to add at run-time new rules governing the application behavior and new pieces of code that will take care of the new situations.

For example, it allows for the selection of alternative code, based on the current unexpected data values. The application can therefore evolve over time without being stopped. It can also cope later with the new situations without being altered. This refinement process is therefore lasting over time and the obsolescence of the code greatly reduced.

The new codes are defined and inserted in the application workflow using the standard specification approach used by YAWL (Figure 7). This is implemented by YAWL so-called exlets that are in charge of exception and error handling. They can be inserted and invoked at run-time in cas of task failure.

For example (Figure 24, at the end of this paper) if a workflow is specified as a sequence of tasks T0, T1 and T2 and that a failure occurs for task T1, an exlet is automatically invoked and takes the form of a dynamic insertion of a set of tasks that cope for the error (Error Handler, Restore and Ignore). It is based on a pre-defined or dynamically provided scenario by the user. The Error Handler task then triggers the Restore task or the Ignore task, based on appropriate decisions made, depending on parameters values or user interactions. In case the Restore task is invoked, the scenario then backtracks the execution of the workflow to the nearest checkpoint CHKPT before the failed task T1. In contrast, if the decision is made to ignore the error, the control is passed to the task immediately following T1 in the original scenario, i.e., T2.

Because it is important that monitoring long-running applications be closely controlled by the users, this dynamic selection and exception handling mechanism also requires a user-defined probing mechanism that provides them with the ability to suspend, evolve and restart the code dynamically.

For example, if the output pressure of an air-conditioning pipe is clearly off limits during a simulation run, the user must be able to suspend it as soon as he is aware of that situation. He can then take corrective actions, e.g., suspending the simulation, modifying some parameters or value ranges and restarting the process.

C. Fault-tolerance

The fault-tolerance mechanism provided by the underlying middleware copes with job and communication failures. Job failures or time-outs are handled by reassignment and re-execution. Communication failures are handled by re-sending appropriate messages. Also, hardware breakdowns are handled by re-assigning running jobs to other resources, implying possible data movements to the corresponding resources. This is standard for most middleware [START_REF] Baude | Programming, composing, deploying for the grid[END_REF].

D. Assymetric Checkpoints

Asymmetric checkpoints are defined by the users at significant execution locations in the application workflows. They are used to avoid the systematic insertion of checkpoints at all potential failure points. They are userdefined at specific critical locations, depending only on the application logic. Clearly, the applications designers and users are the only ones that have the expertise necessary to insert the appropriate checkpoints. In contrast with middleware fault-tolerance which can re-submit jobs and resend data packets, no automatic procedure can be implemented here. It is therefore based on a dynamically evolving set of heuristic rules.

As such, this approach significantly reduces the number of necessary checkpoints to better concentrate on only those that have a critical impact on the applications runs.

For example (Figure 4):

• The checkpoints can be chosen by the users among those that follow long-running components and large data transfers. • Alternatively, those that precede series of small components executions. The base rule set on which the asymmetric checkpoints are characterized is the following:

• R1: no output backup for specified join operations • R2: only one output backup for fork operations • R3: no intermediate result backup for user-specified sequences of operations • R4: no backup for user-specified local operations • R5: systematic backup for remote inputs This rule set can be evolved by the user dynamically, at any time during the application life-time, depending on the specific application requirements.

V. IMPLEMENTATION

A. The YAWL workflow management system

Workflows systems are the support for many e-Science applications [START_REF] Deelman | Managing Large-Scale Scientific Workflows in Distributed Environments: Experiences and Challenges[END_REF][8] [START_REF] Simmha | Building the Trident Scientific Workflow Workbench for Data Management in the Cloud[END_REF]. Among the most popular systems are Taverna, Kepler, Pegasus, Bonita and many others [START_REF] Wang | A highlevel distributed execution framework for scientific workflows[END_REF] [START_REF] Ramakrishna | VGrADS: Enabling e-Science workflows on grids and clouds with fault tolerance[END_REF]. They complement scientific software environments like Dakota, Scilab and Matlab in their ability to provide complex application factories that can be shared, reused and evolved. Further, they support the incremental composition of hierarchic composite applications. Providing a control flow approach, they also complement the usual dataflow approach used in programming toolboxes. Another bonus is that they provide seamless user interfaces, masking technicalities of distributed, programming and administrative layers, thus allowing the users and experts to concentrate on their areas of interest.

The OPALE project at INRIA [START_REF]OPALE project[END_REF] is investigating the use of the YAWL workflow management system for distributed multidiscipline optimization [START_REF] Nguyên | Dynamic Resilient Workflows for Collaborative Design[END_REF]. The goal is to develop a resilient workflow system for large-scale optimization applications. It is based on extensions to the YAWL system to add resilience and remote computing facilities for deployment on high-performance distributed infrastructures. This includes large-PC clusters connected to broadband networks. It also includes interfaces with the Scilab scientific computing toolbox [START_REF] Baudin | Introduction to Scilab[END_REF] and the middleware [START_REF] Baude | Programming, composing, deploying for the grid[END_REF]. Provided as an open-source software, YAWL is implemented in Java. It is based on an Apache server using Tomcat and Apache's Derby relational database system for persistence (Figure 5). YAWL is developed by the University of Eindhoven (NL) and the University of Brisbane (Australia). It runs on Linux, Windows and MacOS platforms [START_REF]YAWL foundation[END_REF]. It allows complex workflows to be defined and supports high-level constructs (e.g., XOR-and OR-splits and joins, loops, conditional control flow based on application variables values, composite tasks, parallel execution of multiple instances of tasks, etc) through highlevel user interfaces (Figure 10, at the end of this paper).

Formally, it is based on a sound and proven operational semantics extending the workflow patterns of the Workflow Management Coalition [START_REF]The Workflow Management Coalition[END_REF] [START_REF] Yildiz | Towards Scientific Workflow Patterns[END_REF]. It is implemented and proved by colored Petri nets. This allows for sophisticated verifications of workflow specifications at design time: fairness, termination, completeness, deadlocks, etc (Figure 5-left).

In contrast, workflow systems which are based on the Business Process Management Notation (BPMN) [START_REF]Object Management Group / Business Process Management Initiative, BPMN Specifications[END_REF] and the Business Process Execution Language (BPEL) [START_REF] Emmerich | Grid Service Orchestration using the Business Process Execution Language (BPEL)[END_REF] are usually not supported by a proven formal semantics. Further, they usually implement only specific and/or proprietary versions of the BPMN and the BPEL specifications (Figure 17, at the end of this paper). There are indeed over 73 (supposedly compliant) implementations of the BPMN, as of February 2011, and several others are currently being implemented [START_REF]Object Management Group / Business Process Management Initiative, BPMN Specifications[END_REF]. In addition, there are more than 20 existing BPEL engines. However, BPEL supports the execution of long running processes required by simulation applications, with compensation and undo actions for exception handling and fault-tolerance, as well as concurrent flows and advanced synchronization mechanisms [START_REF] Emmerich | Grid Service Orchestration using the Business Process Execution Language (BPEL)[END_REF].

Designed as an open platform, YAWL supports natively interactions with external and existing software and application codes written in any programming languages, through shell scripts invocations, as well as distributed computing through Web Services (Figure 6).

It includes a native Web Services interface, custom services invocations through codelets, as well as rules, powerful exception handling facilities, and monitoring of workflow executions [START_REF] Adams | Facilitating Flexibility and Dynamic Exception Handling in Workflows through Worklets[END_REF].

Further, it supports dynamic evolution of the applications by extensions to the existing workflows through worklets, i.e., on-line inclusion of new workflow components during execution [START_REF] Adams | The worklet custom service for YAWL, Installation and User Manual[END_REF].

It supports automatic and step-by-step execution of the workflows, as well as persistence of (possibly partial) executions of the workflows for later resuming, using its internal database system. It also features extensive event logging for later analysis, simulation, configuration and tuning of the application workflows.

Additionally, YAWL supports extensive organizations modeling, allowing complex collaborative projects and teams to be defined with sophisticated privilege management: access rights and granting capabilities to the various projects members (organized as networked teams of roles and capabilities owners) on the project workflows, down to individual components, e.g., edit, launch, pause, restart and abort workitems, as well as processing tools and facilities (Figure 5-right) [START_REF]YAWL foundation[END_REF].

Current experiments include industrial testcases, involving the connection of the Matlab, Scilab, Python, ParaView and OpenFOAM software to the YAWL platform [START_REF] Nguyên | Dynamic Resilient Workflows for Collaborative Design[END_REF]. The YAWL workflow system is used to define the optimization processes, include the testcases and control their execution: this includes reading the input data (StarCCM+ files), the automatic invocation of the external software and automatic control passing between the various application components, e.g., Matlab scripts, OpenFOAM, ParaView (Figure 11, at the end of this paper).

B. Exception handling

The exception handlers are automatically tested by the YAWL workflow engine when the corresponding tasks are invoked. This is standard in YAWL and constraint checking can be activated and deactivated by the users [START_REF] Van Der Aalst | Modern Business Process Automation: YAWL and its support environment[END_REF].

For example, if a particular workflow task WT invokes an external EXEC code through a shell script SH (Figure 7) using a standard YAWL codelet, an exception handler EX can be implemented to prevent from undesirable situations, e.g., infinite loops, unresponsive programs, long network delays, etc. Application variables can be tested, allowing for very close monitoring of the applications behavior, e.g., unexpected values, convergence rates for optimization programs, threshold transgressions, etc.

A set of rules (RDR) is defined in a standard YAWL exlet attached to the task WT and defines the exception handler EX. It is composed here of a constraint checker CK, which is automatically tested when executing the task WT. A compensation action CP triggered when a constraint is violated and a notifier RE warning the user of the exception. This is used to implement resilience (Section V. C.).

The constraint violations are defined by the users and are part of the standard exception handling mechanism provided by YAWL. They can attach sophisticated exception handlers in the form of specific exlets that are automatically triggered at runtime when particular user-defined constraints are violated. These constraints are part of the RDR attached to the workflow tasks.

Resilience is the ability for applications to handle unexpected behavior, e.g., erratic computations, abnormal result values, etc. It is inherent to the applications logic and programming. It is therefore different from systems or hardware errors and failures. The usual fault-tolerance mechanisms are therefore inappropriate here. They only cope with late symptoms, at best.

C. Resilience

Resilience is the ability for applications to handle unexpected behavior, e.g., erratic computations, abnormal result values, etc. It lies at the level of application logic and programming, not at systems or hardware level. The usual fault-tolerance mechanisms are therefore inappropriate here. They only cope with very late symptoms, at best.

New mechanisms are therefore required to handle logic discrepancies in the applications, most of which are only discovered at run-time.

It is therefore important to provide the users with powerful monitoring features and complement them with dynamic tools to evolve the applications according to the erratic behavior observed. This is supported here using the YAWL workflow system so called "dynamic selection and exception handling mechanism". It supports: • Application update using dynamically added rules specifying new codes to be executed, based on application data values, constraints and exceptions. • The persistence of these new rules to allow applications to handle correctly future occurrences of the new case. • The dynamic extension of these sets of rules.

• The definition of the new codes to be executed using the framework provided by the YAWL application specification tool: the new codes are new workflows included in the global application workflow specification.

• Component workflows invoke external programs written in any programming language through shell scripts, custom service invocations and Web Services. In order to implement resilience, two particular YAWL features are used:

• Ripple-down-rules (RDR) which are handlers for exception management, • Worklets, which are particular workflow actions to be taken when exceptions or specific events occur. The RDR define the decision process which is run to decide which worklet to use in specific circumstances.

D. Distributed workflows

The distributed workflow is based on an interface between the YAWL engine and the underlying middleware (Figure 8). At the application level, users provide a specification of the simulation applications using the YAWL Editor. It supports a high-level abstract description of the simulation processes. These processes are decomposed into components which can be other workflows or basic workitems. The basic workitems invoke executable tasks, e.g., shell scripts or custom services. These custom services are specific execution units that call user-defined YAWL services. They support interactions with external and remote codes. In this particular platform, the external services are invoked through the middleware interface.

This interface delegates the distributed execution of the remote tasks to the middleware [START_REF] Baude | Programming, composing, deploying for the grid[END_REF]. The middleware is in charge of the distributed resources allocation to the individual jobs, their scheduling, and the coordinated execution and result gathering of the individual tasks composing the jobs. It also takes in charge the fault-tolerance related to hardware, communications and system failures. The resilience, i.e., the application-level fault-tolerance is handled using the rules described in the previous Sections. The remote executions invoke the middleware functionalities through a Java API. The various modules invoked are the middleware Scheduler, the Jobs definition module and the tasks which compose the jobs. The jobs are allocated to the distributed computing resources based upon the scheduler policy. The tasks are dispatched based on the job scheduling and invoke Java executables, possibly wrapping code written in other programming languages, e.g., Matlab, Scilab, Python, or calling other programs, e.g., CATIA, STAR-CCM+, ParaView, etc.

Optionally, the workflow can invoke local tasks using shell scripts and remote tasks using Web Services. These options are standard in YAWL.

E. Secured access

In contrast with the use of middleware, there is also a need to preserve and comply with the reservation and scheduling policies on the various HPC resources and clusters that are used. This is the case for national, e.g., IDRIS and CINES in France, and transnational HPC centers, e.g., PRACE in Europe.

Because some of the software run on proprietary resources and are not publicly accessible, some privileged connections must also be implemented through secured X11 tunnels to remote high-performance clusters (Figure 13). This also allows for fast access to software needing almost real-time answers, avoiding the constraints associated with the middleware overhead. It also allows running parallel optimization software on large HPC clusters. In this perspective, a both-ways SSH tunnel infrastructure has been implemented for the invocation of remote optimization software running on high-performance clusters and for fast result gathering.

Using the specific ports used by the communication protocol (5000) and YAWL (8080), a fast communication infrastructure is implemented for remote invocation of testcase optimizers between several different locations on a high-speed (10 GB/s) network at INRIA. This is also accessible through standard Internet connections using the same secured tunnels.

Current tests have been implemented monitoring from Grenoble in France a set of optimizers software running on HPC clusters in Sophia-Antipolis near Nice. The optimizers are invoked as custom YAWL services from the application workflow. The data and results are transparently transferred through secured SSH tunnels.

In addition t the previous interfaces, direct local access to numeric software, e.g., SciLab and OpenFOAM, is always available through the standard YAWL custom services using the 8080 communication port and shell script invocations. Therefore, truly heterogeneous and distributed environments can be built here in a unified workflow framework.

F. Interfaces

To summarize, the simulation platform which is based on the YAWL workflow management system for the application specification, execution and monitoring, provides three complementary interfaces that suit all potential performance, security, portability and interoperability requirements of the current sophisticated simulation environments.

These interfaces run concurrently and are used transparently for the parallel execution of the different parts of the workflows (Figure 14). These interfaces are:

• The direct access to numeric software through YAWL custom services that invoke Java executables and shell scripts that trigger numeric software, e.g., OpenFOAM, and visualization tools, e.g., ParaView (Figure 2) • The remote access to high-performance clusters running parallel software, e.g., optimizers, through secured SSH tunnels, using remote invocations of custom services (Figure 13) • The access to wide-area networks through a grid middleware, e.g., Grid5000, for distributed resource reservation and job scheduling (Figure 9)

G. Service orchestration

The YAWL system provides a native Web service interface. This is a very powerful standard interface to distributed service execution, although it might impact HPC concerns. This is the reason why a comprehensive set of interfaces are provided by the platform (Section F, above).

Combined altogether and offered to the users, this rich set of functionalities is intended to support most application requirements, in terms of performance, heterogeneity and standardization.

Basically, an application workflow specifies general services orchestration. General services include here not only Web services, but also shell scripts, YAWL custom services implemented by Java class executables and highlevel operators, as defined in the workflow control flow patterns of the Workflow Management Coalition [START_REF] Russel | Workflow Control Flow Patterns, A Revised View[END_REF][21], e.g., AND-joins, XOR-joins, conditional branchings, etc.

The approach implemented here therefore not only fulfills sound and semantically proved operators for task specification, deployment, invocation, execution and. synchronization. It also fulfills the stringent requirements for heterogeneous distributed and HPC codes to be deployed and executed in a unified framework. This provides the users with high-level GUIs and hides the technicalities of distributed, and HPC software combination, synchronization and orchestration.

Further, because resilience mechanisms are implemented at the application level (Section C), on top of the middleware, network and OS fault-tolerance features, a secured and fault resilient HPC environment is provided, based on high-level constructs for complex and large-scale simulations [START_REF] Trifan | A Dynamic Workflow Simulation Platform[END_REF].

The interface between the workflow tasks and the actual simulation codes can therefore be implemented as Web Services, YAWL custom services, and shell scripts through secured communication channels. This is a unique set of possibilities offered by our approach (Figure 14).

H. Dataflow and control flow

The dual requirements for the dataflow and control flow properties are preserved. Both aspects are important and address different requirements [START_REF] Deelman | Managing Large-Scale Scientific Workflows in Distributed Environments: Experiences and Challenges[END_REF]. The control flow aspect addresses the need for user control over the workflow tasks execution. The dataflow aspect addresses the need for highperformance and parallel algorithms to be implemented effectively.

The control flow aspect is required in order to provide the users with control over the synchronization and execution of the various heterogeneous and remote software that run in parallel and contribute to the application results. This aspect is exemplified in the previous sections (Secttion III) where multiple software contribute to the application results and visualization. This is natively supported by YAWL.

The dataflow aspect is also preserved here in two complementary ways:

• the workflow data is transparently managed by the YAWL engine to ensure the proper synchronization, triggering and stopping of the tasks and complex operators among the different parallel branches of the workflows, e.g., AND joins, OR and XOR forks, conditional branchings. This includes a unique YAWL feature called "cancellation set" that refers to a subset of a workflow that is frozen when another designated task is triggered [START_REF] Nguyên | Dynamic Resilient Workflows for Collaborative Design[END_REF] • the data synchronization and dataflow scheme implemented by the specific numeric software invoked remain unchanged using a separation of concerns policy, as explained below The various software with dataflow dependencies are wrapped in adequate YAWL workflow tasks, so that the workflow engine does not interfere with the dataflow policies they implement.

This allows high-performance concerns to be taken into consideration along with the users concerns and expectations concerning the sophisticated algorithms associated with these programs.

Also, this preserves the global control flow approach over the applications which is necessary for heterogeneous software to cooperate in the workflow.

As a bonus, it allows user interactions during the workflow execution in order to cope with unexpected situations (Section IV). This would otherwise be very difficult to implement because when unexpected situations occur while using a pure dataflow approach, it requires stopping the running processes or threads in the midst of possibly parallel and remote running calculations, while (possibly remote) running processes are also waiting for incoming data produced by (possibly parallel and remote) erratic predecessors in the workflow. This might cause intractable situations even if the errors are due to rather simple events, e.g., network data transfers or execution timeouts.

Note that so far, because basic tasks cannot be divided into remote components in the workflow, the dataflow control is not supported between remotely located software. This also avoids large uncontrolled data transfers on the underlying network. Thus, only collocated software, i.e., using the same computing resources or running on the same cluster, can use dataflow control on the platform. They are wrapped by workflow tasks which are controlled by the YAWL engine as standard workflow tasks. For example, the dataflow controlled codes D0, D1 and D2 depicted Figure 15 are wrapped by the composite task CT which is a genuine YAWL task that invokes a shell script SH to trigger them. Specific performance improvements can therefore be expected from dataflow controlled sets of programs running on large HPC clusters. This is fully compatible with the control flow approach implemented at the application (i.e., workflow) specification level. Incidentally, this also avoids the streaming of large data collections of intermediate results through network connections. It therefore alleviates bandwidth congestion.

The platform interfaces are illustrated by Figure 16, at the end of this paper. Once the orchestration of local and distributed codes is specified at the application (workflow) level, their invocation is transparent to the user, whatever their localization.

I. Other experiments

This distributed and heterogeneous platform is also tested with the FAMOSA optimization suite developed at INRIA by project OPALE [START_REF] Duvigneau | A three-level parallelization strategy for robust design in aerodynamics[END_REF]. It is deployed on a HPC cluster and invoked from a remote workflow running on a Linux workstation (Figure 18, at the end of this paper).

FAMOSA is an acronym for "Fully Adaptive Multilevel Optimization Shape Algorithms" and includes C++ components for:

• CAD generation,

• mesh generation,

• domain partitioning,

• parallel CFD solvers using MPI, and • post-processors

The input is a design vector and the output is a set of simulation results (Figure 19, at the end of this paper). The components also include other software for mesh generation, e.g., Gmsh [START_REF]Gmsh[END_REF], partitioning, e.g., Metis [START_REF] Metis | [END_REF] and solvers, e.g., Num3sis [39]. They are remotely invoked from the YAWL application workflow by shell scripts (Figure 18). FAMOSA is currently tested by an auto manufacturer (Figure 21, at the end of this paper) and ONERA (the French National Aerospace Research Office) for aerodynamics problem solving (Figure 25 and26).

The various errors that are taken into account by the resilience algorithm include run-time errors in the solvers, inconsistent CAD and mesh generation files, and execution time-outs.

The FAMOSA components are here triggered by remote shell scripts running PBS invocations for each one on the HPC cluster. The shell scripts are called by YAWL custom service invocations from the user workflow running on the workstation (Figure 18). Additionally, another experiment described by Figure 20 illustrates the distributed simulation platform used for testing the heterogeneity of the application codes running on various hardware and software environments. It includes four remote computing resources that are connected by a high-speed network. One site is a HPC cluster (Site 4). Another site is a standard Linux server (Site 1). The two other sites are remote virtualized computing resources running Windows and Linux operating systems on different VirtualBox virtual machines that interface the underlying middleware (Sites 3 an 4). This platform has been tested against the testcases described in Section III.

VI. CONCLUSION

The requirements for large-scale simulation make it necessary to deploy various software components on heterogeneous distributed computing infrastructures [START_REF] Simon | Future directions in High-Performance Computing 2009-2018[END_REF][START_REF] Ostermann | Extending Grids with Cloud Resource Management for Scientific Computing[END_REF]. These environments are often required to be distributed among a number of project partners for administrative and collaborative purposes. This paper presents an experiment for deploying a distributed simulation platform. It uses a network of highperformance computers connected by a middleware layer. Users interact dynamically with the applications using a workflow management system. It allows them to define, deploy and control the application executions interactively.

In contrast with choreography of services, where autonomous software interact in a controlled manner, but where resilience and fault-tolerance are difficult to implement, the approach used here is an orchestration of heterogeneous and distributed software components that interact in a dynamic way under the user control, in order to contribute to the application results [START_REF] Sherp | Using UNICORE and WS-BPEL for Scientific Workfow execution in Grid Environments[END_REF]. This allows the dynamic interaction with the users in case of errors and erratic application behavior. This approach is also fully compatible with both the dataflow and control flow approaches which are often described as poorly compatible [START_REF] Ludäscher | Scientific Workflows: Business as usual ?[END_REF][31] [START_REF] Yildiz | Towards Scientific Workflow Patterns[END_REF] and are extensively used in numeric software platforms. The underlying interface to the distributed components is a middleware providing resource allocation and job scheduling [START_REF] Baude | Programming, composing, deploying for the grid[END_REF]. Because of the heterogeneity of the software and resources used, the platform also combines secured access to remote HPC clusters and local software in a unified workflow framework (Figure 20, at the end of this paper).

This approach is also proved to combine in an elegant way the dataflow control used by many HPC software and the control flow approach required by complex and distributed application execution and monitoring.

A significant bonus of this approach is that besides faulttolerance provided by the middleware, which handles communication, hardware and job failures, the users can define and handle application logic failures at the workflow specification level. This means that a new abstraction layer is introduced to cope with application-level errors at run-time. Indeed, these errors do not necessarily result from programming and design errors. They may also result from unforeseen situations, data values and limit conditions that could not be envisaged. This is often the case for simulations due to their experimental nature, e.g., discovering the behavior of the system being simulated. This provides support to resiliency using an asymmetric checkpoint mechanism. This feature allows for efficient handling mechanisms to restart only those parts of an application that are characterized by the users as necessary for overcoming erratic behavior.

Further, this approach can be evolved dynamically, i.e., when applications are running. This uses the dynamic selection and exception handling mechanism in the YAWL workflow system. It allows for new rules and new exception handling to be added on-line if unexpected situations occur at run-time.

Figure 1 .

 1 Figure 1. Flow pressure (left) and speed (right) in an air-conditioner pipe (ParaView screenshots).

Figure 2 .

 2 Figure 2. The YAWL workflow interface to the 2D testcase.

Figure 3 .

 3 Figure 3. The virtualized infrastructure.

Figure 4 .

 4 Figure 4. Asymmetric checkpoints.

Figure 5 .

 5 Figure 5. The user interfaces: YAWL Editor (left) and YAWL Control Center (right).

Figure 6 .

 6 Figure 6. YAWL architecture.

Figure 7 .

 7 Figure 7. Exception handler associated with a workflow task.

Figure 8 .

 8 Figure 8. The distributed simulation platform.

Figure 9 .

 9 Figure 9. The YAWL workflow and middleware interface.

Figure 13 .

 13 Figure 13. High-speed infrastructure for remote cluster access.

Figure 14 .

 14 Figure 14. External services interfaces.

Figure 15 .

 15 Figure 15. Dataflow tasks wrapped by a composite YAWL task.

Figure 10 .

 10 Figure 10. The YAWL testcase and workflow editor deployed on a virtual machine: Linux Fedora Core 12 host running VirtualBox and Windows XP guest.

Figure 11 .

 11 Figure 11. Parameter sweeping and YAWL interface to remote simulation codes.

Figure 12 .

 12 Figure 12. The 3D testcase visualization (ParaView screenshot).

Figure 16 .

 16 Figure 16. The platform interfaces to local and distributed codes.

Figure 17 .

 17 Figure 17. BPMN metamodel -© 2007 OMG.

Figure 18 .

 18 Figure 18. A distributed optimization experiment.

Figure 19 .

 19 Figure 19. The FAMOSA optimization suite -© 2010 Régis Duvigneau -Project OPALE.

Figure 20 .

 20 Figure 20. The distributed simulation platform.

Figure 21 .

 21 Figure 21. Mesh for vehicle aerodynamics simulation (Gmsh screenshot).

Figure 22 .

 22 Figure 22. Examples of resilience and fault-tolerance software.

Figure 23 .

 23 Figure 23. Asymmetric checkpoints software for application resilience.

Figure 24 .

 24 Figure 24. Error handler for task T1 error: YAWL exlet.

Figure 25 .

 25 Figure 25. Pressure on a NACA airfoil (OpenFOAM testcase).

Figure 26 .

 26 Figure 26. Pressure over a 2D airfoil (OpenFOAM testcase).

ACKNOWLEDGMENT This work is supported by the European Commission FP7 Cooperation Program "Transport (incl. aeronautics)", for the GRAIN Coordination and Support Action ("Greener Aeronautics International Networking"), grant ACS0-GA-2010-266184. It is also supported by the French National Research Agency ANR (Agence Nationale de la Recherche), OMD2 project (Optimisation Multi-Discipline Distribuée), grant ANR-08-COSI-007, program COSINUS (Conception et Simulation).


Fawaz Paraiso

Gabriel Hermosillo

Romain Rouvoy

Philippe Merle

Lionel Seinturier

A Middleware Platform to Federate Complex Event Processing

Keywords: CEP, SOA, middleware, federation, component

Distributed systems like crisis management are subject to the dissemination of a huge volume of heterogeneous events, ranging from low level network data to high level crisis management intelligence, depending on the role of the rescue teams involved. In such systems, Complex Event Processing (CEP) has emerged as a solution to detect and react (in real-time) to complex events, which are correlations of more primitive events. Although various CEP engines implement the support for dealing with the business heterogeneity of events, the technological integration of these events remains uncovered. Therefore, in this paper we introduce DiCEPE (Distributed Complex Event Processing Engine), a platform which focuses on the integration of CEP engines in distributed systems. DiCEPE provides a native support for various communication protocols in order to federate CEP engines and ease the deployment of complex systems-of-systems. We illustrate our proposal using a nuclear crisis management scenario and show how DiCEPE leverages the coordination and the federation of different CEP engines.

I. INTRODUCTION

Nowadays, there is a huge amount of information sources everywhere around us, and heterogeneity is the rule among them. These events are used by distributed applications such as Air-Traffic Control Systems [START_REF] Richard | Document processing and data distribution system for an air traffic control Environment[END_REF], Automated Banking Systems [START_REF] Covert | Automated banking machine and system[END_REF], Tracking Roaming Cellular Telephones [START_REF] Della-Torre | Tracking Roaming Cellular Telephones Calls For Anti-Fraud and Other Purposes[END_REF], Retail Point-of-Sale Terminals [START_REF] Christopher | Clustering of retail terminals[END_REF] or Global Positioning Systems [START_REF] Spencer | Global Positioning System: a field guide for the social sciences[END_REF], that require them to be filtered and correlated for complex pattern detection and transformed to new events that reach a semantic level appropriate for higher-level applications.

The need for real-time processing of information is relevant for many systems, as not only the content but also the context in which it was created, determine its value. Having such contextual information in real-time may substantially increase the performance of an application, improving its liveliness since it can better react to the constantly changing situations. One example of such an application can be seen in a nuclear crisis management scenario, where there are several factors that may alter the outcome, and where a lot of information needs to be analyzed and processed (e.g., weather forecasts and radiation surveys), while at the same time, several different teams (e.g., policemen, firemen, medics) are cooperating in the same zone and depend on the decisions made using that information.

One of the problems in that kind of scenarios is the overwhelming amount of data that has to be considered at the same time, which may lead to information overload. The way in which data is handled is very important, and processing it using only one Complex Event Processing (CEP) engine may be a dangerous practice [START_REF] Luckham | The Power of Events: An Introduction to Complex Event Processing in Distributed Enterprise Systems[END_REF], as it may miss some important information due to being overwhelmed by all the events. Moreover, in a crisis management scenario we can see that there are several groups that are sharing the same information to achieve a common goal, however, the way in which each of them manages and interprets the information is different from the others. This could lead to incompatibilities among the groups.

In this paper we present DiCEPE, a platform that focuses on the integration of CEP engines in distributed systems, and which is capable of using various communication protocols in order to federate CEP engines and ease the deployment of complex systems-of-systems. DiCEPE uses the configuration capabilities provided by OASIS's Service Component Architecture (SCA) [START_REF]Organization for the Advancement of Structured Information Standards.Web Services Business Process Execution Language Version 2.0 Standard[END_REF] to create a distributed environment of CEP engines that can prevent information overloading by sharing the load among different CEP engines. This approach also allows multiple domain-specific CEP engines to be included in the environment and to be managed directly by the domain experts, while at the same time sharing the global-level events and processing.

The reminder of this paper is structured as follows. In Section II we give an overview of some background concepts that we use in our proposal. Next, Section III presents the motivation of our work. In Section IV we introduce the DiCEPE platform. Then, in Section V we describe the integration of different CEP engines with DiCEPE. Next, in Section VI we present the validation of our approach. In Section VII we discuss some related work. Finally, we conclude our work in Section VIII.

II. BACKGROUND

In this section we give a brief introduction to some of the concepts and technologies that we use throughout this paper, in order to facilitate the understanding of how they are used with the DiCEPE platform.

A. Complex Event Processing

Complex Event Processing (CEP) has been proposed as a new paradigm for real-time event-based applications [START_REF] Luckham | The Power of Events: An Introduction to Complex Event Processing in Distributed Enterprise Systems[END_REF]. It is designed to match event correlations, called event patterns, from a constant flow of events, called an event stream. One key aspect of CEP is the ability to define reactive rules corresponding to event pattern definition and triggering a process in real-time when an event pattern is detected. CEP is used in a wide diversity of applications, which have to deal with voluminous streams of incoming data, the complexity of processing and time response [START_REF]SL RTView platform[END_REF], [22], [START_REF]WSO2 Complex Event Processing Server[END_REF]. There are many domains in which CEP is actually used, like network management, traffic monitoring, or fraud detection, and it has drawn the attention of many research projects [START_REF] Chakravarthy | Stream data processing: a quality of service perspective : modeling, scheduling, load shedding, and complex event processing[END_REF], [START_REF] Kasi | Cost analysis for complex innetwork event processing in heterogeneous wireless sensor networks[END_REF] and commercial products [START_REF]SL RTView platform[END_REF], [22].

B. SOA and SCA

Service-Oriented Architecture (SOA) is a paradigm for the realization and maintenance of business processes that span large distributed systems [START_REF]Reference Model for Service Oriented Architecture 1.0[END_REF]. It is independent of the development technology and comprises loosely coupled, highly inter-operable application services, which interoperate based on a formal interface definition independent of the underlying platforms and programming languages. The DiCEPE platform uses a programming model based on the SOA paradigm, called Service Component Architecture (SCA).

SCA [START_REF]Organization for the Advancement of Structured Information Standards.Web Services Business Process Execution Language Version 2.0 Standard[END_REF] is a set of specifications for building distributed applications and systems using SOA. It is neutral with respect to programming languages, and can be implemented in any language that supports it. SCA targets the heterogeneous composition of various interface definition languages (e.g., WSDL, Java), implementation technologies (e.g., Java, Spring, BPEL, JavaEE, C++, COBOL, C), and binding technologies (e.g., Web Services, JMS). As illustrated in Figure 1, the basic SCA building blocks are software components [START_REF] Szyperski | Component Software: Beyond Object-Oriented Programming[END_REF], which provide services, re-quire references and expose properties. The references and services are connected by wires. SCA specifies a hierarchical component model, which means that components can be implemented either by primitive language entities or by subcomponents. In the latter case the components are called composites. Any provided services or required references contained within a composite can be exposed by the composite itself by means of promotion links. To support serviceoriented interactions via different communication protocols, SCA provides the notion of binding. For SCA references, a binding describes the access mechanism used to invoke a remote service. In the case of services, a binding describes the access mechanism that clients use to invoke the service. DiCEPE is implemented on top of an SCA platform called FraSCAti, which we describe below.

C. FraSCAti

FraSCAti [START_REF] Seinturier | A Component-Based Middleware Platform for Reconfigurable Service-Oriented Architectures[END_REF] is an open source platform for deploying and executing SCA-based applications. It introduces reflective capabilities to the SCA programming model, and allows dynamic introspection and reconfiguration via a specialization of the Fractal component model [START_REF] Bruneton | The FRACTAL component model and its support in Java: Experiences with Auto-adaptive and Reconfigurable Systems[END_REF]. These features open new perspectives for bringing agility to SOA and for the run-time management of SCA applications. The platform itself is built as an SCA application (its different subsystems are implemented as SCA components). This provides an homogeneous view of a middleware software stack where the platform, the non-functional services, and the applications are uniformly designed and implemented with the same component-based and service-oriented paradigm. With FraSCAti, the structure of an SCA application can be discovered at run-time using introspection, and modified dynamically to add new services, or even reconfigured to take into account new operating conditions.

The component-based approach of the FraSCAti platform has the following characteristics:

• Reusability: Components can be re-used in different composites.

• Substitutability: Alternative implementations are easy to insert, specified interfaces are available, run-time component replacement mechanisms exist, and it has the ability to verify and validate substitutions.

• Composability: A subset of different components can be combined to create complex functional solutions that meet application requirements and system constraints, without the need to load all the existing components every time.

• Extensibility: The platform can be easily extended by adding components that provide additional functionality in a seamless way. Finally, FraSCAti provides a unified way to build applications that communicate among each other in an heterogeneous way, as it supports a wide variety of communication protocols (e.g. Web Services, REST, JMS, JNA, UPnP).

III. MOTIVATION

An emerging characteristic of event processing is its ubiquity: events are everywhere. An example of this can be found in a nuclear crisis management scenario. In this scenario, the multiplicity and diversity of the actors involved, as well as the volume and heterogeneity of information, the critical dependencies between actions, and the dynamics of the scenario create complex situations. These situations can be modeled as patterns that represent topics of interest which need to be detected nearly in real-time (e.g., evacuation of the perimeter, distribution of iodine capsules, information to the public by the media). Events from heterogeneous sources (e.g., firemen, policemen, army) can be combined to detect further situations of interest (i.e., complex events).

For instance, let us consider a scenario where, after a nuclear crisis alert, the evacuation of the population goes smoothly. There are no victims, there is no change in the force or direction of the wind, no rain, nor any alert regarding high radioactive levels. Suddenly, the nuclear plant teams detect a radioactive leak, thanks to an alert given by a high pressure sensor. The throttle valve is open and is not responding to the remote commands to get it closed. The teams realize that there is a risk of radioactive gas leaking into the atmosphere, so they alert the manager of the nuclear plant. This use case is studied in the context of the ANR SocEDA project1 .

Figure 2. Nuclear Crisis Scenario

In this case, we are particularly interested in decision and operation activities that occur outside the plant, as illustrated in Figure 2. The manager of the nuclear plant informs the representative of the national authority, who activates the Emergency Intervention Plan [START_REF]Crisis Management Case Study[END_REF]. The manager also informs the representative of the electric power company, and a crisis cell is formed, led by the local authority. The crisis cell alerts the field actors (e.g., firemen, policemen, army, Emergency Medical Services) and calls for support from the Radiation Survey Network (RSN) and the National Weather Service (NWS) for measurements. The crisis cell also alerts the media and sets off the siren so that the population can learn that they have to stay indoors and listen to the media. The field actors are deployed and regular updates are given to the media, so that the population may remain informed.

There are several factors that may affect the way in which the crisis management is carried out, and so, in order to make the correct decisions during the execution of the Emergency Intervention Plan in this scenario, we need a continuous monitoring of the following information: radioactivity level, wind force, wind direction and precipitation. A change in the wind speed or direction would mean that the evacuation perimeter and routes would have to be readjusted. A change in the radioactivity levels may trigger additional security measures to be considered and more aggressive actions to be taken into account.

As for the actors of the scenario, we can separate them in two main groups according to their role: Management and Operation. The Management group is in charge of evaluating all the information surrounding the situation, analyzing the possible solutions and deciding the better way to proceed. It is formed by the local authority and the members of the crisis cell. These actors are the heads of the whole process and are only there to manage the situation. They do not interact directly with the crisis site, however, any situation that may require a change in the normal execution of the plant has to be analyzed and decided by them. On the other hand, the actors in the Operation group are the ones that interact directly with the crisis site, such as policemen, firemen, the army, and the Emergency Medical Services. They execute any orders received from the Management group. Even through they have a pre-established sequence of tasks to execute, their behavior can be modified by a decision making actor if the situation changes.

Each subgroup of actors (e.g., the army or the police) will have their own CEP engine to manage the events related to their domain. A domain represents a complete run-time configuration, potentially distributed over multiple run-time nodes where statement rules were managed and processed. Each subgroup creates their own rules and complex events according to the goals of their domain. The problem is that when different actors are working together to achieve a common goal, we can no longer maintain a separate event management for each domain. In this case, we need to combine the events from all the actors involved in order to improve the performance of the collaboration, but at the same time, we need to keep the events that only concern a single domain from spamming the global event management, and leave the control of domain-specific rules and events to be managed separately, as is illustrated in Figure 3. Moreover, we also need to allow some sort of hierarchy, so that the Management group may receive the events that it needs for the decision making process, from the Operation groups and from external sources (e.g., RSN, NWS), and be able to modify the global event management of the Operation groups according to their decisions, without interfering with domain-specific rules.

Given the presented scenario, we find that there are many challenges that need to be faced, in order to accomplish a successful federation of complex event processing systems.

• Communication heterogeneity: The nuclear crisis management puts together a large number of actors and CEP engines that need to interact through the network, according to the scenario previously described (see Figure 2). The communication among them is essential, and for different circumstances we may need different types of communication protocols. For instance, when interacting in a private network, an asynchronous protocol, such as JMS, may be much faster and as reliable as a synchronous one. However, when leaving the private network, it may be blocked due to firewall restrictions, and its reliability may suffer as well. In that case, a synchronous approach like REST could become useful.

• Heterogeneous CEP: The actors have different tasks which may generate multiple complex events. These events can then be processed by other CEP engines that may belong to the same or to different domains. Each group of actors can have different kinds of complex event processing engines. In this point, the challenge consists in supporting the interaction among heterogeneous complex event processing engines.

• Scalability: This distributed system will process a huge volume of events, ranging from low level network data to high level crisis management intelligence. Enabling the efficient detection of such situations in real-time and under critical conditions is crucial.

• Adaptability: In the scenario, the Emergency Interven-tion Plan can change during the crisis. Therefore, the possibility to deploy new rules at run-time is essential for the correct federation of complex event processing systems. The domain-specific and engine-constraint rules will ensure the processing of new conditions in the network nodes. This paper brings forward a solution to these challenges, called DiCEPE (Distributed Complex Event Processing Engine). With DiCEPE, we allow the federation of distributed CEP engines that can deal with the heterogeneity of the event's sources and communication protocols, address scalability, provide the means to hierarchically control, and adapt the rules of the different CEP engines in the distributed environment. This approach will allow a domain-specific engine to be managed by the experts of the domain, while providing at the same time a way to create collaborations among different domains.

IV. THE DICEPE PLATFORM

In this section we present the DiCEPE platform. We begin by giving an overview of the solution and then we present the platform's architecture. Finally, in the following section we will describe how DiCEPE can integrate different CEP engines, thanks to its architecture.

A. DiCEPE Overview

DiCEPE is based on the concepts introduced by the Event Processing Network (EPN) [START_REF] Luckham | The Power of Events: An Introduction to Complex Event Processing in Distributed Enterprise Systems[END_REF], however it evolves those concepts by incorporating the advantages of SCA. DiCEPE inherits the flexibility and adaptation facilities provided by FraSCAti, which allow it to provide a complete solution to federate complex event processing systems.

To understand the concepts that rely under the construction of DiCEPE, we begin by explaining how an EPN works. An EPN is a conceptual model, that describes a set of Event Processing Agents (EPA), Event Producers and Event Consumers all connected by a set of Event Channels (EC) [START_REF] Etzion | Event Processing in Action[END_REF]. The event producers are entities that create events, while the event consumers are the entities that receive such events. The EPA behaves as both, consuming events generated by producers, and then forwarding them or creating new events to other consumers. The goal of the EPA is to filter, match and derivate events, according to the business rules. A complex event processing engine is an example of an EPA. Basically, the EPN describes how events received from producers are processed by the agents, who perform transformations, validations or enrichment on them, and finally directed to the consumers. An example of such a network is presented in Figure 4.

The EPN is an abstract model, since it abstracts the features of the input, processing and output elements of an event processing system. Guided by these concepts, our DiCEPE architecture identifies the abstract elements or components and then provides a concrete and flexible way to federate complex event processing engines. In our approach, all the components of the EPN are transformed into DiCEPE components, the different instances of EPA are wrapped by a DiCEPE composite and the communication among them is managed using SCA bindings, as is presented in Figure 5. This approach allows us to overcome some of the weak points of the EPN model, such as heterogeneity and adaptability, as we will present in the following sections.

B. DICEPE Architecture

DiCEPE is a platform for federating CEP engines, with synchronous and asynchronous communication protocols. Its architecture is based on the EPN, but it improves it by filling the gaps that are still present in the EPN model. Since our solution deals with many CEP engines, we describe it in terms of a concrete SOA platform. The proposed solution is implemented on top of FraSCAti [START_REF] Seinturier | A Component-Based Middleware Platform for Reconfigurable Service-Oriented Architectures[END_REF]. However, most of the concepts used in DiCEPE are generic, and we believe a similar implementation can be done on top of other component frameworks as well. The architecture of DiCEPE is composed of four parts: (i) Engine, (ii) Statement, (iii) Listener, and (iv) Context; as presented in Figure 6.

• Engine: This component acts as the engine instance, by which Statement components, events, and outputs (Listener component) are registered.

• Statement: A Statement component is used for quering the inbound event streams. This component is registered within the Engine component. The Engine component is connected to one or many Statement components. • Context: A Context component collects information, like the number of statement rules deployed in the engine at run-time. In addition, DiCEPE supports both synchronous and asynchronous communication. This is useful, for example in the crisis scenario, where in a private network the JMS protocol is sometimes more appropriate for communication among CEP engines given its speed and reliability [START_REF] Eggen | Efficiency of Soap Versus JMS[END_REF]. But, when we scale the scenario to a city using the Internet, the REST protocol is the better suited, since it is less likely to get blocked by a Firewall (since it uses HTTP), and also has great reliability. To retrieve information from the NWS and the RSN, we can use Web Services. There is no universal communication protocol that takes into consideration all those aspects. However, the DiCEPE platform provides access to several communication protocols which can be used as needed.

To summarize, by allowing the use of both communication protocols, synchronous and asynchronous, DiCEPE provides a solution to the Communication heterogeneity challenge identified in Section III.

The deployment of statement rules at run-time is very important for the crisis management plan, because it can change depending on the circumstances. The DiCEPE platform inherits its reconfiguration capabilities from FraS-CAti [START_REF] Seinturier | A Component-Based Middleware Platform for Reconfigurable Service-Oriented Architectures[END_REF], so a component can be stopped, modified and restarted at run-time, without service downtimes. Therefore a domain-specific rule can be changed and deployed at runtime as well, allowing it to be adapted to the new context. With this capability, DiCEPE addresses the Adaptability challenge presented in Section III.

V. INTEGRATION

This section describes how the integration of different CEP engines with the DiCEPE platform is achieved. As described in Section II, FraSCAti provides a componentbased programming model which simplifies the development, assembly, deployment and management of composite applications. The DiCEPE platform facilitates the integration of complex event processing engines. Moreover, the DiCEPE platform also encapsulates the event processing technology, so that each node in the network is running a DiCEPE runtime environment. In the following subsection we describe the integration of DiCEPE with two open source CEP engines (Esper, Etalis).

A. Integration with the Esper CEP Engine

Esper 2 is an open source engine that combines Event Stream Processing (ESP) and CEP capabilities. It is available as Java source and C# .Net source (NEsper). Their Event Processing Language (EPL) is used to express filtering, aggregation and joins, over multiple event streams, while the pattern language is used to define more complex patterns on different types of events. Esper also includes a historical data access layer to connect to the most popular databases, making the possible to combine historical data with real time data in one single query.

To describe how the Esper engine is integrated into DiCEPE, it's important to understand Esper's architecture, which is shown in Figure 7. 1) The EPServiceProvider, which acts as the Engine.

2) The Configuration, which sets the engine configurations.

3) The Event object, which is an object that represents an event. 4) The EPLStatement, which are queries written in EPL. 5) The UpdateListener, which receives updated data as soon as it is processed by the statement. These elements are integrated into DiCEPE in the following way: The Engine composite encapsulates both the EsperServiceProvider (1) and the Configuration components (2), having the EventObject as property (3). The Statement component is associated to the EPLStatement (4). Finally, the Listener component represents the UpdateListener [START_REF] Etzion | Event Processing in Action[END_REF]. The Engine component exposes the engine services with different communication protocols (e.g., REST, WS-Notification, JMS).

All the relationships in an SCA composite are expressed using the SCA Assembly Model [START_REF]SCA Service Component Architecture -Assembly Model Specification[END_REF]. Here is a slightly simplified example of how this file might look for the integration of the Esper CEP engine shown above: Like all SCA Assembly Model configurations, this one wraps its contents in an SCA composite (cf. lines 5-46). In the example shown here, a component element describes each of the two components in this composite. The first component is implemented in Java as indicated in lines 7-9. The second is embedded as a component as indicated in lines 30-31. Even though each component has it's services (cf. lines 18-25) and references (cf. lines 32-44) shown in the diagram, some are explicitly specified in these component elements, while others are not. Instead, the FraSCAti runtime can discover them and then choose the appropriate wires.

After both components have been defined, the service provided by the composite itself is specified using the service element (cf. line 18).

B. Integration with the Etalis CEP Engine

Etalis 3 is an open source system for Complex Event Processing which includes two languages: the Etalis Language for Events (ELE) and the Event Processing SPARQL (EP-SPARQL). The Etalis CEP engine is implemented in Prolog, and supports reasoning about events, contexts, and real-time complex situations (i.e., Knowledge-based Event Processing).

The DiCEPE platform provides a convenient way to access native libraries with pure Java code, using the binding of JNA in FraSCAti with the Prolog environment. In our case, we built an adapter component for the Etalis Engine, which provides a convenient way for getting events in and out of the Etalis Prolog core, as shown in Figure 8. The Engine component embeds the PrologEngineWrapper, while the Listener component is associated to the EtalisEventListener.

VI. VALIDATION

This section describes our simulation results, which study the heterogeneity of events and their interactions via different communication protocols, as well as the performance and scalability of DiCEPE. Given the complexity of crisis management and to facilitate the understanding, we focus on one single scenario.

A. Context

This use case consists in the construction of a simulation tool to validate the DiCEPE platform. The simulation tool was created based on Siafu 4 , an open source context simulator written in Java, which allows to integrate real and simulated contexts to enrich the testing capabilities. To obtain a simulated context, we need to follow three steps. First, the behavior of individual agents is modeled in a separate agent model. Then, the environment is handled in a world model, including possible random events. Finally, we get a context model which is used for defining how context data is simulated. In our simulation tool we designed a small city, where we have police stations, fire stations and an army camp. There are also two nuclear centrals nearby which have several sensors attached to them. Finally, the National Weather Service (NWS) and the Radiation Survey Network (RSN) are also reachable for information. The case study shown in Section III (cf. Figure 2) illustrates this scenario.

To evaluate our solution, we try to find an answer to the following questions: (i) Does DiCEPE introduce much overhead? (ii) What is the scalability of the system? To focus on the real performance of the DiCEPE platform, without the overhead of Internet lag, all the benchmark experiments were performed on a local machine, using an HP Z4000 Workstation with a 2.67 GHz Intel(R) Xeon processor, 16 GB RAM, Ubuntu Server 3.0.0-12 64 bit and Oracle Java 1.6. In addition, another version of this implementation was also deployed to a cloud environment, as described in the following subsection.

B. DiCEPE deployment

Our solution was deployed on a full cloud environment, that is publicly accessible on the Web5 . For this, we used a public Platform as a Service (PaaS) provider, called CloudBees6 , which is basically a Virtual Machine running a Linux distribution, with a Java virtual machine and a Web application container, all hosted in the cloud and offered as a service. In this case, the PaaS hardware resources (e.g., RAM, CPU, Storage) are provided by an external Infrastructure as a Service (IaaS) provider, independent from the PaaS. In Table I we describe the characteristics of the PaaS that we used to deploy the different instances of the DiCEPE platform that are required to run our scenario 7 .

C. DiCEPE Cost Analysis

To evaluate the overhead of the DiCEPE platform, 1,001,946 events were collected from our simulation tools, which represents about 1GB of data to be processed by the CEP engines. We then fed these events to DiCEPE in order to determine which houses were not covered by the firemen. We evaluated two implementations of this scenario where the events were processed: i) natively with the Esper8 CEP engine, and ii) with DiCEPE integrating Esper and implemented on top of FRASCATI 1.5. The scenario was executed ten times on each of the two implementations. As shown in Figure 9, the execution time of each request is stable for both implementations, which means that they are both deterministic.

In Table II, we present the results of the average execution time for each implementation, as well as the mean overhead introduced by the SCA run-time used by DiCEPE. This benchmark shows that there is an overhead introduced by adding the FraSCAti layer, the execution time is still acceptable and the benefits provided by the platform (cf. Section II) outweigh the difference in the execution time. Even if FraSCAti could still be optimized in order to reduce its overhead, we can already consider that DiCEPE is ready for intensive real-time distributed complex event processing applications.

D. DiCEPE Scalability

Our second evaluation of DiCEPE was lead towards scalability. For this, we used the scenario of our case study (cf. Section III), and we wrote a benchmark that mines firemen events. We simulated several firemen inside one big Fire Station using our simulation tool. Each fireman had devices which received notifications and sent events to their DiCEPE platform, which embedded one CEP engine (for this benchmark the CEP engine used was Esper). The firemen's DiCEPE received events, processed them and generated complex events, which were sent to the DiCEPE broker. This is done for the rest of the actors as well, as shown in Figure 10. The goal of the broker is to process the events of the specialized DiCEPE entities to find information which is relevant to the whole situation. Also, thanks to the capabilities provided by DiCEPE to deal with heterogeneous communication, all the entities can interact using different types of communication. To evaluate this scenario, we used two different datasets of event generators: one with 10,000 (see Figure 11) and a second one with 15,000 (see Figure 12), which generated around 500,000 and 750,000 events respectively (50 events per generator, without considering the start and finish notifications). As can be seen in Table III, the processing time for each event remained stable and very low during both benchmarks (around a 10th of a millisecond), despite the fact that the average number of simultaneous sessions had a significant increase of about 50% (from 89 with the first dataset, to 135 with the second). It should also be noted that there were no failures nor events lost during the executions of these tests. Overall, these benchmarks show that the response time is negligible despite the scalability, knowing that in real life we could not have more than 10,000 firemen inside one single Fire Station. For this case study we did not use any cache, which means that the response time can also be improved. With this data, we may conclude that the DiCEPE platform scales well, which addresses the Scalability challenge presented in Section III. We already introduced the Event Processing Network (EPN) in Section IV, and explained how the DiCEPE architecture is based on its four components: the Event Producer (EP), the Event Consumer, the Event Processing Agent (EPA), and the Event Channel [START_REF] Rabinovich | Analyzing the behavior of event processing applications[END_REF], [START_REF] Sharon | Event-processing network model and implementation[END_REF]. However, the EPN model is abstract and only contemplates the independent interaction of isolated EPAs, and does not take into account a distributed architecture.

There are already some works that start dealing with Distributed CEP. For instance, in [START_REF] Schultz-Møller | Distributed complex event processing with query rewriting[END_REF] the authors present a distributed approach for distributed event detection, however they do not deal with CEP heterogeneity, and they do not provide the dynamic rule adaptation provided by DiCEPE. Another approach was proposed by [START_REF] Schilling | Distributed heterogeneous event processing: enhancing scalability and interoperability of CEP in an industrial context[END_REF], which takes into account the heterogeneity of event processing engines in their distribution techniques. However, their solution does not address scalability and does not provide a real way to interconnect heterogeneous components within a common platform.

CEP systems in general, both open research and commercial, propose an event channel to connect the event consumers, providers and processing agents [START_REF] Luckham | Event Processing Glossary -Version 1[END_REF]. However, none of them provide synchronous and asynchronous communication protocols, and as we previously presented in this paper, distributed systems perform better when both protocols are available. Communication in distributed systems is always based on low-level message passing, as offered by the underlying network. DiCEPE makes it easier to deal with the numerous levels of communication as well as with all the issues involved since it supports both, synchronous and asynchronous communication protocols (e.g., JMS, SOAP, REST).

VIII. CONCLUSION

In this article we presented DiCEPE, a platform that offers interoperability for Distributed Complex Event Processing engines, via federation. This platform focuses on providing a very flexible component architecture, which supports the interaction of different complex event processing engines simultaneously, while enabling communication among them with a distributed system and deployment. Thanks to its distributed nature, DiCEPE also offers real scalability, and the evaluation results show that, despite the additional overhead generated by SCA, the execution times are still acceptable.

As of future work, we plan to continue our research in the following directions. First, since the Event Query Language (EQL) is not the same for all CEP engines solutions, we will integrate a Domain Specific Language (DSL) to express statement rules, with which we will be able to translate the DSL to the CEP specific language. Second, in the case of a geographically distributed CEP, the cloud is a good candidate. However, there are still some vendor lock-in problems with the existing solutions. This means that the deployment of the DiCEPE platform on heterogeneous cloud environments, either PaaS (Platform as a Service) or IaaS (Infrastructure as a Service), is still a challenge. As a third direction, we plan to add error handling capabilities to the DiCEPE platform to deal with distributed environments. Finally, we plan to design and evaluate a variety of distributed systems in heterogeneous cloud environments with large scale scenarios.

Figure 1 .

 1 Figure 1. Overview of an SCA Diagram

Figure 3 .

 3 Figure 3. Communication between domains

Figure 4 .

 4 Figure 4. Event Processing Network

Figure 5 .

 5 Figure 5. Distributed Complex Event Processing Engine

Figure 6 .

 6 Figure 6. Overview of the DiCEPE Architecture

Figure 7 .

 7 Figure 7. Overview of Esper Engine Architecture

Figure 8 .

 8 Figure 8. Overview of the Etalis adapter

Figure 9 .

 9 Figure 9. Cost analysis

Figure 10 .

 10 Figure 10. Distributed CEP engine in a crisis management

Figure 11 .

 11 Figure 11. Response time during simulation with 10,000 firemen

Table II EXECUTION

 II TIME AND OVERHEAD

	Implementation	Avg. exec. time	SCA overhead
	Esper	27 sec	-
	DiCEPE (Esper + FraSCAti)	30 sec	11%

SocEDA (SOCial Event Driven Architecture) http://tinyurl.com/SocEDA

http://esper.codehaus.org

http://code.google.com/p/etalis

http://siafusimulator.sourceforge.net

http://dicepe-broker.soceda.cloudbees.net

http://www.cloudbees.com

The IaaS hardware information (CPU, RAM, Storage and OS) are not available to end users.

http://esper.codehaus.org

ACKNOWLEDGMENT

This work is partially funded by the French Ministry of Higher Education and Research, Nord-Pas de Calais Regional Council and FEDER through the Contrat de Projets Etat Region Campus Intelligence Ambiante (CPER-CIA) 2007-2013, and the ANR (French National Research Agency) ARPEGE SocEDA project.