Patent Description:
As data storage devices continue to advance, such as with key-value based drives including key-value solid state drives (KVSSDs), the speed at which the storage devices are able operate continues to increase. The speed of these storage devices may limit the rate at which a data storage system (e.g., a non-volatile memory express (NVME) over fabrics storage server) is able to operate. However, due to the improved performance of the storage devices, and also due to the increased scale at which storage systems must operate, one or more central processing units (CPUs) may be overloaded. The CPUs may be for processing input/output (IO) requests and for coordinating data transfer within the system among multiple storage devices and multiple clients and hosts. The overloading of the CPUs may cause a bottleneck in the system at the CPU.

The above information disclosed in this Background section is only for enhancement of understanding of the background of the disclosure, and therefore may contain information that does not form the prior art.

<CIT> discloses: A storage controller includes a distribution core, a plurality of sorting cores, and a request processing core. The three types of cores are separately configured to distribute an input/output (IO) request to different sorting cores, generate a processing sorting index for each IO request, and process the IO request according to a value of the processing sorting index of each IO request to flexibly schedule the IO request received by the storage controller.

<CIT> discloses: A method and framework for scheduling receive-side processing of data streams received from a remote requesting client by a multiprocessor system computer is disclosed. The method receives data packets from the remote requesting client via a network and, for each data packet, applies a cryptographically secure hashing function to portions of the received data packet yielding a hash value. The method further applies the hash value to a processor selection policy to identify a processor in the multiprocessor system as a selected processor to perform receive-side processing of the data packet. The method queues the received data packet for processing by the selected processor and invokes a procedure call to initiate processing of the data packet.

<CIT> discloses: A device may be configured to store virtual identifier information indicating virtual identifiers associated with servers. The virtual identifier information may associate a quantity of virtual identifiers with each respective server of the servers based on a weight associated with the respective server. The device may receive an object identifier identifying an object to be processed by at least one of the servers. The device may calculate hash values for the virtual identifiers based on the object identifier. The device may determine a virtual identifier associated with a hash value that satisfies a particular condition. The device may select a server associated with the virtual identifier. The device may send an instruction to the server to process the object.

Embodiments described herein provide improvements to data storage.

According to an aspect, there is provided a method of packet processing, the method including receiving an input/output (IO) request from a host, selecting a drive corresponding to the IO request using a hashing algorithm or a round-robin technique, and establishing a connection between the host and the drive.

According to an aspect, there is provided a system for packet processing, the system including a processor including a plurality of cores, and a drive-processing module, wherein one of the cores is configured to receive an input/output (IO) request from a host, the drive-processing module is configured to select a drive corresponding to the IO request using a hashing algorithm or a round-robin technique, and the processor is configured to establish a connection between the host and the drive.

According to an aspect, there is provided a non-transitory computer readable medium implemented on a system for packet processing, the non-transitory computer readable medium having computer code that, when executed on a processor, implements a method of packet processing, the method including receiving an input/output (IO) request from a host, selecting a drive corresponding to the IO request using a hashing algorithm or a round-robin technique, and establishing a connection between the host and the drive.

Accordingly, the system and method of embodiments of the present disclosure is able reduce or eliminate CPU bottlenecking to improve data storage by balancing loads among CPU cores and storage devices.

Non-limiting and non-exhaustive embodiments of the present embodiments are described with reference to the following figures, wherein like reference numerals refer to like parts throughout the various views unless otherwise specified.

Corresponding reference characters indicate corresponding components throughout the several views of the drawings. Skilled artisans will appreciate that elements in the figures are illustrated for simplicity and clarity, and have not necessarily been drawn to scale. For example, the dimensions of some of the elements, layers, and regions in the figures may be exaggerated relative to other elements, layers, and regions to help to improve clarity and understanding of various embodiments. Also, common but well-understood elements and parts not related to the description of the embodiments might not be shown in order to facilitate a less obstructed view of these various embodiments and to make the description clear.

Features of the inventive concept and methods of accomplishing the same may be understood more readily by reference to the detailed description of embodiments and the accompanying drawings. Hereinafter, embodiments will be described in more detail with reference to the accompanying drawings. The described embodiments, however, may be embodied in various different forms, and should not be construed as being limited to only the illustrated embodiments herein. Rather, these embodiments are provided as examples so that this disclosure will be thorough and complete, and will fully convey the aspects and features of the present inventive concept to those skilled in the art. Accordingly, processes, elements, and techniques that are not necessary to those having ordinary skill in the art for a complete understanding of the aspects and features of the present inventive concept may not be described.

Unless otherwise noted, like reference numerals denote like elements throughout the attached drawings and the written description, and thus, descriptions thereof will not be repeated. Further, parts not related to the description of the embodiments might not be shown to make the description clear.

In the detailed description, for the purposes of explanation, numerous specific details are set forth to provide a thorough understanding of various embodiments. In other instances, well-known structures and devices are shown in block diagram form in order to avoid unnecessarily obscuring various embodiments.

It will be understood that, although the terms "first," "second," "third," etc., may be used herein to describe various elements, components, regions, layers and/or sections, these elements, components, regions, layers and/or sections should not be limited by these terms.

It will be further understood that the terms "comprises," "comprising," "have," "having," "includes," and "including," when used in this specification, specify the presence of the stated features, integers, steps, operations, elements, and/or components, but do not preclude the presence or addition of one or more other features, integers, steps, operations, elements, components, and/or groups thereof.

The electronic or electric devices and/or any other relevant devices or components according to embodiments of the present disclosure described herein may be implemented utilizing any suitable hardware, firmware (e.g. an application-specific integrated circuit), software, or a combination of software, firmware, and hardware. For example, the various components of these devices may be formed on one integrated circuit (IC) chip or on separate IC chips. Further, the various components of these devices may be implemented on a flexible printed circuit film, a tape carrier package (TCP), a printed circuit board (PCB), or formed on one substrate.

Further, the various components of these devices may be a process or thread, running on one or more processors, in one or more computing devices, executing computer program instructions and interacting with other system components for performing the various functionalities described herein. The computer program instructions are stored in a memory which may be implemented in a computing device using a standard memory device, such as, for example, a random access memory (RAM). The computer program instructions may also be stored in other non-transitory computer readable media such as, for example, a CD-ROM, flash drive, or the like. Also, a person of skill in the art should recognize that the functionality of various computing devices may be combined or integrated into a single computing device, or the functionality of a particular computing device may be distributed across one or more other computing devices without departing from the scope of the embodiments of the present disclosure.

Unless otherwise defined, all terms (including technical and scientific terms) used herein have the same meaning as commonly understood by one of ordinary skill in the art to which the present inventive concept belongs.

In a storage server, when a host establishes a connection, one or more cores of a CPU of the server may be used for processing all requests from the connection, and may forward all of the requests to a single storage drive. Accordingly, the connection established with the host may be able to access all of the drives of a storage system. That is, once a connection is established with a qualified name, then any of the multiple drives can be accessed.

Because a large number of drives are being served, the drives may handle millions of input/output (IO) requests. However, the CPU itself (e.g., the core) may be able to process only a limited number of requests per second (e.g., one million requests per second). Upon reaching this limit, the CPU/core may be a bottleneck of the system if a client tries to access the same connection because the CPU is unable to process more requests at the same time. That is, even if the system supports a <NUM> Gigabyte network, such throughput may be unable to be achieved due to the bottleneck at the CPU, and some of the available cores may be underutilized.

Accordingly, some of the embodiments disclosed herein provide an architecture for providing improved packet processing using a message-passing architecture.

<FIG> shows a block diagram depicting a connection between a host and a storage device using a multi-CPU storage appliance for storing data on multiple storage devices, according to some embodiments of the present disclosure.

Referring to <FIG>, as mentioned above, as storage device performance increases, bottlenecks in a storage system (e.g., a non-volatile memory express (NVME) over fabrics storage server) may more frequently occur at the central processing unit(s) (CPU(s)) responsible for coordinating IO requests and data transfer associated with multiple IO threads from one or more host applications. As described below, embodiments of the present disclosure improve data storage technology by reducing inefficiencies associated with the coordination handled by the CPU(s), thereby improving overall network performance.

An example memory application or software, which is used as a target of the some embodiments of the present disclosure, may be designed for storage devices, such as block drives <NUM>. As shown, the storage system or network includes a CPU <NUM> having multiple cores <NUM>, a network interface controller (NIC) <NUM>, a memory <NUM>, and multiple drives <NUM>. The drives <NUM> may include SSDs, Ethernet SSDs, KVSSDs, etc., although it should be noted that Ethernet SSDs and KVSSDs are subsets of the different types of devices that can be used in the storage system. The memory <NUM> may include multiple volatile dynamic random-access memory (DRAM) modules <NUM> (e.g., see <FIG>).

When a host establishes a connection via the NIC <NUM>, one of the cores 102a of the CPU <NUM> may be used for processing all of the requests <NUM> (e.g., IO requests) from the connection to the host, and may forward all of the requests <NUM> to a single drive 108a. Accordingly, the host is able to access all of the drives <NUM> via the connection. That is, once a connection by the host is established with a qualified name, then any of the multiple drives <NUM> can be accessed.

Because a relatively large number of storage devices or drives <NUM> may be served, the drives <NUM> may handle a quantity (e.g., millions) of IO requests <NUM> from various hosts. However, the CPU <NUM> itself (e.g., one or more cores <NUM> of the CPU <NUM>) may be able to only process a limited number of requests <NUM> per second (e.g., up to one million requests per second). Upon reaching this limit, the CPU <NUM>, or the core <NUM>, may effectively become a bottleneck in the system, because the CPU <NUM> may be unable to concurrently process further additional requests <NUM> if a client attempts to access the same connection via the NIC <NUM>. Accordingly, even if the storage system supports, for example, a <NUM>-Gigabyte network, such throughput might not be able to be fully achieved due to the bottleneck at the CPU <NUM>. Accordingly, even though many cores <NUM> are available, some of the cores <NUM> may be underutilized.

Accordingly, as described below, embodiments of the present disclosure provide improved methods and systems for improving latency, and for reducing bottlenecking at the CPU <NUM>, by using modules having allocated CPU resources to enable load balancing among CPU cores and storage devices (e.g., to ensure that none of the CPU cores or storage devices have significantly higher workloads than others of the CPU cores or storage devices while some of the CPU cores or storage devices are underutilized).

<FIG> shows a block diagram of a server including two CPUs in a multi-CPU storage appliance for storing data on multiple storage devices, according to some embodiments of the present disclosure, and <FIG> depicts an architecture for a multi-CPU storage appliance for storing data on multiple storage devices, according to some embodiments of the present disclosure.

Referring to <FIG>, in the present example, a server <NUM> may provide hardware architecture representing two different non-uniform memory access (NUMA) nodes <NUM>, each NUMA node <NUM> corresponding to a NIC <NUM>, a memory <NUM> comprising multiple memory modules <NUM> (e.g., DRAM modules), a CPU <NUM>, and multiple storage devices <NUM>.

NUMA may refer to a method that may be used in a symmetric multiprocessing (SMP) system, where the memory access time is influenced by the location of the memory relative to the processor. NUMA may allow for configuring a cluster of microprocessors in a multiprocessing system so that the microprocessors can share memory locally. Under NUMA, a processor may access its own local memory faster than it is able to access non-local memory (e.g., faster than it is able to access memory that is local to another processor, or memory that is shared between processors). Benefits of NUMA may be most notable for workloads on servers where the data is often associated strongly with certain tasks or users. Accordingly, NUMA may improve performance and enable expansion of the system to be expanded.

Accordingly, the server <NUM> may have multiple CPU cores at each of the respective CPUs <NUM> and multiple storage devices <NUM> (e.g., drives). As described further with respect to <FIG>, some embodiments of the present disclosure provide one or more protocols or mechanisms that may be implemented to ensure that the workload experienced by the cores and the storage is balanced. That is, as described below, the disclosed embodiments prevent one or more of the cores and/or one or more of the storage devices <NUM> from being overloaded as a result of multiple IO requests while others of the cores and/or storage devices <NUM> are unused or underutilized. By avoiding the overloading of some cores and/or storage devices <NUM> and the underutilization of other cores and/or storage devices <NUM>, bottlenecks at the corresponding CPU <NUM> that may otherwise restrict data flow may be reduced or eliminated.

Referring to <FIG>, embodiments of the present disclosure may provide a software architecture for providing high-performance packet processing using a message-passing architecture using the hardware architecture shown in <FIG> and <FIG>. A system of embodiments of the present disclosure may include multiple CPU sockets. The CPU sockets have memory <NUM> that is bound to each NUMA socket to ensure that each IO request <NUM> is processed despite bottleneck in the CPU <NUM>. In the present example, a server for implementing embodiments of the present disclosure may have two NUMA sockets. However, it should be noted that the design disclosed herein could be extended to a server including additional NUMA sockets.

Software for implementing embodiments of the present disclosure may be designed to process packets from the network including a storage system. Also, the software may be distributed across multiple NVME drives <NUM>. For example, for KV-based drives, a drive <NUM> may be selected by a drive-processing module <NUM> (described further below) of the CPU <NUM> using any suitable hashing algorithm such as a Rendezvous hashing algorithm (e.g., a highest random weight (HRW) algorithm). The Rendezvous hashing algorithm may use two parameters - a key (e.g., an object key corresponding to a value to be stored on a drive <NUM>), and a drive ID (e.g., a unique identifier or an input/output (IO) identification corresponding to the drive <NUM> on which the data is to be stored). The suitable hashing algorithm may be used when an associated request processing has associated metadata that is sought to be preserved across multiple IO request to a same key.

For example, there may be a list of storage devices or drives <NUM>, with one of the drives <NUM> to be ultimately selected for use for access or storage of a given key. Accordingly, for each drive <NUM>, there may be an identification/string that represents the drive id, each drive <NUM> having a unique drive id. To select a drive <NUM> for storing the key, a Rendezvous hash may be calculated for each of the drives <NUM> on the list based on the input of the key and each respective drive ID. Similarly, for any incoming request <NUM>, the request <NUM> may be identified by the hash calculated by the Rendezvous hashing algorithm, which will be unique for every key-drive combination. Accordingly, the system can avoid implementing a separate table stored on a drive for mapping which drives store which keys, thereby further improving system efficiency.

Accordingly, if data is written, and a readback command for the data is issued, the Rendezvous hashing algorithm ensures that the same data is accessed by ensuring access the same drive <NUM>. In more detail, this may be accomplished by computing a number (e.g., a number between <NUM> and <NUM>) for each drive <NUM> corresponding to a hash of a combination of the key of the IO request <NUM> and the respective drive ID of each drive <NUM>, by then comparing all of the numbers, and by then selecting the highest computed number as corresponding to the drive <NUM> that is to be selected for that particular IO request <NUM>. Although a Rendezvous hashing algorithm is discussed herein, it should be noted that other hashing algorithms or functions may be used in accordance with other embodiments of the present disclosure.

The software running on the CPU <NUM> may be sub-divided into different components. Each component may be tasked with the responsibility of perform a given action or set of actions on all of the packets processed in the storage system. Two main components may be <NUM>) the module <NUM> that processes the IO requests <NUM> arriving at the NIC <NUM> (e.g., for performing network processing), and <NUM>) the module <NUM> that processes the NVME queues of the drives <NUM> (e.g., for performing drive processing or drive IO processing). Herein, these modules may be respectively referred to as a network-processing module <NUM> and a drive-processing module <NUM>. The network-processing module <NUM> may use an RDMA (remote direct memory access) protocol for processing requests, while the drive-processing module <NUM> may use TCP (transmission control protocol) for data processing.

Accordingly, by using the individual modules, embodiments of the present disclosure may improve IO performance by reducing a likelihood of a bottleneck restricting data flow due to the limitations of the CPU <NUM>. Each of the modules <NUM>, <NUM> may use an amount (e.g., a specified amount) of bounded CPU resources. That is, each of the modules <NUM>, <NUM> may allocate a respective set of cores <NUM> to be used by the modules <NUM>, <NUM> to achieve improved or optimized performance.

For instance, in the present example, each NIC <NUM> may use about four cores running at <NUM> to provide the full throughput of a <NUM> network while transferring 2MB objects using NVME over TCP. It should be noted, however, that the number used in the present example may change according to the specifications of the system.

For inbound network connections, a given number of cores <NUM> of the CPU <NUM> may be preallocated for an NIC <NUM> with affinity to a NUMA node <NUM> at which the NIC <NUM> is attached. The NUMA nodes <NUM> may be referred to as a respective CPU-memory couples. For example, the CPU socket and the selected memory <NUM>/bank of drives <NUM> build a respective NUMA node <NUM>. Conventionally, whenever a CPU sought to access the memory of a different NUMA node, the CPU could not directly access the memory of the other NUMA node, and was instead required to access the memory of the other NUMA node by going through a separate CPU that owned the memory.

In embodiments of the present disclosure, however, all connections may be processed by using a set of preallocated cores <NUM> of the CPU <NUM>. The network-processing module <NUM> for processing the requests <NUM> arriving at the NIC <NUM> may respectively allocate all of the connections by using a round-robin allocation scheme combined with usage monitoring (e.g., for monitoring the number of connections on each core <NUM>) to enable stateless processing. That is, each of the connections corresponding to the requests <NUM> may be assigned to the cores <NUM> on a round-robin basis to balance the loading of the cores <NUM> to thereby improve bandwidth from the drives <NUM>. If the drive-processing module <NUM> determines that a given core <NUM> is underperforming or is otherwise unable to handle any additional connections, the drive-processing module <NUM> can ensure no additional connections are made with the given core <NUM>. Accordingly, assuming each client is trying to share the bandwidth of the network equally, the storage system can ensure that the load is balanced across all available cores <NUM>.

Similarly, for processing the requests <NUM> to the NVME queues, a respective drive-processing module <NUM> may run on a dedicated set of cores <NUM> on both of the NUMA sockets of the server (e.g., in the storage server, the drives <NUM> will be on both of the NUMA sockets). Because each request <NUM> may be distributed to any drive <NUM>, some otherwise undesirable cross-NUMA memory copying by the drives <NUM> may occur (e.g., between the memory <NUM> of the first NUMA node 314a and the drives <NUM> of the second NUMA node 314b, and vice versa). However, because the Rendezvous hashing algorithm is used for selecting the drive <NUM>, the distribution may be balanced for normal usage scenarios.

Accordingly, a system with cross-NUMA transfer bandwidth (corresponding to cross-NUMA transfer requests <NUM>) that is more than half of the NIC bandwidth (corresponding to the IO requests <NUM>) may be able to run without degradation in system performance. An IO module (e.g., the drive-processing module <NUM>) may handle requests to the drives <NUM> from all of the connections. Also, the drive-processing module <NUM> may use a bounded number of CPU cores <NUM>, which may be based on the number of available drives <NUM> on the server.

Further, the requests <NUM> that are forwarded to IO threads <NUM> (e.g., the threads <NUM> shown in <FIG>) may be selected by any suitable algorithm such as a round-robin algorithm. For example, the requests <NUM> may be forwarded between the network-processing module <NUM> and the drive-processing module <NUM> using an atomic ring buffer <NUM> (see <FIG>). That is, once one of the modules <NUM>, <NUM> has completed processing of data of a request <NUM> in accordance with its role, the request <NUM> may be forwarded to the other one of the modules <NUM>, <NUM>. The atomic ring buffer <NUM> enables a lockless design, thereby achieving relatively high throughput. The atomic ring buffer <NUM> uses a cooperative thread-scheduling without blocking calls in any function.

Embodiments of the present disclosure may also accommodate the addition of other modules for performing other tasks. Further, a different algorithm may be used to distribute the requests <NUM> across the threads <NUM> of a given module. Also, an entirety of a design of embodiments of the present disclosure may be based on a logical framework that allows different modules to run on the same core <NUM>, which may be useful if the core <NUM> has a sufficient number of CPU cycles allotted for processing a number of requests <NUM> that exceeds that which is required by the module.

As described above, because each IO thread <NUM> can communicate directly to all of the queues of the drives <NUM>, and because there may be another set of queues for each core <NUM>, embodiments of the present disclosure are able to omit locking synchronization that would otherwise may be needed, as each IO thread <NUM> is able to access its own individual queues to the drives <NUM>.

Further, when any new IO request <NUM> occurs, the system is able to select between the numerous IO cores <NUM>. In a NUMA-based system, according to some embodiments, selection of one of the IO cores <NUM> having a smallest physical distance to the corresponding memory/DRAM <NUM> may be a criteria to help ensure sufficient performance.

Accordingly, by going in a round robin fashion, or by balancing across all of the cores <NUM>, improved bandwidth of the drives <NUM> may be achieved.

Additionally, because each CPU <NUM> is provided with a channel to access each associated memory <NUM> and bank of drives <NUM>, the system is able to avoid the limited bandwidth otherwise associated with the bottlenecking that comes with cross-NUMA channel access <NUM>.

<FIG> shows a flowchart depicting a method of packet processing.

Referring to <FIG>, at S401, a core of a processor may receive an input/output (IO) request from a host (e.g., one of the cores 102a of the CPU <NUM> of <FIG> may receive the IO request <NUM> of <FIG>). The core of the processor may receive the IO request at a network-processing module associated with a processor for establishing the connection (e.g., the network-processing module <NUM> of <FIG>). At S402, a drive-processing module (e.g., the drive-processing module <NUM> of <FIG>) may select a drive corresponding to the IO request (e.g., one of the drives <NUM> or <NUM> of <FIG> and <FIG>) using a hashing algorithm or a round-robin technique. In some embodiments, the hashing algorithm that is used may be a Rendezvous hashing algorithm based on a key corresponding to the IO request and a drive ID of the drive. At S403, a connection may be established between the host and the selected drive.

At S404, the network-processing module may forward the IO request associated with the processor to the drive-processing module associated with the processor that is configured to select the drive corresponding to the IO request using the hashing algorithm. At S405, the network-processing module may use a remote direct memory access protocol for processing the IO request. At S406, the drive-processing module may use a transmission control protocol for processing data corresponding to the IO request.

At S407, an atomic ring buffer (e.g., the atomic ring buffer <NUM> of <FIG>) may forward the IO request between the network-processing module and a drive-processing module.

At S408, the network-processing module may assign other IO requests to respective cores of the processor using the round-robin technique to balance connections between one or more hosts and one or more drives and to balance a loading of the cores of the CPU.

Thus, embodiments of the present disclosure are able to reduce or eliminate CPU bottlenecking by using modules having allocated CPU resources to balance loads among CPU cores and storage devices, thereby improving data storage technology.

While the present disclosure has been particularly shown and described with reference to some example embodiments thereof, it will be understood by those of ordinary skill in the art that various changes in form and details may be made therein without departing from the scope of the present disclosure as set forth in the following claims and their equivalents.

Claim 1:
A method of packet processing, in a system for packet processing comprising a processor comprising a plurality of cores (<NUM>), wherein a first subset of the plurality of cores (<NUM>) operates as a network-processing module (<NUM>) and a second subset of the plurality of cores operates as a drive-processing module (<NUM>), the method comprising:
receiving (S401), by the network-processing module (<NUM>), an input/output, IO, request from a host;
selecting (S402), by the drive-processing module (<NUM>), a drive corresponding to the IO request using a hashing algorithm or a round-robin technique;
establishing (S403) a connection between the host and the drive; and
assigning (S408), by the network-processing module (<NUM>), one or more other IO requests to one or more respective cores (<NUM>) of the processor using the round-robin technique to balance one or more connections between one or more hosts and one or more drives, and to balance a loading of the cores (<NUM>) of the processor.