This paper presents a novel approach to multi-robot collision avoidance that integrates global path planning with local navigation strategies, utilizing attentive graph neural networks to manage dynamic interactions among agents. We introduce a local navigation model that leverages pre-planned global paths, allowing robots to adhere to optimal routes while dynamically adjusting to environmental changes. The model’s robustness is enhanced through the introduction of noise during training, resulting in superior performance in complex, dynamic environments. Our approach is evaluated against established baselines, including NH-ORCA, DRL-NAV, and GA3C-CADRL, across various structurally diverse simulated scenarios. The results demonstrate that our model achieves consistently higher success rates, lower collision rates, and more efficient navigation, particularly in challenging scenarios where baseline models struggle. This work offers an advancement in multi-robot navigation, with implications for robust performance in complex, dynamic environments with varying degrees of complexity, such as those encountered in logistics, where adaptability is essential for accommodating unforeseen obstacles and unpredictable changes.

In recent years, journalists have expressed concerns about the increasing trend of news article avoidance, especially within specific domains. This issue has been exacerbated by the rise of recommender systems. Our research indicates that recommender systems should consider avoidance as a fundamental factor. We argue that news articles can be characterized by three principal elements: exposure, relevance, and avoidance, all of which are closely interconnected. To address these challenges, we introduce AWRS, an Avoidance-Aware Recommender System. This framework incorporates avoidance awareness when recommending news, based on the premise that news article avoidance conveys significant information about user preferences. Evaluation results on three news datasets in different languages (English, Norwegian, and Japanese) demonstrate that our method outperforms existing approaches.

In modern materials science, effective and high-volume data management across leading-edge experimental facilities and world-class supercomputers is indispensable for cutting-edge research. However, existing integrated systems that handle data from these resources have primarily focused just on smaller-scale cross-institutional or single-domain operations. As a result, they often lack the scalability, efficiency, agility, and interdisciplinarity, needed for handling substantial volumes of data from various researchersIn this paper, we introduce ARIM-mdx data system1, aiming at a nationwide data platform for materials science in Japan. Currently in its trial phase, the platform has been involving 11 universities and institutes all over Japan, and it is utilized by over 800 researchers from around 140 organizations in academia and industry, being intended to gradually expand its reach. The ARIM-mdx data system, as a pioneering nationwide data platform, has the potential to contribute to the creation of new research communities and accelerate innovations.

We present the Evolving Graph Fourier Transform (EFT), the first invertible spectral transform that captures evolving representations on temporal graphs. We motivate our work by the inadequacy of existing methods for capturing the evolving graph spectra, which are also computationally expensive due to the temporal aspect along with the graph vertex domain. We view the problem as an optimization over the Laplacian of the continuous time dynamic graph. Additionally, we propose pseudo-spectrum relaxations that decompose the transformation process, making it highly computationally efficient. The EFT method adeptly captures the evolving graph's structural and positional properties, making it effective for downstream tasks on evolving graphs. Hence, as a reference implementation, we develop a simple neural model induced with \eft for capturing evolving graph spectra. We empirically validate our theoretical findings on a number of large-scale and standard temporal graph benchmarks and demonstrate that our model achieves state-of-the-art performance.

Audience Expansion, a technique for identifying new audiences with similar behaviors to the original target or seed users. The major challenges include a heterogeneous user base, intricate marketing campaigns, constraints imposed by sparsity, and limited seed users, which lead to overfitting. In this context, we propose a novel solution named AudienceLinkNet, specifically designed to address the challenges associated with audience expansion in the context of Rakuten's diverse services and its clients. Our approach formulates the audience expansion problem as a graph problem and explores the combination of a Pre-trained Knowledge Graph Embedding Model and a Graph Convolutional Networks (GCNs). It emphasizes the structural retention properties of GCNs, enabling the model to overcome challenges related to cross-service data usage, sparsity and limited seed data. AudienceLinkNet simplifies the targeting process for small and large marketing campaigns and better utilizes demographics and behavioral attributes for targeting. Extensive experiments on our advertising platform, Rakuten AIris Target Prospecting, demonstrate the effectiveness of our audience expansion model. Additionally, we present the limitations of AudienceLinkNet.

Learning on evolving(dynamic) graphs has caught the attention of researchers as static methods exhibit limited performance in this setting. The existing methods for dynamic graphs learn spatial features by local neighborhood aggregation, which essentially only captures the low pass signals and local interactions. In this work, we go beyond current approaches to incorporate global features for effectively learning representations of a dynamically evolving graph. We propose to do so by capturing the spectrum of the dynamic graph. Since static methods to learn the graph spectrum would not consider the history of the evolution of the spectrum as the graph evolves with time, we propose an approach to learn the graph wavelets to capture this evolving spectra. Further, we propose a framework that integrates the dynamically captured spectra in the form of these learnable wavelets into spatial features for incorporating local and global interactions. Experiments on eight standard datasets show that our method significantly outperforms related methods on various tasks for dynamic graphs.

Traffic forecasting as a canonical task of multivariate time series forecasting has been a significant research topic in AI community. To address the spatio-temporal heterogeneity and non-stationarity implied in the traffic stream, in this study, we propose Spatio-Temporal Meta-Graph Learning as a novel Graph Structure Learning mechanism on spatio-temporal data. Specifically, we implement this idea into Meta-Graph Convolutional Recurrent Network (MegaCRN) by plugging the Meta-Graph Learner powered by a Meta-Node Bank into GCRN encoder-decoder. We conduct a comprehensive evaluation on two benchmark datasets (i.e., METR-LA and PEMS-BAY) and a new large-scale traffic speed dataset called EXPY-TKY that covers 1843 expressway road links in Tokyo. Our model outperformed the state-of-the-arts on all three datasets. Besides, through a series of qualitative evaluations, we demonstrate that our model can explicitly disentangle the road links and time slots with different patterns and be robustly adaptive to any anomalous traffic situations. Codes and datasets are available at https://github.com/deepkashiwa20/MegaCRN.

Next Point-of-Interest (POI) recommendation plays a crucial role in urban mobility applications. Recently, POI recommendation models based on Graph Neural Networks (GNN) have been extensively studied and achieved, however, the effective incorporation of both spatial and temporal information into such GNN-based models remains challenging. Temporal information is extracted from users' trajectories, while spatial information is obtained from POIs. Extracting distinct fine-grained features unique to each piece of information is difficult since temporal information often includes spatial information, as users tend to visit nearby POIs. To address the challenge, we propose Mobility Graph Transformer (MobGT) that enables us to fully leverage graphs to capture both the spatial and temporal features in users' mobility patterns. MobGT combines individual spatial and temporal graph encoders to capture unique features and global user-location relations. Additionally, it incorporates a mobility encoder based on Graph Transformer to extract higher-order information between POIs. To address the long-tailed problem in spatial-temporal data, MobGT introduces a novel loss function, Tail Loss. Experimental results demonstrate that MobGT outperforms state-of-the-art models on various datasets and metrics, achieving 24% improvement on average. Our codes are available at https://github.com/Yukayo/MobGT.

Precisely recommending candidate news articles to users has always been a core challenge for personalized news recommendation systems. Most recent works primarily focus on using advanced natural language processing techniques to extract semantic information from rich textual data, employing content-based methods derived from local historical news. However, this approach lacks a global perspective, failing to account for users’ hidden motivations and behaviors beyond semantic information. To address this challenge, we propose a novel model called GLORY (Global-LOcal news Recommendation sYstem), which combines global representations learned from other users with local representations to enhance personalized recommendation systems. We accomplish this by constructing a Global-aware Historical News Encoder, which includes a global news graph and employs gated graph neural networks to enrich news representations, thereby fusing historical news representations by a historical news aggregator. Similarly, we extend this approach to a Global Candidate News Encoder, utilizing a global entity graph and a candidate news aggregator to enhance candidate news representation. Evaluation results on two public news datasets demonstrate that our method outperforms existing approaches. Furthermore, our model offers more diverse recommendations1.

Lookalike models are based on the assumption that user similarity plays an important role towards product selling and enhancing the existing advertising campaigns from a very large user base. Challenges associated to these models reside on the heterogeneity of the user base and its sparsity. In this work, we propose a novel framework that unifies the customers' different behaviors or features such as demographics, buying behaviors on different platforms, customer loyalty behaviors and build a lookalike model to improve customer targeting for Rakuten Group, Inc. Extensive experiments on real e-commerce and travel datasets demonstrate the effectiveness of our proposed lookalike model for user targeting task.

In this paper we present a novel method, Knowledge Persistence (), for faster evaluation of Knowledge Graph (KG) completion approaches. Current ranking-based evaluation is quadratic in the size of the KG, leading to long evaluation times and consequently a high carbon footprint. addresses this by representing the topology of the KG completion methods through the lens of topological data analysis, concretely using persistent homology. The characteristics of persistent homology allow to evaluate the quality of the KG completion looking only at a fraction of the data. Experimental results on standard datasets show that the proposed metric is highly correlated with ranking metrics (Hits@N, MR, MRR). Performance evaluation shows that is computationally efficient: In some cases, the evaluation time (validation+test) of a KG completion method has been reduced from 18 hours (using Hits@10) to 27 seconds (using ), and on average (across methods & data) reduces the evaluation time (validation+test) by ≈ 99.96%.

Maintaining guaranteed service level agreements on distributed graph processing for concurrent query execution is challenging because graph processing by nature is an unbalanced problem. In this paper we investigate on maintaining predefined service level agreements for graph processing workload mixtures taking triangle counting as the example. We develop a Graph Query Scheduler Mechanism (GQSM) which maintains a guaranteed service level agreement in terms of overall latency on top of JasmineGraph distributed graph database server. The proposed GQSM model is implemented using the queuing theory. Main component of GQSM is a job scheduler which is responsible for listening to an incoming job queue and scheduling the jobs received. The proposed model has a calibration phase where the Service Level Agreement (SLA) data, load average curve data, and maximum load average which can be handled by the hosts participating in the cluster without violating SLA is captured for the graphs in the system. Results show that for a single host system the SLA is successfully maintained when the total number of users is less than 6.

We present distributed algorithms for training dynamic Graph Neural Networks (GNN) on large scale graphs spanning multi-node, multi-GPU systems. To the best of our knowledge, this is the first scaling study on dynamic GNN. We devise mechanisms for reducing the GPU memory usage and identify two execution time bottlenecks: CPU-GPU data transfer; and communication volume. Exploiting properties of dynamic graphs, we design a graph difference-based strategy to significantly reduce the transfer time. We develop a simple, but effective data distribution technique under which the communication volume remains fixed and linear in the input size, for any number of GPUs. Our experiments using billion-size graphs on a system of 128 GPUs shows that: (i) the distribution scheme achieves up to 30x speedup on 128 GPUs; (ii) the graph-difference technique reduces the transfer time by a factor of up to 4.1x and the overall execution time by up to 40%.

Graph Convolutional Networks (GCN) have become a popular means of performing link prediction due to the high accuracy offered by them. However, scaling such link prediction into large graphs of billions of vertices and edges with rich types of attributes is a significant issue to be addressed due to the storage and computation limitations of the machines. In this paper we present a scalable link prediction approach which conducts GCN training and link prediction on top of a distributed graph database server called JasmineGraph. We partition graph data and persist them in multiple workers. We implement parallel graph node embedding generation using GraphSAGE algorithm in multiple workers. Our approach avoids facing performance bottlenecks in GCN training using an intelligent scheduling algorithm. We show our approach scales well with an increasing number of partitions (2,4,8, and 16) using four real world data sets called Twitter, Amazon, Reddit, and DBLP-V11. JasmineGraph was able to train a GCN from the largest dataset DBLP-V11 (> 9.3GB) in 11 hours and 40 minutes time using 16 workers on a single server while the original GraphSAGE implementation could not process it at all. The original GraphSAGE implementation processed the second largest dataset Reddit in 238 minutes while JasmineGraph took only 100 minutes on the same hardware with 16 workers leading to 2.4 times improved performance.

Graph representation learning resurges as a trending research subject owing to the widespread use of deep learning for Euclidean data, which inspire various creative designs of neural networks in the non-Euclidean domain, particularly graphs. With the success of these graph neural networks (GNN) in the static setting, we approach further practical scenarios where the graph dynamically evolves. Existing approaches typically resort to node embeddings and use a recurrent neural network (RNN, broadly speaking) to regulate the embeddings and learn the temporal dynamics. These methods require the knowledge of a node in the full time span (including both training and testing) and are less applicable to the frequent change of the node set. In some extreme scenarios, the node sets at different time steps may completely differ. To resolve this challenge, we propose EvolveGCN, which adapts the graph convolutional network (GCN) model along the temporal dimension without resorting to node embeddings. The proposed approach captures the dynamism of the graph sequence through using an RNN to evolve the GCN parameters. Two architectures are considered for the parameter evolution. We evaluate the proposed approach on tasks including link prediction, edge classification, and node classification. The experimental results indicate a generally higher performance of EvolveGCN compared with related approaches. The code is available at https://github.com/IBM/EvolveGCN.

Graph Convolutional Networks (GCN) have found multiple applications of graph-based machine learning. However, training GCNs on large graphs of billions of nodes and edges with rich node attributes consume significant amount of time and memory resources. This makes it impossible to train such GCNs on general purpose commodity hardware. Such use cases demand high-end servers with accelerators and ample amounts of memory. In this paper we implement a memory efficient GCN based link prediction on top of a distributed graph database server called JasmineGraph1. Our approach is based on federated training on partitioned graphs with multiple parallel workers. We conduct experiments with three real world graph datasets called DBLP-V11, Reddit, and Twitter. We demonstrate that our approach produces optimal performance for a given hardware setting. JasmineGraph was able to train a GCN on the largest dataset DBLP-V11(>10GB) in 20 hours and 24 minutes for 5 training rounds and 3 epochs by partitioning it into 16 partitions with 2 workers on a single server while the conventional training method could not process it at all due to lack of memory. The second largest dataset Reddit took 9 hours 8 minutes to train with conventional training while JasmineGraph took only 3 hours and 11 minutes with 8 partitions-4 workers in the same hardware giving 3 times improved performance. In case of Twitter dataset JasmineGraph was able to give 5 times improved performance. (10 hours 31 minutes vs 2 hours 6 minutes;16 partitions-16 workers).

As COVID-19 transmissions spread worldwide, governments have announced and enforced travel restrictions to prevent further infections. Such restrictions have a direct effect on the volume of international flights among these countries, resulting in extensive social and economic costs. To better understand the situation in a quantitative manner, we analyzed the OpenSky Network data to clarify flight patterns and flight densities around the world. Then we observed relationships between flight numbers with new infection cases and the economy (the unemployment rate) in Barcelona. We found that the number of daily flights gradually decreased and then suddenly dropped 64% during the second half of March in 2020 after the United States and Europe enacted travel restrictions. We also observed a 51% decrease in the global flight network density decreased during this period. Regarding new COVID-19 cases, the United States had an unexpected surge regardless of travel restrictions. Finally, the layoffs for temporary workers in the tourism and airplane business increased by 4.3 fold in the weeks following Spain's decision to close its borders.

Summary Multiple distributed stream queries, which are executed on stream processing systems, need to be fine‐tuned to the compute cluster in order to harness the full potential of the hardware they run on. In this paper, we describe an automatic technique for conducting such stream query optimization in the presence of multiple stream jobs. During this autotuning process, we identify the structure of each program and conduct automatic program transformation to generate optimized unified streaming jobs. The operators on the unified secondary sample application are grouped into processing elements considering their performance characteristics and the stream graph topology structure to produce high performance stream query network. We implemented this multiple stream query optimization technique on a mechanism called Tahitica. We demonstrate our approach's ability for producing optimized stream query performance by comparing it to naive deployments using two real‐world stream processing applications in the domains of health care and search advertising. Our stream query optimization approach reported 7.1% throughput performance improvement compared to a naive deployment.

Invited Talk

We propose Distributed Neighbor Expansion (Distributed NE), a parallel and distributed graph partitioning method that can scale to trillion-edge graphs while providing high partitioning quality. Distributed NE is based on a new heuristic, called parallel expansion, where each partition is constructed in parallel by greedily expanding its edge set from a single vertex in such a way that the increase of the vertex cuts becomes local minimal. We theoretically prove that the proposed method has the upper bound in the partitioning quality. The empirical evaluation with various graphs shows that the proposed method produces higher-quality partitions than the state-of-the-art distributed graph partitioning algorithms. The performance evaluation shows that the space efficiency of the proposed method is an order-of-magnitude better than the existing algorithms, keeping its time efficiency comparable. As a result, Distributed NE can partition a trillion-edge graph using only 256 machines within 70 minutes.

Graph pattern matching algorithms to handle million-scale dynamic graphs are widely used in many applications such as social network analytics and suspicious transaction detections from financial networks. On the other hand, the computation complexity of many graph pattern matching algorithms is expensive, and it is not affordable to extract patterns from million-scale graphs. Moreover, most real-world networks are time-evolving, updating their structures continuously, which makes it harder to update and output newly matched patterns in real time. Many incremental graph pattern matching algorithms which reduce the number of updates have been proposed to handle such dynamic graphs. However, it is still challenging to recompute vertices in the incremental graph pattern matching algorithms in a single process, and that prevents the real-time analysis. We propose an incremental graph pattern matching algorithm to deal with time-evolving graph data and also propose an adaptive optimization system based on reinforcement learning to recompute vertices in the incremental process more efficiently. Then we discuss the qualitative efficiency of our system with several types of data graphs and pattern graphs. We evaluate the performance using million-scale attributed and time-evolving social graphs. Our incremental algorithm is up to 10.1 times faster than an existing graph pattern matching and 1.95 times faster with the adaptive systems in a computation node than naive incremental processing.

Deep neural networks are representation learning techniques. During training, a deep net is capable of generating a descriptive language of unprecedented size and detail in machine learning. Extracting the descriptive language coded within a trained CNN model (in the case of image data), and reusing it for other purposes is a field of interest, as it provides access to the visual descriptors previously learnt by the CNN after processing millions of images, without requiring an expensive training phase. Contributions to this field (commonly known as feature representation transfer or transfer learning) have been purely empirical so far, extracting all CNN features from a single layer close to the output and testing their performance by feeding them to a classifier. This approach has provided consistent results, although its relevance is limited to classification tasks. In a completely different approach, in this paper we statistically measure the discriminative power of every single feature found within a deep CNN, when used for characterizing every class of 11 datasets. We seek to provide new insights into the behavior of CNN features, particularly the ones from convolutional layers, as this can be relevant for their application to knowledge representation and reasoning. Our results confirm that low and middle level features may behave differently to high level features, but only under certain conditions. We find that all CNN features can be used for knowledge representation purposes both by their presence or by their absence, doubling the information a single CNN feature may provide. We also study how much noise these features may include, and propose a thresholding approach to discard most of it. All these insights have a direct application to the generation of CNN embedding spaces.

How to efficiently detect communities for dynamic graphs has attracted significant attention due to its widespread applications, such as “Recommended System” in social networking or “Anti-Money Laundry” for banks. However, with the growing size and the increasing of changing frequency for dynamic graphs, it is better to detect communities incrementally than applying community detection algorithm for each of the graph. In this paper, we propose a generalized bottom-up community detection framework to help standard community detection algorithms to detect communities in highly dynamic graphs incrementally. The evaluations on real data sets show that our generalized framework does have the ability to accelerate standard community detection methods for highly dynamic graphs (up to 96%). In addition, for super hubs in large scale graph, we also propose an approximate process to meet the requirement of real-time processing. By giving “mathematical proof”, “efficiency” and “accuracy” evaluations from real data sets, we demonstrate this approximate process can not only accelerate the community detection process but also can preserve the accuracy, as the normalized mutual information for regular communities and approximate communities is larger than 92%.

When graphs change over time, it is important to make the changes trackable for many graph-based applications. We propose an implementation of OLTP-oriented graph database that supports time-versioning. There has been a few snapshot-based approaches for supporting time-versions, but they usually require the full-restoration of the graph, and lack the resolution of the time space. Using a B-tree as the datastructure for the backend storage, our database allow fast and scalable support for restoring the arbitrary part of the graph, without slowing down the normal accesses to the current graph. Experimental results show that our scheme is much efficient than the straightforward solutions, in terms of space and performance.

Taking advantage of computing capabilities offered by modern parallel and distributed architectures is fundamental to run large-scale simulation models based on the Parallel Discrete Event Simulation (PDES) paradigm. By relying on this computing organization, it is possible to effectively overcome both the power and the memory wall, which are core limiting aspects to deliver high-performance simulations. This is even more the case when relying on the speculative Time Warp synchronization protocol, which could be particularly memory greedy. At the same time, some form of coordination, such as the computation of the Global Virtual Time (GVT), is required by Time Warp Systems. These coordination points could easily become the bottleneck of large-scale simulations, hindering an efficient exploitation of the computing power offered by large supercomputing facilities. In this paper we present ORCHESTRA, a coordination algorithm which is both wait-free and asynchronous. The nature of this algorithm allows any computing node to carry on simulation activities while the global agreement is reached, thus offering an effective building block to achieve scalable PDES. We claim that the general organization of ORCHESTRA could be adopted by different high-performance computing applications, thus paving the way to a more effective usage of modern computing infrastructures.

Streaming graph data mining has become a significant issue in high performance graph mining due to the increasing appearance of graph data sets as streams. In this paper we propose Acacia-Stream which is a scalable distributed streaming graph database engine developed with X10 programming language. Graph streams are partitioned using a streaming graph partitioner algorithm in Acacia-Stream and streaming graph processing queries are run on the graph streams. The partitioned data sets are persisted on secondary storage across X10 places. We investigate on the use of three different streaming graph partitioner algorithms called hash, Linear Deterministic Greedy, and Fennel algorithms and report their performance. Furthermore, to demonstrate Acacia-Stream's streaming graph processing capabilities we implement streaming triangle counting with Acacia-Stream. We present performance results gathered from Acacia-Stream with different large scale streaming data sets in both horizontal and vertical scalability experiments. Furthermore, we compare streaming graph loading performance of Acacia-Stream with Neo4j and Oracle's PGX graph database servers. From these experiments we observed that Acacia-Stream's Fennel partitioner based graph uploader can upload a 948MB rmat22 graph in 1283.42 seconds which is 38% faster than PGX graph database server and 12.8 times faster than Neo4j database server. Acacia-Stream's Streaming Partitioner's batch size adjustments based optimizations reduced the time used by the network communications almost by half.

Continuous collision detection (CCD) is a process to interpolate the trajectory of polygons and detect collisions between successive time steps. However, primitive-level CCD is a very time-consuming process especially for a large number of moving polygons. Over the years, a number of approaches have been proposed to improve the computational efficiency of CCD by culling out the non-colliding primitives before exact overlap tests. These approaches have two fundamental disadvantages. First, they are mainly designed for self-and pairwise CCD and thus the performance gain would be limited when they are applied to large-scale scenes that contain thousands of moving polygons. Second, they are designed as sequential processes appropriate for execution on a single processor. Therefore, deploying them on high-performance parallel computing systems would not increase their computational efficiency.

There are many large-scale graphs in real world such as Web graphs and social graphs. The interest in large-scale graph analysis is growing in recent years. Breadth-First Search (BFS) is one of the most fundamental graph algorithms used as a component of many graph algorithms. Our new method for distributed parallel BFS can compute BFS for one trillion vertices graph within half a second, using large supercomputers such as the K-Computer. By the use of our proposed algorithm, the K-Computer was ranked 1st in Graph500 using all the 82,944 nodes available on June and November 2015 and June 2016 38,621.4 GTEPS. Based on the hybrid BFS algorithm by Beamer (Proceedings of the 2013 IEEE 27th International Symposium on Parallel and Distributed Processing Workshops and PhD Forum, IPDPSW ’13, IEEE Computer Society, Washington, 2013 ), we devise sets of optimizations for scaling to extreme number of nodes, including a new efficient graph data structure and several optimization techniques such as vertex reordering and load balancing. Our performance evaluation on K-Computer shows that our new BFS is 3.19 times faster on 30,720 nodes than the base version using the previously known best techniques.

Linked data mining has become one of the key questions in High Performance graph mining in recent years. However, the existing Resource Description Framework (RDF) database engines are not scalable and are less reliable in heterogeneous clouds. In this paper we describe the design and implementation of Acacia-RDF which is a scalable distributed RDF graph database engine developed with X10 programming language to solve this issue. Acacia-RDF partitions the RDF data sets into subgraphs following vertex cut paradigm. The partitioned data sets are persisted on secondary storage across X10 places. We developed a scalable SPARQL (an RDF query language) processor for Acacia-RDF which operates on top of partitioned RDF data. Furthermore, we designed and implemented a replication based fault tolerance mechanism for Acacia-RDF. We present performance results gathered from Acacia with different scales of LUBM (Lehigh University Benchmark) RDF benchmark data sets. We make a comparison of Acacia-RDF's performance against Neo4j graph database server. From the scalability experiments conducted upto 16 X10 places, we observed that Acacia-RDF scales well with LUBM data sets. Acacia-RDF reported less than ten seconds elapsed times on 16 places for running the first and the third queries of the LUBM benchmark on LUBM 160 universities data set with 3.6 million vertices and 28.5 million edges which was 1.7GB in size. Through this work we describe and demonstrate the use of X10 language for development of scalable RDF graph data management systems.

Community detections for large-scale real world networks have been more popular in social analytics. In particular, dynamically growing network analyses become important to find long-term trends and detect anomalies. In order to analyze such networks, we need to obtain many snapshots and apply same analytic methods to them. However, it is inefficient to extract communities from these whole newly generated networks with little differences every time, and then it is impossible to follow the network growths in the real time. We proposed an incremental community detection algorithm for high-volume graph streams. It is based on the top of a well-known batch-oriented algorithm named DEMON [1]. We also evaluated performance and precisions of our proposed incremental algorithm with real-world big networks with up to 410,236 vertices and 2,439,437 edges and computed in less than one second to detect communities in an incremental fashion - which achieves up to 107 times faster than the original algorithm without sacrificing accuracies.

Breadth-First Search(BFS) is one of the most fundamental graph algorithms used as a component of many graph algorithms. Our new method for distributed parallel BFS can compute BFS for one trillion vertices graph within half a second, using large supercomputers such as the K-Computer. By the use of our proposed algorithm, the K-Computer was ranked 1st in Graph500 using all the 82,944 nodes available on June and November 2015 and June 2016 38,621.4 GTEPS. Based on the hybrid-BFS algorithm by Beamer[3], we devise sets of optimizations for scaling to extreme number of nodes, including a new efficient graph data structure and optimization techniques such as vertex reordering and load balancing. Performance evaluation on the K shows our new BFS is 3.19 times faster on 30,720 nodes than the base version using the previously-known best techniques.

Recently, large-scale graph analytics has become a very popular topic owing to the emergence of gigantic graphs whose number of vertices and edges is in millions, billions or even trillions. Many graph analytics libraries and frameworks have been proposed with various computational models and programming languages to deal with such graphs. X10 programming language is a PGAS language that aims at both software performance and programmer's productivity. We introduce ScaleGraph library developed using X10 programming to illustrate the use of X10 for large-scale graph analytics. ScaleGraph library provides XPregel framework that is inspired by Google's Pregel computation model, serving as a building block for implementing graph kernels. We also optimized X10 runtime in some parts such as collective communication and memory management. We evaluated the performance and scalability of ScaleGraph libraries. The result shows that most graph kernels have good performance and scalability. ScaleGraph library is 9.4 times faster than Giraph in the experiment of PageRank with 16 machine nodes. To the best of our knowledge, ScaleGraph is the first X10-based library to address performance, scalability and productivity issues in dealing with large-scale graph analytics.

In this paper, we develop a pipelining schema for image deep learning on GPU cluster to leverage heavy workload of training procedure. In addition, it is usually necessary to train multiple models to obtain a good deep learning model due to the limited a priori knowledge on deep neural network structure. Therefore, adopting parallel and distributed computing appears is an obvious path forward, but the mileage varies depending on how amenable a deep network can be parallelized and the availability of rapid prototyping capabilities with low cost of entry. In this work, we propose a framework to organize the training procedures of multiple deep learning models into a pipeline on a GPU cluster, where each stage is handled by a particular GPU with a partition of the training dataset. Instead of frequently migrating data among the disks, CPUs, and GPUs, our framework only moves partially trained models to reduce bandwidth consumption and to leverage the full computation capability of the cluster. In this paper, we deploy the proposed framework on popular image recognition tasks using deep learning, and the experiments show that the proposed method reduces overall training time up to dozens of hours compared to the baseline method.

Analyzing large-scale traffics by simulation needs repeating execution many times with various patterns of scenarios or parameters. Such repeating execution brings about big redundancy because the change from a prior scenario to a later scenario is very minor in most cases, for example, blocking only one of roads or changing the speed limit of several roads. In this paper, we propose a new redundancy reduction technique, called exact-differential simulation, which enables to simulate only changing scenarios in later execution while keeping exactly same results as in the case of whole simulation. The paper consists of two main efforts: (i) a key idea and algorithm of the exact-differential simulation, (ii) a method to build large-scale traffic simulation on the top of the exact-differential simulation. In experiments of Tokyo traffic simulation, the exact-differential simulation shows 7.26 times as much elapsed time improvement in average and 2.26 times improvement even in the worst case as the whole simulation.

Using the Cloud for large-scale distributed simulations, such as agent-based traffic simulations, sounds like a good idea, as it is possible to provision and release easily processing nodes (e.g., Virtual machines) in the Cloud. However, the question is complex as it involves users' objectives, such as, time to process the simulation and cost of the simulation, and because the workload evolves in distributed simulations, in each node and the whole system, and this impact the resource provisioning plans. This paper proposes two main contributions: (i) a method for efficient utilization of computational resources for distributed agent-based simulations, providing a mechanism that adapts the resource provisioning to users' objectives and workload evolution, and (ii) a staged asynchronous migration technique to limit the migration overhead when the number of workers change. Our preliminary experimental results on a 24 hour scenario of traffic in the city of Tokyo show that our system outperforms a static provisioning by 12% in average and 23% during periods when workload changes a lot.

This paper introduces some of the example applications handling extremely big data with supercomputers such as large-scale network analysis, X10-based large-scale graph analytics library, Graph500 benchmark, and billion-scale social simulation.

Fast processing for extremely large-scale graph is becoming increasingly important in various domains such as health care, social networks, intelligence, system biology, and electric power grids. The GIM-V algorithm based on MapReduce programing model is designed as a general graph processing method for supporting petabyte-scale graph data. On the other hand, recent large-scale data-intensive computing systems tend to employ GPU accelerators to gain good peak performance and high memory bandwidth, however, the validity of acceleration, including optimization techniques, of the GIM-V algorithm using GPUs is an open problem. To address the problem, we implemented a multi-GPU-based GIM-V application with load balance optimization between GPU devices. Our implementation extends the existing MapReduce library for supporting multi-GPU-environments using the MPI library and optimizes load balance between GPU devices by employing task scheduling-based graph partitioning. We conducted our implementation on the TSUBAME2.0 supercomputer using 256 nodes (6144 hyper-threaded CPU cores, 768 GPUs). The results exhibit that our GPU-based implementation performed 87.04 ME/s on 230(1.07 billion) vertices and 234(17.2 billion) edges, and 1.52 times faster than the CPU-based naive implementation with 2^29 vertices and 233edges. We also studied the performance characteristics of our implementation and load balance optimization technique.

Betweenness centrality is a measure that determines the relative importance of a vertex (or an edge) within a graph based on shortest paths. Recently, large-scale graphs have emerged in many different domains, as social networks, road networks, protein interaction networks, etc., and they are too large to fit into the memory of a single SMP. The algorithm proposed by Edmonds et al. [1] is capable of running on distributed memory systems. However, the algorithm does not expose intra-node parallelism. In this paper we investigated the inter- and intra-node parallelism of computing betweenness centrality on distributed memory systems. We developed the implementation based on the algorithm proposed by Edmonds et al. using X10 programming language [2]. We further improved the performance of the implementation by optimizing the network transport of the X10 runtime. We thoroughly evaluated the performance of our implementation on synthetic graphs of various scales against the existing implementation of Edmonds' algorithm from PBGL. We estimated the betweenness centrality of the huge Twitter networks [3] and found that its distribution follows a power law.

In this paper we propose a highly optimized parallel and distributed BFS on GPU for Graph500 benchmark. We evaluate the performance of our implementation using TSUBAME2.0 supercomputer. We achieve 317 GTEPS (billion traversed edges per second) with scale 35 (a large graph with 34.4 billion vertices and 550 billion edges) using 1366 nodes and 4096 GPUs. With this score, TSUBAME2.0 supercomputer is ranked fourth in the ranking list announced in June 2012. We analyze the performance of our implementation and the result shows that inter-node communication limits the performance of our GPU implementation. We also propose SIMD Variable-Length Quantity (VLQ) encoding for compression of communication data with GPU.

In recent years many online graph database service providers have started their operations in public clouds due to the growing demand for graph data storage and analysis. While this trend continues to rise there is little support available for benchmarking graph database systems in cloud environments. In this paper we introduce XGDBench which is a graph database benchmarking platform for cloud computing systems. XGDBench has been designed to operate in current cloud service infrastructures as well as on future exascale clouds. We extend the famous Yahoo! Cloud Serving Benchmark to the domain of graph database benchmarking. The benchmark platform is written in X10 which is a PGAS language intended for programming future exascale systems. We describe the architecture of the XGDBench focusing on its importance for exascale clouds. We did a cluster analysis to compare the realistic nature of XGDBench data generator and saw that the community structures of synthetic graphs produced by XGDBench outperforms RMAT. We also conducted performance evaluation of four famous graph data stores AllegroGraph, Fuseki, Neo4j, an OrientDB using XGDBench on Tsubame 2.0 HPC cloud environment.

This paper describes highly scalable X10-based agent simulation platform called XAXIS. XAXIS is designed to handle millions or billions of agents on recent highly distributed and parallel computing environments with more than hundreds of CPU cores. To make the runtime scalable on such environments, we need to redesign and implement the simulation middleware. In this paper, we propose the software design, implementation on X10, one of the state-of-the-art PGAS language, and then application to large-scale traffic simulation. By using 192 CPU cores in distributed memory computing environment, the performance scalability is achieved with a traffic simulation.

Scalable analysis of massive graphs has become a challenging issue in high performance computing environments. ScaleGraph is an X10 library aimed for large scale graph analysis scenarios. This paper evaluates scalability of ScaleGraph library for degree distribution calculation, betweeness centrality, and spectral clustering algorithms. We make scalability evaluation by analyzing a synthetic Kronecker graph with 40.3 million edges (for all the three algorithms), and a real social network with 69 million edges (for degree distribution calculation) on Tsubame 2.0 distributed memory environment.

Large graph analysis has become a widely studied area in recent years. Clustering is one of the most important types of analysis that has versatile applications such as community detection in social networks, image segmentation, graph partitioning, etc. However, existing clustering algorithms do not intend for large scale graphs. To solve this problem, we implemented spectral clustering in X10, that is a programming language aimed for developing highly scalable applications on Post- Petascale supercomputers. Our spectral clustering is based on the algorithm proposed by Shi and Malik. After evaluating scalability and precision, we found that our implementations are scalable in terms of execution time and precise for analyzing real data.

Stream computing has emerged as a new processing paradigm that processes incoming data streams from large numbers of sensors in real time. At the same time, many recent efforts have shown the suitability of GPGPU s for batch-typed long-running applications. However, few studies conduct the applicability of GPGPU to stream computing and also our first experiment shows that the performance does not scale as expected if one introduces the GPGPU to stream computing applications. This paper presents the workload characterization of GPGPU-based stream computing. We especially focus on computing SVD (Singular Value Decomposition) with GPGPUs for small-sized matrix since it can be widely applicable to real-time stream-based data mining applications such as real-time anomaly detection. In this paper, we not only show the workload characterization of GPGPU-based stream computing but also propose the optimization approach of SVD for stream computing applications called “GPU task Parallelism” to leverage the sufficient capability of GPGPUs. This optimization offers new levels of scalability for real-time operations with large numbers of sensors. Our experimental results show that GPU task parallelism provides roughly four-fold performance gains against a quad-core CPU for matrixes of 300 × 300 to 500 × 500 data values. We also implemented a stream-based change-point and anomaly detection system based on the optimized SVD and stream computing system called System S. By porting the optimized version of SVD to the distributed stream computing system, it was easy to exploit multiple GPUs on multiple nodes. The performance results showed performance around 7.6 times faster than CPUs. The scalability of our proposed system was tested up to 1,525 sensors, which were simultaneously handled for change-point detection every 5 seconds.

Graph500 is a new benchmark to rank supercomputers with a large-scale graph search problem. We found that the provided reference implementations are not scalable in a large distributed environment. We devised an optimized method based on 2D partitioning and other methods such as communication compression and vertex sorting. Our optimized implementation can handle BFS (Breadth First Search) of a large graph with 236 (68.7 billion vertices) and 240 (1.1 trillion) edges in 10.58 seconds while using 1366 nodes and 16,392 CPU cores. This performance corresponds to 103.9 GE/s. We also studied the performance characteristics of our optimized implementation and reference implementations on a large distributed memory supercomputer with a Fat-Tree-based Infiniband network.

Stream programs have to be crafted carefully to maximize the performance gain that can be obtained from stream processing environments. Manual fine tuning of a stream program is a very difficult process which requires considerable amount of programmer time and expertise. In this paper we present Hirundo, which is a mechanism for automatically generating optimized stream programs that are tailored for the environment they run. Hirundo analyzes, identifies the structure of a stream program, and transforms it to many different sample programs with same semantics using the notions of Tri-Operator Transformation, Transformer Blocks, and Operator Blocks Fusion. Then it uses empirical optimization information to identify a small subset of generated sample programs that could deliver high performance. It runs the selected sample programs in the run-time environment for a short period of time to obtain their performance information. Hirundo utilizes these information to output a ranked list of optimized stream programs that are tailored for a particular run-time environment. Hirundo has been developed using Python as a prototype application for optimizing SPADE programs, which run on System S stream processing run-time. Using three example real world stream processing applications we demonstrate effectiveness of our approach, and discuss how well it generalizes for automatic stream program performance optimization.

Stream computing, also known as data stream processing, has emerged as a new processing paradigm that processes incoming data streams from tremendous numbers of sensors in a real-time fashion. Data stream applications must have low latency even when the incoming data rate fluctuates wildly. This is almost impossible with a local stream computing environment because its computational resources are finite. To address this kind of problem, we have devised a method and an architecture that transfers data stream processing to a Cloud environment as required in response to the changes of the data rate in the input data stream. Since a trade-off exists between application's latency and the economic costs when using the Cloud environment, we treat it as an optimization problem that minimizes the economic cost of using the Cloud. We implemented a prototype system using Amazon EC2 and an IBM System S stream computing system to evaluate the effectiveness of our approach. Our experimental results show that our approach reduces the costs by 80% while keeping the application's response latency low.

A new trend involves Web services such as Twitter beginning to publish streaming Web APIs that enable partners and end users to retrieve streaming data. By combining such push-based Web services and existing pull-based Web services, it is now possible for us to understand the current status or trends of the world in a more real-time way, such as real-time tracking of infectious disease, real-time crime prediction, or real-time marketing, and so various innovative business services are possible. For a system architecture to implement such services, the services are normally built from the scratch, and the performance and scalability depend upon the engineers' skills. In this paper we propose a real-time Web monitoring system called "StreamWeb" on top of a stream computing system called System S developed by IBM Research. The Stream Web system allows developers to easily describe their analytical algorithms for a variety of kinds of Web streaming data without worrying about the performance and scalability, and provides real-time and scalable Web monitoring for massive amounts of data. As an experimental proof-of-concept application, we built an application that monitors a list of keywords in the Twitter streaming data, and that displays any messages including the specified keywords onto a map of the physical location (from Google) where the message was posted. Our system can handle nearly 30 thousand Twitter messages per second on a system with 8 computing nodes. This prototype application confirms that we can build real-time Web monitoring systems while satisfying the needs for high software productivity and for system scalability.

ETL (Extract-Transform-Load) processing is filling an increasingly critical role in analyzing business data and in taking appropriate business actions based on the results. As the volume of business data to be analyzed increases and quick responses are more critical for business success, there are strong demands for scalable high-performance ETL processors. In this paper, we evaluate a distributed data stream processing engine called System S for those purposes. Based on the original motivation of building System S as a data stream processing engine, we first perform a qualitative study to see if the programming model of System S is suitable for representing an ETL workflow. Second we did performance studies with a representative ETL scenario. Through our series of experiments, we found that the SPADE programming model and its runtime environment naturally fits the requirements of handling massive amounts of ETL data in a highly scalable manner.

Programmers who develop Web applications often use dynamic scripting languages such as Perl, PHP, Python, and Ruby. For general purpose scripting language usage, interpreter-based implementations are efficient and popular but the server-side usage for Web application development implies an opportunity to significantly enhance Web server throughput. This paper summarizes a study of the optimization of PHP script processing. We developed a PHP processor, P9, by adapting an existing production-quality just-in-time (JIT) compiler for a Java virtual machine, for which optimization technologies have been well-established, especially for server-side application. This paper describes and contrasts microbenchmarks and SPECweb2005 benchmark results for a well-tuned configuration of a traditional PHP interpreter and our JIT compiler-based implementation, P9. Experimental results with the microbenchmarks show 2.5-9.5x advantage with P9, and the SPECweb2005 measurements show 20-30 % improvements. These results show that the acceleration of dynamic scripting language processing does matter in a realistic Web application server environment. CPU usage profiling shows our simple JIT compiler introduction reduces the PHP core runtime overhead from 45 % to 13 % for a SPECweb2005 scenario, implying that further improvements of dynamic compilers would provide little additional return unless other major overheads such as heavy memory copy between the language runtime and Web server frontend are reduced.

The performance of server-side applications is becoming increasingly important as more applications exploit the Web application model. Extensive work has been done to improve the performance of individual software components such as Web servers and programming language runtimes. This paper describes a novel approach to boost Web application performance by improving inter-process communication between a programming language runtime and Web server runtime. The approach reduces redundant processing for memory copying and the context switch overhead between user space and kernel space by exploiting the zero-copy data transfer methodology, such as the sendfile system call. In order to transparently utilize this optimization feature with existing Web applications, we propose enhancements of the PHP runtime, FastCGI protocol, and Web server. Our proposed approach achieves a 126% performance improvement with micro-benchmarks and a 44% performance improvement for a standard Web benchmark, SPECweb2005.

Web applications often use HTML templates to separate the webpage presentation from its underlying business logic and objects. This is now the de facto standard programming model for Web application development. This paper proposes a novel implementation for existing server-side template engines, FlyingTemplate, for (a) reduced bandwidth consumption in Web application servers, and (b) off-loading HTML generation tasks to Web clients. Instead of producing a fully-generated HTML page, the proposed template engine produces a skeletal script which includes only the dynamic values of the template parameters and the bootstrap code that runs on a Web browser at the client side. It retrieves a client-side template engine and the payload templates separately. With the goals of efficiency, implementation transparency, security, and standards compliance in mind, we developed FlyingTemplate with two design principles: effective browser cache usage, and reasonable compromises which restrict the template usage patterns and relax the security policies slightly but in a controllable way. This approach allows typical template-based Web applications to run effectively with FlyingTemplate. As an experiment, we tested the SPECweb2005 banking application using FlyingTemplate without any other modifications and saw throughput improvements from 1.6x to 2.0x in its best mode. In addition, FlyingTemplate can enforce compliance with a simple security policy, thus addressing the security problems of client-server partitioning in the Web environment.

PHP is well known as a programming language in the Web 2.0 era enabling agile server-side software development. It has officially supported SOAP messaging since version 5 through a C-based built-in library. In this paper we perform a thorough study of the capability of PHP as a Web service engine in both qualitative and quantitative aspects while comparing it with other Web service engines implemented in Java and C. We used Axis2 for this purpose as it is an open source web service engine whose implementation is available both in Java and C. We report that PHP as a web service engine performs competitively with Axis2 Java for Web services involving small payloads, and greatly outperforms it for larger payloads by 5-17 times. As the authors expected, Axis2 C performs best, but the experimental results demonstrate that PHP performance is closer to Axis2 C with larger payloads. This performance difference comes from the fact that the SOAP engine within the PHP runtime is implemented in C with a monolithic architecture, whereas Axis2 uses a more modular architecture for the flexible insertation of handlers for an assorted set of WS-* standards, and also that Axis2 uses a different data binding mechanism known as ADB (Axis2 Data binding). This paper is the first attempt to compare Web services engines implemented in PHP, Java and C, and the authors believe that this boosts the development of SOAP-based Web services in PHP by letting people know its decent performance score and high productivity characteristics.

Web services technology has emerged as a key infrastructure that enables business entities to interact with each other without any human inventions. In order for the technology to be widely used, especially in any field where a large volume of transactions may be processed, it is highly desirable that the Web services engine should tolerate such environments. In this paper, we present a novel approach for improving Web services performance. We first focus on the fundamental characteristics of the Web services in that the SOAP messages on the wire are mostly generated by machines and have a lot of similarities among the processed messages. By making use of these features and eliminating the redundant processing, we propose a new deserialization mechanism that reuses matching regions from the previously deserialized application objects from earlier messages, and only performs deserialization for a new region that would not be processed before. Through our experiments in this paper, we observed that our approach obtained a 288% performance gain (maximum) by incorporating the differential deserialization into the Axis SOAP engine.

XML (Extensible Markup Language) processing can incur significant runtime overhead in XML-based infrastructural middleware such as Web service application servers. This paper proposes a novel mechanism for efficiently processing similar XML documents. Given a new XML document as a byte sequence, the XML parser proposed in this paper normally avoids syntactic analysis but simply matches the document with previously processed ones, reusing those results. Our parser is adaptive since it partially parses and then remembers XML document fragments that it has not met before. Moreover, it processes safely since its partial parsing correctly checks the well-formedness of documents. Our implementation of the proposed parser complies with the JSR 63 standard of the Java API for XML Processing (JAXP) 1.1 specification. We evaluated Deltarser performance with messages using Google Web services. Comparing to Piccolo (and Apache Xerces), it effectively parses 35 % (106%) faster in a server-side use-case scenario, and 73 % (126%) faster in a client-side use-case scenario.

JiPANG(A Jini-based Portal Augmenting Grids) is a portal system and a toolkit which provides uniform access interface layer to a variety of Grid systems, and is built on top of Jini distributed object technology. JiPANG performs uniform higher-level management of the computing services and resources being managed by individual Grid systems such as Ninf, NetSolve, Globus, etc. In order to give the user a uniform interface to the Grids JiPANG provides a set of simple Java APIs called the JiPANG Toolkits, and furthermore, allows the user to interact with Grid systems, again in a uniform way, using the JiPANG Browser application. With JiPANG, users need not install any client packages before-hand to interact with Grid systems, nor be concerned about updating to the latest version. Such uniform, transparent services available in a ubiquitous manner we believe is essential for the success of Grid as a viable computing platform for the next generation.

Recent developments of global computing systems such as Ninf, NetSolve and Globus have opened up the opportunities for providing high-performance computing services over wide-area networks. However, most research focused on the individual architectural aspects of the system, or application deployment examples, instead of the necessary characteristics such systems should intrinsically satisfy, nor how such systems relate with each other. Our comparative study performs deployment of example applications of network-based libraries using Ninf, NetSolve, and CORBA systems. There, we discover that dedicated systems for global computing such as Ninf and NetSolve have management, programmability, and it does not suffer performance disadvantages over more generic distributed computing capabilities provided by CORBA. Such results indicate the advantage of dedicated global computing systems over general systems, stemming further basic research is necessary across multiple systems to identify the ideal software architectures for global computing.