Patent Description:
There are two broad types of replication - incremental replication and full replication. In incremental replication, each time data on the primary computing system changes, a change instruction is provided to the replica computing system(s) (though sometimes change instructions can accumulate for a short time for efficiency in transfer). Thus, the data on the replica computing system(s) is updated incrementally in relative real-time, as changes are made on the primary computing system.

On the other hand, a full replication operation involves all of the data on the primary computing system being replicated in a large data transfer to a replica computing system. Full replication is appropriate where the replica computing system is being started up, or if the replica computing system has lost all of its data. As an example, if the data on the replica computing system is represented in volatile memory, a full replication would be performed upon powering up a replica computing system after a power cycle. Full replications of large amounts of data can take significant time, computing resources, and network bandwidth.

Rather, this background is only provided to illustrate one exemplary technology area where some embodiments describe herein may be practiced.

<CIT> describes performing data management operations on replicated data in a computer network. Log entries are generated for data management operations of an application executing on a source system. Consistency point entries are used to indicate a time of a known good, or recoverable, state of the application. A destination system is configured to process a copy of the log and consistency point entries to replicate data in a replication volume, the replicated data being a copy of the application data on the source system. When the replicated data represents a known good state of the application, as determined by the consistency point entries, the destination system(s) may perform a storage operation (e.g., snapshot, backup) to copy the replicated data and to logically associate the copied data with a time information (e.g., time stamp) indicative of the source system time when the application was in the known good state.

The principles described herein relate to the replication of data from a primary computing system to a secondary computing system. The replication is single-threaded or multi-threaded depending on one or more characteristics of the data to be replicated. As an example, the characteristics could include the type of data being replicated and/or the variability of that data. Also, the multi-threading capabilities of the primary and secondary computing systems are determined. Then, the primary computing system decides whether to perform multi-threaded replication and what the multi-threading parameters of the replication will be based on the one or more characteristics of that data, as well as based on the multi-threading capabilities of the primary and secondary computing systems.

As an example, for data that can be partitioned into approximately equal size, multi-threaded replication is accomplished by assigning partitions to each of the threads. Corresponding threads at the secondary computing system gather the partitions. Furthermore, the secondary computing system assembles those partitions into the replicated data. Thus, multi-threaded replication allows for more rapid and efficient replication from the primary computing system to the secondary computing system. This is especially advantageous for full replication, in which large amounts of data are replicated. Full replication can thus be done in less time by tailoring the replication according to the nature of the data being replicated.

<FIG> illustrates an environment <NUM> in which the principles described herein may be employed. The environment <NUM> includes a primary computing system <NUM> (also called herein a "primary system" or simply "the primary") and a secondary computing system <NUM> (also called herein a "secondary system" or simply "the secondary"). The primary system <NUM> replicates its data to the secondary system <NUM>.

As an example, the primary system <NUM> and the secondary system <NUM> may each be physical systems. Alternatively, one or both of the primary system <NUM> and the secondary system <NUM> may be a virtual machine thereby emulating a computing system. As an example, the primary system <NUM> and the secondary system <NUM> may be structured as described below for the computing system <NUM> or <FIG>, or may be structured to emulate such a computing system. In addition, the replica component <NUM> on the primary system <NUM> and the replica component <NUM> on the secondary system <NUM> may each be structured as described below for the executable component <NUM> of <FIG>.

As represented by bi-directional arrow A, the primary system <NUM> interfaces with a user computing system <NUM> (also called herein a "user system"). The primary system <NUM> receives requests from the user system <NUM>, as appropriate provides responses or notifications to the user system <NUM>, and as appropriate alters data <NUM> held by the primary system <NUM>. Thus, the data <NUM> may be altered as the primary system <NUM> interfaces with the user system <NUM>. Because of this interaction with the user system <NUM>, the primary system <NUM> is modified by the term "primary".

On the other hand, the secondary system <NUM> is modified by the term "secondary" as it is not presently interfacing directly with the user system <NUM>. Instead, the secondary system <NUM> stands at the ready to become primary in case the current primary system <NUM> is no longer available or has reduced functionality. To ensure that the secondary system <NUM> can do this, at least some of the data <NUM> on the primary system <NUM> is replicated (as represented by arrow B) to the data <NUM> on the secondary system <NUM>. A replica component <NUM> on the primary system <NUM> and a replica component <NUM> on the secondary system collaborate to accomplish this replication.

The replication could be full replication of the data <NUM> from the primary system to the secondary system <NUM>. That is, in full replication, a replication scope is defined, and all data within that replication scope on the primary system is replicated to the secondary system. As an example, the replication scope could be an address range, or perhaps could be a large piece of structured data, such as perhaps a database.

<FIG> illustrates an example of a data range <NUM> that is to be replicated. In this example, the data range is represented linearly from A to B. The data range <NUM> will be referred to as an example of data to be replicated hereinafter. In the case of the secondary system <NUM> receiving a full replication of the primary system <NUM>, the secondary system <NUM> may also be referred to as a "replica" system. Although the environment <NUM> illustrates replication occurring from a primary system to a single secondary system, the principles described herein may also be performed to replicate data from a primary computing system to multiple secondary computing systems.

In accordance with the principles described herein, the replication occurs in a manner that depends on one or more characteristics of the data to be replicated as part of the replication. <FIG> illustrates a flowchart of a method <NUM> for performing replication in a manner that depends on one or more characteristics of the data to be replicated, in accordance with the principles described herein. The method <NUM> may be performed to replicate the data <NUM> of <FIG> from the primary system <NUM> to the secondary system <NUM> in <FIG>. Accordingly, the method <NUM> will now be described with respect to the environment <NUM> of <FIG> and the data <NUM> of <FIG>.

The method <NUM> includes acts that are performed by the primary system (e.g., the primary system <NUM>), which are illustrated in the left column of <FIG> under the heading "Primary" and have reference numbers in the <NUM>'s. Furthermore, the method <NUM> includes acts that are performed by the secondary system (e.g., the secondary system <NUM> of <FIG>) which are illustrated in the right column of <FIG> under the heading "Secondary" and have reference numbers in the <NUM>'s. As an example, the acts performed by the primary system may be performed by the replica component <NUM> of the primary system <NUM>, and the acts performed by the secondary system may be performed by the replication component <NUM> of the secondary system <NUM>.

The method <NUM> includes determining that a replication of data is to be performed (act <NUM>). As an example, the data range <NUM> of <FIG> may be the data to be replicated. The primary system <NUM> may initiate the replication in response to a request from the secondary system <NUM>. Alternatively, the primary system <NUM> may initiate the replication without such a request. In any case, the replication component <NUM> on the primary system <NUM> determines that a replication of data is to be performed.

In accordance with the principles described herein, the replication is performed using different multi-threading parameters, depending on characteristics of the data to be replicated. Accordingly, the primary system determines one or more characteristics of the data to be replicated (act <NUM>). As an example, the data <NUM> of <FIG> is the data to be replicated. The right cross-hatching represents that this data <NUM> has a particular set of characteristics.

In <FIG>, the act <NUM> is shown in parallel with the act <NUM>, which emphasizes that there is no time dependency between when the time that the one or more characteristics of the data are identified (act <NUM>), and the time that the replication is determined to occur (act <NUM>). For example, the data of that data region may be monitored continuously, periodically, or intermittently, such that the one or more characteristics of the data of the data range are known even prior to determining that the replication is to occur. On the other hand, the identification of the one or more characteristics may be deferred until after it is determined to replicate the data.

As an example, the characteristics could be variability in size of the partitionable portions of the data to be replicated. The characteristics could also be the data type of the data to be replicated in cases in which the data type is correlated with variability in the size of partitionable portions of the data of that data type. As an example, some key-value pairs may have predictable sizes in the value field (e.g., one kilobyte). On the other hand, some values may represent hash values and be of very large and unpredictable size.

Also, the primary system determines the multi-threading capabilities of the primary system and the secondary system (act <NUM>). As an example, the multi-threading capabilities may be connected to the number of cores that the respective systems have. For instance, suppose that the primary system <NUM> has eight cores, but the secondary system <NUM> has four cores. From this, the primary system can conclude that the highest number of threads possible for the replication is four, governed by the computing system with the least number of cores. The primary and secondary systems agree upon the replication buffer size and number of buffers. For example, the secondary system might have lower memory available, say <NUM> MB, than the primary, say <NUM> MB. Both the systems will agree on using the lower value of <NUM> MB to replicate the data. During replication, the primary system also sends additional information for the size of data filled in a partition where the partition could be partially filled or empty. Note that the act <NUM> is shown in parallel with the acts <NUM> and <NUM> to emphasize that there is no time dependency between when the capabilities of the primary and secondary systems are determined (act <NUM>), the time that the replication is determined to occur (act <NUM>) and the time that the one or more characteristics of the data to be replicated are identified (act <NUM>).

Based on the identified characteristic(s) of the data region and the multi-threading capabilities of both the primary system and the secondary system, the primary system determines whether to perform replication by multi-threading the replication (decision block <NUM>). If the primary system determines to perform single-threaded replication ("Single" in decision block <NUM>), then the primary system replicates the data using a single thread (act <NUM>). Likewise, the secondary system is to receive the data using a single thread.

On the other hand, if the primary system determines to perform multi-threaded replication ("Multi" in decision block <NUM>), the primary system identifies one or more multi-threading parameters that will be used to perform replication of the data region (act <NUM>), communicates the multi-threading parameters to the secondary system (act <NUM>), and also performs replication using the multi-threading parameters (act <NUM>). From the perspective of the secondary system, the secondary system receives the multi-threading parameters (act <NUM>), and also receives the replication using the same multi-threading parameters (act <NUM>).

A simple example of multi-threading parameters is a number of threads to use when replicating. In one example, less variability in the size of the partition portions of data to be replicated tends the decision towards more aggressively using multi-threading (of course constrained at the upper end by the capabilities of the primary and secondary systems). Similarly, in this simple example, less variability in the size of partitions of data to be replicated tends the decision towards less aggressive multi-threading, or even single threaded replication.

<FIG> illustrates a primary system <NUM> replicating data to a single secondary system <NUM>. However, the principles described herein may also be used to replicate from the primary system <NUM> to multiple secondary systems. Since the decision on whether to perform single-threaded or multi-threaded replication depends at least in part upon the capabilities of the secondary system that is the target of replication, it is possible that replication of the same region of data occurs by single threaded replication to one secondary system, and multi-threaded replication to another secondary system. Likewise, it is also possible that the same region of data is replicated by multi-threading, albeit with different multi-threading parameters, depending on the capabilities of the respective secondary systems. For instance, there may be one secondary with four cores, and thus multi-threaded replication may occur using four cores for that secondary system. However, multi-threaded replication might occur to another secondary using only two cores where that secondary only has two cores.

<FIG> illustrates an example showing single threaded replication of data <NUM> from the volatile memory <NUM> of a primary system (such as the primary system <NUM>) of <FIG> to the volatile memory <NUM> of a secondary system (such as the secondary system <NUM>) of <FIG>. The volatile memory <NUM> may also be referred to as the "primary memory", and the volatile memory <NUM> may also be referred to as the "secondary memory". In the example of <FIG>, the string "test1, test2, test3" is to be replicated from the primary memory <NUM> to the secondary memory <NUM> in a buffer capable of holding <NUM> characters. In accordance with the method <NUM>, the primary system decided to perform single threaded replication ("Single" in decision block <NUM>), and negotiated this with the secondary system.

In <FIG>, the buffer 405A is populated with the first eight characters "test1, t", transferred over a network, and the data from the buffer provided as data 412A to the secondary memory <NUM>. In <FIG>, the buffer 405B is populated with the next eight characters "est2, tes", is transferred over the network, and the characters are appended to the data 412A within the secondary memory <NUM> to formulate data 412B. In <FIG>, the buffer 405C is populated with the final characters "t3", is transferred over the network, and the characters are appended within to the previous data the secondary memory <NUM> to formulate the final data <NUM>. The final data <NUM> within the secondary memory <NUM> is a replicated form of the original data <NUM> that is in the primary memory <NUM>. This replication occurred using three network transfers over one connection.

<FIG> illustrate an example showing multi-threaded replication of data <NUM> from the primary memory <NUM> to the secondary memory <NUM>. In the example of <FIG>, the data <NUM> in the form of values "test", "rest", "best" and "west" are to be replicated from the primary memory <NUM> to the secondary memory <NUM> in buffers that are partitioned in four parts. In this example, the primary system decides to perform multi-threaded replication ("Multi" in decision block <NUM>), determines multi-threading parameter(s) in the form of the number of threads being four (act <NUM>), and negotiates this number of threads with the secondary system (act <NUM>). Thus, the primary system will use four threads to perform the replication, and likewise the secondary system will use four threads to perform the replication.

In <FIG>, the buffer 505A is partitioned into four parts 505A1, 505A2, 505A3 and 505A4, being equal in number to the number of threads to be used by each system for the replication. Accordingly, the primary system spins up four threads, and the secondary system spins up four threads. On the primary system, a thread is assigned to each of the parts of the buffer 505A. A first thread populates the first part 505A1 of the buffer 505A with the first two characters of the value "test" (in other words "te"). A second thread populates the second part 505A2 of the buffer 505A with the first two characters of the value "rest" (in other words "re"). A third thread populates the third part 505A3 of the buffer 505A with the first two characters of the value "best" (in other words "be"). A fourth thread populates the fourth part 505A4 of the buffer 505A with the first two characters of the value "west" (in other words "we").

The four threads transfer the appropriate parts of the buffer 505A. A first thread on the secondary system takes the content of part 505A1 ("te") and populates that into secondary memory <NUM>. A second thread on the secondary system takes the content of part 505A2 ("re") and populates that into secondary memory <NUM>. A third thread on the secondary system takes the content of part 505A3 ("be") and populates that into secondary memory <NUM>. A fourth thread on the secondary system takes the content of part 505A4 ("we") and populates that into the secondary memory <NUM>. The result of this first transfer over the network is the data 512A. Alternatively, as shown in <FIG>, each thread on the secondary system holds its respective data for assembling with other data prior to populating of that data into the secondary system.

In the next transfer, the first thread populates the first part 505B1 of the buffer 505B with the last two characters of the value "test" (in other words "st"). The second thread populates the second part 505B2 of the buffer 505B with the last two characters of the value "rest" (in other words "st"). The third thread populates the third part 505B3 of the buffer 505B with the last two characters of the value "best" (in other words "st"). The fourth thread populates the fourth part 505B4 of the buffer 505B with the last two characters of the value "west" (in other words "st"). The four threads of the primary system transfer the appropriate parts of the buffer 505B to the secondary system. The four threads of the secondary system each take the content of respective parts 505B1, 505B2, 505B3 and 505B4 and append that content onto the respective values to form data <NUM>. The result of this second transfer over the network is the data <NUM>, which is a replicated form of the original data <NUM>.

<FIG> illustrates an example showing multi-threaded replication of data <NUM> from the primary memory <NUM> to the secondary memory <NUM>. Again, the data <NUM> in the form of values "test", "rest", "best" and "west" are to be replicated from the primary memory <NUM> to the secondary memory <NUM> using multiple buffers 505A and 505B. But now, the four threads populate the buffers concurrently. The buffers 505A and 505B are each partitioned in four parts. For example, like the <FIG>, buffer 505A is partitioned into parts 505A1 through 505A4, and buffer 505B is partitioned into parts 505B1 through 505B4.

In this example, the primary system decides to perform multi-threaded replication ("Multi" in decision block <NUM>), and determines multi-threading parameter(s) in the form of the number of threads being four and the number of buffers per transfer being two (act <NUM>), and negotiates this number of threads and buffers with the secondary system (act <NUM>). Thus, the primary system will use four threads and two buffers to perform the replication as a single transfer, and likewise the secondary system will use four threads and two buffers to receive the replication.

A first thread populates the first part 505A1 of the first buffer 505A with the first two characters of the value "test" (in other words "te"), and the first part 505B1 of the second buffer 505B with the last two characters of the value "test" (in other words "st"). A second thread populates the second part 505A2 of the first buffer 505A with the first two characters of the value "rest" (in other words "re") and the second part 505B2 of the second buffer 505B with the last two characters of the value "rest" (in other words "st"). A third thread populates the third part 505A3 of the first buffer 505A with the first two characters of the value "best" (in other words "be"), and the third part 505B3 of the second buffer 505B with the last two characters of the value "best" (in other words "st"). A fourth thread populates the fourth part 505A4 of the first buffer 505A with the first two characters of the value "west" (in other words "we"), and the fourth part 505B1 of the buffer 505B with the last two characters of the value "west" (in other words "st").

The four threads transfer the appropriate parts of the buffers 505A and 505B. A first thread on the secondary system takes the content of part 505A1 ("te") and part 505B1 ("st"), assembles the content into value "test", and populates that value into secondary memory <NUM>. A second thread on the secondary system takes the content of part 505A2 ("re") and part 505B2 ("st"), assembles the content into value "rest", and populates that value into secondary memory <NUM>. A third thread on the secondary system takes the content of part 505A3 ("be") and part 505B3 ("st"), assembles the content into value "best", and populates that value into secondary memory <NUM>. A fourth thread on the secondary system takes the content of part 505A4 ("we") and part 505B4 ("st"), assembles the content into value "west", and populates that value into the secondary memory <NUM>. The result of this is the data <NUM>, which is a replicated form of the original data <NUM>.

In one embodiment of <FIG>, each buffer part is sent as soon as it is filled. For instance, buffer part 505A2 might be sent as soon as it is filled, regardless of whether the other buffer parts are populated with respective data. In another embodiment of <FIG>, each buffer is sent as soon as it is filled with the data. For instance, buffer 505A may be sent even though buffer 505B is not yet ready to be sent.

In another modification of <FIG>, instead of using only one connection, there are multiple connections (e.g., SSL connections) used to transfer the various parts of the buffers. The number of connections may be negotiated in advance between the primary system and the secondary system. Here, the buffers 505A and 505B may be communicated concurrently over their distinct connections. Other examples of multi-threading parameters that may be communicated includes whether the data is an ordered set or an unordered set, and if ordered how the ordering is determined. Thus, the secondary computing system can reassemble ordered data in the proper order. The use of multiple connections helps to scale out the encryption and decryption load across multiple sockets.

Because the data <NUM> of <FIG> has the same set of characteristic(s), the decisions of whether to perform single-threaded or multi-threaded replication may be performed only once and applied across the entire data range of the data <NUM> for the given secondary system. Furthermore, if multi-threaded replication is to be performed, the same multi-threading parameters may be used to replicate the entire data range of the data <NUM> for that secondary system. However, <FIG> illustrates data <NUM> that has different characteristics. For example, the data <NUM> ranges from point A to B. A first part <NUM> of that data <NUM> ranging from point A to C has a first set of one or more characteristics as represented by the rightward facing cross hatching. However, a second part <NUM> of the data <NUM> ranging from point C to B has a second set of one or more characteristics as represented by the leftward facing cross hatching.

Thus, the method <NUM> may be performed once for the first part <NUM>, and once for the second part <NUM>. Accordingly, the primary system may decide to replicate the first part <NUM> using single-threaded replication, and the second part <NUM> using multi-threaded replication, or vice versa. Alternatively, the primary system may decide to replicate the first part <NUM> using a first set of multi-threading parameters, and the second part <NUM> using a second set of multi-threading parameters.

Accordingly, what has been described is the replication of data from a primary computing system to a secondary computing system in a manner that the threading of the replication depends on one or more characteristics of the data to be replicated. This allows for replication to be performed more efficiently and in manner that is more suitable given the characteristics of the data being replicated. Of course, depending on the characteristics of the data within parts <NUM> and <NUM>, the entirety of the data might be replicated in a single thread.

Because the principles described herein are performed in the context of a computing system, some introductory discussion of a computing system will be described with respect to <FIG>.

As illustrated in <FIG>, in its most basic configuration, a computing system <NUM> includes at least one hardware processing unit <NUM> and memory <NUM>. The processing unit <NUM> includes a general-purpose processor. Although not required, the processing unit <NUM> may also include a field programmable gate array (FPGA), an application specific integrated circuit (ASIC), or any other specialized circuit. In one embodiment, the memory <NUM> includes a physical system memory. That physical system memory may be volatile, non-volatile, or some combination of the two. In a second embodiment, the memory is non-volatile mass storage such as physical storage media. If the computing system is distributed, the processing, memory and/or storage capability may be distributed as well.

The computing system <NUM> also has thereon multiple structures often referred to as an "executable component". For instance, the memory <NUM> of the computing system <NUM> is illustrated as including executable component <NUM>. The term "executable component" is the name for a structure that is well understood to one of ordinary skill in the art in the field of computing as being a structure that can be software, hardware, or a combination thereof. For instance, when implemented in software, one of ordinary skill in the art would understand that the structure of an executable component may include software objects, routines, methods (and so forth) that may be executed on the computing system. Such an executable component exists in the heap of a computing system, in computer-readable storage media, or a combination.

One of ordinary skill in the art will recognize that the structure of the executable component exists on a computer-readable medium such that, when interpreted by one or more processors of a computing system (e.g., by a processor thread), the computing system is caused to perform a function. Such structure may be computer readable directly by the processors (as is the case if the executable component were binary). Alternatively, the structure may be structured to be interpretable and/or compiled (whether in a single stage or in multiple stages) so as to generate such binary that is directly interpretable by the processors. Such an understanding of example structures of an executable component is well within the understanding of one of ordinary skill in the art of computing when using the term "executable component".

While not all computing systems require a user interface, in some embodiments, the computing system <NUM> includes a user interface system <NUM> for use in interfacing with a user. The user interface system <NUM> may include output mechanisms 812A as well as input mechanisms 812B. The principles described herein are not limited to the precise output mechanisms 812A or input mechanisms 812B as such will depend on the nature of the device. However, output mechanisms 812A might include, for instance, speakers, displays, tactile output, virtual or augmented reality, holograms and so forth. Examples of input mechanisms 812B might include, for instance, microphones, touchscreens, virtual or augmented reality, holograms, cameras, keyboards, mouse or other pointer input, sensors of any type, and so forth.

For the processes and methods disclosed herein, the operations performed in the processes and methods may be implemented in differing order. Furthermore, the outlined operations are only provided as examples, and some of the operations may be optional, combined into fewer steps and operations, supplemented with further operations, or expanded into additional operations without detracting from the essence of the disclosed embodiments.

Claim 1:
A method for performing a replication from a primary computing system (<NUM>; <NUM>) to a secondary computing system (<NUM>; <NUM>) in a manner that depends on one or more characteristics of at least some of the data (<NUM>) to be replicated as part of the replication, by performing the following:
identifying (<NUM>) one or more characteristics of a region (<NUM>; <NUM>) of data (<NUM>) to be replicated within at least part of the replication from the primary computing system (<NUM>; <NUM>) to the secondary computing system (<NUM>; <NUM>), and characterized in that the method further comprises performing the following:
determining (<NUM>) multi-threading capabilities of the primary computing system (<NUM>; <NUM>), and multi-threading capabilities of the secondary computing system (<NUM>; <NUM>);
based on the identified one or more characteristics of the data (<NUM>) of the region (<NUM>; <NUM>) of data (<NUM>) and the multi-threading capabilities of both the primary computing system (<NUM>; <NUM>) and the secondary computing system (<NUM>; <NUM>), deciding (<NUM>) to perform multi-threaded replication of the region (<NUM>; <NUM>) of data (<NUM>) using one or more multi-threading parameters (<NUM>), and communicating (<NUM>) the one or more multi-threading parameters to the secondary computing system (<NUM>; <NUM>); and
in response to the decision to perform multi-threaded replication of the region (<NUM>; <NUM>) of data (<NUM>) using the one or more multi-threading parameters, performing (<NUM>) multi-threaded replication of the region (<NUM>; <NUM>) of data (<NUM>) using the one or more multi-threading parameters.