A technique includes receiving a user input in an array-oriented database. The user input indicates a database operation and processing a plurality of chunks of data stored by the database to perform the operation. The processing in dudes selectively distributing the processing of the plurality of chunks between a first group of at least one central processing unit and a second group of at least one co-processor.

BACKGROUND

Array processing has wide application in many areas including machine learning, graph analysis and image processing. The importance of such arrays has led to new storage and analysis systems, such as array-oriented databases (AODBs). An AODB is organized based on a multi-dimensional array data model and supports structured query language (SQL)-type queries with mathematical operators to be performed on arrays, such as operations to join arrays, operations to filter an array, and so forth. AODBs have been applied to a wide range of applications, including seismic analysis, genome sequencing, algorithmic trading and insurance coverage analysis.

DETAILED DESCRIPTION

An array-oriented database (AODB) may be relatively more efficient than a traditional database for complex multi-dimensional analyses, such as analyses that involve dense matrix multiplication, K-means clustering, sparse matrix computation and image processing, just to name a few. The AODB may, however, become overwhelmed by the complexity of the algorithms and the dataset size. Systems and techniques are disclosed herein for purposes of efficiently processing queries to an AODB-based system by distributing the processing of the queries among central processing units (CPUs) and co-processors.

A co-processor, in general, is supervised by a CPU, as the co-processor may be limited in its ability to perform some CPU-like functions (such as retrieving instructions from system memory, for example). However, the inclusion of one or multiple co-processors in the processing of queries to an AODB-based system takes advantage of the co-processor's ability to perform array-based computations. In this manner, a co-processor may have a relatively large number of processing cores, as compared to a CPU. For example, a co-processor such as the NVIDIA Tesla M2090 graphics processing unit (GPU) may have 16 multi-processors, with each having 32 processing cores for a total of 512 processing cores. This is in comparison to a given CPU, which may have, for example, 8 or 16 processing cores. Although a given CPU processing core may possess significantly more processing power than a given co-processor processing core, the relatively large number of processing cores of the co-processor combined with the ability of the co-processor's processing cores to process data in parallel make the co-processor quite suitable for array computations, which often involve performing the same operations on a large number of array entries.

For example implementations disclosed herein, the co-processor is a graphics processing unit (GPU), although other types of co-processors (digital signal processing (DSP) co-processers, floating-point arithmetic co-processors, and so forth) may be used, in accordance with further implementations.

In accordance with example implementations, the GPU(s)and CPU(s) of an AODB system maybe disposed on at least one computer (a server, a client, an ultrabook computer, a desktop computer, and so forth). More specifically, the GPU may be disposed on an expansion card of the computer and may communicate with components of the computer over an expansion bus, such as a Peripheral Component Interconnect Express (PCIe) bus, for example. The expansion card may contain a local memory, which is separate from the main system memory of the computer; and a CPU of the computer may use the PCIe bus for purposes of transferring data and instructions to the GPU's local memory so that the GPU may access the instructions and data for processing. Moreover, when the GPU produces data as a result of this processing, the data is stored in the GPU'S local memory; and a CPU may likewise use PCIe bus communications to instruct the transfer of data from the GPU's local memory to the system memory.

The GPU may be located on a bus other than a PCIe bus in further implementations. Moreover, in farther implementations, the GPU may be a chip or chip set that is integrated into the computer, and as such, the GPU may not be disposed on an expansion card.

FIG. 1depicts an example implementation of an AODB-based database system100according to an example implementation. The system100is constructed to process a user input150that describes an array-based operation. As an example, in accordance with example implementations, the system100may be constructed to process SciDB-type queries, where “SciDB” refers to a specific open source array management and analytics database, in this manner, the user input150may be, in accordance with some example implementations, an array query language (AQL) query (similar to a SQL query but specifying mathematical operations) or an array functional language (AFL) query. Moreover, the user input150maybe generated, for example, by an array-based programming language, such as R.

In general, the user input150may be a query or a user defined function. Regardless of its particular form, the user input150defines an operation to be performed by the database system100. In this manner, a query, in general, may use operators that are part of the set of operators defined by the AODB, where as the user-defined function allows the user to specify custom algorithms and/or operations on array data.

A given user input150may be associated with one or multiple units of data called “data chunks” herein. As an example, a given array operation that is described by a user input150may be associated with partitions of one or multiple arrays, and each chunk corresponds to one of the partitions. The system100distributes the compute tasks for the data chunks among one or multiple CPUs112and one or multiple GPUs114of the system100. In this context, a “compute task” maybe viewed as the compute kernel for a given data chunk. Each CPU112may have one or multiple processing cores (8 or 16 processing cores, as an example); and each CPU processing core is a potential candidate for executing a thread to perform a given compute task. Each GPU114may also contain one or multiple processing cores (512 processing cores, as an example); and the processing cores of the GPU114may perform a given compute task assigned to the GPU114in parallel.

For the foregoing example, it is assumed that the AODB system100is formed from one or multiple physical machines110, such as example physical machine110-1. In general, the physical machines110are actual machines that are made up of actual hardware and actual machine executable instructions, or “software.” In this regard, as depicted inFIG. 1, the physical machine110-1includes such hardware as one or multiple CPUs112; one or multiple GPUs114; a main system memory130(i.e., the working memory for the machine110-1); a storage interface116that communicates with storage117(one or multiple hard disk drives, solid state drives, optical drives, and so forth); a network interface, and so forth, as can be appreciated by the skilled artisan.

As depicted inFIG. 1, each GPU114has a local memory115which receives (via PCIe bus transfers, for example) instructions and data chunks to be processed by the GPU114torn the system memory130and stores data chunks resulting from the GPU's processing, which are transferred back (via PCIe bus transfers, for example) into the system memory130. Moreover, one or more of the CPUs112may execute machine executable instructions to form modules, or components, of an AODB-based database120for purposes of processing the user input150.

For the example implementation depicted inFIG. 1, the AODB database120includes a parser122that parses the user input150; and as a result of this parsing, the parser122identifies one or multiple data chunks to be processed and one or compute tasks to perform on the data chunk(s). The AODB database120further includes a scheduler134that schedules the compute tasks to be performed by the CPU(s)112and GPU(s)114. In this manner, the scheduler134places data indicative of the compute tasks in a queue127of an executor126and tags this data to indicate which compute tasks are to be performed by the CPU(s)112and which compute tasks are to be performed by the GPU(s)114.

Based on the schedule indicated by the data in the queue127, the executor126retrieves corresponding data chunks118from the storage117and stores the chunks118in the system memory130. For a CPU-executed compute task, the executor126initiates execution of the compute task by the CPU(s)112; and the CPU(s)112access the data chunks from the system memory130for purposes of performing the associated compute tasks. For a GPU-executed task, the executor126may transfer the appropriate data chunks from the system memory130into the GPU's local memory115(via a PCIe bus transfer, for example).

The AODB database120further includes a size regulator, or size optimizer124, that regulates the data chunk sizes for compute task processing. In this manner, although the data chunks118may be sized for efficient transfer of the chunks118from the storage117(and for efficient transfer of processed data chunks to the storage117), the size of the data chunk118may not be optimal for processing by a CPU112or a GPU114. Moreover, the optimal size of the data chunk for CPU processing may be different than the optimal size of the data chunk for GPU processing.

In accordance with some implementations, the AODB database120recognizes that the chunk size influences the performance of the compute task processing. In this manner, for efficient GPU processing, relatively large chunks may be beneficial due to (as examples) the reduction in data transfer overhead, as relatively larger chunks are more efficiently transferred into and out of the GPU's local memory115(via PCIe bus transfers, for example); and relatively larger chunks enhances GPU processing efficiency, as the GPU's processing cores having a relatively large amount of data to process in parallel. This is to be contrasted to the chunk size for CPU processing, as a smaller chunk size may enhance allocating data locality and reduce the overhead of accessing data to be processed among CPU112threads.

The size optimizer124regulates the data chunk size based on the processing entity that performs the related compute task on that chunk. For example, the size optimizer124may load relatively large data chunks118from the storage117and store relatively large data chunks in the storage117for purposes of expediting communication of this data to and from the storage117. The size optimizer124selectively merges and partitions the data chunks118to produce modified size data chunks based on the processing entity that processes these chunks. In this manner, in accordance with an example implementation, the size optimizer124partitions the data chunks118into multiple smaller data chunks when these chunks correspond to compute tasks that are performed by a CPU112and stores these partitioned blocks along with the corresponding CPU tags in the queue127. To the contrary, the size optimizer124may merge two or multiple data chunks118together to produce a relatively larger data chunk for GPU-based processing; and the size optimizer124may store this merged chunk in the queue127along with the appropriate GPU tag.

FIG. 3is an illustration300of the relative CPU and GPU response times versus chunk size according to an example implementation. In this regard, the bars302ofFIG. 3illustrate the CPU response times for different chunk sizes; and the bars304represent the corresponding GPU response times for the same chunk sizes. As can be seen by trends320and330for CPU and GPU processing, respectively, in general, the trend330for the GPU processing indicates that the response times for the GPU processing decrease with chunk size, whereas the trend320for CPU processing depicts the response times for the CPU processing increase with chunk size.

In accordance with example implementations, the executor126may further decode, or convert, the data chunk into a format that is suitable for the processing entity that performs the related compute task. For example, the data chunks118maybe stored in the storage117in a triplet format. An example triplet format400is depicted inFIG. 4. In the example triplet format400, the data is arranged as an array of structures402, which may not be a suitable format by processing by a GPU114but may be a suitable format by processing by a CPU112. Therefore, if a given data chunk is to be processed by a CPU112, the executor126may not perform any further format conversion. However, if the data chunk is to be processed by a GPU114, in accordance with example implementations, the executor126may convert the data format into one that is suitable for the GPU114. Using the example ofFIG. 4, the executor128may convert the triplet form at400ofFIG. 4into a structure500of arrays502(depicted inFIG. 5), which is suitable for parallel processing by the processing cores of the GPU114.

Referring back toFIG. 1, in accordance with example implementations, the scheduler134may assign compute tasks to the CPU(s)112and GPU(s)114based on static criteria. For example, the scheduler134may assign a fixed percentage of compute tasks to the GPU(s)114and assign the remaining compute tasks to the CPU(s)112.

In accordance with further implementations, the scheduler134may employ a dynamic assignment policy based on metrics that are provided by a monitor128of the AODB database120. In this manner, the monitor128may monitor such metrics as CPU utilization, CPU compute task processing time, GPU utilization, GPU compute task processing time, the number of concurrent GPU tasks and so forth; and based on these monitored metrics, the scheduler134dynamically assigns the compute tasks, which provides the scheduler134the flexibility to tune performance at runtime. In accordance with example implementations, the scheduler134may make the assignment decisions based on the metrics provided by the monitor128and static policies. For example, the scheduler134may assign a certain percentage of compute tasks to the GPU(s)114until a fixed limit on the number of concurrent GPU tasks are reached or until the GPU compute task processing time decreases below a certain threshold. Thus, in accordance with some implementations, the scheduler134may exhibit a bias toward assigning compute tasks to the GPU(s)114. This bias, in turn, takes advantage of a potentially faster compute task processing time by the GPU114.

In this manner,FIG. 6depicts an illustration of an observed relative speedup multiplier associated with using GPU-based compute task processing versus CPU-based compute task processing for different operations. These are shown by speedup multipliers604,606and608for image processing, dense matrix multiplication and page rank calculations, respectively. As can be seen fromFIG. 6, the GPU provides different speedup multipliers depending on the data type, and for the example ofFIG. 6, the maximum speed multiplier occurs for dense matrix multiplication.

Referring toFIG. 2, to summarize, in accordance with an example implementation, the AODB database120establishes a work flow200for distributing compute tasks among the CPU(s)112and GPU(s)114. The workflow200includes retrieving data chunks118from the storage117and selectively assigning corresponding compute tasks between the CPU(s)112and GPU(s)114, which results in GPU and CPU tasks, or jobs. The workflow200includes selectively merging and partitioning the data chunks118as disclosed herein to form partitioned chunks210for the illustrated CPU jobs ofFIG. 2and merged chunks218for the illustrated GPU job ofFIG. 2.

The CPU(s)112process the data chunks210to form corresponding chunks212that are communicated back to the storage117. The data chunks218for the GPU job may be further decoded, or reformatted (as indicated by reference numeral220), to produce corresponding reformatted data chunks221that are moved in (as illustrated by reference numeral222) into the GPU's memory115(via a PCIe bus transfer, for example) to form local blocks223to be processed by the GPU (s)114. After GPU processing224that produces data blocks225, the work flow200includes moving out the blocks225from the GPU local memory115(as indicated at reference numeral226), such as by a PCIe bus transfer, which produces blocks227and encoding (as Indicated by reference numeral228) the blocks227(using the CPU, for example) to produce reformatted blocks230that are then transferred to the storage117.

Thus, referring toFIG. 7, to generalize, in accordance with an example implementation, a technique700generally includes receiving (block702) a user input in an array-oriented database. Pursuant to the technique700, tasks for processing the chunks are selectively assigned (block704) among one or more CPUs and one or more GPUs.

More specifically,FIG. 8depicts a technique800that may be performed in accordance with example implementations. Pursuant to the technique800, a user input is received, pursuant to block802and tasks for processing of data chunks associated with the user input are assigned (block804) based on at least one monitored CPU and/or GPU performance metric. The data chunks may be retrieved from storage using a first chunk size optimized for the retrieval, pursuant to block806; and then the chunks may be selectively partitioned/merged based on the processing entity that processes the chunks, pursuant to block810. The technique800also includes communicating (block812) the partitioned/merged chunks to the CPU(s) and GPU(s) according to the assignments.

While a limited number of examples have been disclosed herein, those skilled in the art, having the benefit of this disclosure, will appreciate numerous modifications and variations therefrom. It is intended that the appended claims cover all such modifications and variations.