Patent Description:
A graph database is a type of database that uses vertices and edges to represent and store data. A graph database can be used for many different types of problem domains. For example, graph databases can be used to store data for social, business, traffic, biology or other natural sciences applications. A graph database can store information about connected entities. An entity can be represented in the graph as a vertex. A connection (e.g., relationship) between entities can be represented in the graph as an edge between two vertices.

<CIT> describes a graph analytics appliance. The graph analytics appliance includes a router, a worklist scheduler, a processing unit, and an input/output unit. The router receives an abstraction program including a plurality of parallel algorithms for a query request from an abstraction program compiler residing on computational node or the graph analytics appliance. The worklist scheduler generates a prioritized plurality of parallel threads for executing the query request from the plurality of parallel algorithms. The processing unit executes multiple threads selected from the prioritized plurality of parallel threads. The input/output unit communicates with a graph database.

<CIT> describes techniques for fast processing of path-finding queries in large graph databases. A computer system receives a graph search request to find a set of result paths between one or more source vertices of a graph and one or more target vertices of the graph. The graph comprises vertices connected by edges. During a first pass, the computer system performs one or more breadth-first searches to identify a subset of edges of the graph. The one or more breadth-first searches originate at the one or more source vertices. After the first pass and during a second pass, the computer system performs one or more depth-first searches to identify the set of result paths. The one or more depth-first searches originate at the one or more target vertices. The one or more depth-first searches traverse at most the subset of edges of the graph.

<NPL>) describes empirical evaluation of various computing platforms including an Intel Xeon E5 CPU, a Nvidia Geforce GTX1070 GPU and an Xeon Phi <NUM> processor codenamed Knights Landing (KNL) in the domain of parallel graph processing.

The invention is defined by the independent claims, while preferred embodiments form the subject of the dependent claims.

The present disclosure involves systems, software, and computer implemented methods for configuring graph query parallelism for high system throughput. An example method includes: receiving a graph query to be executed against a graph database; determining system properties of a system in which the graph query is to be executed; determining algorithmic properties of at least one graph algorithm to be used to execute the graph query against the graph database; determining graph data statistics for the graph database; and for a first iteration of the graph query: determining first graph traversal estimations for the first iteration; determining a first estimated cost model for the first iteration based on the first graph traversal estimations; determining first estimated thread boundaries for performing parallel execution of the first iteration; generating first work packages of vertices to be processed during the execution of the first iteration based on the first estimated cost model; and providing the first work packages to a work package scheduler for scheduling the first work packages to execute the first iteration.

Implementations can include one or more of the following features. The work package scheduler can determine which of the first work packages are executed in parallel and which of the first work packages are executed sequentially. A next set of vertices to process in a second iteration can be determined and second graph traversal estimations, a second cost model, and second estimated thread boundaries for the second iteration can be determined. Second work packages can be generated for the second iteration and provided to the work package scheduler. The first estimated cost model can be used for the second and subsequent iterations and the second cost model is not determined for the second iteration. Generating the work packages can include creating a set of work packages that each have a substantially similar cost. The system properties can include cache sizes and memory access times. The algorithmic properties can include memory consumption and amount of memory accessed by atomic and non-atomic operations. The graph data statistics can include vertex degrees and vertex degree distribution information. The estimated graph traversal information can include estimates for number of newly-visited vertices and number of touched vertices for the first iteration. When algorithmic properties are determined for more than one graph algorithm, a first estimated cost model can be determined for each graph algorithm and a graph algorithm with a lowest first estimated cost model can be selected for further evaluation.

While generally described as computer-implemented software embodied on tangible media that processes and transforms the respective data, some or all of the aspects may be computer-implemented methods or further included in respective systems or other devices for performing this described functionality. The details of these and other aspects and embodiments of the present disclosure are set forth in the accompanying drawings and the description below.

Graph based data sets can be used to manage vast amounts of data for various types of applications, such as social networks, financial networks, traffic networks, and biology or other natural science applications. Graph based data sets can include, for example, thousands or even billions of vertices and edges. Applications that utilize a graph database may perform various types of queries, such as larger, more complex queries or smaller, less complex queries, either interactively or as batch jobs. For instance, a graph-based application such as a route prediction service for road networks may operate on relatively smaller graphs, but may process a larger number of concurrent queries as compared to other applications.

Graph processing can be inherently complex, since data sets can substantially differ and different types of graphs can be used. For example, scale-free graphs can have a power-law-based degree distribution with a long tail, such as for social networks, while other graphs can exhibit a relatively constant degree among most or all vertices, such as for road networks or electrical power grids. Furthermore, the behavior of different graph algorithms can be fundamentally different. For instance, some algorithms (e.g., breadth-first search (BFS) algorithms) can operate on a relatively small subset of the overall graph at a given point in time, while other algorithms (e.g., page rank (PR) algorithms) can operate more frequently on a complete graph. Additionally, performance objectives of graph applications may vary. For example, if only single queries are processed, latency may be a key performance metric. As another example, for other applications which perform a larger amount of concurrent queries, throughput may be a key performance metric.

Given the various aspects and complexities of graph databases, it can be difficult to implement high-performance graph computations. Improving performance of either multi-query or single query processing can be difficult. Single query optimizations can include algorithm selection (e.g., push or pull) and parallelization (e.g., multi-threading for faster processing). However, selecting a best algorithm or configuring parallelization to achieve performance gains may be challenging before queries are actually executed, for example, due to the algorithm and data set differences mentioned above. Additionally, optimizations can result in additional overhead, such as instance synchronization for parallel execution. Overhead may diminish performance gains or even result in lower performance. Optimizations such as NUMA (Non-Uniform Memory Access) awareness and code generation may reduce but likely will not completely remove overhead. For multi-query execution (e.g., concurrent queries), complexity is further increased as concurrent queries need to be scheduled and amounts of queries to be executed concurrently need to be configured.

With graph processing, increasing a number of threads can increase performance. However, if too many threads are created (e.g., beyond a "sweet spot" number of threads), contention for same data items among multiple threads can occur, which can decrease performance. Contention can occur, for example, when a given vertex may be visited from multiple other vertices. Different threads, each processing one of the multiple other vertices, may each want to visit the given vertex at a same time, resulting in contention. As another example, too many threads might result in work packages that are too small to compensate for overhead costs, thus harming rather than improving performance.

Graph database performance can be improved by automatically controlling a degree of parallelization during multi-query execution. Although multi-query execution is described, intra-query parallelism can also be controlled for single query execution. Controlling multi-query parallelization for graph databases can result in improved system utilization, lower synchronization cost, and efficiencies gained from a higher degree of concurrent execution. Automatically controlling multi-query parallelism can include: (<NUM>) sampling of vertices to determine graph statistics; (<NUM>) deriving parallelization constraints from algorithm and system properties; and (<NUM>) generating work packages of suitable sizes based on the graph statistics and the parallelization constraints. The multi-query parallelization approach can improve performance for different types of queries and different data sets with low overhead. Additionally, performance achieved through automatic multi-query parallelism can be more significant than performance achieved through manual optimization approaches.

For instance, overhead can be reduced, as compared to other approaches, by use of efficient generation of statistics and cost models. For example, a cost estimator model can incorporate traversal behavior estimators (e.g., based on an estimated new vertex set and estimated memory access). Overhead that may be incurred can be outweighed by resulting improvement of query execution. For instance, efficient methods can be used that result in minimal overhead during execution but that are accurate enough for good scheduling decisions. With the improved approach, resulting query execution efficiency can be increased, with an increased number of accumulated operations per time unit, for a variety of scenarios, such as for different types of graph data (e.g., data size, data type), different algorithms (e.g., local, non-local), and different types of concurrency (e.g., single-query or multi-query execution). In summary, the improved approach can include design and implementation of a runtime resource control system that schedules graph queries during execution based on latency-aware parallelization, cost-based work packaging, and selective sequential execution. The improved approach can be used for different types of algorithms, such as different BFS or PR algorithms, and for different amounts of concurrency.

<FIG> is a block diagram illustrating an example system <NUM> for configuring graph query parallelism for high system throughput. Specifically, the illustrated system <NUM> includes or is communicably coupled with a server <NUM>, a client device <NUM>, and a network <NUM>. Although shown separately, in some implementations, functionality of two or more systems or servers may be provided by a single system or server. In some implementations, the functionality of one illustrated system, server, or component may be provided by multiple systems, servers, or components, respectively.

A client application <NUM> running on the client device <NUM> can send a query to the server <NUM> to query a graph database <NUM>. A graph engine <NUM> can process the query. As described in more detail below, the graph engine <NUM> can intelligently configure parallel execution of the query (e.g. intra query parallelism) as well as inter-query parallelism.

The graph engine <NUM> can be designed to achieve three main goals: <NUM>) efficient system utilization; <NUM>) synchronization cost reduction; and <NUM>) concurrent execution efficiency. As described in more detail below with respect to <FIG>, the graph engine <NUM> can use different techniques to achieve these goals, including: (<NUM>) latency-aware parallelization; (<NUM>) cost-based work packaging; and (<NUM>) selective sequential execution. Additionally, the graph engine <NUM> can be designed for predictable behavior so that small changes (e.g., in a number of vertices or edges) between queries does not result in significantly higher or lower performance. The graph engine <NUM> can be configured to perform work in parallel such that parallel execution is faster than a sequential processing alternative. The graph engine <NUM> can be configured to share resources with other engines that may be configured. For example, the graph engine <NUM> may be part of a database system <NUM> and the database system <NUM> may have other running components, such as a relational engine. Accordingly, the graph engine <NUM> can be configured so as to not utilize all system resources (e.g., so that other processing of the database system <NUM> is not unacceptably adversely affected).

Graph computational iterations can have small elapsed times. Accordingly, computations that are employed by the graph engine <NUM> to predict behavior during execution can have a low overhead so that the time used for prediction is an acceptable portion (e.g., a small fraction) of the elapsed time of an iteration of a graph computation. The graph engine <NUM> can generate acceptable predictions while having the low, acceptable overhead.

In further detail, the graph engine <NUM> includes an algorithm building block / operator component <NUM>, an algorithm preparation component <NUM>, and a dynamic runtime backend component <NUM>. The algorithm building block / operator component <NUM> can be used for graph operations such as for BFS and/or page rank approaches, and can be used for multiple iterations of an algorithm.

The algorithm preparation component <NUM> can be used to determine (e.g., preselect) useful parameters in a preparation phase based on available data and assumptions. Algorithm preparation includes parameter and cost estimation performed by a parameter and cost estimator <NUM> and work package generation performed by a work package generator <NUM>. Algorithm preparation can be performed for each iteration of an algorithm. Different types of algorithms can be executed, such as breadth-first-search and page-rank algorithms. The parameter and cost estimator <NUM> can generate cost estimates using system properties <NUM>, algorithm properties <NUM>, and graph data statistics <NUM>, when building a cost model <NUM>. The system properties <NUM>, algorithm properties <NUM>, graph data statistics <NUM>, and generating a cost model are described in more detail below.

For latency-aware parallelization, a performance model for graph traversal algorithms can be applied by the parameter and cost estimator <NUM> when determining system parameters, such as a number of threads to use for parallel processing. Different numbers of threads may be optimal in terms of performance depending on the primitive operations performed by an algorithm and the size of intermediate data. As mentioned above, parallel execution may in some cases harm performance. Accordingly, the parameter and cost estimator <NUM> can use the cost model <NUM> to determine an upper and lower bound for parallel execution of a specific query.

The work package generator <NUM> can generate work packages of suitable sizes of vertices to be processed together, based on the graph data statistics <NUM> and the thread number constraints generated by the parameter and cost estimator <NUM>. That is, the work package generator <NUM> can generate packages given an upper thread boundary and cost, from the cost and parameter estimation process. Given the vertices to be processed in an iteration, the cost model includes an estimated cost per vertex. The work package generator <NUM> can generate work packages that have substantially a same cost.

The dynamic runtime backend component <NUM> includes a work package scheduler <NUM> that schedules work packages that are received from the work package generator <NUM>. The work package scheduler <NUM> can schedule work packages for execution by a job executor <NUM>. For example, work package(s) can be assigned to threads. A thread can then perform processing on a subset of vertices processed during a query iteration. As described in more detail below, the work package scheduler <NUM> can control execution of a core graph algorithm and can handle conditions that are unknown prior to execution of the core graph algorithm.

As used in the present disclosure, the term "computer" is intended to encompass any suitable processing device. For example, although <FIG> illustrates a single server <NUM>, and a single client device <NUM>, the system <NUM> can be implemented using a single, stand-alone computing device, two or more servers <NUM>, or two or more client devices <NUM>. Indeed, the server <NUM> and the client device <NUM> may be any computer or processing device such as, for example, a blade server, general-purpose personal computer (PC), Mac®, workstation, UNIX-based workstation, or any other suitable device. In other words, the present disclosure contemplates computers other than general purpose computers, as well as computers without conventional operating systems. Further, the server <NUM> and the client device <NUM> may be adapted to execute any operating system, including Linux, UNIX, Windows, Mac OS®, Java™, Android™, iOS or any other suitable operating system. According to one implementation, the server <NUM> may also include or be communicably coupled with an e-mail server, a Web server, a caching server, a streaming data server, and/or other suitable server.

Interfaces <NUM> and <NUM> are used by the client device <NUM> and the server <NUM>, respectively, for communicating with other systems in a distributed environment - including within the system <NUM> - connected to the network <NUM>. Generally, the interfaces <NUM> and <NUM> each comprise logic encoded in software and/or hardware in a suitable combination and operable to communicate with the network <NUM>. More specifically, the interfaces <NUM> and <NUM> may each comprise software supporting one or more communication protocols associated with communications such that the network <NUM> or interface's hardware is operable to communicate physical signals within and outside of the illustrated system <NUM>.

The server <NUM> includes one or more processors <NUM>. Each processor <NUM> may be a central processing unit (CPU), a blade, an application specific integrated circuit (ASIC), a field-programmable gate array (FPGA), or another suitable component. Generally, each processor <NUM> executes instructions and manipulates data to perform the operations of the server <NUM>. Specifically, each processor <NUM> executes the functionality required to receive and respond to requests from the client device <NUM>, for example.

Regardless of the particular implementation, "software" may include computer-readable instructions, firmware, wired and/or programmed hardware, or any combination thereof on a tangible medium (transitory or non-transitory, as appropriate) operable when executed to perform at least the processes and operations described herein. Indeed, each software component may be fully or partially written or described in any appropriate computer language including C, C++, Java™, JavaScript®, Visual Basic, assembler, Perl®, any suitable version of 4GL, as well as others. While portions of the software illustrated in <FIG> are shown as individual modules that implement the various features and functionality through various objects, methods, or other processes, the software may instead include a number of sub-modules, third-party services, components, libraries, and such, as appropriate. Conversely, the features and functionality of various components can be combined into single components as appropriate.

The server <NUM> includes memory <NUM>. In some implementations, the server <NUM> includes multiple memories. The memory <NUM> may include any type of memory or database module and may take the form of volatile and/or non-volatile memory including, without limitation, magnetic media, optical media, random access memory (RAM), read-only memory (ROM), removable media, or any other suitable local or remote memory component. The memory <NUM> may store various objects or data, including caches, classes, frameworks, applications, backup data, business objects, jobs, web pages, web page templates, database tables, database queries, repositories storing business and/or dynamic information, and any other appropriate information including any parameters, variables, algorithms, instructions, rules, constraints, or references thereto associated with the purposes of the server <NUM>.

The client device <NUM> may generally be any computing device operable to connect to or communicate with the server <NUM> via the network <NUM> using a wireline or wireless connection. In general, the client device <NUM> comprises an electronic computer device operable to receive, transmit, process, and store any appropriate data associated with the system <NUM> of <FIG>. The client device <NUM> can include one or more client applications, including the client application <NUM>. A client application is any type of application that allows the client device <NUM> to request and view content on the client device <NUM>. In some implementations, a client application can use parameters, metadata, and other information received at launch to access a particular set of data from the server <NUM>. In some instances, a client application may be an agent or client-side version of the one or more enterprise applications running on an enterprise server (not shown).

The client device <NUM> further includes one or more processors <NUM>. Each processor <NUM> included in the client device <NUM> may be a central processing unit (CPU), an application specific integrated circuit (ASIC), a field-programmable gate array (FPGA), or another suitable component. Generally, each processor <NUM> included in the client device <NUM> executes instructions and manipulates data to perform the operations of the client device <NUM>. Specifically, each processor <NUM> included in the client device <NUM> executes the functionality required to send requests to the server <NUM> and to receive and process responses from the server <NUM>.

The client device <NUM> is generally intended to encompass any client computing device such as a laptop/notebook computer, wireless data port, smart phone, personal data assistant (PDA), tablet computing device, one or more processors within these devices, or any other suitable processing device. For example, the client device <NUM> may comprise a computer that includes an input device, such as a keypad, touch screen, or other device that can accept user information, and an output device that conveys information associated with the operation of the server <NUM>, or the client device <NUM> itself, including digital data, visual information, or a GUI <NUM>.

The GUI <NUM> of the client device <NUM> interfaces with at least a portion of the system <NUM> for any suitable purpose, including generating a visual representation of the client application <NUM>. In particular, the GUI <NUM> may be used to view and navigate various Web pages, or other user interfaces. Generally, the GUI <NUM> provides the user with an efficient and user-friendly presentation of business data provided by or communicated within the system. The GUI <NUM> may comprise a plurality of customizable frames or views having interactive fields, pull-down lists, and buttons operated by the user. The GUI <NUM> contemplates any suitable graphical user interface, such as a combination of a generic web browser, intelligent engine, and command line interface (CLI) that processes information and efficiently presents the results to the user visually.

Memory <NUM> included in the client device <NUM> may include any memory or database module and may take the form of volatile or non-volatile memory including, without limitation, magnetic media, optical media, random access memory (RAM), read-only memory (ROM), removable media, or any other suitable local or remote memory component. The memory <NUM> may store various objects or data, including user selections, caches, classes, frameworks, applications, backup data, business objects, jobs, web pages, web page templates, database tables, repositories storing business and/or dynamic information, and any other appropriate information including any parameters, variables, algorithms, instructions, rules, constraints, or references thereto associated with the purposes of the client device <NUM>.

There may be any number of client devices <NUM> associated with, or external to, the system <NUM>. For example, while the illustrated system <NUM> includes one client device <NUM>, alternative implementations of the system <NUM> may include multiple client devices <NUM> communicably coupled to the server <NUM> and/or the network <NUM>, or any other number suitable to the purposes of the system <NUM>. Additionally, there may also be one or more additional client devices <NUM> external to the illustrated portion of system <NUM> that are capable of interacting with the system <NUM> via the network <NUM>. Further, the term "client", "client device" and "user" may be used interchangeably as appropriate without departing from the scope of this disclosure. Moreover, while the client device <NUM> is described in terms of being used by a single user, this disclosure contemplates that many users may use one computer, or that one user may use multiple computers.

<FIG> illustrates an example system <NUM> that shows interactions between graph processing engine and scheduler components. The system <NUM> illustrates how the result from operations at one component is used in a succeeding component, and how certain boundaries are set. <FIG> also illustrates which properties of system, algorithm, and data are taken into consideration by which subcomponent. In <FIG>, different types of lines highlight the flow of information between components, with dashed lines highlighting access to upfront prepared data, solid lines highlighting data flow between components, and dotted lines highlighting data for next iterations.

In further detail, a preparation operation performed by a parameter and cost estimator <NUM> (which can be the parameter and cost estimator <NUM> of <FIG>) generates information for latency-aware parallelization and therefore about a most-suitable execution mode. Cost-based work packaging performed by a work package generator <NUM> (which may be the work package generator <NUM> of <FIG>) generates work packages that fulfill a goal of equal work distribution with a sufficient amount of work per package. A work package scheduler <NUM> (which can be the work package scheduler <NUM> of <FIG>) can use work package information <NUM> received from the work package generator <NUM> for scheduling work packages, either in parallel or, in some cases, with selective sequential execution. For instance, the work package scheduler <NUM> can take into account dynamic system behavior that is not considered by the parameter and cost estimator <NUM> or the work package generator <NUM>.

In further detail, the parameter and cost estimator <NUM> can generate a cost estimation using a cost model <NUM> that incorporates system properties <NUM>, algorithmic properties <NUM>, and graph data statistics <NUM>. System properties <NUM> can include, for example, cache sizes <NUM> and access times <NUM>. Dynamic system properties like the access times <NUM> can be determined prior to experiments using a degree count benchmark. Similarly, static system properties such as the cache sizes <NUM> can also be determined prior to experiments using appropriate tools such as a CPU analyzer tool.

The algorithmic properties <NUM> can include, for example, memory consumption information <NUM> and processing costs <NUM>. Algorithmic properties <NUM> can be obtained by counting, for example, by a query compiler, respective operations and touched memory. The graph data statistics <NUM> can include, for example, information about vertex degrees <NUM> and degree distribution <NUM>.

The parameter and cost estimator <NUM>, as part of a latency-aware parallelization approach, can generate parameters to be used by other components. For example, the parameter and cost estimator <NUM> can generate thread boundaries <NUM> (e.g., a minimum and maximum number of threads) to be provided as runtime configuration information to the work package scheduler <NUM>. As another example, the parameter and cost estimator <NUM> can determine a target number of work packages <NUM>, to be provided to the work package generator <NUM>.

While latency-aware parallelization tries to provide optimal parameters, such as thread count bounds for parallel execution and number of work packages, cost-based work packaging performed by the work package generator <NUM> involves optimizing the work packages. On a large scale, the average degree and the related work of sets of vertices can be described statistically. But if the variance of the edge degrees is high, or the number of vertices in a partition is small, the potential work per partition might be non-uniformly distributed between different partitions, resulting in inefficiencies due to the bulk-synchronous execution. Accordingly, some work packages may take much longer than others. To address the problem of non-uniform distribution of work between packages, input data statistics (e.g., the graph data statistics) can be used. For instance, for cases with a high vertex degree variance and a low numbers of vertices, work packages can be generated that are based on a vertex and edge performance model, in which vertices are iterated over and the out degree of vertices are obtained until a work share is exceeded for a particular work package. A number of work packages can be limited to a predefined multiple (e.g., <NUM> times) of the maximum usable level of parallelism to avoid over-parallelization and resulting effects such as contention. Additionally, work packages can be ordered (e.g., in an execution order <NUM>) so that work packages with a high cost (e.g., due to a single dominating vertex) are executed first. For cases where the number of vertices is high or the variance is low, for efficiency reasons, a static partitioning can be used (e.g., of equally-sized partitions). In general, the number of work packages is much larger than the used number of cores, which allows the runtime to react on dynamic execution behavior.

While the work page generator <NUM> can generate intelligently-sized work packages, the work package generator <NUM>, since not operating at runtime, cannot address runtime effects such as the dynamic aspect of contention. The work package scheduler <NUM>, however, can be configured to deal with these dynamic aspects and to handle the fact that in some cases sequential execution may be more efficient than parallel execution. The work package scheduler <NUM> performs, for example, the following functions: <NUM>) assigning of work to worker threads; and <NUM>) controlling if the work is executed sequentially or in parallel. When the execution of a task starts, the work package scheduler <NUM> requests worker threads from the system according to the upper thread boundary. When a worker thread gets assigned, the worker thread registers itself with the work package scheduler <NUM> and requests a work package. The work package scheduler <NUM> checks if the number of registered worker threads is higher than the minimum boundary for parallel execution. If the number of registered worker threads is higher than the minimum boundary for parallel execution, the work package scheduler <NUM> assigns a work package to the registered worker threads for parallel execution. If the number of registered worker threads is not higher than the minimum boundary for parallel execution, the work package scheduler <NUM> assigns one worker thread to execute a package sequentially, while the other worker threads wait until the package is completed. Then, the work package scheduler <NUM> reevaluates the worker thread situation. This is repeated for a limited number of sequential packages after which the work package scheduler <NUM> releases all but one worker thread and completes the execution sequentially. This approach avoids a central scheduler that needs to deal with many different tasks which might run a very short time and might be of different types, such as relational and graph tasks. The work package scheduler <NUM> also executes the tasks using an optimal parallelism strategy, taking into account concurrently-executed tasks. Especially for scenarios with concurrent queries, the work package scheduler <NUM> can schedule to prefer sequential execution of a single query over parallel execution of a single query, to avoid over-parallelization.

The work package scheduler <NUM> schedules packages for a current iteration of the query. In some implementations, the work package scheduler <NUM> provides data about (e.g., for) a next iteration <NUM>. For example, the next set of vertices to be processed in the next iteration can be provided to the parameter and cost generator <NUM>, for generation of a next code model and next parameters for the next iteration. As another example, actual measured performance costs of execution of the current iteration can be provided to the parameter and cost estimator <NUM>, and, in some implementations, the parameter and cost estimator <NUM> can use the current iteration performance information for generating next estimates.

<FIG> illustrates an example system <NUM> that shows interactions and dependencies between statistical values, estimations, and work package generation. The system <NUM> includes a vertex sampling and statistic generation portion <NUM>, a thread number selection and static cost estimation portion <NUM>, and a work package generation portion <NUM>. The vertex sampling and statistic generation portion <NUM> includes details for estimators that can approximate the size of certain vertex sets. The thread number selection and static cost estimation portion <NUM> can include details for estimating static costs per vertex, including use of a cost model for processing vertices and criteria for determining a number of threads (which may be limited by determined thread boundaries) to use for processing those vertices. The work package generation portion <NUM> includes details for generating work packages.

The vertex sampling and statistics generation portion <NUM> can include generation of graph statistics <NUM> based on a frontier queue <NUM> that includes all of the vertices that may be processed and an adjacency list <NUM>. Specific graph statistics <NUM> are described in more detail below with respect to <FIG>. Vertex sampling and statistics generation can vary based on a type of graph algorithm. For example, for topology-centric algorithm such as page rank, where the vertices stay the same for all iterations, preprocessing of statistics generation and vertex sampling can be performed once, up front. As another example, for data-driven algorithm such as breadth-first algorithms, the preprocessing can be performed for every iteration as for every iteration a different set of vertices may be processed.

Good statistical data related to the vertices to be processed can be desired, since the vertices, their related degree and the related work can vary between iterations. Global statistics generation can be performed so as to keep overhead to an acceptable level. An inexpensive way to gather statistical data is to collect statistics at creation time from the index data structure for the graph topology (e.g., the adjacency list <NUM>). Here, statistics for edges and vertices can be gathered and stored, including the mean and maximum vertex out degrees. Vtouched <NUM>, Elocal <NUM>, and Flocal <NUM> can be derived, as described below. Flocal <NUM> can also be referred to as Vnew, e.g., vertices that are newly found after each iteration.

At run time, a determination can be made as to whether the global statistics are to be used for cost estimation or if local statistics are to be computed and used for the current iteration. The determination can be based on a ratio of maximum vertex degree to mean vertex degree, as a variance metric that reflects variance. The variance metric can be compared to a threshold. For instance, from experimentation, a threshold of <NUM> can be determined to be effective as a variance threshold. If the variance metric is below the threshold (e.g., low variance) then the cost estimation can use the global statistics. If the variance metric is above the threshold (e.g., high variance), then local statistics can be computed and used. For example, |Vtouched| and |Vnew| values can be computed on a subset of current data (e.g., up to the first <NUM>,<NUM> vertices) using real vertex degrees, and global values can be extrapolated from the subset. The computation of local statistics can be parallelized to minimize overhead.

For graph traversal algorithms, the number of starting vertices and processed edges are typically important cost factors. For cost modeling, two important factors are the number of first time newly visited vertices (|Vnew|) and the number of vertices that will be touched in the next iteration (|Vtouched|). The to-be-visited vertices are directly related to the found vertices, while touched vertices relate to the amount of memory that is shared, such as duplicate filters during a graph traversal. Graph processing can involve traversal operations that traverse a graph from vertex to vertex, using connected edges. The touched vertices are vertices that have been visited already from another vertex. The new found vertices are vertices that are newly visited in each iteration of the graph traversal.

Obtaining exact Vnew and Vtouched values would require executing the graph algorithm. Rather than executing the algorithm, estimations can be performed in which whether a vertex is visited or touched is modeled using a conditional probability process. A model for estimating traversal behavior can be based on a general assumption that the probability for all vertices to be visited by an edge is identical. Other assumptions for the model that follow include assuming that none of the following apply: <NUM>) a correlation between different vertices; <NUM>) structural effects like a rich club effect (e.g., level-dependent); or the graph being a multigraph (e.g., with increased probability for some vertices).

<FIG> is a table <NUM> that includes information about parameters for traversal behavior estimation. A parameter V <NUM> represents all vertices in the graph. A parameter Vtouched <NUM> represents vertices that are touched via edges. A parameter Vnew <NUM> represents vertices that are newly found after each iteration. A parameter Vreach represents vertices that are reachable via a graph traversal (e.g., vertices that are neither isolated nor without an incoming edge). A parameter Vlocal <NUM> represents a set of vertices in the current (local) queue. A parameter Flocal <NUM> represents a set of found vertices. As mentioned, Flocal <NUM> can correspond to Vnew <NUM>. A parameter Elocal <NUM> represents a set of vertex edges for the local queue Vlocal. A parameter Pv visits <NUM> represents a probability that a vertex will be visited by a vertex v. A parameter Vno visit <NUM> represents vertices that have not been visited before. A parameter Pno visit <NUM> represents a probability that a vertex has not been visited.

<FIG> illustrates equations <NUM> for estimating counts of touched vertices. An Equation (<NUM>) <NUM> illustrates that an estimation for |Vtouched| <NUM> can be based on a probability for a reachable vertex to be reached at least once from a vertex from a frontier queue v ∈ Vqueue <NUM>, aggregated for a set of all reachable vertices Vreach <NUM>.

Based on an assumption that the graph is not a multigraph and that each outgoing edge of a given vertex v is connected to another vertex, the probability for a specific vertex to be reached is pv visits <NUM>. As illustrated in an Equation (<NUM>) <NUM>, pv visits <NUM> can be calculated using calculation <NUM> of <MAT>, with deg+(v) being the out degree of a vertex v.

In some instances, it can be too costly to compute the probability for each vertex of the frontier queue, as the queue can be of substantial size. To reduce the costs of computing a probability for each vertex, a difference between the maximum and mean value of outgoing edges per vertex can be used. If the difference between the maximum and mean value of outgoing edges per vertex is small (e.g., less than a threshold), the mean value can be used as an approximation, as illustrated in an Equation (<NUM>) <NUM>. If the difference between the maximum and mean value of outgoing edges per vertex is not small (e.g., not less than the threshold), an extrapolation approach can be used that includes extrapolating the product of the probabilities from a sample of vertices in the queue at the beginning of each cost calculation. The sampling approach can use up to a predefined number of vertices (e.g., a first <NUM>,<NUM> vertices can be sampled). The predefined number can be selected to achieve a low latency and sufficient quality.

<FIG> illustrates equations <NUM> for estimating counts of newly found vertices. An Equation (<NUM>) <NUM>, an Equation (<NUM>) <NUM>, and an Equation (<NUM>) are similar to the Equation (<NUM>) <NUM>, the Equation (<NUM>) <NUM>, and the Equation (<NUM>) <NUM> described above with respect to <FIG>. That is, for the estimation of |Vnew|, a similar approach is used as for the estimation |Vtouched |, but in addition to being reachable each specific vertex needs also to be unvisited. Accordingly, Pno visit <NUM> is included in Equation (<NUM>) <NUM> and corresponding specific calculations <NUM> and <NUM> of <MAT> are included in Equation (<NUM>) <NUM> and Equation (<NUM>) <NUM>, respectively.

Referring again to <FIG>, for the thread number selection and static cost estimation portion <NUM>, in order to model performance, an assumption can be made that the cost is proportional to the number of vertices |Vqueue|that have to be processed, the related edges e of these vertices that have to be traversed, and the new vertices f ∈ Vnew that are found as a result of this traversal. Using the estimations about traversal, different cost estimations can be calculated in the thread number selection and static cost estimation portion <NUM>. For example, an estimate of an amount of memory that may be involved in the estimated traversals and operations can be determined.

In further detail, cost estimation be performed using a linear model that includes system properties <NUM>, algorithmic properties <NUM>, and the graph statistics <NUM> (e.g., statistics about the processed graph data). As mentioned, the system properties <NUM> can include cache sizes and memory access times. The algorithmic properties <NUM> can include an amount of touched data and the number of operations per edge and vertex, including computations, memory accesses, and atomic updates. For example, the algorithmic properties <NUM> can be used to determine Csub values <NUM> which each represent a sub cost for a given item (e.g., vertex or edge). Ctotal values <NUM>, <NUM>, and <NUM> can be determined based on respective Csub values <NUM>. The linear model can be used to estimate a resulting memory footprint M <NUM> that is used to determine the cache level the problem of the current traversal can fit into. A latency of the determined cache level can in turn be used as a parameter for a memory cost model Lmem <NUM> (e.g., for non-atomic operations). Additionally, the determined cache level can be used to compute Latomic(T) values <NUM> using the related thread boundaries. Latomic(T) values <NUM> can represent an amount of atomic latency that depends on an amount of memory M.

Estimated thread boundaries <NUM> and <NUM> can be determined through interpolation, as described in more detail below. As the estimated thread boundaries <NUM> and <NUM> are worst cases, the system can work well with contention minimization techniques such as small local buffers even without adaptations. The thread number selection and static cost estimation portion <NUM> can involve use of other parameters and calculations, as described in more detail below.

<FIG> is a table <NUM> that includes information about cost model parameters. A parameter i <NUM> represents an item of generic type, including vertex (v), (new) found vertex (f) or edge (e). A parameter I <NUM> represents a set of items of a generic type, including a set of vertices in the current (local) queue (Vlocal), a set of found vertices (Flocal), or a set of vertex edges for the local queue (Elocal). A parameter |I| <NUM> represents a number of elements in a given set. A parameter T <NUM> represents a number of threads used. A parameter M <NUM> represents an amount of accessed data. A parameter Csub(i,T,M) <NUM> represents a sub-cost of execution for a given item (e.g., processing only a vertex v or f or an edge e). A parameter Ctotal(v,T,M) <NUM> represents a total cost of parallel execution for a vertex v. A parameter Lop <NUM> represents a latency of an arithmetic operation. A parameter Lmem(M) <NUM> represents a latency of a non-atomic memory access, depending on the size M of the accessed data. A parameter Latomic(T, M) <NUM> represents a latency of an atomic operation, depending on the size M of the accessed data and the amount of threads T. A parameter Nops(i) <NUM> represents a number of arithmetic operations used to process item i. A parameter Nmem(i) <NUM> represents a number of memory operations (e.g., non-atomic load and stores) used to process item i. A parameter Natomics(i) <NUM> represents a number of atomic operations used to process item i.

Regarding the Latomic parameter <NUM>, an assumption can be made that parallel and sequential implementations are identical as the parallel code protects critical sections using atomic operations, while the sequential code can instead simply employ plain memory operations. This assumption can be modeled by setting the atomic update latency for a single thread Latomic(T = <NUM>, M) equal to the memory access latency Lmem(M) <NUM>, with both being dependent on the amount of accessed memory M <NUM>.

Cost estimation can include three main cost subcomponents: <NUM>) computations (e.g., operations), regular memory access operations (e.g. "mem") and atomic memory operations (e.g., atomics). A sub-cost Csub (i, T, M) <NUM> for the processing of a given item i can be computed, as described in more detail below.

<FIG> illustrates equations <NUM> related to cost estimation. An Equation (<NUM>) <NUM> illustrates computations for calculating Csub(i, T, M) <NUM> for a given item i. In order to compare the estimates for the parallel and the sequential cases, a total cost per vertex Ctotal(T,M) <NUM> can be used, which can be calculated as illustrated in an Equation (<NUM>). The cost Ctotal(T, M) <NUM> can be computed as a sum of the costs to process the vertex itself (e.g., equation portion <NUM>), the cost of its share of edges (e.g., equation portion <NUM>), and the newly found vertices (e.g., equation portion <NUM>).

Referring again to <FIG>, the work package generation portion <NUM> includes generation of a set of work packages W <NUM> that partitions work being parallelized, with respective work packages represented as wi <NUM>. As described above, the work package generator can be provided a target work package count (e.g., |W| <NUM>). The value |W| can be calculated as <MAT> where |Wmax| is the maximum work package count.

<FIG> is a table <NUM> that includes information about parameters for work package and thread boundaries estimation. A parameter CT overhead <NUM> represents a start cost for a single thread. A parameter CT min <NUM> represents a minimum work per thread (e.g., that is larger than CT overhead. A parameter Cpara startup <NUM> represents a start cost for parallel execution. A parameter P <NUM> represents a maximum number of cores. Parameters Tmin and Tmax <NUM> represent a minimum and maximum thread bound, respectively. Parameters Jmin and Jmax <NUM> represent a minimum and maximum thread bound per cache level, respectively. A parameter W <NUM> represents a work package set that partitions the work being parallelized. A parameter wi <NUM> represents a particular work package of a number of vertices that are assigned together to a thread.

<FIG> illustrates equations <NUM> related to work package generation and thread boundary estimation. An Equation (<NUM>) <NUM> illustrates computation of a Vmin for parallel value <NUM> that can be determined based on the cost model. The Vmin for parallel value <NUM> can be used to determine whether it is profitable to execute the algorithm code in parallel and for which thread ranges (Tmin <= T <= Tmax). For example, if the count of all vertices in the graph (e.g., |V|) is at least equal to the Vmin for parallel value <NUM>, then parallel execution can be profitable.

A minimum work per work package Cw min, a start cost for parallel execution Cpara startup <NUM> and a start cost per thread CT overhead <NUM> can be empirically determined, ensuring that the overhead remains reasonable in case of a high load.

The computation of the thread boundaries can be expensive, due to thread-count-dependent memory latency being non-linear. To compute thread boundaries, an optimization problem can be solved for a smallest number of threads (Tmin) as shown in an Equation (<NUM>) <NUM>, under some side conditions.

<FIG> illustrates an algorithm <NUM> that can be performed to produce a time-efficient solution to the optimization problem for computing thread boundaries. The algorithm <NUM> includes code that iteratively doubles a number of threads and checks at each iteration if a valid upper and lower thread bound have been reached.

<FIG> illustrates equations <NUM> for modeling update contention. While in some regards the compute cores of a multi-core system can be considered independent, in reality resources like a CPU (Central Processing Unit) internal interconnection network (e.g., system interconnect), cache hierarchies, or a memory controller can be shared among the cores. Furthermore, some modules can be limited in parallelism (e.g., banked caches), or require serialization (e.g., a memory controller). Typically a shared use of resources increases contention and thus the delay of a particular operation. As a result, the CPU has to wait longer until an operation like a memory access or a data transfer completes. An example case of resource contention is the use of a same memory address by multiple cores. Read-only access to a range of addresses causes virtually no contention as multiple shared copies of a given memory address can be installed in different caches. However, shared copies are not allowed in the case of a write access, thus write operations can result in a large amount of invalidations, subsequent cache misses, and ultimately, contention. Similar contention issues can apply to atomic operations like mutually exclusive read-modify-write operations, with contention occurring due to corresponding cache lines being locked during the operation and no other requests on the same cache line able to be served.

For memory-sensitive algorithms such as many graph algorithms, memory contention has a significant impact on overall performance. In general, there can be many sources on different levels that influence contention and its impact on the overall performance. For example, on a hardware level, there can be, for example, contention sources such as CPU cores, system interconnects, memory accesses, cache organization and the specification and implementation of atomic operations. On a software level, there can lower-level contention sources such as thread-to-physical core mapping and higher-level sources such as algorithmic design. Push-based graph algorithms can be prone to contention due to updating of a same address. Pull-based algorithms are not generally prone to contention. Furthermore, the data that is being processed can be a source of contention.

Between the various different types of contention sources, complex interactions can exist that can be highly dynamic. In practice, the aforementioned contention sources can pose challenges for modeling contention and how contention affects latency and throughput of updates. However, to predict and optimize the performance of parallel graph algorithms, an accurate prediction of update performance is necessary.

A solution for predicting update performance can be to omit analytical modeling of contention, and instead create a model based on experiments and measurements. For example, a parametric model can be trained on a given system using a reference algorithm / problem and differently-sized data sets. Such a training can be performed (e.g., in at least a semi-automatic fashion) each time a new hardware configuration is used. Little's law can be applied for the parametric model by assuming that throughput and latency are interchangeable. For prediction of latency (e.g., during the cost estimation process described above) the prediction can be mapped to the reference problem. For a reference algorithm, degree count can be used, because degree count can be varied almost arbitrarily to model different scenarios. Furthermore, use of degree count can be comparable to many push-based graph algorithms. The reference algorithm can count the occurrence of vertex IDs of the vertex set V in an edge list, either as a source or target vertex, using fetch-and-add atomic operations on a single counter array. When executed in parallel, the input edge list can be partitioned in non-overlapping parts (e.g., of a predetermined size, such as <NUM> edges each). If the partitioning results in fewer partitions than cores, we exclude this setting from the experiments. The partitions can be dynamically dispatched as work packages to a set of worker threads. The reference data set selected can be of a type that causes high contention and is a representative data set to real world applications. An appropriate and useful reference data set can be a synthetic data set generated using a recursive matrix (RMAT) approach, with a RMAT data set being representative of many graph problems. A scale-free degree distribution of RMAT graphs can cause high contention on vertices with high degree count. Furthermore, contention can vary with the number of vertices and the related counter array size, as memory accesses will be accordingly distributed.

For modeling update contention, two parameters can be examined: <NUM>) the total number of threads; and <NUM>) the amount of touched memory, which is the unique set of shared addresses accessed by any memory operation. To ensure that thread counts are representative with regard to later use in inference, a thread count used for experimentation can be the total number of threads successively divided by two, so that modeled thread counts are exponentially spaced. As illustrated in an Equation (<NUM>) <NUM>, with counter <NUM> being a single counter of a counter array, an amount of touched memory, e.g., a counter array size Mcounters <NUM> can be computed as sizeof(counter) · |V|.

<FIG> is a graph <NUM> that illustrates experimental results for modeling update time. The graph <NUM> plots mean update time <NUM> as a function of counter array size <NUM> (e.g., a size parameter), with different data types <NUM> for a counter, for a constant thread count T of <NUM>. The experimental results illustrated in <FIG> show that update time is mainly a function of MCounters, particularly as the data type is varied. Accordingly, a conclusion can be made that there is no dependency to graph parameters as with the number of vertices or edges, which enables limiting of experimentation to counter array size and to generalize from such experimentation. Accordingly, a set of measurements can be obtained that describe a mean update time respectively latency L(M,T) which is a function of memory set size M and number of threads T. An additional observation can be made with respect to <FIG> is that with an increasing MCounters value the update time decreases as the contention is distributed across more memory locations, which appears to be an effect of a function of the logarithm of the counter array size rather than depending linearly on the counter array size.

Referring again to <FIG>, to derive a suitable heuristic that predicts L(M,T) for a given data set size M and thread count T, bounds can be identified for the memory access costs. As cache levels are discrete, we use the highest memory hierarchy level l that can fit the data set of size M. (e.g., l = min { x : Mx > M}, with Mx being a capacity of memory hierarchy level x. To approximate the effective access latency, a polynomial interpolation between the cache level l and u = l - <NUM> can be used, with a rationale that higher cache levels will also observe some cache hits. Memory M > Mm can be excluded, with Mm referring to main memory size. The rationale that higher cache levels will also observe some cache hits holds true for l referring to main memory. For l = <NUM> a special case can be handled, in which the problem fits into a L1 cache and a setting of u = l is performed, effectively configuring an identical lower and upper bound. As illustrated in an Equation (<NUM>) <NUM>, a (M) value <NUM> can describe the polynomial interpolation, depending on data set size M. Note that Equation (<NUM>) uses the logarithm of the data set size, according to the observation made in <FIG>.

An Equation (<NUM>) <NUM> shows calculations to determine a difference δL(T, l) <NUM> between the latency of the two memory hierarchy levels l and u, with u derived from l as described above. An Equation (<NUM>) <NUM> shows calculations to determine an update time prediction Lpredict(M, T) <NUM> as a function of the access cost of a lower bound L(Ml, T) <NUM>, but effectively reduced by a δL(T) value <NUM> multiplied with the previous interpolation S(M, T) <NUM> cubed. Cubing S(M, T) be be performed based on results that have been empirically derived from experiments on multiple systems that showed a best fit in different regressions.

<FIG> is a graph <NUM> that plots relative cost <NUM> for atomics as a function of thread count <NUM> and counter array size. Another important factor for the update time prediction are dynamic effects that depend on the number of threads. The graph <NUM> illustrates a clear dependency between the number of threads and the atomic update time. Furthermore, the graph <NUM> indicates that when limiting the problem to higher cache levels (e.g., by adapting problem size), thread count has a much higher impact on atomic update time. To address this dependency, the atomic update time can be estimated using the measured access times with the anticipated number of threads.

<FIG> is a flowchart of an example method for configuring graph query parallelism for high system throughput. It will be understood that method <NUM> and related methods may be performed, for example, by any suitable system, environment, software, and hardware, or a combination of systems, environments, software, and hardware, as appropriate. For example, one or more of a client, a server, or other computing device can be used to execute method <NUM> and related methods and obtain any data from the memory of a client, the server, or the other computing device. In some implementations, the method <NUM> and related methods are executed by one or more components of the system <NUM> described above with respect to <FIG>. For example, the method <NUM> and related methods can be executed by the graph engine <NUM> of <FIG>.

At <NUM>, a graph query to be executed against a graph database is received. The graph query can be received from a client device, for example.

At <NUM>, system properties are determined of a system in which the graph query is to be executed. The system properties can include cache sizes and memory access times.

At <NUM>, algorithmic properties are determined of at least one graph algorithm to be used to execute the graph query against the graph database. The algorithmic properties can include memory consumption and amount of memory accessed by atomic and non-atomic operations.

At <NUM>, graph data statistics for the graph database are determined. The graph data statistics can include vertex degrees and vertex degree distribution information.

At <NUM>, first graph traversal estimations are determined for a first iteration of the graph query.

At <NUM>, a first estimated cost model is determined for the first iteration based on the first graph traversal estimations. The cost model estimation step can be skipped after the first iteration in some cases, such as when it is known that the processed data is not changed and no runtime feedback is used (e.g., for page rank scenarios). The estimated graph traversal information can include estimates for number of newly-visited vertices and number of touched vertices for the first iteration. When more than one graph algorithm is being evaluated, a first estimated cost model can be determined for each algorithm. An algorithm that has a lowest cost model can be identified as a selected algorithm for further evaluation.

At <NUM>, first estimated thread boundaries are determined for performing parallel execution of the first iteration.

At <NUM>, based on the first estimated cost model, first work packages are generated that include vertices to be processed during the execution of the first iteration. Generating the work packages can include creating a set of work packages that each have a substantially similar cost.

At <NUM>, the first work packages are provided to a work package scheduler for scheduling the first work packages to execute the first iteration. The work package scheduler can determine which of the first work packages are executed in parallel and which of the first work packages are executed sequentially.

At <NUM>, data is determined for a next iteration. For example, a next set of vertices to process in a second iteration can be determined. Second graph traversal estimations, a second cost model, and second estimated thread boundaries can be determined for the second iteration. Second work packages can be generated for the second iteration and the second work packages can be provided to the work package scheduler. As mentioned, in some cases, cost model estimation can be performed only for the first iteration and not for subsequent iterations.

In an evaluation phase, the system was evaluated based on various variants of PR and BFS algorithms. PR algorithms were evaluated as a topology-centric algorithm variant using two different types - push and pull. In the push variant, the updates to compute the ranks are pushed to the target vertices, which can involve an atomic operation for each update. The pull variant, which does not require atomic operations, can involve gathering data from all source vertices to compute a rank. BFS algorithms can be considered data-driven algorithm and can be implemented as a top-down variant. Each type of evaluated algorithm can use different schedulers (e.g., sequential, simple-parallel, system scheduler). A sequential scheduler can execute code sequentially. Simple-parallel scheduling can involve executing code in parallel using a simple work partitioning that partitions the frontier queue in equal-sized packages. A common package size can be determined by a maximum number of threads value and a lower limit. The system scheduler uses the scheduler described above. Sequentially scheduled variants can serve in concurrent query settings as a baseline, since under high concurrency, a per-query sequential processing is typically preferable. Since the behavior of graph computations can be highly data-dependent, different synthetic data sets are evaluated with different scale factors (SF), and various real-world data sets are evaluated. Synthetic data sets can be generated using a recursive matrix (RMAT) approach. For scale factors, the following equations can be relevant: |V| = <NUM>SF, |E| = <NUM> * |V|.

As described in more detail below, for both PR and BFS evaluations, the use of the system scheduler resulted in efficient executions, with the system scheduler being the most efficient or close in efficiency to the best performing alternative, for different types of data sets and configurations. Particularly for PR evaluations, system scheduler pull and system scheduler push approaches were substantially faster than alternatives. Additionally, for BFS evaluations, the system scheduler approach performs better or similarly to sequential and simple parallel approaches, for different types of data sets.

<FIG> are evaluation graphs <NUM> and <NUM> that illustrate performance results for page rank algorithms for single query execution. During evaluation, single query performance can be evaluated to identify scheduling overhead. For instance, the evaluation graph <NUM> plots performance scaling on a Y-axis <NUM> in MPEPS (millions of processed edges per second), for different PR implementations across synthetic RMAT graphs of different sizes (e.g., different PR implementations are presented on an X-axis <NUM>, according to scale factor). In the evaluation graph <NUM> in <FIG>, lines <NUM>, <NUM>, <NUM>, and <NUM> plot performance for a sequential push evaluation <NUM>, a sequential pull evaluation <NUM>, a system scheduler push evaluation <NUM>, and a system scheduler pull evaluation <NUM>, respectively.

Single-query PR performance can depend on problem size and algorithm, although pull versions may generally perform better than push algorithms. In general, push and pull algorithms may have substantially different properties can therefore also differ in respective cost models. As illustrated by lines <NUM> and <NUM> in the evaluation graph <NUM>, sequential processing is faster for small problem sizes (e.g., SF <= <NUM> for push and SF <= <NUM> for pull), otherwise parallel processing is faster. The overhead of the system scheduling method can be assessed by comparing system scheduler performance to sequential performance. For example, system scheduler push results can be compared to sequential push results. In comparing the lines <NUM> and <NUM>, for lower scale factors, the system scheduler variant behaves similarly to the sequential variant, with only a small (e.g., <NUM>% at SF17 - <NUM>% at SF of <NUM>) reduction in performance). Thus, in spite of fundamental differences between push and pull variants, in particular with regard to the use of atomics, the system scheduler chooses the right execution strategy, so that the overhead never dominates the elapsed time.

The evaluation graph <NUM> in <FIG> plots performance scaling on a Y-axis <NUM> for different real-world data sets <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, and <NUM> that are plotted on a X-axis <NUM>. For each data set, evaluation results are plotted as bar graphs for a sequential push evaluation <NUM>, a sequential pull evaluation <NUM>, a simple push evaluation <NUM>, a simple pull evaluation <NUM>, a system scheduler push evaluation <NUM>, and a system scheduler pull evaluation <NUM>, respectively.

The evaluation graph <NUM> shows that for different data sets and a single query PR, differences between values from the simple push evaluation <NUM> and scheduling-optimized counterpart values (e.g., from the system scheduler push evaluation <NUM>) is negligible, with either the simple push values or the system scheduler push values having only small advantages compared to the other, depending on the data set. The differences for pull variants are similarly small. Thus, for PR algorithms, the system scheduler is stable across different data sets and the scheduling overhead is small.

<FIG> are evaluation graphs <NUM> and <NUM> that illustrate performance results for breadth-first search algorithms for single query execution. Similar to the evaluation graph <NUM>, the evaluation graph <NUM> plots performance scaling on a Y-axis <NUM> in MPEPS for different BFS implementations across RMAT graphs of different sizes (e.g., different BFS implementations are presented on an X-axis <NUM>, according to scale factor). In the evaluation graph <NUM> in <FIG>, lines <NUM>, <NUM>, and <NUM> plot performance for a sequential evaluation <NUM>, a simple evaluation <NUM>, and a system scheduler evaluation <NUM>, respectively.

Similar to results for PR algorithms, data set size may determine a preferred algorithm choice. For example, for SF <= <NUM> and as illustrated by the line <NUM>, sequential processing was fastest. For higher scale factors (e.g., SF > <NUM>) and as respectively illustrated by the lines <NUM> and <NUM>, simple (e.g., straight-forward range partitioning of the frontier queue) and system scheduler algorithms were faster. For the smaller scale factors (e.g., SF < <NUM>), the system scheduler algorithm outperformed the simple algorithm, due to reducing false invalidations by reducing concurrent writes to same memory locations. The similarity in the lines <NUM> and <NUM> (e.g., especially for SF > <NUM>) illustrate that the scheduling optimizations from the system scheduler approach come with negligible overhead in comparison to simple parallelization.

The evaluation graph <NUM> in <FIG> illustrates breadth-first search performance on real-world data sets data set <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, and <NUM>, for a sequential evaluation <NUM>, a simple evaluation <NUM>, and a system scheduler evaluation <NUM>. For single query execution, BFS performance can be highly data-dependent. Accordingly, the system scheduler can, for some data sets, be slower than the simple approach, which is based on naive parallelization, albeit with only small differences and usually only if a difference to between the system scheduler approach and a sequential approach is substantial. System scheduling overhead can be generally constant even when BFS processing time is highly dependent on the data set. Accordingly, for a small BFS execution time, scheduling overhead can be more significant than for longer BFS execution times. Furthermore, results suggest that for data sets that resulted in lower performance (e.g., the data set <NUM> and the data set <NUM>), the system scheduler approach outperformed the simple approach, such as due to complex dependencies (e.g., between edges) and fewer parallelization opportunities as compared to other data sets.

As shown in <FIG>, the system scheduler approach is either a best performing approach or close to the best performing approach, independent of particular data set characteristics. In summary, scheduling overhead for the system scheduler was negligible and scheduling optimizations can be beneficial for overall performance. As expected with a variety of data sets, graph computations (and resulting performance) can be highly dependent on a selected algorithm and the graph data of a particular data set. Despite the differences of algorithm variants and data sets, the performance of the system scheduler was generally stable across the various evaluated algorithm variants and data sets.

<FIG> illustrates evaluation graphs <NUM> that illustrate performance results for page rank algorithms for multiple sessions for synthetic data sets. The evaluation graphs <NUM> include graphs <NUM>, <NUM>, and <NUM> that plot results for a sequential push evaluation <NUM>, a system scheduler push evaluation <NUM>, a sequential pull evaluation <NUM>, and a system scheduler pull evaluation <NUM>, for a first data set <NUM>, a second data set <NUM>, and a third data set <NUM>, respectively. Each of the first data set <NUM>, the second data set <NUM>, and the third data set <NUM> correspond to different scale factors. Lines <NUM>, <NUM>, and <NUM> illustrate that the system scheduler approach provided the fastest performance among evaluated approaches. Performance advantages can depend on data set size and concurrency (e.g., number of sessions). For example and as shown by a line <NUM>, only for the smallest data set size (e.g., the first data set <NUM> with SF of <NUM>) and highest number of sessions is the sequential pull approach a fastest approach, which can be due to data sets in these scenarios providing few opportunities for parallelism. A break-even point between the system scheduler approach and the sequential pull approach can move towards larger amounts of concurrency as the data set size increases. For example, with increasing data set size an overhead of parallel execution becomes more negligible.

<FIG> illustrates evaluation graphs <NUM> that illustrate performance results for BFS algorithms for multiple sessions for synthetic data sets. The evaluation graphs <NUM> include graphs <NUM>, <NUM>, and <NUM> that plot results for a sequential evaluation <NUM>, a simple evaluation <NUM>, and a system scheduler evaluation <NUM>, for a first data set <NUM>, a second data set <NUM>, and a third data set <NUM>, respectively. Each of the first data set <NUM>, the second data set <NUM>, and the third data set <NUM> correspond to different scale factors.

As illustrated by a line <NUM>, for the first data <NUM>, sequential processing is the fastest, in particular with growing concurrency. As illustrated by lines <NUM> and <NUM>, as data set sizes increase, the system scheduler performance improves. With smaller data set sizes, parallel processing can be less efficient but with larger data set sizes, parallel processing becomes more efficient.

<FIG> illustrates evaluation graphs <NUM> that illustrate performance results for page rank algorithms for multiple sessions for real world data sets. The evaluation graphs <NUM> include graphs <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, and <NUM> that plot results for a sequential push evaluation <NUM>, a scheduler push evaluation <NUM>, a sequential pull evaluation <NUM>, and a system scheduler pull evaluation <NUM>, for a first data set <NUM>, a second data set <NUM>, a third data set <NUM>, a fourth data set <NUM>, a fifth data set <NUM>, a sixth data set <NUM>, and a seventh data set <NUM>, respectively.

Lines <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, and <NUM> illustrate that the system scheduler pull approach is a fastest approach, independent of data set and concurrency. While the performance of the system scheduler pull approach is almost constant with regard to concurrency, performance of alternative approaches (e.g., sequential push and sequential pull) is usually scaling linearly with concurrency, which may suggest that for some data sets (e.g., the third data set <NUM>, the fifth data set <NUM>, and the sixth data set <NUM>), sequential alternative approaches can be similar to the performance of the system scheduler approach given sufficient concurrency. However, for other data sets (e.g., the fourth data set <NUM> and the seventh data set <NUM>), a performance increase by concurrency for alternative sequential approaches is not enough to reach the performance of the system scheduler implementation (e.g., due to a different internal level of parallelism that these data sets allow).

<FIG> illustrates evaluation graphs <NUM> that illustrate performance results for bread-first search algorithms for multiple sessions for real world data sets. The evaluation graphs <NUM> include graphs <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>, and <NUM> that plot results for a sequential evaluation <NUM>, a simple evaluation <NUM>, and a system scheduler evaluation <NUM>, for a first data set <NUM>, a second data set <NUM>, a third data set <NUM>, a fourth data set <NUM>, a fifth data set <NUM>, a sixth data set <NUM>, and a seventh data set <NUM>, respectively. BFS algorithm performance can be highly dependent on data set and concurrency, and three general patterns can be observed.

In a first pattern, for the first data set <NUM> and the seventh data set <NUM>, the simple parallel approach and the system scheduler approach are similarly fast, with the sequential approach substantially slower. However, the difference between the sequential approach and the other approaches decreases with increasing concurrency, which can be that concurrencies higher than those evaluated might result in a break-even point.

In a second pattern, for the second data set <NUM> and the sixth data set <NUM>, performance of the three approaches is similar but with high variance depending on concurrency. Performance of the sequential approach, when slower at lower concurrencies tends to improve as concurrencies increase, sometimes even outperforming the other algorithms.

In a third pattern, for the third data set <NUM> and the fifth data set <NUM>, performance for the sequential approach and the system scheduler approach are similar, with small advantages for sequential, and scaling is almost linear with concurrency. Simple parallel has a constant performance, resulting in a growing performance gap with the sequential and system scheduler approaches as concurrency increases.

The preceding figures and accompanying description illustrate example processes and computer-implementable techniques. But system <NUM> (or its software or other components) contemplates using, implementing, or executing any suitable technique for performing these and other tasks. It will be understood that these processes are for illustration purposes only and that the described or similar techniques may be performed at any appropriate time, including concurrently, individually, or in combination. In addition, many of the operations in these processes may take place simultaneously, concurrently, and/or in different orders than as shown. Moreover, system <NUM> may use processes with additional operations, fewer operations, and/or different operations, so long as the methods remain appropriate.

Claim 1:
A computer-implemented method comprising:
receiving a graph query to be executed against a graph database (<NUM>);
determining system properties (<NUM>, <NUM>, <NUM>) of a system (<NUM>, <NUM>, <NUM>) in which the graph query is to be executed;
determining algorithmic properties (<NUM>, <NUM>) of at least one graph algorithm to be used to execute the graph query against the graph database (<NUM>);
determining graph data statistics (<NUM>, <NUM>) for the graph database (<NUM>); and
for a first iteration of the graph query:
determining first graph traversal estimations for the first iteration;
determining a first estimated cost model for the first iteration based on the first graph traversal estimations;
determining first estimated thread boundaries for performing parallel execution of the first iteration, wherein the first estimated thread boundaries comprise a first minimum boundary for parallel execution as a minimum number of threads and a first upper thread boundary as a maximum number of threads;
generating first work packages of vertices to be processed during the execution of the first iteration based on the first estimated cost model; and
providing the first work packages to a work package scheduler (<NUM>, <NUM>) for scheduling the first work packages to execute the first iteration,
requesting, by the work package scheduler (<NUM>, <NUM>), worker threads from the system (<NUM>, <NUM>, <NUM>) according to the first upper thread boundary,
when a worker thread gets assigned, registering, by the worker thread, with the work package scheduler (<NUM>, <NUM>) and requesting, by the worker thread, a first work package,
checking, by the work package scheduler (<NUM>, <NUM>), if a number of registered worker threads is higher than the first minimum boundary,
if the number of registered worker threads is higher than the first minimum boundary, assigning, by the work package scheduler (<NUM>, <NUM>), a first work package to the registered worker threads for parallel execution,
if the number of registered worker threads is not higher than the first minimum boundary, assigning, by the work package scheduler (<NUM>, <NUM>), one worker thread to execute a first work package sequentially, while the other worker threads wait until the first work package is completed.