While detailed resource usage monitoring is possible on the low-level using proper tools, associating such usage with higher-level abstractions in the application layer that actually cause the resource usage in the first place presents a number of challenges. Suppose a large-scale scientific data analysis workflow is run using a distributed execution environment such as a compute cluster or cloud environment and we want to analyze the I/O behaviour of it to find and alleviate potential bottlenecks. Different tasks of the workflow can be assigned to arbitrary compute nodes and may even share the same compute nodes. Thus, locally observed resource usage is not directly associated with the individual workflow tasks. By acquiring resource usage profiles of the involved nodes, we seek to correlate the trace data to the workflow and its individual tasks. To accomplish that, we select the proper set of metadata associated with low-level traces that let us associate them with higher-level task information obtained from log files of the workflow execution as well as the job management using a task orchestrator such as Kubernetes with its container management. Ensuring a proper information chain allows the classification of observed I/O on a logical task level and may reveal the most costly or inefficient tasks of a scientific workflow that are most promising for optimization.

Segmented regression is a statistical method that approximates a function $f$ by a piecewise function $\hat{f}$ using noisy data samples. Min-$\epsilon$ approaches aim to reduce the regression function’s mean squared error (MSE) for a given number of $k$ segments. An optimal solution for min-$\epsilon$ segmented regression is found in $\mathcal{O}(n^2)$ time (Bai & Perron, 1998; Yamamoto & Perron, 2013) for $n$ samples. For large datasets, current heuristics improve time complexity to $\mathcal{O}(n\log{n})$ (Acharya et al., 2016) but can result in large errors, especially when exactly $k$ segments are used. We present a method for min-$\epsilon$ segmented regression that combines the scalability of top existing heuristic solutions with a statistical efficiency similar to the optimal solution. This is achieved by using a new method to merge an initial set of segments using precomputed matrices from samples, allowing both merging and error calculation in constant time. Our approach, using the same samples and parameter $k$, produces segments with up to 1,000 times lower MSE compared to Acharya et al. (2016) in about 100 times less runtime on data sets over $10^4$ samples.

In recent years, scientific workflows have become increasingly popular. However, their tasks are often seen as black boxes, making it difficult to optimize them or identify bottlenecks due to the complex relationships between tasks. Several factors impact task progress, including input data availability, computing power, data transfer speed and network connectivity. During task execution, resource requirements may change significantly. We propose a new method to model task requirements over their lifetime. Using these models, we predict resource consumption over time and execution duration based on a given allocation strategy with low overhead. This method enables computationally simple and fast performance predictions, including bottleneck analysis during workflow runtime. We derive a piecewise-defined bottleneck function from the discrete intersections of the task models' limiting functions. This allows us to predict potential performance gains when mitigating bottlenecks and aids in better resource allocation and workflow execution.

ABSTRACT The positional population count operation pospopcnt counts for an array of ‐bit words how often each of the bits was set. Various applications in bioinformatics, database engineering, and digital processing exist. Building on earlier work by Klarqvist et al., we show how positional population counts can be rapidly computed using SIMD techniques with good performance from the first byte, approaching memory‐bound speeds for input arrays of as little as 4 KiB. Improvements include an improved algorithm structure, better handling of unaligned and very short arrays, as well as faster bit‐parallel accumulation of intermediate results. We provide a generic algorithm description as well as implementations for various SIMD instruction set extensions, including Intel AVX2, AVX‐512, and ARM ASIMD, and discuss the adaptation of our algorithm to other platforms.

Correlating detailed local resource utilization data with the high-level concepts of distributed scientific workflow systems eventually causing it is challenging. When running a large-scale scientific data analysis workflow across a distributed execution environment, we want to analyze its I/O behaviour to identify potential bottlenecks. Since tasks are assigned to any available nodes, local resource usage on a node does not directly show which tasks are causing it. We acquire resource usage profiles of the involved nodes to link them to the individual workflow tasks. This is done by properly associating low-level trace metadata with high-level task information from log files and job management systems like Kubernetes. This information helps identifying areas of the workflow on a logical task level where improvements can make the biggest impact.

Porting a scientific data analysis workflow (DAW) to a cluster infrastructure, a new software stack, or even only a new dataset with some notably different properties is often challenging. Despite the structured definition of the steps (tasks) and their interdependencies during a complex data analysis in the DAW specification, relevant assumptions may remain unspecified and implicit. Such hidden assumptions often lead to crashing tasks without a reasonable error message, poor performance in general, non-terminating executions, or silent wrong results of the DAW, to name only a few possible consequences. Searching for the causes of such errors and drawbacks in a distributed compute cluster managed by a complex infrastructure stack, where DAWs for large datasets typically are executed, can be tedious and time-consuming. We propose validity constraints (VCs) as a new concept for DAW languages to alleviate this situation. A VC is a constraint specifying logical conditions that must be fulfilled at certain times for DAW executions to be valid. When defined together with a DAW, VCs help to improve the portability, adaptability, and reusability of DAWs by making implicit assumptions explicit. Once specified, VCs can be controlled automatically by the DAW infrastructure, and violations can lead to meaningful error messages and graceful behaviour (e.g., termination or invocation of repair mechanisms). We provide a broad list of possible VCs, classify them along multiple dimensions, and compare them to similar concepts one can find in related fields. We also provide a proof-of-concept implementation for the workflow system Nextflow.

Scientific workflows have received increasing interest and are used in many scientific fields to gather, analyze, and process significant amounts of data. However, their tasks are usually treated as black boxes, and their behavior remains unconsidered for resource allocations, which can lead to subpar resource allocations with typical scheduling. Although not done yet, it should be possible to observe such tasks, learn their behavior, and use this knowledge to improve future executions. As workflows and their tasks are often executed multiple times on a massive scale, even a slight improvement per execution may save hours of execution time and significant amounts of energy.To achieve this goal, we develop an innovative approach to model task executions and predict resource usage. The prediction is embedded in a feedback loop to repeatedly improve the models and to closely track workflow executions to make predictions and resource allocations accurate.

In the recent years, scientific workflows gained more and more popularity. In scientific workflows, tasks are typically treated as black boxes. Dealing with their complex interrelations to identify optimization potentials and bottlenecks is therefore inherently hard. The progress of a scientific workflow depends on several factors, including the available input data, the available computational power, and the I/O and network bandwidth. Here, we tackle the problem of predicting the workflow progress with very low overhead. To this end, we look at suitable formalizations for the key parameters and their interactions which are sufficiently flexible to de scribe the input da ta consumption, the computational effort and the output production of the workflow’s tasks. At the same time they allow for computationally simple and fast performance predictions, including a bottleneck analysis over the workflow runtime. A piecewise-defined bottleneck function is derived from the discrete intersections of the task models’ limiting functions. This allows to estimate potential performance gains from overcoming the bottlenecks and can be used as a basis for optimized resource allocation and workflow execution.

Checkpoint/restart (C/R) makes large-scale parallel jobs resilient against multiple node failures but typically takes considerable time and storage space. Efficient C/R strategies try to gain high levels of fault-tolerance while keeping the involved I/O and computation low. By combining XOR and partner checkpointing, two relatively weak C/R strategies, we develop and evaluate a stable, scalable, and fast C/R approach (including initialization, checkpointing, version consensus, and recovery mechanisms) that outperforms other C/R methods such as Reed-Solomon checkpointing in terms of stability and performance.

We present a novel way to judge the performance of IDA* heuristics. With this measure of heuristic quality η, different heuristics for the same problem space can be compared objectively without regards to a particular problem instance. We show how η can be used to model the performance expectations of PDB heuristics. By drawing histograms of the contributions of different parts of the search space to η, we show what parts are most critical to the quality of a heuristic and contribute to the long-standing question on what h values are most critical to the performance of an IDA* heuristic.

Summary Achieving efficient many‐to‐many communication on a given network topology is a challenging task when many data streams from different sources have to be scattered concurrently to many destinations with low variance in arrival times. In such scenarios, it is critical to saturate but not to congest the bisectional bandwidth of the network topology in order to achieve a good aggregate throughput. When there are many concurrent point‐to‐point connections, the communication pattern needs to be dynamically scheduled in a fine‐grained manner to avoid network congestion (links, switches), overload in the node's incoming links, and receive buffer overflow. Motivated by the use case of the Compressed Baryonic Matter experiment (CBM), we study the performance and variance of such communication patterns on a Cray XC40 with different routing schemes and scheduling approaches. We present a distributed Data Flow Scheduler (DFS) that reduces the variance of arrival times from all sources at least 30 times and increases the achieved aggregate bandwidth by up to 50%.

Building consensus sequences based on distributed, fault-tolerant consensus, as used for replicated state machines, typically requires a separate distributed state for every new consensus instance. Allocating and maintaining this state causes significant overhead. In particular, freeing the distributed, outdated states in a fault-tolerant way is not trivial and adds further complexity and cost to the system. In this article, we propose an extension to the single-decree Paxos protocol that can learn a sequence of consensus decisions `in-place', i.e., with a single set of distributed states. Our protocol does not require dynamic log structures and hence has no need for distributed log pruning, snapshotting, compaction, or dynamic resource allocation. The protocol builds a fault-tolerant atomic register that supports arbitrary read-modify-write operations. We use the concept of consistent quorums to detect whether the previous consensus still needs to be consolidated or is already finished so that the next consensus value can be safely proposed. Reading a consolidated consensus is done without state modifications and is thereby free of concurrency control and demand for serialisation. A proposer that is not interrupted reaches agreement on consecutive consensus decisions within a single message round-trip per decision by preparing the acceptors eagerly with the previous request.

General solutions of state machine replication have to ensure that all replicas apply the same commands in the same order, even in the presence of failures. Such strict ordering incurs high synchronization costs due to the use of distributed consensus or a leader. This paper presents a protocol for linearizable state machine replication of conflict-free replicated data types (CRDTs) that neither requires consensus nor a leader. By leveraging the properties of state-based CRDTs---in particular the monotonic growth of a join semilattice---synchronization overhead is greatly reduced. In addition, updates just need a single round trip and modify the state 'in-place' without the need for a log. Furthermore, the message size overhead for coordination consists of a single counter per message. While reads in the presence of concurrent updates are not wait-free without a coordinator, we show that more than 97\,% of reads can be handled in one or two round trips under highly concurrent accesses. Our protocol achieves high throughput without auxiliary processes such as command log management or leader election. It is well suited for all practical scenarios that need linearizable access on CRDT data on a fine-granular scale.

New generations of high-performance computing applications depend on an increasing number of components to satisfy their growing demand for computation. On such large systems, the execution of long-running jobs is more likely affected by component failures. Failure classes vary from frequent transient memory faults to rather rare correlated node errors. Multilevel checkpoint/restart has been introduced to proactively cope with failures at different levels. Writing checkpoints on slower stable devices, which survive fatal failures, causes more overhead than writing them on fast devices (main memory or local SSD), which, however, only protect against light faults. Given a graph of the components of a particular storage hierarchy mapping their fault-domains and their expected mean time to failure (MTTF), we optimize the checkpoint frequencies for each level of the storage hierarchy (multilevel checkpointing) to minimize the overhead and runtime of a given job. We reduce the checkpoint/restart overhead of large data-intensive jobs compared to state-of-the-art solutions on multilevel checkpointing by up to 10 percent in the investigated cases. The improvement increases further with growing checkpoint sizes.

The objective comparison of hash-based set reconciliation algorithms for distributed replica repair is challenging. Each algorithm's behaviour can be tuned for a given use case, e.g. low bandwidth or computational overhead, using different sets of parameters. Changes on these parameters, however, often also influence the algorithm's accuracy in recognising differences between replicas and thus hinder objective comparisons. We develop models to deduce parameters for equally accurate set reconciliation algorithms for replica repair in a distributed system and compare equally accurate instances of two trivial hash-based algorithms, an algorithm using Bloom filters and a Merkle tree based algorithm. Instead of using a large fixed hash size for Merkle trees we propose to use dynamic hash sizes to align the transfer overhead with the desired accuracy. We evaluate (a) the transferred volume of data with respect to different entropy levels, data and failure distributions on the set of items, and (b) the scalability in the number of items. Our results allow to easily choose an efficient algorithm for practical set reconciliation tasks based on the required level of accuracy. Our way to find equally accurate configuration parameters for different algorithms can also be adopted to other set reconciliation algorithms and allows to rate their respective performance in an objective manner.

Distributed key-value stores are horizontally scalable by design. However, structured data with links between values may raise hotspots or bottlenecks caused by popular keys and large index objects. These hotspots typically reduce the scalability of the key-value store, especially for operations changing data. Relational database management systems, on the other hand, are designed to handle relational data efficiently, but generally do not scale horizontally in a cost-efficient way. Combining the best of both worlds, would be great. With a wiki as a demonstrator, we map a relational database schema to a distributed transactional key-value store. This includes solutions for typical constraints key-value stores impose on applications due to their limited query expressibility. It also includes the mapping of dependent tables and secondary indices to a single key-value namespace. We evaluate and identify hotspots and bottlenecks and propose improved mappings. We reduce the effects of the most prominent hotspots, i.e. secondary indices, by applying advanced partitioning schemes which both reduce the size of the indices and allow more concurrent write accesses in transactional contexts. These optimisations are generic and help to map relational schemas and corresponding applications to transactional key-value stores in a way to preserve their horizontal scalability. With our data models for key-value stores, we get the best of two worlds for the wiki application: a horizontally scalable database serving a moderately complex relational schema. Our optimisations give up to 96% fewer transaction aborts for data change operations and an up to 25-fold latency improvement for the overall operations mix, i.e. reading, changing, and creating data, compared to the basic mapping, when replaying an access trace of the Wikipedia on our system.

Key/value stores which are built on structured overlay networks often lack support for atomic transactions and strong data consistency among replicas. This is unfortunate, because consistency guarantees and transactions would allow a wide range of additional application domains to benefit from the inherent scalability and fault-tolerance of DHTs. The Scalaris key/value store supports strong data consistency and atomic transactions. It uses an enhanced Paxos Commit protocol with only four communication steps rather than six. This improvement was possible by exploiting information from the replica distribution in the DHT. Scalaris enables implementation of more reliable and scalable infrastructure for collaborative Web services that require strong consistency and atomic changes across multiple items.

We present a distributed algorithm for identifying the location of data centers and their relative sizes. This topology information can be used in P2P systems to improve the routing performance, replica placement, or job scheduling. The algorithm uses gossiping with local agglomerative clustering. It is robust to failures and it correctly identifies outliers that are caused, e.g., by temporarily overloaded nodes or network failures. We present empirical results on the Grid 5000 testbed.

Abstract We present a scheme for reserving job resources with imprecise requests. Typical parameters such as the estimated runtime, the start time or the type or number of required CPUs need not be fixed at submission time but can be kept fuzzy in some aspects. Users may specify a list of preferences which guide the system in determining the best matching resources for the given job. Originally, the impetus for our work came from the need for efficient co‐reservation mechanisms in the Grid where rigid constraints on multiple job components often make it difficult to find a feasible solution. Our method for handling fuzzy reservation requests gives the users more freedom to specify the requirements and it gives the Grid Reservation Service more flexibility to find optimal solutions. In the future, we will extend our methods to process co‐reservations. We evaluated our algorithms with real workload traces from a large supercomputer site. The results indicate that our scheme greatly improves the flexibility of the solution process without having much affect on the overall workload of a site. From a user's perspective, only about 10% of the non‐reservation jobs have a longer response time, and from a site administrator's view, the makespan of the original workload is extended by only 8% in the worst case. Copyright © 2006 John Wiley & Sons, Ltd.

Core Grid technologies are rapidly maturing, but there remains a shortage of real Grid applications. One important reason is the lack of a simple and high-level application programming toolkit, bridging the gap between existing Grid middleware and application-level needs. The Grid Application Toolkit (GAT), as currently developed by the EC-funded project GridLab, provides this missing functionality. As seen from the application, the GAT provides a unified simple programming interface to the Grid infrastructure, tailored to the needs of Grid application programmers and users. A uniform programming interface will be needed for application developers to create a new generation of "Grid-aware" applications. The GAT implementation handles both the complexity and the variety of existing Grid middleware services via so-called adaptors. Complementing existing Grid middleware, GridLab also provides high-level services to implement the GAT functionality. We present the GridLab software architecture, consisting of the GAT, environment-specific adaptors, and GridLab services. We elaborate the concepts underlying the GAT and outline the corresponding application programming interface. We present the functionality of GridLab's high-level services and demonstrate how a dynamic Grid application can easily benefit from the GAT. All GridLab software is open source and can be downloaded from the project Web site.

We present a method for the efficient access to parts of remote files. The efficiency is achieved by using a file format independent compact pattern description, that allows us to request several parts of a file in a single operation. This results in a drastically reduced number of remote operations and network latencies if compared to common solutions. We measured the time to access parts of remote files with compact patterns, compared it with normal GridFTP remote partial file access and observed a significant performance increase. Further we discuss how the presented pattern access can be used for an efficient read from multiple replicas and how this can be integrated into a data management system to support the storage of partial replicas for large scale simulations.

Clusters provide an outstanding cost/performance ratio, but their efficient orchestration, i.e. their cooperative management, maintenance, and use, still poses difficulties. Moreover, many sites operate multiple clusters, each possible running under a different cluster management system. In this paper, we present an architectural scheme for the coordinated management of multiple clusters in a fabric. Our scheme allows different cluster management systems to interact with each other via adaptors, thereby providing interoperability within a single administrative entity, the fabric. Using adaptors, various jobs (grid jobs, local jobs, system maintenance jobs) can be served by the same methods, and existing cluster management software (like LSF, PBS, CCS, etc.) can be extended by additional functions without much modification effort.

The Testbed and Applications working group of the European Grid Forum (EGrid) is actively building and experimenting with a grid infrastructure connecting several research-based supercomputing sites located in Europe. The paper reports on our first feasibility study: running a self-migrating version of the Cactus simulation code across the European grid testbed, including "live" remote data visualization and steering from different demonstration booths at Supercomputing 2000, in Dallas, TX. We report on the problems that had to be resolved for this endeavour and identify open research challenges for building production-grade grid environments.