In this paper we present a distributed, message passing implementation of a dynamic dictionary structure for range queries. The structure is based on a distributed fine-grain implementation of skip lists that can scale across a cluster of multicore machines. Our implementation makes use of the unique features of Fine-Grain MPI and introduces novel algorithms and techniques to achieve scalable performance on a cluster of multicore machines. Unlike concurrent data structures the distributed skip list operations are deterministic and atomic. Range-queries are implemented in a way that parallelizes the operation and takes advantage of the recursive properties of the skip list structure. We report on the performance of the skip list for range-queries, on a medium sized cluster with two hundred cores.

One of only two new transport protocols introduced in the last 30 years is the Stream Control Transmission Protocol (SCTP). SCTP enables capabilities like additional throughput and fault tolerance for multihomed hosts. An SCTP implementation is included with the Linux kernel and another implementation called sctplib functions successfully in userspace on several platforms but unfortunately neither of these implementations have all of the latest features nor do they perform as well as the FreeBSD kernel implementation of SCTP. We were motivated to produce a portable implementation of the FreeBSD kernel SCTP stack that operates in userspace of any system because of both our desires to obtain a higher performance SCTP stack for Linux as well as to exploit recent developments in hardware virtualization and transport protocol onloading. Unlike any other userspace transport implementation for TCP or SCTP, our userspace SCTP stack simultaneously achieves similar throughput and latency as the Linux kernel TCP stack, without compromising on any of the transport's features as well as maintaining true portability across multiple operating systems and devices. We create a callback API and implement a threshold to control its usage; our userspace SCTP stack with these optimizations obtains higher throughput than the Linux kernel implementation of SCTP. We describe our userspace SCTP stack's design and demonstrate how it gives similar throughput and latency on Linux as the kernel TCP implementation, with the benefits of the new features of SCTP.

MPI implementations typically equate an MPI process with an OS-process, resulting in a coarse-grain programming model where MPI processes are bound to the physical cores. Fine-Grain (FG-MPI) extends the MPICH2 implementation of MPI and implements an integrated runtime system to allow multiple MPI processes to execute concurrently inside an OS-process. FG-MPI's integrated approach makes it possible to add more concurrency than available parallelism, while minimizing the overheads related to context switches, scheduling and synchronization. In this paper we evaluate the benefits of added concurrency for cache awareness and message size and show that performance gains are possible by using FG-MPI to adjust the grain-size of a program to better fit the cache and potential advantages in passing smaller versus larger messages. We evaluate the use of FG-MPI on the complete set of the NAS parallel benchmarks over large problem sizes, where we show significant performance improvement (20%-30%) for three of the eight benchmarks. We discuss the characteristics of the benchmarks with regards to trade-offs between the added costs and benefits.

As the number of cores inside compute clusters continues to grow, the scalability of MPI (Message Passing Interface) is important to ensure that programs can continue to execute on an ever-increasing number of cores. One important scalability issue for MPI is the implementation of communicators and groups. Communicators and groups are an integral part of MPI and play an essential role in the design and use of libraries. It is challenging to create an MPI implementation to support communicators and groups to scale to the hundreds of thousands of processes that are possible in today's clusters. In this paper we present the design and evaluation of techniques to support the scalability of communicators and groups in MPI.

Programs that execute in parallel across a network often use the Message Passing Interface (MPI) library for communication. The network requirements of an MPI program are often unclear because of the difficulty in exploring alternative network configurations as well as obtaining packet level information about the communication. MPI-NeTSim is an execution environment to emulate MPI programs on simulated networks to allow users to better explore the impact of the network on the performance of MPI programs. We describe the design of MPI-NeTSim and the integration of OMNeT++’s INET framework into MPICH2’s MPI middleware. We introduce a novel technique for uniformly slowing down the execution of the system to allow the discrete event network simulator to keep up with the execution and provide a consistent view of the communication. We validate our technique with synthetic programs as well as the standard NAS benchmarks. We demonstrate MPI-NeTSim’s usefulness in analyzing the effect of the network on communication by using our environment to study the impact of a slow-link on the NAS benchmarks.

We describe the design of a lightweight library using MPI to support stream-processing on acyclic process structures. The design can be used to connect together arbitrary modules where each module can be its own parallel MPI program. We make extensive use of MPI groups and communicators to increase the flexibility of the library, and to make the library easier and safer to use. The notion of a communication context in MPI ensures that libraries do not conflict where a message from one library is mistakenly received by another. The library is not required to be part of any larger workflow environment and is compatible with existing MPI execution environments. The library is part of MarketMiner, a system for executing financial workflows.

This paper describes a high-frequency pair trading strategy that exploits the power of MarketMiner, a high-performance analytics platform that enables a real-time, market-wide search for short-term correlation breakdowns across multiple markets and asset classes. The main theme of this paper is to discuss the computational requirements of model formulation and back-testing, and how a scalable solution built using a modular, MPI-based infrastructure can assist quantitative model and strategy developers by increasing the scale of their experiments or decreasing the time it takes to thoroughly test different parameters. We describe our work to date which is the design of a canonical pair trading algorithm, illustrating how fast and efficient backtesting can be performed using MarketMiner. Preliminary results are given based on a small set of stocks, parameter sets and correlation measures.

Remote direct memory access (RDMA) and point- to-point network fabrics both have their own advantages. MPI middleware implementations typically use one or the other, however, the appearance of the Internet Wide Area RDMA Protocol (iWARP), RDMA over IP, and protocol off-load devices introduces the opportunity to use a hybrid design for MPI middleware that uses both iWARP and a transport protocol directly. We explore the design of a new MPICH2 channel device based on iWARP and the stream control transmission protocol (SCTP) that uses SCTP for all point-to-point MPI routines and iWARP for all remote memory access routines (i.e., one-sided communication). The design extends the Ohio Supercomputer Center software- based iWARP stack and our MPICH2 SCTP-based channel device. The hybrid channel device aligns the semantics of the MPI routine with the underlying protocol that best supports the routine and also allows the MPI API to exploit the potential performance benefits of the underlying hardware more directly. We describe the design and issues related to the progress engine design and connection setup. We demonstrate how to implement iWARP over SCTP rather than TCP and discuss its advantages and disadvantages. We are not aware of any other software implementations of iWARP over SCTP, nor MPI middleware that uses both iWARP verbs and the SCTP API.

We investigate the design and implementation of a parallel workflow environment targeted towards the financial industry. The system performs real-time correlation analysis and clustering to identify trends within streaming high-frequency intra-day trading data. Our system utilizes state-of-the-art methods to optimize the delivery of computationally-expensive real-time stock market data analysis, with direct applications in automated/algorithmic trading as well as knowledge discovery in high-throughput electronic exchanges. This paper describes the design of the system including the key online parallel algorithms for robust correlation calculation and clique-based clustering using stochastic local search. We evaluate the performance and scalability of the system, followed by a preliminary analysis of the results using data from the Toronto Stock Exchange.

A difficulty in using heterogeneous collections of geographically distributed machines across wide area networks for parallel computing is the huge variability in message latency that is orders of magnitude larger than parallel programs executing on dedicated systems. This variability is in part due to the underlying network bandwidth and latency which can vary dramatically according to network conditions. Although such an environment is not suitable for many message passing programs there are those programs that can take advantage of it. Using SCTP (Stream Control Transmission Protocol) for MPI, we show how to reduce the effect of latency on task farm programs to allow them to effectively execute in high latency environments. SCTP is a recently standardized transport level protocol that has a number of features that make it well-suited to MPI and our goal is to reduce the effect of latency on MPI programs in wide area networks. We take advantage of SCTP's improved congestion control as well as its ability to have multiple independent message streams over a single connection to eliminate the head of line blocking that can occur in TCP-based middleware. The use of streams required a novel use of MPI tags to identify independent streams rather than different types of messages. We describe the design of a task farm template that exploits streams, uses buffering and pipelining of task requests to improve its performance under network loss and variable latency. We use these techniques to improve the performance of two real-world MPI programs: a robust correlation matrix computation and mpiBLAST

MPI middleware glues together the components necessary for execution. Almost all implementations have a communication component also called a message progression layer that progresses outstanding messages and maintains their state. The goal of this work is to thin or eliminate this communication component by pushing the functionality down onto the standard IP stack in order to take advantage of potential advances in commodity networking. We introduce a TCP-based design that successfully eliminates the communication component. We discuss how this eliminated TCP-based design doesn't scale and show a more scalable design based on the stream control transmission protocol (SCTP) that has a thinned communication component. We compare the designs showing why SCTP one-to-many sockets in their current form can only thin the communication component. We show what additional features would be required of SCTP to enable a practical design with a fully eliminated communication component

SCTP (Stream Control Transmission Protocol) is a recently standardized transport level protocol with several features that better support the communication requirements of parallel applications; these features are not present in traditional TCP (Transmission Control Protocol). These features make SCTP a good candidate as a transport level protocol for MPI (Message Passing Interface). MPI is a message passing middleware that is widely used to parallelize scientific and compute intensive applications. TCP is often used as the transport protocol for MPI in both local area and wide area networks. Prior to this work, SCTP has not been used for MPI. We compared and evaluated the benefits of using SCTP instead of TCP as the underlying transport protocol for MPI. We re-designed LAM-MPI, a public domain version of MPI, to use SCTP.We describe the advantages and disadvantages of using SCTP, the necessary modifications to the MPI middleware to use SCTP, and the performance of SCTP as compared to the stock implementation that uses TCP.

The computation of covariance and correlation matrices are critical to many data mining applications and processes. Unfortunately the classical covariance and correlation matrices are very sensitive to outliers. Robust methods, such as QC and the Maronna method, have been proposed. However, existing algorithms for QC only give acceptable performance when the dimensionality of the matrix is in the hundreds; and the Maronna method is rarely used in practice because of its high computational cost.In this paper, we develop parallel algorithms for both QC and the Maronna method. We evaluate these parallel algorithms using a real data set of the gene expression of over 6,000 genes, giving rise to a matrix of over 18 million entries. In our experimental evaluation, we explore scalability in dimensionality and in the number of processors. We also compare the parallel behaviours of the two methods. After thorough experimentation, we conclude that for many data mining applications, both QC and Maronna are viable options. Less robust, but faster, QC is the recommended choice for small parallel platforms. On the other hand, the Maronna method is the recommended choice when a high degree of robustness is required, or when the parallel platform features a high number of processors.

An important family of graphs is the n-dimensional hypercube, the graph with $2^{n}$ nodes labelled $0,1,\cdots, 2^{n}-1$, and an edge joining two nodes whenever their binary representation differs in a single coordinate. The problem of deciding if a given source graph is a partial subgraph of an n-dimensional cube has recently been shown to be NP-complete. In this paper the same problem on a very restricted family of source graphs, trees, is considered. It is shown that the problem of determining for a given tree T and integer k if T is a partial subgraph of a k-dimensional cube is NP-complete.