Abstract:
Systems and methods of capturing application data locally on a source data management system and performing a high performance, bandwidth optimized replication the data from the source data management system to a remote data management system either (a) in parallel with copying data from an application to the source data management system, or (b) serially by replicating the data to the remote data management system after the data is copied to the source data management system, wherein the determination to operate in parallel or serially is based on at least one of a characteristic of the data and user specified tuning parameters.

Description:
CROSS REFERENCE TO RELATED APPLICATIONS 
       [0001]    This application claims priority to U.S. Provisional Application No. 62/344,662, filed on Jun. 2, 2016, entitled “STREAMING AND SEQUENTIAL DATA REPLICATION” which is incorporated herein by reference. 
     
    
     TECHNICAL FIELD 
       [0002]    The present disclosure relates generally to data replication and, more particularly, to techniques for improving the process of replicating data by deciding when replication should be performed serially or in parallel with making a local copy. 
       BACKGROUND 
       [0003]    Data management systems support replicating data to a remote system. The purpose of the replication is to have a copy of the data at a remote location, as a safeguard if something were to go wrong with the original location. The replication is performed by capturing a consistent image of the data to be protected at the source location, and transmitting this data to the remote location. Drawbacks of known data management systems is not accounting for changing transmission conditions and types of data received to both improve the speed of getting data transferred, and reducing the bandwidth requirements for replication. 
       SUMMARY OF THE INVENTION 
       [0004]    In accordance with the disclosed subject matter, systems, methods, and non-transitory computer-readable media are provided for improving the process of replicating data by deciding when replication should be performed serially or in parallel with making a local copy. In some embodiments, systems, methods, and non-transitory computer-readable media for capturing application data locally on a source data management system and performing a high performance, bandwidth optimized replication the data from the source data management system to a remote data management system either (a) in parallel with copying data from an application to the source data management system, or (b) serially by replicating the data to the remote data management system after the data is copied to the source data management system, wherein the determination to operate in parallel or serially is based on at least one of a characteristic of the data and user specified tuning parameters. In some embodiments, a source computing device requests application data from a host where an application is running. In some embodiments, the application data comprises at least one of a full copy of the application data, and changes associated with the application data, the changes including changed elements of the application data since a previous data capture of the application data from a host, and a list of changed elements associated with the application data. In some embodiments, the source computing device receives the application data and determines a parameter associated with an ordering of the changed elements. In some embodiments, the source computing device begins to copy the application data to a location within the source computing device. In some embodiments, if the parameter is indicative of a non-overlapping order, the source computing device begins to replicate the application data to a remote computing device in parallel with copying the application data within the source computing device. In some embodiments, the source computing device and the remote computing device are connected by a network. In some embodiments, if the parameter is indicative of an overlapping order, the source computing device begins to replicate serially the application data to the remote computing device after copying the application data within the source computing device is complete. 
         [0005]    In some embodiments, the full copy of the application data is requested when no previous captured version of the application data exists on the source computing device. In some embodiments, the full copy is automatically associated with a parameter indicating a non-overlapping order. In some embodiments, a source computing device replicates the full copy from the source computing device to the remote computing device in parallel with copying the full copy from the host to the source computing device. In some embodiments, the list of changed elements is a list of changed extents within a block device, wherein each extent describes a range of modified bytes or modified blocks. In some embodiments, replicating the application data from the source computing device to the remote computing device includes combining the extent list with each previous extent list associated with each previous data capture of the application data since a last successful replication operation. In some embodiments, the list of changed elements is a bitmap, the bitmap including bit values of zero and one to represent whether a block or a grain within the application data is associated with a changed element, the grain being associated with a number of blocks. In some embodiments, the remote computing device stores the application data alongside previously replicated application data on the remote computing device to create a complete representation of the changed application data and the previously replicated application data. In some embodiments, the source computing device replicates additional application data to the remote computing device when the remote computing device includes no previously replicated application data from the source computing device. In some embodiments, replicating the application data from the source computing device in parallel with copying the application data to the source computing device comprises: a) copying, by the source computing system, an element of changed application data from the host application to the source computing device; b) replicating, from the source computing device to the remote computing device, the element of changed application data; and repeating steps a) and b) until each element of the changed application data is copied to the source computing device and replicated to the remote computing device. 
         [0006]    In some embodiments, a speed associated with copying each of the elements to the source computing device is substantially similar to a speed associated with replicating to the remote computing device. In some embodiments, a speed associated with replicating each of the elements to the remote data management system is substantially slower than a speed associated with copying each of the elements to the source computing device. In some embodiments, the source computing device determines when all of the application data is copied to the source computing device, stops the replication of the application data to the remote staging disk in response to determining that all of the application data is copied to the source staging disk, determines an offset associated with a data element of the application data being replicated prior to the stop, and resumes replication of the application data occurring after the offset from the location associated with the copied application data being stored in the source computing device to the remote computing device. In some embodiments, the source computing device determines a bandwidth level of the network, the bandwidth level indicative of the network being at least one of a high bandwidth network and a low bandwidth network, enables an application of deduplication and compression to the application data prior to replicating the application data to the remote computing device when the bandwidth level is indicative of the network being a low bandwidth network, and disables the application of deduplication and compression to the application data prior to replicating the application data to the remote computing device when the bandwidth level is indicative of the network being a high bandwidth network. In some embodiments, a frequency associated with the copying of the application data to a location within the source computing device is higher than a frequency associated with replicating the application data from the source computing device to the remote computing device. 
         [0007]    These and other capabilities of the disclosed subject matter will be more fully understood after a review of the following figures, detailed description, and claims. It is to be understood that the phraseology and terminology employed herein are for the purpose of description and should not be regarded as limiting. 
     
    
     
       BRIEF DESCRIPTION OF FIGURES 
         [0008]    Various objectives, features, and advantages of the disclosed subject matter can be more fully appreciated with reference to the following detailed description of the disclosed subject matter when considered in connection with the following drawings, in which like reference numerals identify like elements. 
           [0009]      FIG. 1  is a system diagram showing the creation of a local and a remote copy of application data, according to some embodiments of the present disclosure. 
           [0010]      FIG. 2  is a system diagram showing an incremental local copy and replication operation, according to some embodiments of the present disclosure. 
           [0011]      FIG. 3  is a block diagram showing representations of difference information, according to some embodiments of the present disclosure. 
           [0012]      FIG. 4  is a system diagram showing an incremental local copy and replication run sequentially, according to some embodiments of the present disclosure. 
           [0013]      FIG. 5  is a system diagram showing an incremental local copy and replication run in parallel, according to some embodiments of the present disclosure. 
           [0014]      FIG. 6  is a system diagram showing an incremental local copy and replication run in parallel, according to some embodiments of the present disclosure. 
           [0015]      FIG. 7  is a system diagram showing in detail a first phase of streaming replication, according to some embodiments of the present disclosure. 
           [0016]      FIG. 8  is a system diagram showing in detail a second phase of streaming replication, according to some embodiments. 
           [0017]      FIG. 9  is a flowchart illustrating a control flow of streaming replication run in streaming mode, according to some embodiments of the present disclosure. 
           [0018]      FIG. 10  is a system diagram showing a full ingest of a chained streaming replication job, as described in some embodiments of the present disclosure. 
           [0019]      FIG. 11  is a system diagram showing a first incremental snapshot operation of a chained streaming replication job, as described in some embodiments of the present disclosure. 
           [0020]      FIG. 12  is a system diagram showing an incremental operation of a chained streaming replication job including snapshot and streaming operations, as described in some embodiments of the present disclosure. 
           [0021]      FIG. 13  is a system diagram showing an incremental operation of a chained streaming replication job including snapshot and deduplication operations, as described in some embodiments of the present disclosure. 
           [0022]      FIG. 14  is a flowchart illustrating a chained job including snapshot, deduplication and streaming replication policies, according to some embodiments of the present disclosure. 
           [0023]      FIG. 15  is a flowchart showing a process for choosing streaming versus sequential replication, according to some embodiments of the present disclosure. 
       
    
    
     DETAILED DESCRIPTION 
       [0024]    With regard to getting a copy of the data to the remote location using data management systems, including advanced copy data management systems, such as the Actifio Copy Data Management System and Actifio Sky, there are competing concerns or requirements: The speed with which the data is made available at the remote location, and the bandwidth of communications required and consumed by the transfer. Optimizing one is often at the expense of the other. If you want to increase the speed with which the image of the data becomes available on the remote system, you will usually have to invest in higher quality communications infrastructure, and this has a higher cost. 
         [0025]    Various techniques have been developed to improve the speed of getting the data transferred, and/or reducing the bandwidth requirements for replication. Data compression is one such approach, where patterns in the data are recognized, and specified in a shortened notation, so that the number of bytes required to transmit the data are reduced. Deduplication is another technique, where large blocks of identical or similar data are reduced in size by referencing previously transmitted data. 
         [0026]    Incremental replication can be used to improve both the speed and reduce the bandwidth requirements when the data to be replicated is related to data that was transmitted at a previous time. Incremental replication identifies parts of the data that are identical to a recent, previous replication, and sends only the data that is different, along with a description of which parts have changed. The receiving side can reconstruct a complete copy of the data by combining the unchanged parts of the previous data with changes received. When only a small fraction of the data has changed as might be the case with a database application from one day to the next, this technique can provide huge gains. 
         [0027]    Sometimes, in addition to replicating a copy of valuable data to a remote location, there is a requirement to retain a copy of the data locally, to protect against damage to the local production copy of the data, or for various other purposes. 
         [0028]    The requirement to retain an image on the local system can introduce additional challenges to the process of replication. If the local copy is created in full first, then the data is transmitted, it delays the copy on the remote location. If the local copy and replication are done simultaneously, one can slow down the other one, complicating data capture by extending the duration that the stable image of production data has to be retained. 
         [0029]    Systems and methods are described herein for improving the process of replicating data while maintaining a local copy, by deciding when replication should be performed serially after a local copy, or in parallel with the local copy, or even a combination, where replication begins in parallel with the local copy, but then switches to a using the local copy after that step completes. In some embodiments, making a local copy is also referred to herein as copying data to a location within the source computing device. 
         [0030]    The techniques described here can be independent of optimization techniques such as compression, deduplication and incremental replication. It applies to all of these and other optimizations applied to the local copy and replication operations. 
         [0031]    Systems and methods are also described herein for optimizing data replication for faster replication (also referred to herein as streaming replication) by leveraging a high network bandwidth. Replication can be run in streaming mode where replication is performed in parallel to a local backup. Alternately, replication can be run as a chained replication job that uses an existing snapshot as a source for replication and avoiding the redundant work of copying application data to staging disks and creating additional snapshots to use as a source for replication. 
         [0032]    In some embodiments, streaming replication can bypass deduplication so as to optimize the time required to transfer the data versus the minimizing the bandwidth used. The operation creates a point in time copy of the application, and copies it to the local staging disk, and to the remote system. 
         [0033]    In some embodiments, streaming replication can be operated in two modes—streaming and chaining. Streaming mode optimizes the time from the freezing of the application to getting the data to the remote system. Chaining allows for scheduling of the replication at a later time, based on the availability of the WAN bandwidth or to meet other goals. The ability to schedule the replication at a later time can take advantage of differential pricing of bandwidth at different times of the day, between peak and off peak hours. It can also alleviate unpredictable demand for CPU or memory in the source or remote systems. Both of these modes of operation can include some redundant operations when a local snapshot of the application is also desired in addition to the remote copy. 
         [0034]    In streaming mode, a streaming job can create an application snapshot and copy data from application snapshot to staging disks. At the same time, the data can also be replicated in parallel being streamed to the remote system, eliminating the wait for the copy to the local staging to complete, thus completing the replication quicker. If the remote replication operation can keep pace with the local copy, the replication will complete at the same time as the local copy. If remote copy lags behind the local copy, the local copy is allowed to complete, and then the replication continues from the local staging disk. Once the local copy is complete, a checkpoint of data replicated to this point is recorded, and the ongoing replication is stopped. The data left to replicate is computed, and the replication resumes, using the freshly snapshotted staging disk as the source. Switching the replication to the snapshot of the staging disk enables the application snapshot to be deleted, minimizing time for which the snapshot is kept around. 
         [0035]    In chaining mode, a streaming replication job runs in two separate phases. There is a local snapshot job, which takes a snapshot of the application, and copies it to the local staging disk. At a later time, when the snapshot job is complete, the replication portion of the job is started, and it uses the staging disk (or a snapshot of the staging disk) as the source for the replication. It is similar to the second phase of streaming replication, but copies all blocks changed between a previous streaming replication instance and the time of starting the latest successful local backup. 
         [0036]      FIG. 1  is a system diagram showing the creation of a local and a remote copy of application data, according to some embodiments of the present disclosure.  FIG. 1  shows a source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , application snapshot  1005 , application data  1006 , local copy  1007 , network  1009 , remote copy  1010 , and data transfer  1011 . 
         [0037]    The production host  1003  runs the production application  1004 , which generates the application data  1006  which needs to be copied locally, and replicated to the remote location. 
         [0038]    The Source Data Management system  1001  is the data management system close to the location of the production host; it creates and manages local copies of the application. 
         [0039]    The remote data management system  1002  is the remote system which receives the replicated data and create the remote image. 
         [0040]    The network  1009  connects the source data management system to the remote data management system. 
         [0041]    The process shown in  FIG. 1  creates a consistent image of application data  1006  belonging to application  1004 , running on production host  1003  on the source data management system  1001  and at the remote data management system  1002 . In some embodiments, the process includes taking a stable image of the application data  1006  in the form of an application snapshot  1005  and then copying this data using a data transfer mechanism  1011  to the source data management system 1   1001  to create the local copy  1007  and through the network  1009  to the remote data management system  1002  to create remote copy  1010 . The data transfer mechanism can include  1011  copying data over a socket or pipe using Transmission Control Protocol (TCP) or User Datagram Protocol (UDP). The data transfer can also involve having shared disks that can directly be read by the source data management system; it may also include other data transfer techniques such as Direct Memory Access, or any other technique. 
         [0042]      FIG. 2  is a system diagram showing an incremental local copy and replication operation, according to some embodiments of the present disclosure.  FIG. 2  shows a source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , application snapshot  1005 , application data  1006 , network  1009 , previous application snapshot  2001 , new local copy  2002 , previous local copy  2003 , new remote copy  2004 , previous remote copy  2005 , and data transfer  2006 . 
         [0043]    Application  1004  on production host  1003  includes application data  1006 . Application data  1006  is being incrementally copied locally to the source data management system  1001  and replicated to the remote data management system  1002 . 
         [0044]    The process illustrated in  FIG. 2  includes computing the difference information  2007  between the current application snapshot  1005 , and the previous application snapshot  1007 . In some embodiments, there does not have to be a physical copy of the previous application snapshot  1007 . The snapshot  1007  existed at some point in time in the past and was used during a previous cycle of replication. The difference information can be computed from comparing snapshots, but it can also be computed by keeping track of changed blocks or changed files, or any other technique. 
         [0045]    The combination of the difference information and the changed elements, that is, the actual data that has changed is transferred using the data transfer mechanism  2006 . This changed data and difference information is then saved in the new local copy  2002  and the new remote copy  2004 . Here again, the exact mechanism for saving the changed data and difference information is not material to the process. The changed data may be saved as an incremental backup; it can be converted to a synthetic full backup; it can be combined with the previous local copy  2003  in the form of two snapshots, or using any other technique. 
         [0046]      FIG. 3  is a block diagram showing representations of difference information, according to some embodiments of the present disclosure.  FIG. 3  shows a byte-oriented extent list  3001 , file list  3002 , bitmap  3003 , and multi part byte-oriented extent lists  3004 . 
         [0047]      3001  is a single byte-oriented list. It includes a table which contains rows of byte offsets and byte counts, which indicate the regions of an object (such as a device, logical volume or file) that have changed from one snapshot to the other. The table is followed by a string of bytes that represent the new data in those regions. 
         [0048]      3002  is a single file list. It consists of a list of filenames which are files that have changed. The list is followed by the changed data of the files. The table may also contain file lengths, or offsets and byte counts of regions that have changed within the file. 
         [0049]      3003  is a bitmap. The bitmap has a table with a header that describes the format of the bitmap, including at least the grain size of the bitmap. This header is followed by a string of bits that indicate which grains have changed, and which ones have not. This table is followed by the changed data itself. As used herein, a grain refers to a certain number of blocks. Generally, a bitmap is a string of bits representing the modified portions of data (e.g., volume, file, or some other quantity or organization of data). A bitmap has a granularity, where each bit represents a certain amount of data. A bit can represent a single block or a larger amount of data, such as 8 KB or 64 KB. The amount of data represented by the bit is referred to herein as the “grain-size” of the bitmap. 
         [0050]      3004  is an example of a multi-part changed information representation. In this example, it is a multipart byte-oriented list. This form of difference information consists of a table of changed regions followed by changed data, followed by another table, then more changed data, and so on. This form allows the computation and consumption of difference information in chunks, where each of the chunks is limited in size. 
         [0051]    Many other forms of difference information are possible. There is no requirement that difference information be ordered in any fashion in the tables. In some cases, byte oriented lists or filename lists or multipart lists may include overlaps, where a later entry in a table overlaps a previous entry in the same or an earlier table. The same file or region or an overlapping region may occur in a later table entry that had appeared in an earlier table entry. As long as the changed information is processed in the order it is represented, it will accurately describe all of the changes from the old state to the new one. In some embodiments, the format of the difference information often reflects how the difference information was tracked or collected. Some active collection techniques track every change to the monitored data and record the information by setting a bit in a bitmap in memory. The tracker can then dump the collected list of changes when requested. Other tracking techniques continuously record each of the areas changed as an extent. If the same area is changed again, at a later time, these trackers may generate an extent list where later elements may overlap earlier elements. Still other ways of generating difference information may be passive in the sense that they collect difference data only when requested, by comparing current data with older data, or comparing the signatures of current data with signatures of old data. 
         [0052]    In some embodiments, an extent list for the replication to the remote system is created by combining all of the extent lists for each of the copies to the source computing device since the last replication was performed. For example, assume the following copy, replication, and extent list scheme:
       Day 1. Local copy and replication. Extent list 1   Day 2. Local copy only. Extent list 2   Day 3. Local copy and replication. Extent list 3.
 
On day 3, the local copy uses extent list 3. The replication to the remote system uses a combination of extent lists 2 and 3, since the changes in extent list 2 were not replicated on day 2.
       
 
         [0056]      FIG. 4  is a system diagram showing an incremental local copy and replication run sequentially, according to some embodiments of the present disclosure.  FIG. 4  shows many of the same elements illustrated and described with respect to  FIGS. 1 and 2 .  FIG. 4  also shows local data transfer  4001  and remote data transfer  4002 . 
         [0057]    In the process shown in  FIG. 4 , an incremental data copy from the host  1003  is performed to the source data management system  1001 . After the incremental data copy to the source data management system  1001  is complete, an incremental replication  4002  from the source data management system  1001  is performed to the remote data management system  1002 . 
         [0058]    As described above, application  1004  on production host  1003  includes application data  1006 . Application data  1006  is being incrementally copied locally to the source data management system  1001  and replicated to the remote data management system  1002 . 
         [0059]    The combination of the difference information and the changed elements  2007 , is transferred using the local data transfer mechanism  4001  to create the New Local Copy  2002 . Once that operation is complete, the application snapshot  1005  can be released. 
         [0060]    Only after the new local copy  2002  is created is the remote data transfer mechanism  4002  used to transfer the difference information  2007  from the new local copy  2002  to the new remote copy  2004  through the network  1009 . 
         [0061]    The difference information  2007  is used for both the local copy and then for the remote copy operation. This is the case as long as the previous local copy  2003  and the previous remote copy  2005  are identical in content. The previous local copy  2003  and the previous remote copy  2005  are identical in content if incremental replication was carried out when the previous local copy was created. If the incremental replication was not performed or the operation failed, additional difference information can be computed between the previous remote copy  2005  and the previous local copy  2003 , and this difference information can also be transferred using the remote data transfer  4002  to create the new remote copy  2004 . 
         [0062]    The process of running the local copy  2002  and the remote replication  4002  sequentially allows for the application snapshot  1005  to be released as soon as the new local copy  2002  is created. This minimizes the time for which the snapshot is held. On the other hand, the time for completing the new remote copy  2004  is delayed, because the replication cannot begin until the new local copy  2002  is completed. 
         [0063]      FIG. 5  is a system diagram showing an incremental local copy and replication run in parallel, according to some embodiments of the present disclosure.  FIG. 5  shows many of the same elements illustrated and described with respect to  FIGS. 1 and 2 .  FIG. 5  also shows local data transfer  11001  and remote data transfer  11002 . 
         [0064]    In the process shown in  FIG. 5 , an incremental data copy is performed to the source data management system  1001 , and at the same time, an incremental replication is performed to the remote data management system  1002 . 
         [0065]    As described above, application  1004  on production host  1003  includes application data  1006 . Application data is being incrementally copied locally to the source data management system  1001  and replicated to the remote data management system  1002 . 
         [0066]    The combination of the difference information and the changed elements  2007 , is transferred using the local data transfer mechanism  5001  to create the new local copy,  2002 . At the same time, the difference information and changed elements  2007  are also transferred using the remote data transfer mechanism  5002  to the new remote copy  2004  over the network  1009 . 
         [0067]    After both the new local copy  2002  is completed and the new remote copy  2004  is completed, the application snapshot  1005  is released. 
         [0068]    The process of running the local copy and the remote replication in parallel can improve the speed of the completion of the new remote copy  1013 . However, the application snapshot  1005  cannot be released until the remote copy  1013  has been completed, which can take a long time, if the network is slow. 
         [0069]      FIG. 6  is a system diagram showing an incremental local copy and replication run in parallel, according to some embodiments of the present disclosure.  FIG. 6  shows many of the same elements illustrated and described with respect to  FIGS. 1 and 2 .  FIG. 6  also shows local data transfer  6001 , replication from local copy  6002 , and remote data transfer  6003 . 
         [0070]    In the process shown in  FIG. 6 , an incremental data copy  6001  is performed to the source data management system  1001 , and at the same time, an incremental replication  6003  is performed to the remote data management system  1002 . When the local data copy completes  6001 , the incremental replication is halted and resumed from the local copy  6002 . 
         [0071]    As described above, application  1004  on production host  1003  includes application data  1006 . This data is being incrementally copied locally to the source data management system  1001  and replicated to the remote data management system  1002 . 
         [0072]    The combination of the difference information and the changed elements  2007 , is transferred using the local data transfer mechanism  6001  to create the New Local Copy,  2002 . At the same time, the difference information and changed elements  2007  are also transferred using the remote data transfer mechanism  6003  to the new remote copy  2004  over the network  1009 . 
         [0073]    The local data transfer mechanism  6001  is usually much faster than the remote data transfer mechanism  6003 , and as a result, the new local copy  2002  is complete while the new remote copy  2004  is still in progress. 
         [0074]    When the new local copy  2002  is complete, the remote data transfer is halted, and the application snapshot  1005  is released. The remote data transfer is then resumed, but now it is using the new local copy  2002  as the source, since the application snapshot  1005  is no longer available. The data transfer from the local copy  2002  is resumed from the point in the difference information stream (also referred to herein as checkpoint) where it was halted. The replication from the local copy  2002  continues until the new remote copy  2004  has been completed. 
         [0075]    The process of running the local copy and the remote replication in combination parallel/sequential mode has the advantages of releasing the application snapshot  1005  at the earliest possible time, and also completing the remote replication with a minimum amount of delay. 
         [0076]    As described above, the modes of local copy and remote replication described with reference to  FIG. 6  are not restricted to incremental local copy and remote replication. The same techniques can be equally applied to local copy and replication where no difference information is available, and all of the data from the application snapshot has to be copied and replicated. The techniques described above for a combined parallel/sequential mode will also be referred to herein as streaming replication. 
         [0077]      FIG. 7  is a system diagram showing in detail a first phase of streaming replication, according to some embodiments of the present disclosure.  FIG. 7  shows a source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , application snapshot  1005 , network  1009 , source staging disks  7001 , and remote staging disks  7002 . 
         [0078]    As described above, application  1004  running in production host  1003  is being protected at source data management system  1001 . It is protected by a streaming replication policy to perform replication to the remote data management system  1002  connected over a network  1009 . In some embodiments, the network is a high bandwidth network. In some embodiments, a streaming replication policy is one of the policies within the SLA for an application. It defines when and how frequently the application data is to be copied locally, how frequently it is to be replicated to the remote locations, and how long it is to be retained at each location. When no source staging disks exist in the source data management system  1001 , the source data management system  1001 , as part of a streaming replication process, first creates local staging disks  7001  that are big enough to hold the application data on source data management system  1001 . Source data management system  1001  also creates remote staging disks  7002  on remote data management system  1002 , if none exist. Source data management system  1001  then creates a snapshot  1005  of the source application  1004  to use as a consistent point in time for replication. In some embodiments, application snapshot  1005  can also be generated by the production host  1003 . Changes to the application data that occur between snapshots are copied  7003  from the production host  1003  to the source staging disks  7001 . If this is the first replication job, a full ingest  7003  is performed where all blocks are copied. While changed blocks are copied  7003  to the staging disk, these changed blocks are copied in parallel  7004  to the remote staging disks  7002 . These copy operations  1006  and  1008  continue until all changed blocks are copied from source application snapshot. As referred to herein, a first pass or a first phase of replication includes steps  7003  and  7004  and ends when all of the changed blocks are copied to the local staging disks  7002 . As described in more detail below, a full copy of the application data necessarily indicates that there is no overlap in an extent list associated with the full copy. 
         [0079]      FIG. 8  is a system diagram showing in detail a second phase of streaming replication, according to some embodiments.  FIG. 8  shows source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , application snapshot  1005 , network  1009 , source staging disks  7001 , remote staging disks  7002 , source snapshot  8001 , and remote snapshot  8002 . 
         [0080]    As described above, the first phase of streaming replication, as described herein, ends when all of the changed blocks are copied to the local staging disks  7001 . Application specific reference information from application snapshot  1005  is first saved to obtain changed block information. After saving the application specific reference information, the application snapshot is deleted. As described above with respect to  FIG. 3 , the reference information can include lists or indexes indicating which portions of have changed. Next, a snapshot copy  8001  of source staging disks  7001  is created. Remote copy operation  7004  from the first phase is stopped  8003  and the last checkpoint offset is saved to record the progress of the remote copy. The checkpoint offset determines the offset from where second phase of copy should begin. 
         [0081]    The second phase of copy is started with local snapshot copy  8001  as source and remote staging disks  7002  as target. Instead of copying all the original modified blocks, this operation only copies blocks starting from the previous checkpoint offset at the end of first phase. Once all blocks are copied to the remote staging disks, the second phase of streaming is completed. A remote snapshot copy  8002  is taken from the remote staging disks  1010  as a replication point in time reference. 
         [0082]      FIG. 9  is a flowchart illustrating a control flow of streaming replication run in streaming mode, according to some embodiments of the present disclosure. 
         [0083]    In step  9001 , application snapshot  1005  is created for application  1004 . If there are any changes in application data from previous snapshot, step  9002  copies the changed block to local staging disks. Step  9003  initiates copy the changed blocks to remote staging disks and offset of last successfully copied data block is recorded as a checkpoint. Copy operation continues to check for more changed blocks and perform steps  9002  and  9003  until all blocks are copied to the local staging disk. When there are no more blocks to be copied, step  9004  deletes the application snapshot  1005 . Source data management system  1001  now creates a local snapshot  8001  in source data management system  1001  from source staging disks  7001 . A local backup image is cataloged in the database that acts as a reference to these local snapshot disks for operations like restoring from this point in time image. If the remote copy operations initiated in step  9003  are not completed, they are stopped  8003 . A new remote copy operation to remote staging disks  8004  is started in step  9006  that uses the local snapshot copy  8001  as source and last successful checkpoint offset from  9003  as starting offset for this new copy operation. Step  9007  creates a remote snapshot  8002  from remote staging disks  7002  and catalogs the remote image in the database that acts as a reference to this point in time at the remote data management system  1002 . The remote image cataloged as part of previous streaming replication job is deleted at this point of time, as only the last successful state of the application is maintained at the remote data management system  1002 . 
         [0084]      FIG. 10  is a system diagram showing a full ingest of a chained streaming replication job, as described in some embodiments of the present disclosure.  FIG. 10  shows source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , network  1009 , source application  10001 , remote reference application  10002 , snapshot policy  10003 , dedup policy  10004 , streaming policy  10005 , snapshot image  10007 , dedup image  10010 , and remote image  10011 . 
         [0085]    Application  1004  running in source host  1003  is being protected at source data management system  1001  in a similar manner as described with respect to  FIG. 7 . The protected version of the application within data management system  1001  is referred to herein as source application  10001 . Source application  10001  is protected using various policies that adhere to the service level agreement for protecting this application, as described above. Snapshot policy  10003  is associated with snapshotting the application  1004  at periodic intervals. As described above, the first job includes performing a full ingest  10006  that copies all blocks of source application into data management system  1001 , and a snapshot image  10007  is created that references the point in time image of the source application. Dedup policy  10004  is associated with periodically deduplicating snapshots created in the previous step and the corresponding compressed objects are cataloged as dedup images. As part of the first replication, the dedup policy specifies copying  10009  the snapshot image  10007  of the full ingest into dedup store and catalogs the resulting dedup objects as dedup image  10010 . Streaming policy  10005  is associated with periodically transferring the snapshot created by snapshot policy to the replication target data management system  1002 . The first streaming replication job performs a full ingest that copies all blocks of the local snapshot image  10007  and catalogs the remote images as  10011  that points to the remote copies. A remote reference application  10002  provides a record of remote images corresponding to the incoming replicated data. 
         [0086]      FIG. 11  is a system diagram showing a first incremental snapshot operation of a chained streaming replication job, as described in some embodiments of the present disclosure.  FIG. 11  shows source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , network  1009 , source application  10001 , remote reference application  10002 , snapshot policy  10003 , dedup policy  10004 , streaming policy  10005 , dedup image  10010 , remote image  10011 , delta x  11001 , and snapshot image  11002 . 
         [0087]    As described above, snapshot policy  10003  ensures that snapshots are periodically run for application  10001 . When it is time to run the next snapshot job after the first full ingest, source data management system  1001  calculates the changed blocks from previous snapshot job. Resulting changed blocks, referenced as Delta X  11001 , are copied to source data management system  1001  and a new snapshot image  11002  is cataloged that references the new state of the application at the time of running the incremental snapshot job. 
         [0088]      FIG. 12  is a system diagram showing an incremental operation of a chained streaming replication job including snapshot and streaming operations, as described in some embodiments of the present disclosure.  FIG. 12  shows source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , network  1009 , source application  10001 , remote reference application  10002 , snapshot policy  10003 , dedup policy  10004 , streaming policy  10005 , dedup image  10010 , remote image  10011 , delta y  12001 , delta x+y  12003 , and snapshot image  12004 . 
         [0089]    As described above, streaming policy  10005  ensures that application data is periodically transferred to the remote data management system  1002 . When it is time to perform an incremental snapshot and also replicate it to the remote data management system, a snapshot is performed similar to the first incremental snapshot described above. Source data management system  1001  copies the new changed blocks, referenced as Delta Y  12001  to data management system  1001  and a new snapshot image  12002  is cataloged. A streaming job scheduled at this point of time uses the latest snapshot image  12002  as a source for copying. All of the changed blocks between the previous streaming job and current job, referenced as Delta X+Y  12003 , are copied to remote data management system  1002 . When the copy operation is complete, a new image  12004  is cataloged that references the newly copies remote objects. Previous image  10011  is now deleted form the remote data management system  1002 , as only the latest point in time image of the application is maintained at the remote data management system  1002 . 
         [0090]      FIG. 13  is a system diagram showing an incremental operation of a chained streaming replication job including snapshot and dedup operations, as described in some embodiments of the present disclosure.  FIG. 13  shows source data management system  1001 , a remote data management system  1002 , a production host  1003 , application  1004 , network  1009 , source application  10001 , remote reference application  10002 , snapshot policy  10003 , dedup policy  10004 , streaming policy  10005 , delta z  13001 , snapshot image  13002 , delta x+y+z  13003 , and dedup image  13004 . 
         [0091]    As described above, dedup policy  4004  ensures that application data is periodically transferred to the local dedup store on source data management system  1001 . When it is time to perform an incremental snapshot and also deduplicate it, a new snapshot image  13002  is created by a copy operation that captures the changed blocks referenced as Delta Z  13001  from application. All changed blocks between the previous dedup operation and the current one, referenced as Delta X+Y+Z  13003 , are copied to the dedup store and a dedup image  13004  is cataloged as a reference. 
         [0092]      FIG. 14  is a flowchart illustrating a chained job including snapshot, dedup and streaming replication policies, according to some embodiments of the present disclosure. 
         [0093]    When it is time to schedule a local backup, step  14001  creates an application snapshot. In step  14002 , all changes from application are copied to local staging disks. If it is the first snapshot, a full ingest is performed where are blocks of application data are copied to local disks on source data management system  1001 . When the copy is complete, step  14003  creates local copies and catalogs a local backup image in the database that acts as a reference to the local copies. Application snapshot created in step  14001  is now deleted after saving a reference to it for use in future jobs for obtaining changed block information. 
         [0094]    When it is time to perform a streaming replication, step  14005  picks the last successful snapshot as source. It calculates the changed blocks between a previous image and the latest snapshot and all the changed blocks are copied to remote staging disks. If this is the first job, all blocks are copied from local snapshot to the remote staging disks. At the end of the copy, step  14006  creates remote copies from the remote staging disks and catalogs the remote image in the database that acts as a reference to the point in time of the application on remote data management system  1002 . In step  14007 , previous image is deleted if this is not the first streaming job for this application. 
         [0095]    When it is time to perform a dedup copy, step  14008  picks the latest local snapshot as source. It calculates the changed blocks between the previous dedup image and the latest snapshot and all the changed blocks are copied to the dedup store. If this is the first dedup job, all blocks are copied from local snapshot to the dedup store. At the end of the copy, step  14009  creates local dedup objects and catalogs local dedup image in the database that acts as a reference to the point in time of the application in the dedup store. 
         [0096]      FIG. 15  is a flowchart showing a process for choosing streaming versus sequential replication, according to some embodiments of the present disclosure. 
         [0097]    Step  15001  creates a snapshot of the application, and computes difference information between the current snapshot and any previous snapshots. 
         [0098]    Step  15002  makes a decision on whether difference information is available. If none is available, the system behaves as though there is a single extent list where all extents have changed (step  15005 ), and the system proceeds with replication in streaming mode, step  15006 . 
         [0099]    If there is difference information available, then there is a determination in step  15003  whether the difference information is organized in such a way that streaming replication is advantageous (also referred to herein as being friendly to streaming replication). If it is determined that the difference information is streaming friendly, the system proceeds with streaming replication, step  15006 . 
         [0100]    If it is determined that the difference information is not friendly to streaming, then the system proceeds with running the local copy and replication as sequential operations, step  15004 . 
         [0101]    Difference information is considered to be friendly to stream under these conditions:
       It is known that the difference information is strongly ordered. For example, difference information is strongly ordered when it is identified that one changed extent is before or after any other changed extent.   It is known that changed extents that come later do not overlap with changed extents that come earlier in the ordered list of extents. Overlapping is defined as the two extents include the same part of an object, and include different changed data.       
 
         [0104]    In some embodiments, if either of these conditions is not satisfied (or cannot be determined authoritatively), then the difference information is not friendly to streaming replication. 
         [0105]    In some embodiments, and as described above, a decision to replicate data from a source device to a remote device in parallel or serially with copying the data to a location within the source device can depend on a parameter associated with an ordering of the changed extents or elements. The parameter can indicate whether the ordering of changed elements in the application data is overlapping or non-overlapping. 
         [0106]    In some embodiments, data can be replicated in parallel with copying of the data when the list is overlapping. In some embodiments, this involves sorting and combining overlapping entries in the extent list so the extent list no longer contains overlapping entries, and then can be processed in parallel. In situations where the whole extent list is not available up front, the source device can sort as much of the extent list as is available, process that portion in parallel, then obtain the next chunk of extents, sort and process in parallel, etc. 
         [0107]    In some embodiments, compression and/or deduplication can be applied to the application data by a source device prior to sending the application data to a remote device. The source device can determine when to apply deduplication and/or compression based on network conditions, including latency and bandwidth of the communications between the local and remote systems. In some embodiments, the replication of application data is done directly by sending data from the source computing device to the remote copying device without deduplicating and compressing the data to take advantage of high bandwidth networks, and application data is deduplicated and compressed before sending to the remote system to optimize the replication for low bandwidth networks. Bandwidth can be determined by sending a known amount of data, and measuring the time taken, so as to compute the number of bytes per second that the network can sustain. High bandwidth is when the bandwidth of the network is a large fraction of the rate at which data can be written to a location on the source device. When there is high bandwidth, the time taken to replicate the data is similar to the time taken to replicate the data to the remote system. Low bandwidth is when the rate of transfer of data across the network is substantially slower than the rate of storing the data in another location on the source system. The time taken to complete the replication is substantially more than the time to make a local copy. In this environment, the user is willing to expend memory and CPU resources to perform deduplication and compression to attempt to speed up the replication to the remote system. 
         [0108]    The subject matter described herein can be implemented in digital electronic circuitry, or in computer software, firmware, or hardware, including the structural means disclosed in this specification and structural equivalents thereof, or in combinations of them. The subject matter described herein can be implemented as one or more computer program products, such as one or more computer programs tangibly embodied in an information carrier (e.g., in a machine readable storage device), or embodied in a propagated signal, for execution by, or to control the operation of, data processing apparatus (e.g., a programmable processor, a computer, or multiple computers). A computer program (also known as a program, software, software application, or code) can be written in any form of programming language, including compiled or interpreted languages, and it can be deployed in any form, including as a stand-alone program or as a module, component, subroutine, or other unit suitable for use in a computing environment. A computer program does not necessarily correspond to a file. A program can be stored in a portion of a file that holds other programs or data, in a single file dedicated to the program in question, or in multiple coordinated files (e.g., files that store one or more modules, sub programs, or portions of code). A computer program can be deployed to be executed on one computer or on multiple computers at one site or distributed across multiple sites and interconnected by a communication network. 
         [0109]    The processes and logic flows described in this specification, including the method steps of the subject matter described herein, can be performed by one or more programmable processors executing one or more computer programs to perform functions of the subject matter described herein by operating on input data and generating output. The processes and logic flows can also be performed by, and apparatus of the subject matter described herein can be implemented as, special purpose logic circuitry, e.g., an FPGA (field programmable gate array) or an ASIC (application specific integrated circuit). 
         [0110]    Processors suitable for the execution of a computer program include, by way of example, both general and special purpose microprocessors, and any one or more processor of any kind of digital computer. Generally, a processor will receive instructions and data from a read only memory or a random access memory or both. The essential elements of a computer are a processor for executing instructions and one or more memory devices for storing instructions and data. Generally, a computer will also include, or be operatively coupled to receive data from or transfer data to, or both, one or more mass storage devices for storing data, e.g., magnetic, magneto optical disks, or optical disks. Information carriers suitable for embodying computer program instructions and data include all forms of nonvolatile memory, including by way of example semiconductor memory devices, (e.g., EPROM, EEPROM, and flash memory devices); magnetic disks, (e.g., internal hard disks or removable disks); magneto optical disks; and optical disks (e.g., CD and DVD disks). The processor and the memory can be supplemented by, or incorporated in, special purpose logic circuitry. 
         [0111]    To provide for interaction with a user, the subject matter described herein can be implemented on a computer having a display device, e.g., a CRT (cathode ray tube) or LCD (liquid crystal display) monitor, for displaying information to the user and a keyboard and a pointing device, (e.g., a mouse or a trackball), by which the user can provide input to the computer. Other kinds of devices can be used to provide for interaction with a user as well. For example, feedback provided to the user can be any form of sensory feedback, (e.g., visual feedback, auditory feedback, or tactile feedback), and input from the user can be received in any form, including acoustic, speech, or tactile input. 
         [0112]    The subject matter described herein can be implemented in a computing system that includes a back end component (e.g., a data server), a middleware component (e.g., an application server), or a front end component (e.g., a client computer having a graphical user interface or a web browser through which a user can interact with an implementation of the subject matter described herein), or any combination of such back end, middleware, and front end components. The components of the system can be interconnected by any form or medium of digital data communication, e.g., a communication network. Examples of communication networks include a local area network (“LAN”) and a wide area network (“WAN”), e.g., the Internet. 
         [0113]    It is to be understood that the disclosed subject matter is not limited in its application to the details of construction and to the arrangements of the components set forth in the following description or illustrated in the drawings. The disclosed subject matter is capable of other embodiments and of being practiced and carried out in various ways. Also, it is to be understood that the phraseology and terminology employed herein are for the purpose of description and should not be regarded as limiting. 
         [0114]    As such, those skilled in the art will appreciate that the conception, upon which this disclosure is based, may readily be utilized as a basis for the designing of other structures, methods, and systems for carrying out the several purposes of the disclosed subject matter. It is important, therefore, that the claims be regarded as including such equivalent constructions insofar as they do not depart from the spirit and scope of the disclosed subject matter. 
         [0115]    Although the disclosed subject matter has been described and illustrated in the foregoing exemplary embodiments, it is understood that the present disclosure has been made only by way of example, and that numerous changes in the details of implementation of the disclosed subject matter may be made without departing from the spirit and scope of the disclosed subject matter, which is limited only by the claims which follow.