Graph is a ubiquitous structure in many domains. The rapidly increasing data volume calls for efficient and scalable graph data processing. In recent years, designing distributed graph processing systems has been an increasingly important area to fulfil the demands of processing big graphs in a distributed environment. Though a variety of distributed graph processing systems have been developed, very little attention has been paid to achieving a good combinational system performance in terms of usage simplicity, efficiency and scalability. To contribute to the study of distributed graph processing system, this work tries to fill this gap by designing a simple, speedy and scalable system. Our observation is that enforcing the communication flexibility of a system leads to the gains of both system efficiency and scalability as well as simple usage. We realize our idea in a system Graph3S and conduct extensive experiments with diverse algorithms over big graphs from different domains to test its performance. The results show that, besides simple usage, our system has outstanding performance over various graph algorithms and can even reach up to two orders of magnitude speedup over existing in-memory systems when applying to some algorithms. Also, its scalability is competitive to disk-based systems and even better when less machines are used.
Proposed in 1993 the Scalable Distributed Data Structures (SDDSs) became a profile of basis for the data management on Multi computer. In this paper we propose an organization of a LH* bucket based on the trie hashing in order to improve times of different access request.
Sequence alignment algorithms are a basic and critical component of many bioinformatics fields. With rapid development of sequencing technology, the fast growing reference database volumes and longer length of query sequence become new challenges for sequence alignment. However, the algorithm is prohibitively high in terms of time and space complexity. In this paper, we present DSA, a scalable distributed sequence alignment system that employs Spark to process sequences data in a horizontally scalable distributed environment, and leverages data parallel strategy based on Single Instruction Mul…
This paper proposes Kudu, a distributed execution engine with a well-defined abstraction that can be integrated with existing single-machine graph pattern mining (GPM) systems to provide efficiency and scalability at the same time. The key novelty is the extendable embedding abstraction which can express pattern enumeration algorithms, allow fine-grained task scheduling, and enable low-cost GPM-specific data reuse to reduce communication cost. The effective BFSDFS hybrid exploration generates sufficient concurrent tasks for communication-computation overlapping with bounded memory consumption.…
We describe \emph{Fantom}, a framework for asynchronous distributed systems. \emph{Fantom} is based on the Lachesis Protocol~\cite{lachesis01}, which uses asynchronous event transmission for practical Byzantine fault tolerance (pBFT) to create a leaderless, scalable, asynchronous Directed Acyclic Graph (DAG). We further optimize the \emph{Lachesis Protocol} by introducing a permission-less network for dynamic participation. Root selection cost is further optimized by the introduction of an n-row flag table, as well as optimizing path selection by introducing domination relationships. We propos…
We propose a distributed protocol for a queue, called \textsc{Skueue}, which spreads its data fairly onto multiple processes, avoiding bottlenecks in high throughput scenarios. \textsc{Skueue} can be used in highly dynamic environments, through the addition of join and leave requests to the standard queue operations enqueue and dequeue. Furthermore \textsc{Skueue} satisfies sequential consistency in the asynchronous message passing model. Scalability is achieved by aggregating multiple requests to a batch, which can then be processed in a distributed fashion without hurting the queue semantics…
We propose two protocols for distributed priority queues (for simplicity denoted 'heap') called SKEAP and SEAP. SKEAP realizes a distributed heap for a constant amount of priorities and SEAP one for an arbitrary amount. Both protocols build on an overlay, which induces an aggregation tree on top of which heap operations are aggregated in batches, ensuring that our protocols scale even for a high rate of incoming requests. As part of SEAP we provide a novel distributed protocol for the $k$-selection problem that runs in $O(\log n)$ rounds w.h.p. SKEAP guarantees sequential consistency for its h…
Graphs are a key form of Big Data, and performing scalable analytics over them is invaluable to many domains. As our ability to collect data grows, there is an emerging class of inter-connected data which accumulates or varies over time, and on which novel analytics - both over the network structure and across the time-variant attribute values - is necessary. We introduce the notion of time-series graph analytics and propose Gopher, a scalable programming abstraction to develop algorithms and analytics on such datasets. Our abstraction leverages a sub-graph centric programming model and extend…
As distributed systems grow in scale and complexity, the need for flexible automation of systems management functions also grows. We outline a framework for building tools that provide distributed, scalable, declarative, modular, and continuous automation for distributed systems. We focus on four points of design: 1) a state-management approach that prescribes source-of-truth for configured and discovered system states; 2) a technique to solve the declarative unification problem for a class of automation problems, providing state convergence and modularity; 3) an eventual-consistency approach …
We describe an efficient and fault-tolerant algorithm for distributed cyclic garbage collection. The algorithm imposes few requirements on the local machines and allows for flexibility in the choice of local collector and distributed acyclic garbage collector to use with it. We have emphasized reducing the number and size of network messages without sacrificing the promptness of collection throughout the algorithm. Our proposed collector is a variant of back tracing to avoid extensive synchronization between machines. We have added an explicit forward tracing stage to the standard back tracing…
The in-memory graph layout or organization has a considerable impact on the time and energy efficiency of distributed memory graph computations. It affects memory locality, inter-task load balance, communication time, and overall memory utilization. Graph layout could refer to partitioning or replication of vertex and edge arrays, selective replication of data structures that hold meta-data, and reordering vertex and edge identifiers. In this work, we present DGL, a fast, parallel, and memory-efficient distributed graph layout strategy that is specifically designed for small-world networks (lo…
The rise of worldwide Internet-scale services demands large distributed systems. Indeed, when handling several millions of users, it is common to operate thousands of servers spread across the globe. Here, replication plays a central role, as it contributes to improve the user experience by hiding failures and by providing acceptable latency. In this paper, we claim that atomic multicast, with strong and well-defined properties, is the appropriate abstraction to efficiently design and implement globally scalable distributed systems. We substantiate our claim with the design of two modern onlin…
Recently, Deep Neural Networks (DNNs) have recorded great success in handling medical and other complex classification tasks. However, as the sizes of a DNN model and the available dataset increase, the training process becomes more complex and computationally intensive, which usually takes a longer time to complete. In this work, we have proposed a generic full end-to-end hybrid parallelization approach combining both model and data parallelism for efficiently distributed and scalable training of DNN models. We have also proposed a Genetic Algorithm based heuristic resources allocation mechan…
Service-Oriented Computing (SOC) enables the composition of loosely coupled service agents provided with varying Quality of Service (QoS) levels, effectively forming a multiagent system (MAS). Selecting a (near-)optimal set of services for a composition in terms of QoS is crucial when many functionally equivalent services are available. As the number of distributed services, especially in the cloud, is rising rapidly, the impact of the network on the QoS keeps increasing. Despite this and opposed to most MAS approaches, current service approaches depend on a centralized architecture which cann…
Emerging Deep Learning (DL) applications introduce heavy I/O workloads on computer clusters. The inherent long lasting, repeated, and random file access pattern can easily saturate the metadata and data service and negatively impact other users. In this paper, we present FanStore, a transient runtime file system that optimizes DL I/O on existing hardware/software stacks. FanStore distributes datasets to the local storage of compute nodes, and maintains a global namespace. With the techniques of system call interception, distributed metadata management, and generic data compression, FanStore pr…
In large-scale distributed file systems, efficient meta- data operations are critical since most file operations have to interact with metadata servers first. In existing distributed hash table (DHT) based metadata management systems, the lookup service could be a performance bottleneck due to its significant CPU overhead. Our investigations showed that the lookup service could reduce system throughput by up to 70%, and increase system latency by a factor of up to 8 compared to ideal scenarios. In this paper, we present MetaFlow, a scalable metadata lookup service utilizing software-defined ne…
The implementation of a vast majority of machine learning (ML) algorithms boils down to solving a numerical optimization problem. In this context, Stochastic Gradient Descent (SGD) methods have long proven to provide good results, both in terms of convergence and accuracy. Recently, several parallelization approaches have been proposed in order to scale SGD to solve very large ML problems. At their core, most of these approaches are following a map-reduce scheme. This paper presents a novel parallel updating algorithm for SGD, which utilizes the asynchronous single-sided communication paradigm…
The amount of remote sensing (RS) data has increased at an unexpected scale, due to the rapid progress of earth-observation and the growth of satellite RS and sensor technologies. Traditional relational databases attend their limit to meet the needs of high-resolution and large-scale RS Big Data management. As a result, massive RS data management is currently one of the most imperative topics. To address this problem, this paper describes a distributed architecture for big RS data storage based on a unified metadata file, pyramid model, and Hilbert curve for data composition and indexing using…
This paper presents a new fast, highly scalable distributed matrix multiplication algorithm on Apache Spark, called Stark, based on Strassen's matrix multiplication algorithm. Stark preserves Strassen's 7 multiplications scheme in a distributed environment and thus achieves faster execution. It is based on two new ideas; it creates a recursion tree of computation where each level of such tree corresponds to division and combination of distributed matrix blocks in the form of Resilient Distributed Datasets(RDDs); It processes each divide and combine step in parallel and memorize the sub-matrice…
This paper presents a new framework, namely \emph{\onlay}, for scalable asynchronous distributed systems. In this framework, we propose a consensus protocol $L_ϕ$, which is based on the Lachesis protocol~\cite{lachesis01}. At the core of $L_ϕ$ protocol, it introduces to use layering algorithm to achieve practical Byzantine fault tolerance (pBFT) in leaderless asynchronous Directed Acyclic Graph (DAG). Further, we present new online layering algorithms for the evolutionary DAGs across the nodes. Our new protocol achieves determistic scalable consensus in asynchronous pBFT by using assigned laye…
Counters are an important abstraction in distributed computing, and play a central role in large scale geo-replicated systems, counting events such as web page impressions or social network "likes". Classic distributed counters, strongly consistent, cannot be made both available and partition-tolerant, due to the CAP Theorem, being unsuitable to large scale scenarios. This paper defines Eventually Consistent Distributed Counters (ECDC) and presents an implementation of the concept, Handoff Counters, that is scalable and works over unreliable networks. By giving up the sequencer aspect of class…
We propose two new methods to address the weak scaling problems of KRR: the Balanced KRR (BKRR) and K-means KRR (KKRR). These methods consider alternative ways to partition the input dataset into p different parts, generating p different models, and then selecting the best model among them. Compared to a conventional implementation, KKRR2 (optimized version of KKRR) improves the weak scaling efficiency from 0.32% to 38% and achieves a 591times speedup for getting the same accuracy by using the same data and the same hardware (1536 processors). BKRR2 (optimized version of BKRR) achieves a highe…
This paper introduces a high performance implementation of \texttt{Zolo-SVD} algorithm on distributed memory systems, which is based on the polar decomposition (PD) algorithm via the Zolotarev's function (\texttt{Zolo-PD}), originally proposed by Nakatsukasa and Freund [SIAM Review, 2016]. Our implementation highly relies on the routines of ScaLAPACK and therefore it is portable. Compared with the other PD algorithms such as the QR-based dynamically weighted Halley method (\texttt{QDWH-PD}), \texttt{Zolo-PD} is naturally parallelizable and has better scalability though performs more floating-p…
We consider the problem of computing an aggregation function in a \emph{secure} and \emph{scalable} way. Whereas previous distributed solutions with similar security guarantees have a communication cost of $O(n^3)$, we present a distributed protocol that requires only a communication complexity of $O(n\log^3 n)$, which we prove is near-optimal. Our protocol ensures perfect security against a computationally-bounded adversary, tolerates $(1/2-ε)n$ malicious nodes for any constant $1/2 > ε> 0$ (not depending on $n$), and outputs the exact value of the aggregated function with high probability.
We present here a cost effective framework for a robust scalable and distributed job processing system that adapts to the dynamic computing needs easily with efficient load balancing for heterogeneous systems. The design is such that each of the components are self contained and do not depend on each other. Yet, they are still interconnected through an enterprise message bus so as to ensure safe, secure and reliable communication based on transactional features to avoid duplication as well as data loss. The load balancing, fault-tolerance and failover recovery are built into the system through…
Large-scale distributed training is increasingly becoming communication bound. Many gradient compression algorithms have been proposed to reduce the communication overhead and improve scalability. However, it has been observed that in some cases gradient compression may even harm the performance of distributed training. In this paper, we propose MergeComp, a compression scheduler to optimize the scalability of communication-efficient distributed training. It automatically schedules the compression operations to optimize the performance of compression algorithms without the knowledge of model a…
We present a highly scalable algorithm for multiplying sparse multivariate polynomials represented in a distributed format. This algo- rithm targets not only the shared memory multicore computers, but also computers clusters or specialized hardware attached to a host computer, such as graphics processing units or many-core coprocessors. The scal- ability on the large number of cores is ensured by the lacks of synchro- nizations, locks and false-sharing during the main parallel step.
Advances in deep neural networks (DNN) and computer vision (CV) algorithms have made it feasible to extract meaningful insights from large-scale deployments of urban cameras. Tracking an object of interest across the camera network in near real-time is a canonical problem. However, current tracking platforms have two key limitations: 1) They are monolithic, proprietary and lack the ability to rapidly incorporate sophisticated tracking models; and 2) They are less responsive to dynamism across wide-area computing resources that include edge, fog and cloud abstractions. We address these gaps usi…
The ability to express a program as a hierarchical composition of parts is an essential tool in managing the complexity of software and a key abstraction this provides is to separate the representation of data from the computation. Many current parallel programming models use a shared memory model to provide data abstraction but this doesn't scale well with large numbers of cores due to non-determinism and access latency. This paper proposes a simple programming model that allows scalable parallel programs to be expressed with distributed representations of data and it provides the programmer …
In parallel with big data processing and analysis dominating the usage of distributed and cloud infrastructures, the demand for distributed metadata access and transfer has increased. In many application domains, the volume of data generated exceeds petabytes, while the corresponding metadata amounts to terabytes or even more. This paper proposes a novel solution for efficient and scalable metadata access for distributed applications across wide-area networks, dubbed SMURF. Our solution combines novel pipelining and concurrent transfer mechanisms with reliability, provides distributed continuu…
Trajectory clustering is an important operation of knowledge discovery from mobility data. Especially nowadays, the need for performing advanced analytic operations over massively produced data, such as mobility traces, in efficient and scalable ways is imperative. However, discovering clusters of complete trajectories can overlook significant patterns that exist only for a small portion of their lifespan. In this paper, we address the problem of Distributed Subtrajectory Clustering in an efficient and highly scalable way. The problem is challenging because the subtrajectories to be clustered …
We design and implement a distributed algorithm for balanced $k$-way hypergraph partitioning that minimizes fanout, a fundamental hypergraph quantity also known as the communication volume and ($k-1$)-cut metric, by optimizing a novel objective called probabilistic fanout. This choice allows a simple local search heuristic to achieve comparable solution quality to the best existing hypergraph partitioners. Our algorithm is arbitrarily scalable due to a careful design that controls computational complexity, space complexity, and communication. In practice, we commonly process hypergraphs with b…
One of the keys for deep learning to have made a breakthrough in various fields was to utilize high computing powers centering around GPUs. Enabling the use of further computing abilities by distributed processing is essential not only to make the deep learning bigger and faster but also to tackle unsolved challenges. We present the design, implementation, and evaluation of ChainerMN, the distributed deep learning framework we have developed. We demonstrate that ChainerMN can scale the learning process of the ResNet-50 model to the ImageNet dataset up to 128 GPUs with the parallel efficiency o…
In today's world of computers, dealing with huge amounts of data is not unusual. The need to distribute this data in order to increase its availability and increase the performance of accessing it is more urgent than ever. For these reasons it is necessary to develop scalable distributed data structures. In this paper we propose a TH* distributed variant of the Trie Hashing data structure. First we propose Thsw new version of TH without node Nil in digital tree (trie), then this version will be adapted to multicomputer environment. The simulation results reveal that TH* is scalable in the sens…
This paper describes a new fast and implicitly parallel approach to neighbour-finding in multi-resolution Smoothed Particle Hydrodynamics (SPH) simulations. This new approach is based on hierarchical cell decompositions and sorted interactions, within a task-based formulation. It is shown to be faster than traditional tree-based codes, and to scale better than domain decomposition-based approaches on hybrid shared/distributed-memory parallel architectures, e.g. clusters of multi-cores, achieving a $40\times$ speedup over the Gadget-2 simulation code.
Computer vision is experiencing an AI renaissance, in which machine learning models are expediting important breakthroughs in academic research and commercial applications. Effectively training these models, however, is not trivial due in part to hyperparameters: user-configured values that control a model's ability to learn from data. Existing hyperparameter optimization methods are highly parallel but make no effort to balance the search across heterogeneous hardware or to prioritize searching high-impact spaces. In this paper, we introduce a framework for massively Scalable Hardware-Aware D…
Bizur is a consensus algorithm exposing a key-value interface. It is used by a distributed file-system that scales to 100s of servers, delivering millions of IOPS, both data and metadata, with consistent low-latency. Bizur is aimed for services that require strongly consistent state, but do not require a distributed log; for example, a distributed lock manager or a distributed service locator. By avoiding a distributed log scheme, Bizur outperforms distributed log based consensus algorithms, producing more IOPS and guaranteeing lower latencies during normal operation and especially during fail…
Deep Learning (DL) has had an immense success in the recent past, leading to state-of-the-art results in various domains such as image recognition and natural language processing. One of the reasons for this success is the increasing size of DL models and the proliferation of vast amounts of training data being available. To keep on improving the performance of DL, increasing the scalability of DL systems is necessary. In this survey, we perform a broad and thorough investigation on challenges, techniques and tools for scalable DL on distributed infrastructures. This incorporates infrastructur…
More and more distributed software systems are being developed and deployed today. Like other software, distributed software systems also need very strong quality assurance support. Distributed software is often very large/complex, has distributed components, and does not have a global clock. All these characteristics make it very challenging to analyze the information flow of such systems to support the software quality assurance. One challenge is that existing dynamic analysis techniques hardly scale to large distributed software systems in the real world. It is also challenging to develop c…
Apache Flink is an open-source system for scalable processing of batch and streaming data. Flink does not natively support efficient processing of spatial data streams, which is a requirement of many applications dealing with spatial data. Besides Flink, other scalable spatial data processing platforms including GeoSpark, Spatial Hadoop, etc. do not support streaming workloads and can only handle static/batch workloads. To fill this gap, we present GeoFlink, which extends Apache Flink to support spatial data types, indexes and continuous queries over spatial data streams. To enable the efficie…
This paper proposes a simple and scalable web-based model for grid resource discovery for the Internet. The resource discovery model contains the metadata and resource finder web services. The information of resource finder web services is kept in the repositories that are distributed in the application layer of Internet. The resource finder web services will be discovered by sending queries to the repositories in a similar way as the DNS protocol. The underlying technology for implementation of the two architectures of this model is introduced. These architectures are: Direct and Centralized …
Recently, edge computing has emerged as a promising paradigm to support mobile access in IoT multinetworks. However, coexistence of heterogeneous wireless communication schemes brings about new challenges to the mobility management and access control in such networks. To maintain its computing continuum, Software-Defined Networking (SDN) architecture can be applied to the distributed edge networks, which provides consistent frameworks for mobility management and creates secure mechanisms for access control. In this paper, we first review the state-of-the-art SDN technology and its applications…
It has been experimentally observed that distributed implementations of mini-batch stochastic gradient descent (SGD) algorithms exhibit speedup saturation and decaying generalization ability beyond a particular batch-size. In this work, we present an analysis hinting that high similarity between concurrently processed gradients may be a cause of this performance degradation. We introduce the notion of gradient diversity that measures the dissimilarity between concurrent gradient updates, and show its key role in the performance of mini-batch SGD. We prove that on problems with high gradient di…
Center of Computation & Technology（计算与技术中心）；Louisiana State University（路易斯安那州立大学）；Microsoft Research Montréal（微软研究院蒙特利尔分院）
Although recent scaling up approaches to training deep neural networks have proven to be effective, the computational intensity of large and complex models, as well as the availability of large-scale datasets, require deep learning frameworks to utilize scaling out techniques. Parallelization approaches and distribution requirements are not considered in the preliminary designs of most available distributed deep learning frameworks, and most of them still are not able to perform effective and efficient fine-grained inter-node communication. We present Phylanx that has the potential to alleviat…
We engineer algorithms for sorting huge data sets on massively parallel machines. The algorithms are based on the multiway merging paradigm. We first outline an algorithm whose I/O requirement is close to a lower bound. Thus, in contrast to naive implementations of multiway merging and all other approaches known to us, the algorithm works with just two passes over the data even for the largest conceivable inputs. A second algorithm reduces communication overhead and uses more conventional specifications of the result at the cost of slightly increased I/O requirements. An implementation wins th…
We give a distributed variant of an adaptive histogram estimation procedure previously developed by the first author. The procedure is based on regular pavings and is known to have numerous appealing statistical and arithmetical properties. The distributed version makes it possible to process data sets significantly bigger than previously. We provide prototype implementation under a permissive license.
