Scatter gather using key-value store

Scatter gather operation(s) are performed by accessing a shared memory that is shared amongst nodes interconnected through network(s) and having a CNS shared amongst the nodes. Data is gathered from multiple processes at corresponding multiple nodes into location(s) in the CNS, and tuple(s) having a same tuple name are created in the CNS. The tuple(s) have information referencing the gathered data in the location(s). Alternatively, data that has been gathered using the same tuple name is scattered to multiple processes participating in the CNS. The scattering uses the tuple(s) in the CNS, and is performed from the location(s) into other location(s) at one or multiple nodes for one or multiple processes at the corresponding one or multiple nodes. Both the gathering data and the scattering data may also be performed.

BACKGROUND

This invention relates generally to the scatter gather process and, more specifically, relates to scatter gather using a key-value store.

Memory is typically thought of as using blocks, where each element in a block is adjacent to another element. If the data you want is stored in this manner, then typical memory is easy to use. If, however, the data you want is stored in many different locations (such as the blocks) scattered over the memory, then this data must be gathered before use, and then the results after use scattered back to their original locations. This technique is called “scatter/gather” or (as also used herein) “scatter gather”. For more detail about scatter gather, see Bryon Moyer, “How Does Scatter/Gather Work?”, Electronic Engineering, Feb. 9, 2017.

Scatter/gather is a useful feature that is used widely in many scientific applications. Scatter gather typically must be performed in a single process. If multiple processes are involved, especially if the processes do not exist concurrently, this is challenging. One current approach is to address this is to store the gathered data in files. This adds a layer of complexity, however, as one has to have and use a file system, organize the data within the files, and then have somewhere in the file to store results of operations on the gathered data for subsequent scattering.

SUMMARY

This section is meant to be exemplary and not meant to be limiting.

In an exemplary embodiment, a method includes performing one or more scatter gather operations by accessing a shared memory that is shared amongst multiple nodes interconnected through one or more networks. The shared memory comprises a coordination namespace that is shared amongst the multiple nodes The operations comprise: gathering data from multiple processes at corresponding multiple nodes into a one or more locations in the coordination namespace, and creating one or more tuples having a same tuple name in the coordination namespace, wherein the one or more tuples have information referencing the gathered data in the one or more locations; or scattering data that has been gathered using the same tuple name to multiple processes participating in the coordination namespace, the scattering using the one or more tuples in the coordination namespace, the scattering performed from the one or more locations into other locations at one or multiple nodes for one or multiple processes at the corresponding one or multiple nodes; or performing both the gathering data and the scattering data.

In another exemplary embodiment, an apparatus is disclosed. The apparatus comprises one or more memories having computer-readable code thereon and one or more processors. The one or more processors, in response to retrieval and execution of the computer-readable code, cause the apparatus to perform operations comprising: performing one or more scatter gather operations by accessing a shared memory that is shared amongst multiple nodes interconnected through one or more networks, the shared memory comprising a coordination namespace that is shared amongst the multiple nodes, the operations comprising: gathering data from multiple processes at corresponding multiple nodes into a one or more locations in the coordination namespace, and creating one or more tuples having a same tuple name in the coordination namespace, wherein the one or more tuples have information referencing the gathered data in the one or more locations; or scattering data that has been gathered using the same tuple name to multiple processes participating in the coordination namespace, the scattering using the one or more tuples in the coordination namespace, the scattering performed from the one or more locations into other locations at one or multiple nodes for one or multiple processes at the corresponding one or multiple nodes; or performing both the gathering data and the scattering data.

Another example is a computer program product. The computer program product comprises a computer readable storage medium having program instructions embodied therewith. The program instructions are executable by a device to cause the device to perform at least the following: performing one or more scatter gather operations by accessing a shared memory that is shared amongst multiple nodes interconnected through one or more networks, the shared memory comprising a coordination namespace that is shared amongst the multiple nodes, the operations comprising: gathering data from multiple processes at corresponding multiple nodes into a one or more locations in the coordination namespace, and creating one or more tuples having a same tuple name in the coordination namespace, wherein the one or more tuples have information referencing the gathered data in the one or more locations; or scattering data that has been gathered using the same tuple name to multiple processes participating in the coordination namespace, the scattering using the one or more tuples in the coordination namespace, the scattering performed from the one or more locations into other locations at one or multiple nodes for one or multiple processes at the corresponding one or multiple nodes; or performing both the gathering data and the scattering data.

DETAILED DESCRIPTION

addr address

AGEM Aggregated Global Extended Memory

API Application Programming Interface

cmd command

CNS Coordination Namespace

CPU Central Processing Unit

CS Coordination space as used within tuple records

dest destination

DMA or dma Direct Memory Access or Accessing

DDR Double Data Rate

DRAM Dynamic Random Access Memory

EM Extended Memory

FPGA Field Programmable Gate Array

GPU Graphics Processing Unit

GVAS Global Virtual Address Space

ID or id identification or identifier

LT local tuple

NDE Named Data Element

NH Natural Home

NIC Network Interface Controller

OS operating system

PCI peripheral component interconnect

PR Pending Record, or Process (Pr) such as inFIG. 13

ptr pointer

RT remote tuple

SCM or scm Storage Class Memory

src source

RDMA or rdma Remote Direct Memory Access

Typical scatter gather operations are performed based on address and size. Each operation would require both source and destination addresses, such as a source (src) list and destination (dest) address (for gather) or src address and dest list (for scatter). By contrast, we propose to perform these operations based on name (e.g., and size) using an implementation of tuple space. The examples herein may use a source list and destination tuple name for gather while scatter may use a destination list and tuple name. Tuple space stores tuples based on names and their location is stored in a key-value store such as a hash map. This is described in more detail below.

To perform gather operations, one may assign all the data that needs to be gathered to the same “tuple name”. The scatter/gather engine, referred to as a tuple space manager herein, is given a list of locations in local memory and size along with a tuple name. The tuple space manager collects these data and transfers them to another storage area and records the information in a key-value store under the same tuple name. The user can then scatter the collected data by sending another list of destination locations in local memory along with the tuple name. The tuple space manager searches its key-value store and fmds the location in storage where the tuples are stored and starts transferring them using direct memory addressing (DMA) to the local memory.

Potential use cases include one or more of the following as examples:

1) Sorting an array of data in a different order—gather them and scatter in the specified order.

2) Gathering results from multiple processes and scattering the results to a next set of participating processes.

3) Gathering data from a matrix column-wise (populate gather list with column indices) and scatter results row-wise.

4) Gathering by type (size or field), scattering by type. In this case, the tuples will be stored in a same hash index, but different hash elements per type.

Turning toFIG. 1, this figure depicts a schematic diagram illustrative of a system constructed using a node architecture100. At the conceptual level, this architecture100enables constructing a system from “units” that combine memory pools and processing capability. As shown in theFIG. 1, multiple types of units are possible. Anode110may contain a single unit or multiple units, and four nodes110-1,110-2,110-3, and110-4are illustrated. Examples of units in a node may include a memory service unit (single chip module, SCM, Unit)115, a Sequential Processing unit (DRAM+CPU)120, a Throughput Processing unit (HBM+Graphic Processing Unit (GPU))130, an acceleration unit140, or a Field field-programmable gate array (FPGA) unit150Unlike previous architectures where GPUs and accelerators are dependent on the host processor, units are independent and treated as peers under the architecture. These units may be optimized for specific computational and memory tasks. The architecture depicts a collection of units where intra-node network103provides an efficient coherent interconnect between the units within a single node and inter-node network180interconnecting the nodes110within the system. Similar to a unit, the inter-node network180may also contain memory (Mem)186and associated processing189. The external network identifies access beyond the system.

In some embodiments, a system is constructed from nodes110connected using an inter-node network180. Logically, the inter-node network is an extension of the intra-node network103. The networks differ in latency, bandwidth, and other physical characteristics. The latency optimized intra-node network allows for coherent load/store access between units. The inter-node network has characteristics that enable scaling to an exascale system while also enabling non-coherent load/store accesses between nodes.

The system includes an Extended Memory (EM) architecture for accessing memory beyond a node110. The Extended Memory (EM) architecture includes two methods for accessing memory: the Global Virtual Address Space (GVAS) and the Coordination Namespace (CNS) methods170distributed over the full system. In this document, the CNS methods are used. Nodes110within the system may have one or more of four major characteristics: (1) Capable of being managed by a single operating system; (2) Efficient coherent load/store access to all memory pools within the node; (3) Global Virtual Address Space for referencing memory pools inside and outside the node; and (4) Access to a system ide Coordination Namespace. In this document the CNS is used.

In prior systems, each node110typically has a fixed topology and limited number of configurations. For example, a node may have two (2) general-purpose processors, 256 GB of DRAM, zero (0) to six (6) Graphical Processing Units (GPUs), and one (1) or two (2) network devices. When constructing large systems, this fixed topology may cause an imbalance in resources. For example, if the targeted application requires a GPU to CPU ratio of 12 to 1, the system would end up with 50% of the general-purpose processors not being used. If the ratio was equal to or lower than 6 to 1, a heterogeneous combination of nodes (some with fewer than 6 GPUs) could meet the ratio, but the node would be over designed and GPU resources are not used. For optimal flexibility in large system design, there needs to be a set of units individually connected to a network and the means for dynamically configuring these units into a node. Therefore, there is a need to dynamically create a logical grouping of units to perform the functions of the targeted application.

The Extended Memory architecture views the system as a collection of memory pools with attached processing rather than a collection of computational engines and associated memory. The subtle reordering places focus on memory allowing programmers to define the data organization, layout, and distribution across the various memory pools in the system. The approaches described herein simplify managing the multiple memory pools and the extended memory architecture provides a consistent view of memory across all units in the system or a subset of units in the system. From a conceptual point-of-view, the plurality of nodes110, may be viewed as a single flat network connecting all units together as peers with equal access to all memory pools and compute resources in a consistent manner. The independent nature of the units enables constructing a system with the proper balance of Sequential Processing units and Throughput Processing units at the system level to meet the needs of a variety of applications and workflows. The approach is to present each memory pool and associated computational capability as independent units to software. The units may be, for example, a combination of processors, programmable logic, controllers, or memory. Example Units160contains a list of example units and does not imply any specific limitations on the types of units within a system with many other types possible, the units and devices are, but not limited to, general-purpose processors, special purpose processors, programmable logic devices, controllers, memory, and the like. To dynamically configure a logical group, these units need to appear to software, especially the operating system and device drivers, as if these are all part of a physically connected system within the shared memory space.

This is like how the cores and GPUs of a traditional node are assigned by the OS, but at a system wide level. The extended memory architecture extends the shared memory space (a Global Virtual Address Space) to other nodes110and provides an efficient means for storing data, communications, and coordination within applications and workflows through a separate, system-wide Coordination Namespace. Units are the fundamental building blocks for a system. In an embodiment, these units may run a specialized kernel for local management in addition to an operating system. This structure allows for combining both traditional and specialized units in various ratios to create a system tailored to the needs of a specific application or workflow. The intra-node network connects units within a node while an inter-node network connects a plurality of nodes to create an exascale system. The intra-node network is optimized for coherently connecting units which are physically close. The inter-node network may be a network such as, but not limited to, Ethernet or InfiniBand with optimizations for enabling a Global Virtual Address Space across the connected nodes. As depicted inFIG. 1, the node architecture may include external network connections providing access outside of the system. These external network connections are networks, such as, but not limited to, Ethernet or InfiniBand attached to each node. One or more units within each node act as a bridge from the intra-node network to the industry standard networks.

From a physical point of view, the term memory traditionally refers to the DRAM or other memory associated with a single system. Thus, an operating system in such a system associates real addresses with DRAM locations. A virtual address translation mechanism converts virtual addresses in a user application to these real addresses. During application execution, the operating system may relocate the physical contents pointed to by a virtual address to some other medium like non-volatile memory or disk. In this case, the application's operation stalls when accessing the associated virtual address until the physical contents are moved back into DRAM and address translation is re-established by the operating system. The extended memory architecture extends this concept of memory in two directions. First, the term memory refers both to DRAM and to SCM associated with the node and to DRAM and SCM on remote nodes. This provides the operating system with a larger range of physical memory to which a virtual address can be associated.

It is helpful at this point to provide a brief overview of memory. SCM stands for “storage class memory”, which really is a broad class of storage devices like flash-memory-based non-volatile memory, phase change memory, and the like. They provide larger storage capacity, similar to SSD (solid state drive) hard drives on the order of terabytes per node when DRAM typically is on the order of 100's of gigabytes per node. DRAM also is volatile memory, that is, data is lost when power is lost or the system is rebooted. The latency of storage class memory is higher than DRAM. Hence, the references herein to DRAM being used as a local memory with lower latency. In exemplary embodiments herein, both system memory and storage class memory are used for CNS. System memory is typically DRAM/HBM that is easily accessible by the processor through its built in memory controller. The storage class memory can be, e.g., a PCI-attached card and has an onboard SCM controller that the processor talks to in order to retrieve data. In exemplary embodiments herein (as described in more detail below), the hash table structures for CNS may be stored in the system memory portion while the tuple data themselves may be stored in the SCM, as an example.

The second extension is a complementary method, provided to the programmer, to facilitate access to Named Data Elements (NDEs) anywhere in the system, at any time. In contrast to the byte-level virtual address used to reference data, these NDEs exist in a new namespace and are referenced by a name or a combination of name and datum within the NDE's contents. The combination of these two techniques provides new and innovative mechanisms for accessing memory within a node as well as across nodes. In addition, the Coordination Namespace allows for accessing address spaces corresponding to different applications within a workflow independent of time.

In an example embodiment, two memory models provided by the extended memory architecture are a Global Virtual Address Space and a Coordination Namespace. As previously stated, the Coordination Namespace is the model used in the examples herein. The Coordination Namespace model, hereafter referred to as the Coordination Namespace or CNS, provides an alternate view of extended memory that is separate from a processes' virtual address space. In the Coordination Namespace, references to extended memory use a “name” for accessing a finite, ordered list of immutable values referred to as a Named Data Element (NDE). In an exemplary embodiment, the first field associated with every NDE is its name, a character string with an implementation-dependent maximum length. The “name” references an NDE located in the Coordination Namespace. The “name” can simply be the first field, the name, a search template for any set of the fields in the NDE, and the like and may be referenced herein as a “name,” a “key,” or as a “NDE-name.” The Coordination Namespace allows access to NDEs contained within a distributed object store.

While it is possible for both these memory models to concurrently exist in a system, a given physical memory location is only accessible using one of the models. The NDE access method provides a set of commands to create, read, retrieve, and destroy NDEs in the Coordination Namespace.

The set of commands described herein are for illustrative purposes only where changes, variations, new, and differences are expected in various embodiments of the concepts described herein. In an embodiment, each node contains a CNS controller that provides access to the Coordination Namespace. When accessing the Coordination Namespace, the CNS controller [e.g., Client or Server] may perform a distributed hash function on the NDE-name to locate the data and perform the data movement. The CNS Server allows access to NDEs in a distributed system in a similar way as load-store instructions in a typical instruction set allows access to locations in a virtual address space. Furthermore, these NDEs are located beyond an application's virtual address space. NDEs may persist beyond the tenure of the application.

FIG. 2depicts an example embodiment of a Sequential Processing unit (SPU) referred to as a node110supporting caching remote memories within a local system's storage. The local system has one or more central processing units (CPUs)210accessing memory250via a coherent bus230. A PCI-Host Bridge (PHB)290connects to a Disk295which may be used for paging or for other purposes, such as, loading programs. By way of example, and not limitation, other architectures may be used to perform I/O, such as, the Industry Standard Architecture (ISA) bus, the Micro Channel Architecture (MCA) bus, and the Peripheral Component Interface (PCI). System memory controller240enables regions of local memory250to be used as a cache. The local memory may be, for example, DRAM, HBM, or the like, and function as both system memory and a cache for remote memory or locally attached SCM275(more distant than the local memory250). A large physical address window (equal to or larger than the memory regions allocated for the cache) may be used for mapping remote and local SCM. Smaller blocks of the physical address space are then mapped, using a cache directory, to a smaller physical memory area allocated to the extended memory caches (RM_EM$ or NM_EM$). In an embodiment, the system memory controller240may support multiple independent cache regions dedicated to caching a memory. For example, the “near-memory” cache (NM_EM$)260serves for data stored in the locally attached SCM275and the “Remote-memory” cache (RM_EM$)255is used for data located in remote memories attached to a remote node. In addition, there can be a plurality of each type of cache. When a referenced datum is not available in the NM_EM$, the reference is forwarded directly to the associated “near-memory” SCM Memory Controller270, completing the access without any CPU involvement. When a referenced datum is not available in the RM_EM$, the memory controller sends an Extended Memory (EM) Cache Miss exception to one of the CPU(s)210. A selected CPU may utilize an interrupt vector for handling the EM Cache Miss exception. In an embodiment, a firmware interrupt handler forwards the virtual address causing the exception to an architected network interface to bring a replica of the remote memory into the RM_EM$. When data is returned from the network interface controller (NIC)285and written into the RM_EM$255, the exception handler is notified, and the CPU load operation is re-issued and is serviced from the RM_EM$. The exception is used to: 1) Prevent stalling the CPU load for the entire duration of the network operation. 2) Determine the virtual address associated with the miss. The network controller may be configured to allow the firmware exception handler to fetch remote memory without needing a full-fledged device driver. In an embodiment, an architected, low latency interface for performing remote direct memory accesses (RDMA) is configured to route the RDMA request to the correct unit or node based on a virtual address. RDMA is a standard protocol to move data from remote nodes.

The schematic diagram of the node110is shown inFIG. 2may implement the methods disclosed herein. The node is only one example of a suitable system node and is not intended to suggest any limitation as to the scope of use or functionality of embodiments described herein. The node could be constructed from a single CPU, a single coherent bus, a single system memory controlling accessing a single memory unit, that is, a node consisting of a single Unit. Examples of well-known computing systems, environments, and/or configurations that may be suitable for use with the node include, but are not limited to, personal computer systems, server computer systems, thin clients, thick clients, handheld or laptop devices, multiprocessor systems, microprocessor-based systems, set top boxes, programmable consumer electronics, network PCs, minicomputer systems, mainframe computer systems, and distributed cloud computing environments that include any of the above systems or devices, and the like. The CPUs210may be described in the general context of computer system-executable instructions, such as program modules, being executed by a computer system. Generally, program modules may include routines, programs, objects, components, abstract data types, data structures, and so on that perform tasks or logic. The CPUs210may be practiced in distributed cloud computing environments where tasks are performed by remote processing devices that are linked through a communications network286via the NIC285. In a distributed cloud computing environment, program modules may be in both local and remote computer system storage media including memory storage devices.

The node110may also contain other devices such as, but not limited to, accelerators280, NICs285, and SCM Controllers270connected to the CPUs210. By way of example, and not limitation, these devices can be directly connected to the coherent bus230or through interface architectures such as Open Coherent Accelerator Process Interconnect (OpenCAPI), or Peripheral Component Interconnects Express (PCIe) bus.

The node110uses CNS, e.g., to access data stored in the local SCM275. As such, the node110contains a CNS controller (Cntr)220, which may be implemented as hardware (CNS controller220-1) located intermediate the coherent bus230and the SCM memory controller270, may be integrated as hardware into the SCM memory controller270, or be located as hardware in another location in node110. The CNS controller220may alternatively or additionally be implemented as computer readable code as CNS controller220-2in memory250and retrieved and executed by the one or more CPUs210to cause the node110to perform actions as described herein. The accessing of data via the CNS is described in more detail below.

The tuple space manager500may be implemented in hardware as tuple space manager500-1. The tuple space manager500may also be implemented as alternatively or additionally as computer readable code as tuple space manager500-2in memory250and retrieved and executed by the one or more CPUs210to cause the node110to perform actions as described herein. The tuple space manager500is described in more detail, beginning atFIG. 5.

The coherent bus230represents one or more of any of several types of bus structures, including a memory bus or memory controller, a peripheral bus, an accelerated graphics port, and a processor or local bus using any of a variety of bus architectures.

The node110typically includes a variety of computer system readable media, such as disk295. Such media may be any available media that is accessible by the node, and it includes both volatile and non-volatile media, removable and non-removable media. The memory250may be any system memory that can include computer system readable media in the form of volatile memory, such as, DRAM and/or a cache memory. The node may further include other removable/non-removable, volatile/non-volatile computer system storage media. By way of example only, a storage system can be provided for reading from and writing to a non-removable, non-volatile magnetic media (not shown and typically called a “hard drive”). Although not shown, a magnetic disk drive for reading from and writing to a removable, non-volatile magnetic disk or memory stick, and an optical disk drive for reading from or writing to a removable, non-volatile optical disk such as a CD-ROM, DVD-ROM or other optical media can be provided. In such instances, each can be connected to the bus by one or more data media interfaces. As will be further depicted and described below, the local SCM275may include at least one program product having a set (e.g. at least one) of program modules that are configured to carry out the functions of embodiments of the methods disclosed herein. A program/utility, having the set (at least one) of program modules, may be stored in the SCM by way of example, and not limitation, as well as an operating system, one or more application programs, other program modules, and program data.

Each of the operating systems may have one or more application programs, other program modules, and program data or some combination thereof, and may include an implementation of a networking environment. The program modules generally carry out the functions and/or methodologies of embodiments of the methods as described herein. The node may also communicate with a set of one or more external devices such as a keyboard, a pointing device, a display, a tablet, a digital pen, etc. wherein these one or more devices enable a user to interact with the node and/or any devices (e.g. network card, modem, etc.) that enable the node to communicate with one or more other computing devices. Such communication can occur via Input/Output (I/O) interfaces. These include wireless devices and other devices that may be connected to the node, such as, a universal serial bus (USB) port, which may be used by a tablet device (not shown). Still yet, the node can communicate with one or more networks such as a local area network (LAN), a general wide area network (WAN), and/or a public network (e.g. the Internet) via a network adapter. As depicted, a network285communicates with the other components of the node via the coherent bus230.

The instant techniques use a Coordination Namespace (CNS). This is described in more detail now. Referring toFIG. 3A, this figure shows a flowchart depicting steps taken to send a request to a coordination namespace (CNS) server. At step300an incoming request, such as an out (NDE-name, NDE-value, or key-value) for a named data element (NDE), is received. At step310, a hash calculation is performed on the NDE-name (hashing may result in a node identifier, a table identifier and a hash index for the node). The hash is performed on at least a portion of the NDE-name parameter passed to the NDE request. Using the node identifier, at step320, the incoming request is sent to a node110(e.g., and the CNS controller220therein) identified by hash result and the process ends at step395.

FIG. 3Amakes the process of finding data in a CNS relatively simple, and to an extent, it is. However, there are also multiple details and background information that is helpful to review at his point. The coordination namespace (CNS) model provides an alternative view of memory distributed across the system that is separate from the virtual address space of a process. Memory within a CNS is treated as key-value pairs instead of being byte addressable. That is, the CNS is a form of a key-value pair that can also span all or a subset of units within the system. The CNS allows applications to store a variable amount of information in distributed memory and to retrieve this information using a single “name”. The name is a text string associated with the information. The “name” and associated information is referred to as a tuple. More specifically, memory within the CNS is referenced using the “name”. The “name” (also referred to as a “key”) references a variable sized block of data (e.g., a value) stored within the tiers of memory distributed across the system. The key-value pair (and therefore the name and associated information as a pair) is referred to as a tuple.

FIG. 1shows that multiple different nodes110may be accessed via an intra-node network103.FIG. 2illustrates a node having a local CNS controller285and network interface controller (NIC) that may be used to access a distributed memory with CNS. As mentioned above, the memory associated with a given CNS may be distributed across the memory pools within the system. The local CNS controllers220are responsible for managing the contributed memory and working with other remote CNS controllers220for performing CNS accesses on behalf of the local processes. An AGEM compliant system should have a local CNS controller220associated with each node110.

All access to a CNS requires calling Application Programming Interface (API) routines. To create a tuple, an application calls an API routine and supplies the name and data, or virtual address pointer to the data, as parameters. The API routine sends an instruction to a local CNS controller220to create the tuple of the given name. The local CNS controller determines where the tuple should be created and, in cooperation with other remote CNS controllers, copies the data from the processes address space into the CNS memory. Once the data is copied, a record of the tuple is generated. Once a record of the tuple is created, the applications can retrieve the information associated with the tuple by again calling an API routine with the name of the tuple and a virtual address pointer where to place the data. The API sends an instruction to a local CNS controller220which, in cooperation with other remote CNS controllers, locates the tuple within the CNS and copies or moves the associated information into the calling process's virtual address space.

FIG. 3Bis an illustration of a distributed coordination namespace and includes a pictorial view of the distributed CNS. As shown inFIG. 3B, the CNS330is divided into groups340-1through340-N. Each group340represents the portion of the CNS storage contributed by a node110. Any node110may attach to, and access a CNS330. The unit performing a CNS access is called the agent using the API routines. When an access instruction is received by the local CNS controller220, a hash function is applied to the name, resulting in a hash value. See block310ofFIG. 3A. A pre-defined number of upper bits from the hash value may be used to select the group associated with the name. The selected group is called the “natural home” for the tuple. Once the natural home is determined, the local CNS controller220forwards (see block320ofFIG. 3A) the request to the CNS controller220corresponding to the tuple's natural home (or “target”) CNS controller220. The target CNS controller uses a predefined number of lower bits from the hash value to select an entry within a local hash table350, also called a hash bucket. Depending on the type of reference, the target CNS controller220either searches the linked list360associated with the hash bucket to determine if the referenced tuple exists, or adds an entry to the list if a new tuple is being generated.

Tuples may also be directed to a preferred group rather than being stored at the natural home. A preferred group ID is provided along with the tuple creation request. The tuple is called a “relocated tuple” and the group where the data is located is called the “actual home”. Applications can retrieve a relocated tuple by providing just the name, or name and preferred group. Providing a preferred group when retrieving a tuple causes the CNS controllers to search the hash bucket corresponding to the preferred group first. If the tuple is not found, the request is forwarded to the natural home.

FIG. 4also illustrates this concept.FIG. 4shows a diagram depicting the homing of a named data element (NDE) in a Coordination Namespace (CNS)330. The requesting node (a node110-RN) is the location running the process making the NDE request. The hash algorithm may be applied to the NDE-name to identify the natural home430. The natural home430directly or indirectly indicates the node110-NH where the NDE is created or may be found if no other information is provided. The preferred home410may be provided by the process making the request or by a prediction algorithm in the CNS client220, for example, by an affinity parameter. When supplied, the preferred home410directly or indirectly indicates the node110-PH where the NDE should be created or where to first search for the NDE. The actual home440identifies the node110-AH where the NDE resides.

When creating a NDE, the preferred home410is tried first. If the NDE cannot be created there for some reason, such as the node being out of memory, an alternate home is chosen, and that node becomes the actual home440. When a NDE is created, the natural home430always keeps a record in the local hash table indicating the actual home but does not store the data. When a NDE is requested, the hash table on the preferred home410is searched first. If the NDE is not found, the request is sent to the natural home430for recording the dummy pointers for the associated key. The nodes110-NH, -AH, and -PH identified by the natural, actual, and preferred homes, respectively, can all be different, the same, or any combination. The CNS330is spread over these nodes. In addition, they can also be different or the same as the requesting node110-RN, and therefore the node110-RN may also be part of the CNS330. The communication between the requesting node, the natural home430, the preferred home410, and the actual home440may be performed via the computer network(s)286.

The rest of this document is laid out as follows.FIG. 5presents an exemplary embodiment tuple space manager that is used to perform the scatter gather operations.FIGS. 6-9describe a first approach for scatter gather operations, where a multipart tuple for scatter gather operations is stored in a single contiguous memory location.FIGS. 10-13illustrate a second approach for scatter gather operations, where the multipart tuple for scatter gather operations is stored in multiple memory locations.FIG. 14is an example of possible software API functions that can be implemented and used in an exemplary embodiment.FIG. 15illustrates a special case for scatter gather operations.FIGS. 16-20illustrate examples of implementations of a key-value store for tuples, in accordance with exemplary embodiments.

FIG. 5is a block diagram of a tuple space manager interacting with a key-value store for tuples, in accordance with an exemplary embodiment. The tuple space manager500comprises an input work scheduler520with a work queue521, N tuple engines530-1,530-2, . . . ,530-N, a heap manager540, a messaging unit535, and a DMA unit550. The tuple space manager500is coupled in this example to a system memory505and a storage class memory (SCM)515, e.g., in a node110. The tuple space manager500is distributed, meaning each node110has its own copy of or access to a “local” tuple space manager500. However, exemplary embodiments allow nodes to not have storage class memory515and yet participate in CNS over the network503. The master and client processes (such as threads) issue CS opcodes as gather/scatter commands in reference510. CS and CNS are being used interchangeably. Typically, CNS is used to refer to the whole architecture of coordination namespace and CS is used for a more limited nomenclature for coordination space as used within the tuple records. These come into the work queue, and the input work scheduler520which ensures these commands are addressed. The work queue521is used by the input work scheduler520when scheduling work to the tuple engines530, or when deciding to send requests off node or to DMA and the like. That is, the input work scheduler520issues new work to the different tuple engines530. The heap manager540supports scatter/gather lists, such as allocating/deallocating multiple addresses. The heap manager540therefore allocates/frees memory in storage class memory515. The DMA unit550provides access (via data movement560) into or out of local memory such as system memory505(e.g., see memory250ofFIG. 2) or SCM515(e.g., see local SCM275ofFIG. 2). The requests sent off node, such as remote access of data, will need help from the messaging unit535and the network503, e.g., employing something like an RDMA protocol via outgoing messages504. Similarly, network messages506also can be incoming and be routed to the input work scheduler520, e.g., for causing local access to the SCM515. Incoming messages506would be from the network503, where the outside tuples space feeds directly the work scheduler521and the incoming messages506are injected into the work scheduler521as another process like the client process from reference510. The DMA unit550moves data according to the scatter/gather addresses without the involvement of software. The tuples engines530provide search, sort, and aggregate functions for tuples (as these functions are described below).

The key-value store570is described in part here but is also described in more detail in reference toFIGS. 16-20. The key-value store570comprises a hash element575that has a local tuple (LT) head pointer (ptr)580, which points to (references) the head585of a double-linked circular list590of local tuples. Note that a hash element575may be considered to be what has been previously referred to as a named data element (NDE). The hash element575may also include remote tuple (RT) and/or pending record (PR) linked list pointers, which are described in more detail below. Depending on implementation, there may be only a single tuple (e.g., without a list590) (see, e.g.,FIG. 7) or multiple tuples in the list590(see, e.g.,FIG. 11). The local tuples (LTs) are described inFIGS. 16-20, as are additional types of tuples (RTs and PRs) that are possibly used.

The near memory545can be a separate memory that has lower latency with respect to the tuple engines530or could be a partition within system memory505. The storage class memory can also be another partition within system memory505. The arrows between the input work scheduler520and the tuple engines530and between the tuple engines530and the key value store570could be implemented as ring or bus or multiplexor, as examples.

There are two approaches described herein. The first approach is to store a multipart tuple from a gather operation in a single contiguous memory location, that is a single tuple as head585. Scatter addresses and gather addresses to the tuple space manager500that manages the coordination namespace. A mechanism is provided in the tuple space manager500to move data according to the scatter/gather addresses without the involvement of software. The list of scatter/gather addresses is accessible by the tuple space manager500. The first approach is illustrated byFIGS. 6-9.

A second approach is to store the multipart tuple in multiple memory locations. This approach needs to maintain a linked list (“linkedlist”) such as list590in the key-value store that provides address of these individual gathered tuple records. This approach needs a mechanism to identify this scatter/gather list within the key-value store, and needs a mechanism to identify the order between the scatter/gathered tuples in the tuple space manager.

For the first approach, methods (seeFIGS. 6 and 8) are disclosed to implement gather (FIGS. 6 and 7) and scatter (FIGS. 8 and 9) of data using a tuple space controller500. This makes use of a key-value store570, e.g., in hardware. Explicit gathering into one contiguous block in storage class memory (scm) may be performed. Applications can create list of data in system memory. The addresses are collected in an array and given to the hardware to gather into one location in storage class memory. The tuple space controller500creates only one tuple record for the aggregated data in scm. Data aggregated in storage class memory can be scattered again to different addresses by passing the tuple name, list of addresses and list of size where the data needs to be scattered.

Once gathered in storage class memory, the tuple space manager500can post process and distribute data again into a different method/tuple name as required. The linked list storing the tuple name will be modified accordingly. Scatter/gather is performed into only one tuple record per name. Gather information is stored within the tuple and the data may be redistributed based on the previous gathered information. One possibility is to store the block size for each fragment within the tuple to facilitate this.

Referring toFIGS. 6 and 7,FIG. 6is a flowchart of a method of a first approach to store a multipart tuple from a gather operation in a single contiguous memory location, andFIG. 7is an illustration of tuple space manager interactions for the method inFIG. 6. In block610, individual processes (e.g., threads) generate data in their respective node's local system memory. This is illustrated inFIG. 7by the processes Pr-0710-0, Pr-1710-1, PR-2710-2, through PR-n710-n, which create corresponding data data0720-0, data1720-1, data2720-2, through data-n720-n. Each of these data is at a corresponding address addr0730-0, addr1730-1, addr2730-2, through addr-n730-n. These addresses are the addresses in system memory (e.g.,505inFIG. 5) from which the tuple space manager500is going to gather and store into SCM (e.g.,515inFIG. 5).

In block620, a user process in a user space (e.g., connected to or part of a node110) requests gather of data. In block630, any one process that is participating in a coordination namespace issues a command (e.g., gather_tuple_list), responsive to the request, to a tuple space manager500attached to its node110to cause a gathering of a tuple list. That is, the tuple records of the same name are gathered under the same hash element570, which is a function of the key-value store570. The tuple data is stored in SCM515. Parameters in the command might include a tuple name under which the data is gathered, a list of node and process IDs and addresses where data is present and a list of sizes of each data. This is illustrated inFIG. 7by the tuple space manager500and the command gather_tuple_list with the parameters of tuple name (e.g., “tuple name 1”), a nodeID list of nodes110where data720-0through720-nmight be stored, a processID list (e.g., containing process IDs of the processes710-0through710-n), an addr list (e.g., comprising addresses730-0through730-nwhere the corresponding data720-0through720-nare stored), and a size list (e.g., comprising sizes of the corresponding data720-0through720-n). It is noted that a list herein may also be referred to as an array, since an array is one implementation of a list.

In block640, a tuple space manager500at a preferred home receives the request, and requests a memory (e.g., scm) controller for a location where the data needs to be gathered. The size of data is an aggregate of all data sizes. That is, an accumulation of sizes in the content of the size array. It is noted that the process inFIG. 6might start at a local tuple space manager500that is not at the preferred home, and block640concerns what happens at the preferred home. How the flow might get from the local tuple space manager500to the tuple space manager500at the preferred home is described below, in reference toFIGS. 6A and 6B.

In block650, the tuple space manager500processes the array (having a size) of data list one at a time, performing a remote memory access (e.g., RDMA) load to retrieve the data to a temporary buffer790in the preferred home where this request is being processed. To access data on a remote node, the tuple space controller500uses the process ID, address, and size as part of the RDMA protocol message setup. The DMA operations need not be blocking and can be issued one after another without waiting for completion. That is, a “normal” DMA operation would have to wait until a previous DMA operation was complete, and this DMA operation would therefore be blocked until the previous DMA operation is complete. In this example, however, the DMA operations can be issued effectively in parallel and one DMA operation does not block another.

In block660, from the temporary buffer790on the preferred home, the tuple space manager500then moves the data to the scm address space (e.g., in SCM515). Upon aggregating in block670all the data together to the scm controller space, the tuple space manager500completes the creation of the tuple record and links the tuple record in the hash table linked list. InFIG. 7, blocks660and670are illustrated by the Hash Element 1575-1, having a tuple name of “tuple name 1” and having an LT head ptr580-1to a single tuple740. All of the gathered data is stored in a single memory block that is accessed by this single tuple740. As illustrated in block751ofFIG. 7, the single tuple740has information linking (e.g., referencing) to the data (e.g., the memory block containing the data) in the SCM.

In block680, the preferred node then issues a completion notification to the requesting node where the original request to gather was received. The preferred home also notifies a natural home about the creation of the tuple record. The requesting node's tuple space manager in block690sends the completion notification to the user space, informing the user process that requested the gather of data.

As described above, the flow might get from the local tuple space manager500to the tuple space manager500at the preferred home is described below, in reference toFIGS. 6A and 6B. As an introduction, a preferred home is essentially a user-provided home suggesting where a tuple record should first be stored or sought from. If user does not provide one, then the system defaults to a natural home. It might be the case that user predicts incorrectly the preferred home, and tuple space manager queries for the tuple record location from the natural home and identifies the “actual home” of the tuple record and the associated data. The data itself is on the SCM515attached to the node110where the tuple record exists.

FIG. 6Ais a flowchart of a method for performing a csOut command, which creates a new tuple in CNS. For csOut, a preferred home becomes an actual home. In block602, a user requests a csOut, and this includes a tuple name601. For instance, a user using a node100can make this request, and the node110contacts its tuple space manager500with the request. In block606, the work scheduler620checks if the current node is the preferred home. If not (No), the flow proceeds to block608, where the messaging unit535(e.g., under control of the input work scheduler520) sends a message to the preferred home. The network message604is an incoming message506from another node. There are several types of messages, e.g., from forwarding packets to control messages, and the like, going between nodes in support of CNS. Here, the network message604is referring to the forwarded packet that was sent off node in block608of a requesting node and the messages reaches the destination node and enters the flow in box604. Thus, one node for this csOut performs blocks602,606, and608, and a second node receives the network message604(e.g., comprising the csOut), performs block606then proceed to block612.

If the current node is the preferred home (Yes), the tuple engine530(e.g., one of the tuple engines530-1through530-N) checks the hash and computes a hash entry address, and issues a Read head pointer command to read into the hash table. The blocks that access the DDR memory614are shown in the figure. Block612, for instance illustrates a DDR memory614is accessed. As previously described, in an exemplary embodiment, DDR is where the tuple records are stored and SCM is where tuple data is stored. DDR/DRAM is attached to a processor or processors through, e.g., memory DDR slots while SCM is attached through, e.g., PCI cards, in an exemplary embodiment. The tuple name601is an array of “bits” and a hash is a unique function that reduces these bits into a number. It is possible that more than one tuple name can reduce to the same hash. Hence, there is a need to search within the same hash index, for each hash element to see if the element matches the tuple name. The record retrieved might indicate the presence of the tuple record either locally or the retrieved record might indicate that the tuple record is in another home, called an actual home (i.e., the preferred home provided by a user was incorrect, and the natural home tells the correct home for the tuple record). Some of the blocks below perform this searching.

In block616, the tuple engine530checks the DDR response, computes a next address of the hash element, and issues a Read hash element command, which is used to access the DDR memory614. If a hash element is not found (No), this means there is no hash element for the hash index. The tuple engine530needs to create new hash element record and then goes to block628to create a tuple record.

If the hash element is found (Yes), in block618, the tuple engine530checks the DDR response, checks the tuple name in the hash element, and does a tuple name match request. If the tuple name match request indicates the tuple name is the same as in the user request (tuple name601) of block602(Yes), in block622, the tuple engine530gets the head of a local tuple list, issues a DDR Read head request for a first local tuple, and accesses the DDR memory614. Block622assumes the first local tuple is retrieved. If the tuple name match request indicates no tuple match (No), in block619, it is determined if this is the last hash element. If not (No), the flow proceeds to block616. If it is the last hash element (Yes), the flow proceeds to block623, where the tuple engine530gets a free pointer for a hash element and the flow proceeds to block628.

In block624, the tuple engine530gets the next pointer of the retrieved tuple, issues a DDR read request for the next local tuple in the list, and accesses the DDR memory614. If the next tuple is retrieved (Yes), in block626, the tuple engine530determines if this is the last element in the list. If not (No), the flow proceeds back to624. If so (Yes), in block628, the tuple engine530gets a free pointer for the local tuple record, writes a new tuple record with location of the data in the scm (also referred to as SCM) and accesses the DDR memory614to access the SCM515. The tuple engine530in block632completes processing, notifying the work queue521and the user of completion. The tuple engine530also notifies the natural home of the new record and in block634a messaging unit535(e.g., under control of the input work scheduler520) send one or more messages to the user and/or the natural home.

Referring toFIG. 6B, this figure is a flowchart of a method for performing a csIn command, which retrieves and removes a matching tuple from CNS. In block636, a user issues a request csIn command, including a tuple name601, which is sent to the input work scheduler520. For instance, a user using a node100can make this request, and the node110contacts its tuple space manager500with the request. Blocks604,606,612and616are similar or the same asFIG. 6A. However, for block616inFIG. 6B, there should be no need to create a hash element (as this process inFIG. 6Bretrieves hash elements already created). If the current node110is not (No) the preferred home, the flow proceeds to block638, where the messaging unit535(e.g., under control of the input work scheduler520) sends a message to the preferred home. This is similar to block608ofFIG. 6A, except that the message being sent is a csIn (instead of the csOut of block608).

In block644, the tuple engine530checks the DDR response, checks the tuple name in the hash element, and determines whether the tuple name matches the tuple name601in the request. Note if this is the last element (Is last element), in block642, the tuple engine530sends a message to the natural home. In other words, if the end of hash element linked list (“linkedlist”) has been reached, go to block642, because the element does not exist on this node, and send a request to the natural home. Note that at the natural home, the processing goes through the same flow from beginning, while checking whether the node=natural home. If the node still does not find the entry, the node creates a hash element with a pending record (PR, see below for a description) for this request, expecting a csOut for that tuple name in the future.

If there is no name match, then the flow proceeds back to block616, where another Read Hash element is performed. That is, the next hash element in the linked list is retrieved. If there is a name match (Yes), in block646, the tuple engine530gets a head of a local tuple list, and issues a DDR read request for the first local tuple, and accesses the DDR memory614. In block648, the tuple engine530removes the element from the linked list, updates the hash element to point to the next element in the list, and deletes the hash element if the hash element was the last element. This accesses (e.g., via a read) the DDR memory614. In block652, the tuple engine530informs the DMA to transfer data from the scm (e.g., SCM515) to local memory (e.g., system memory505). The natural home is updated, by the tuple engine530, in block654of the tuple removal. In block656, the tuple engine530completes processing and notifies the input work scheduler520and/or the user of the completion.

The techniques inFIGS. 6A and 6Bapply to the other flowcharts herein. For instance,FIGS. 8, 10, and 12can use one or both of these techniques. Note also that the tuple name601is only one parameter that might be sent in a request. Other parameters are illustrated, for instance, inFIGS. 11 and 13.

Turning toFIGS. 8 and 9,FIG. 8is a flowchart of a method of a first approach to scatter data from a multipart tuple, previously stored from a gather operation in a single contiguous memory location, andFIG. 9is an illustration of tuple space manager interactions for the method inFIG. 8.

In block810, any process that is participating in the coordination namespace issues (e.g., based on a request from a user process) a scatter tuple list command to the tuple space manager attached to its node. Parameters may include tuple name under which the data is gathered, list of addresses where the data needs to be scattered, list of sizes of each data and the process IDs that will receive this data. The order of arrays in each of the lists should be same, matching the corresponding data segment in the gathered data. This is illustrated inFIG. 9by the scatter_tuple_list (tuple name, nodeID[ ], processID[ ], addr[ ], size[ ]). In this example, the tuple name is “tuple name 1” as illustrated by Hash Element 1575-1.

In block820, the requesting node sees this request and sends the request to a preferred home that was part of the request. If the preferred home is not provided, the request is sent to a natural home computed from the tuple name. As previously described, a preferred home is user-provided. It is a definition used for common tuple operations for csOut, csIn too. The user could incorrectly predict where the tuple record may be found, and hence the need to consult with the natural home. If the preferred home is predicted correctly, then the tuple processing can be performed right there and the tuple processing engine updates the natural home accordingly.

In block830, if the natural home received the request, the natural home looks through its hash table for the particular hash ID, and scans the linked list for the given tuple name. The hash ID is computed from the tuple name. When the request for csIn/csOut is issued, either the CNS software layer or CNS hardware computes the hash ID from the tuple name and appends the hash ID as part of the request going further into the CNS tuple processing. The tuple name is an array of “bits” and hash is a unique function that reduces these bits into a number. It is possible that more than one tuple name can reduce to the same hash. Hence, there is a need to search within the same hash index, for each hash element to see if the element matches the tuple name. The record retrieved might indicate the presence of the tuple record either locally or the retrieved record might indicate that the tuple record is in another home, called an actual home (i.e., the preferred home provided by a user was incorrect, and the natural home tells the correct home for the tuple record).

In block840, the request is forwarded to the actual home if it is found that natural home did not have the record locally. In block850, the actual home or natural home where the tuple record is present as a local tuple record starts to process (e.g., via the tuple space manager500) the received request and starts issuing memory (e.g., DMA) transfers to move data from storage class memory (scm) to a temporary buffer in local memory on that node. That is, fromFIG. 8, the tuple space manager500accesses the data in the SCM515based on information from the single tuple740linking (e.g., referencing) the data (e.g., the memory block containing the data) in the SCM515, and moves this data to a temporary buffer990.

The actual/natural home now triggers (see block860) RDMA calls to move data parts to the different addresses that may be on other nodes/processes. Upon completion of moving all the data parts to different scatter addresses from the temporary buffer, the temporary buffer990is released. The tuple space manager500cleans up the tuple record in its hash table. InFIG. 9, this is illustrated by the data being scattered from the single tuple740and the temporary buffer990to the data data0920-0, data1920-1, data2920-2, through data-n920-n, each at a corresponding address addr0930-0, addr1930-1, addr2930-2, through addr-n930-n. The data data0920-0, data1920-1, data2920-2, through data-n920-nalso corresponding to processes Pr-0910-0, Pr-1910-1, PR-2910-2, through PR-n910-n.

In block870, if the actual home did the above processing, the actual home sends a notification to natural home to clean its copy as well. If actual home is the same as the natural home, the actual home cleans up the records itself. The actual home in block880then sends a completion notification to the original requesting node. In block890, the tuple space manager500on the original requesting node issues a completion notification to the user process to indicate scatter completion.

A second approach is to store the multipart tuple in multiple memory locations. This approach needs to maintain a linked list (“linkedlist”) in the key-value store that provides address of these individual gathered tuple records. This is implicit gathering of tuples at a preferred home using a tuple space manager, where data can be from multiple processes. Data can stay fragmented in storage class memory. Only the tuple information is gathered together in one hash element record.

Turning toFIGS. 10 and 11,FIG. 10is a flowchart of a method of a second approach to store a multipart tuple from a gather operation in multiple memory locations, andFIG. 11is an illustration of tuple space manager interactions for the method inFIG. 10. In block1010, the individual processes (e.g., threads) generate data in their respective node's local system memory. This is illustrated inFIG. 11by the processes Pr-01110-0, Pr-11110-1, PR-21110-2, through PR-n1110-n, which create corresponding data data01120-0, data11120-1, data21120-2, through data-n1120-n. Each of these data is at a corresponding address addr01130-0, addr11130-1, addr21130-2, through addr-n1130-n. This is an address in system memory (e.g., system memory505) where the data is currently located in user process space. The process ID and this address are needed to translate from virtual address to real address in the system memory of the given node to access the data. The process ID is required to get permission from the OS to access that address space.

In block1020, each of these processes issues a command to create a new tuple, with a same tuple name. If ordering of requests is needed, the process should include a part number. The command to create a new tuple may be csOut, as illustrated inFIG. 14as an API and as described in reference toFIG. 6A. InFIG. 11, this is illustrated as each thread issuing an “Out” command with parameters of data address, size and a same tuple name, and part number. InFIG. 6A, this is the user request of csOut in block602. The part number is a sequence number of the gathered list if one wishes to maintain this order. It is noted that data can stay fragmented in storage class memory, and that only the tuple information may be gathered together in one hash element record. This is illustrated by block1051, where the tuples have information linking from the tuples to the corresponding data in the SCM515.

There are two options. In option 1 (see block1030), the referred home should not be provided, thus allowing gather at a natural home computed from tuple name; or it should be ensured all processes use a same preferred home. In option 2 (see block1035), the preferred home can be different. This relies on ordering in a natural home of the tuples.

In block1040, since these requests can be generated across multiple nodes/processes, the tuple space manager on each requesting node receives a request. If a preferred home is provided, this is the node where the request is sent. If not, the natural home is computed from the tuple name and this is the node where the tuple record is created. The tuple space manager at the requesting node then takes care of forwarding the request to the preferred home or natural home if no preferred home is provided.

In block1050, the tuple space manager500at the preferred home/natural home receives each of the requests from different nodes, e.g., in random order. The tuple request processes them as individual csOut requests and gathers them as individual tuples with a same name. When the tuple record is created in the local tuple linked list, the part number is used to identify the tuple record in the list. The scm controller may store the data either at a same aggregated location or the data can be spread out. In block1060, data on remote nodes is fetched using remote DMA load to a temporary buffer1190(seeFIG. 11) in the preferred home where this request is being processed. To access data on a remote node, the tuple space manager uses the process id, address, and size as part of the RDMA protocol message setup.

In block1070, from the temporary buffer1190on the preferred home, the tuple space manager then moves the data to the scm address space. In the example ofFIG. 11, this is illustrated by the Hash Element 1575-2, with a tuple name of “tuple name 1”. The Hash Element 1575-2has an LT heat pointer (ptr)580-2that points to (e.g., references) the head1140-0of a double-linked circular list1150, which has tuples1140-0,1140-1,1140-2,1140-3, . . . ,1140-n. That is, tuples1140are the tuple records. These records have information on where the data is physically stored in SCM515. The data got moved from data1120(at their corresponding addresses1130) to SCM515. In other words, the tuples1140do not contain the data1120, but instead contain information linking (e.g., referencing) the data1120in SCM515. The double-linked circular list1150is a list of tuples1140with a same name.

In block1080, the preferred node then issues a completion notification to each of the csOut requests the preferred node received. This technique used a gather-by-tuple-name operation to collect all tuple records in one location even though the data itself may or may not be aggregated within the scm controlled data space. In an exemplary embodiment, linked list support is built in for CNS. Tuple records associated with tuples of the same name automatically get gathered under the same hash element at the natural home/preferred home. Implicit means utilizing this built-in mechanism. The data gets moved from system memory/user process space to scm and may be in separate blocks within scm depending on the address given by the heap manager for each tuple record. No attempt is made to allocate a single large block of SCM memory. Instead, it is possible to gather the data under multiple tuple records belonging to the same hash element. Therefore, the data is now under the master/scm process. The objective is to allow sharing this data with other processes. A different process participating in the CNS can now come and read all the aggregated data by scanning through the linked list (“linkedlist”) associated with this hash element (via a key).

Referring toFIGS. 12 and 13,FIG. 12is a flowchart of a method of a second approach to scatter data from a multipart tuple, previously stored from a gather operation in multiple memory locations, andFIG. 13is an illustration of tuple space manager interactions for the method inFIG. 12.

In block1210, individual processes that need the gathered data back issue a command (e.g., csIn, seeFIG. 14) to retrieve and remove a matching tuple from CNS, including part number and tuple name. These requests all get routed to a location where local tuple records are stored. This is similar to previous sections on preferred/natural node routing. This is illustrated inFIG. 13by each process/thread issuing “In” with addr (address), size (e.g., for the data) and same name, and part number in the tuple space manager500.

In block1220, if the natural home received the request, the natural home looks through its hash table for the particular hash ID, and scans the linked list1150(seeFIG. 13) for the given tuple name. As previously described, the hash ID is computed either in software or in hardware and is a hash of the tuple name. The name is really a set of bits, and a hash is performed on this to generate a number which is used as a hash index. The record retrieved might indicate the presence of the tuple record either locally or it might be that the tuple record is in another home, called an actual home (i.e., the preferred home provided by user was incorrect, and natural home tells the correct home for the tuple). In block1230, the request is forwarded to the actual home if it is found that the natural home did not have the record locally.

In block1240, the actual home or natural home where the individual tuple record is present, as a local tuple record, starts to process the received request. The actual/natural home identifies the tuple in the linked list1150under the tuple name by its part number and starts issuing DMA transfers to move data from storage class memory (scm) to a temporary buffer1390(seeFIG. 13) in local memory on that node. Note that the tuples in the hash element575-2may, as indicated in block1051may have information linking from the tuples to data in SCM515. In block1250, the actual/natural home now triggers RDMA calls to move data to the scatter address (addr) that came in with the csIn request, which is the output address to where the data is copied to. SeeFIGS. 6A and 6B, which explain the csOut and csIn flows, respectively. This data may be on other nodes/processes. Upon completion of moving the data to the remote node address from the temporary buffer, the temporary buffer1390is released. The tuple space manager cleans up the tuple record in its hash table. This process is illustrated inFIG. 13by the data data01320-0, data11320-1, data21320-2, through data-n1320-nbeing output to their corresponding address addr01330-0, addr11330-1, addr21330-2, through addr-n1330-n. Each of these addresses1330corresponds to an individual one of processes Pr-01310-0, Pr-11310-1, PR-21310-2, through PR-n1310-n. The data is now scattered back to the processes1310. It is noted that the processes inFIG. 11andFIG. 13tneed not be the same. The process list for scatter and gather is expected to be different, but typically the number (n) of them is same. If n is not the same, then only the explicit scatter-gather works, as data is now scattered in chunks that are different from the original.

In block1260, if the actual home did the above processing, the actual home sends a notification to natural home to clean its copy as well. If actual home is the same as the natural home, the actual home cleans up the records itself. The actual home in block1270then sends a completion notification to the original requesting nodes that requested the different parts of the gathered tuple records. The tuple space manager500in block1280on the original requesting node issues completion notification to the user process to indicate scatter completion.

As additional examples for the second approach, methods have been descried to implement creation of multi-part tuple using scatter and gather of data. For collection of the multi-part tuple, there is implicit gathering of tuples using a tuple space manager500. The tuples of a same name are gathered together in one linked list. Data can stay scattered in storage class memory. When all parts are received, a tuple is created. This is the exemplary case of implicit gathering. Each process essentially stored the data using csOut with same name, and the data got moved from system memory/local space to storage class memory. The tuple records are gathered under one hash element (also referred to as a named data element), but the data can remain scattered across the SCM. In an exemplary embodiment, no computation is performed on the data in SCM. Thus, the option to aggregate the data in one location in SCM is just an option.

Store ordering may be performed as follows: 1) Store in part order; or 2) Store out of order. For storage out of order, metadata is used to determine order when retrieved. Load ordering may be performed as follows: 1) Load in order as tuple was stored as sequential; or 2) Load by metadata or part number.

Insertion/search in tuple list may be performed according to part number. It is possible to speculatively return parts for loads, e.g., by loading all parts. In other words, if requests to retrieve the gathered tuple record comes in from different processes, then the tuple space manager500starts retrieving the records one-by-one from the hash table. This process takes time. If one can speculatively guess that the gather has initiated based on the first request, it is possible to proactively start retrieving all the tuple records in the same order as part number to hide (e.g., limit) the latency of retrieving linked-list data. Parts refers to the sequence number of the data in the list/array.

Turning toFIG. 14, a table is shown illustrating possible exemplary commands between CNS controllers in accordance with an exemplary embodiment. The following rows are shown: Software API, which lists the software API function called to perform the corresponding action in the “Description” row; Hardware opcode, which lists a corresponding opcode that is implemented in hardware; and Description, which describes the action taken by the hardware in response to the opcode. As examples, the csOut for the software API would create a new tuple in CNS, and the csIn retrieves and removes a matching tuple from CNS.

FIG. 15illustrates a special case of scatter gather in an exemplary embodiment. This uses the example fromFIG. 13, which uses the second approach for scatter gather operations, where the multipart tuple for scatter gather operations is stored in multiple memory locations. However, this special case could also use the first approach (seeFIG. 9) for scatter gather operations, where a multipart tuple for scatter gather operations is stored in a single contiguous memory location.FIG. 15illustrates using the implicit scatter/gather feature to gather data located across different processes into one memory location of a given process. So all the participating processes would gather like before, but for scatter, the data are written back to a same process in adjacent data blocks. This creates an aggregated data. From a user point of view, this is gathering data to one location in system memory.

FIG. 15illustrates this by the command of scatter_tuple_list (tuple name, addr[ ], size[ ]) in the tuple space manager500, which causes the tuple space manager500to transfer data from the linked list1150to the array1510of adjacent addresses indicated by the “addr[ ]” array. This example shows the end result is a single hash element, which is identified by a tuple name, and is indicated by the array1510. The array1510begins at addr01330-0and ends n addresses later. The size[ ] array indicates the size of each corresponding datum in the data data01320-0, data11320-1, data21320-2, . . . , data-n1320-n. One can calculate size from the spacing/striping between addresses, but it is safer to provide the size. The heap manager could have allocated a larger size to the data even though the data size is smaller. The data is written by the tuple space manager500to a single process1310, illustrated as process Pr-x1310-x. As before the temporary buffer1390may be used as a temporary storage location during the scatter process.

FIGS. 16-20illustrate examples of implementations of a key-value store for tuples, in accordance with exemplary embodiments. These figures illustrate field programmable gate array (FPGA) double data rate (DDR) hash structures1600, which is one exemplary hardware implementation for these structures. It should be noted that FPGA is only one example of hardware, and it could be possible to implement some or all of this using computer readable code and corresponding processor(s). Also, DDR is one example of memory, and other memories may be used. Also, DDR memory should be the local memory that is close to the FPGA. SeeFIG. 5, where the key value store570has near memory545. Hash structures are present within this local memory. The local memory can be any type of memory (e.g., DRAM, SDRAM, NVRAM), and DDR is just one such type of memory. The FPGA DDR hash structures1600are one exemplary way a key-value store570(see, e.g.,FIG. 5) may be implemented.

Turning toFIG. 16, this figure illustrates the FPGA DDR hash structures1600and reference numerals associated with the elements of the figure. For clarity, reference numerals that have been used previously are also referred to onFIG. 16, and new reference numerals have been added too. InFIG. 16, there is a hash table350with a hash table pointer (hashtable ptr). The entries in the hash table350are HashElemPtr, hash element pointers, which point to one of the hash elements575. In the previous examples, there was only one hash element575, but this example has multiple hash elements575-1,575-2,575-3, and575-4, which are formed in a linked list (hash element575-1points to hash element571-2, which points to hash element575-3, which points to hash element575-4, and finally hash element575-4points to hash element575-1, in a circular manner). The element at hastable ptr+index points to the hash element575-1of HashElement1. There are a number of free lists per type in the upper right-hand side: hash elements1605; local tuples (LTs)1610; remote tuples (RTs)1620; and pending records (PRs)1630.

Each of the hash elements575includes an LT Head Ptr580, an RT Head Ptr1640, and a PR Head Ptr1650, although these may not be active. For the hash element575-1, this has an LT Head Ptr that points to the head of a double-linked list1660-1and an RT Head Ptr that points to a head of a double-linked list1670-1. The PR Head Ptr is not active. The natural home/actual home, local and remote tuples may be accessed via the lists1660-1and1670-1, respectively. The hash element575-2has an LT Head Ptr that points to the head of a double-linked list1660-2and the RT Head Ptr and PR Head Ptr are not active. The actual home, local tuples only may be accessed via the list1660-2. The hash element575-3has a PR Head Ptr that points to the head of a double-linked list1680and the LT Head Ptr and RT Head Ptr are not active. The natural home, pending record tuples only may be accessed via the list1680. The hash element575-4has an RT Head Ptr that points to the head of a double-linked list1670-2and the LT Head Ptr and PR Head Ptr are not active. The natural home, remote tuples only may be accessed via the list1670-2.

In more detail, the hash table350contains pointers to hash element linked lists. The index into the hash table is computed from a tuple name hash. Multiple tuple names can hash to a same index. And they are linked as a linked list of hash elements. Each hash element in that list would be for a unique tuple name. It is possible to have multiple tuples for the same name. A hash element575has pointers to a local tuple, remote tuples and pending records, all for the same tuple name. Each of the local tuples/remote tuples/pending records connected to a given hash element is connected to themselves as circular double-linked structures.

There are four possible combinations of allocation of tuple records in memory: 1) local and remote tuples associated with a tuple name hash element if it's a natural home or actual home; 2) if only local tuples are present, this indicates this is the actual home; 3) if only pending records present for a given tuple, this indicates it's the natural home, and PR cannot be present in actual homes; 4) if only remote tuple list is present for a given tuple name, this is the natural home for that tuple.

There are four types of free list memory buffers (1605,1610,1620,1630), one for each type that is needed to form these linked list structures. When the tuple engine needs to create an entry in these linked structures, the tuple engine picks up the entry from the free lists1605,1610,1620,1630of the given type. These are described on the next figures.

Referring toFIG. 17, this figure illustrates fields for a hash element in the hash elements1605: the address of the next hash element (addr of next HashElem); the address of the previous hash element (addr of prev HashElem); the address of the hast table parent (addr of HashTable parent); the address of a pending request (addr of PendingReq); the address of the local tuple (addr of LocalTuple); the address of the remote tuple (addr of RemoteTuple); and 80 characters (name) (e.g., for the tuple name).

Referring toFIG. 18, this figure illustrates fields for an element in the local tuples1610: address of the next local tuple (addr of next LocalTuple); address of the previous local tuple (addr of prev LocalTuple); address of the hash element parent (addr of HashElem parent); address of the actual tuple (addr of actual tuple); the size of the actual tuple (size of actual tuple): and the address of the natural home (NH) remote tuple (addr of NH RemoteTuple).

Referring toFIG. 19, this figure illustrates fields for an element in the remote tuples1620: address of the next remote tuple (addr of next RemoteTuple); address of the previous remote tuple (addr of prev RemoteTuple); address of the hash element parent (addr of HashElem parent); actual home of the tuple (Actual home unit of tuple); and the address of the local tuple at the (actual) home (addr of LocalTuple at home).

Referring toFIG. 20, this figure illustrates fields for an element in the pending records1630: address of the next pending request (addr of next PendingReq); address of the previous pending request (addr of prev PendingReq); address of the hash element parent (addr of HashFlem parent); the requesting unit (Requesting unit); the requesting process ID (Requesting pid); the requesting address (Requesting addr); the requesting size (Requesting size); the requesting queue tag (Requesting queue tag); and the requesting type (Request type (RD/IN)). RD is short for csRd (read tuple without removing record from coordination namespace), seeFIG. 14. Additionally, IN is short for csIn (read tuple and remove from coordination namespace), see alsoFIG. 14. Further, OUT is short for csOut (write tuple to coordination namespace), as described above. Pending records are created only for csRd and csIn operation when they are issued before the csOut was issued. If a matching tuple already exists, no pending record is created. This disclosure specifically does not require pending records, but such records are needed as part of CNS.