Abstract:
Exemplary embodiments provide a way to manage data recovery in a distributed system having multiple data store nodes. A storage system comprises: a first node including a first processor; and a plurality of second nodes coupled to the first node, each of the plurality of second nodes including a second processor and one or more second storage devices. The first processor is configured to control to store data and replication of the data in the second storage devices of two or more second nodes. If at least one of the second nodes has failed and a storage capacity of the plurality of second nodes is below a given threshold, one of the second nodes is configured to receive a first data, which is replication of data stored in a failed second node, from another of the second nodes, and create parity data based on the received first data.

Description:
BACKGROUND OF THE INVENTION 
       [0001]    The present invention relates generally to storage systems and, more particularly, to distributed storage system and method for managing data recovery. 
         [0002]    In a distributed storage system, methods for replicating data to other nodes and failure recovery of a storage node are known. The distributed storage system performs to mirror twice or triple replication data to recover composited disks, nodes and/or network failures. 
         [0003]    When many nodes fail, the distributed storage system changes the layout of replication data which are stored to the failure nodes to healthy nodes and it copies data to healthy nodes to keep redundancy. However, if too many nodes have failed, physical capacity is insufficient and replication or host write will fail. 
       BRIEF SUMMARY OF THE INVENTION 
       [0004]    Exemplary embodiments of the invention provide a way to manage data recovery in a distributed system when many nodes fail. The distributed storage system has multiple data store nodes. The distributed storage system monitors the total physical capacity of healthy nodes and continues to operate normally when the total physical capacity of healthy nodes is larger than a threshold capacity which is required for twice or triple mirroring capacity of stored data. When enough nodes fail and the total physical capacity of healthy nodes is lower than the threshold capacity, the distributed storage system temporarily changes the redundancy method. In a first approach, data store nodes create parity from two mirroring data sets which are stored newly. If the total physical capacity of healthy nodes continues to drop, in a second approach, data store nodes read an existing mirroring data and creates parity from the existing mirroring data and a mirroring data which is stored newly. The distributed objects storage system manages data redundancy level temporarily to prevent data loss and to prevent reaching full physical capacity. 
         [0005]    In accordance with an aspect of the present invention, a storage system comprises: a first node including a first processor; and a plurality of second nodes coupled to the first node, each of the plurality of second nodes including a second processor and one or more second storage devices. The first processor is configured to control to store data and replication of the data in the second storage devices of two or more second nodes. If at least one of the second nodes has failed and a storage capacity of the plurality of second nodes is below a given threshold, one of the second nodes is configured to receive a first data, which is replication of data stored in a failed second node, from another of the second nodes, and create parity data based on the received first data. 
         [0006]    In some embodiments, the first node is configured to determine, based on the storage capacity, a coding mode for the second nodes. If the storage capacity is below a first threshold and above a second threshold which is lower than the first threshold, the coding mode is a first coding mode of parity between re-replicated data from multiple second nodes. If the storage capacity is below the second threshold and above a third threshold which is lower than the second threshold, the coding mode is a second coding mode of parity with existing replication data. If the storage capacity is below the third threshold, the coding mode is a higher parity mode having a higher parity than the first coding mode and the second coding mode. If the storage capacity is at or above a fourth threshold equal to or higher than the first threshold, the coding mode is a normal coding mode under which the second node which has created parity data is configured to restore data, which were used to create the parity data, to the other second nodes. 
         [0007]    In specific embodiments, the first node is configured to issue a first write request to a first target second node to store primary data and, if the first target second node cannot store the primary data or the first target second node has failed, issue the first write request to another first target second node to store the primary data. The first node is configured to issue a second write request to a second target second node to store mirror data corresponding to the primary data and, if the second target second node cannot store the mirror data or the second target second node has failed, issue the second write request to another second target second node to store the mirror data. 
         [0008]    In some embodiments, the first node is configured to determine, based on the storage capacity, a coding mode for the second nodes. If the storage capacity is below a first threshold and above a second threshold which is lower than the first threshold, the coding mode is a first coding mode of parity between re-replicated data from multiple second nodes. If the coding mode is the first coding mode, the second node which has received the first data is configured to receive a second data, which is replication of data stored in a failed second node, and to create parity data based on the received first data and second data. 
         [0009]    In specific embodiments, the first node is configured to determine, based on the storage capacity, a coding mode for the second nodes. If the storage capacity is below a first threshold and above a second threshold which is lower than the first threshold, the coding mode is a first coding mode of parity between re-replicated data from multiple second nodes. If the storage capacity is below the second threshold, the coding mode is a second coding mode of parity with existing replication data. If the coding mode is the second coding mode, the second node which receives the first data includes an existing third data which is replication of data stored in another second node, and is configured to create parity data based on the received first data and the existing third data, to overwrite the created parity data to the existing third data, and to delete the received first data. 
         [0010]    In some embodiments, the first node is configured to issue a first write request to a first target second node to store primary data. The first target second node is configured to replicate the primary data to the second node which stores the received first data. The second node which stores the received first data is configured to create parity data between the received first data and the replicated primary data. 
         [0011]    In specific embodiments, the first node is configured to issue a first read request to a first target second node which stores non-parity data of the read request. If the first target second node cannot access the non-parity data or the first target second node has failed, the first node is configured to search a second target second node which stores parity data created using the non-parity data, and the second target second node is configured to issue a second read request to a destination second node which has parity restore data for the parity data, to receive the parity restore data from the destination second node, and to restore recovery data between the received parity restore data from the destination second node and the parity data. 
         [0012]    Another aspect of the invention is directed to a method for managing data storage in a storage system which includes a first node having a first processor and a plurality of second nodes coupled to the first node, each of the plurality of second nodes having a second processor and one or more second storage devices. The method comprises: controlling to store data and replication of the data in the second storage devices of two or more second nodes; and if at least one of the second nodes has failed and a storage capacity of the plurality of second nodes is below a given threshold, receiving by one of the second nodes a first data, which is replication of data stored in a failed second node, from another of the second nodes, and creating parity data based on the received first data. 
         [0013]    Another aspect of the invention is directed to a computer-readable storage medium storing a plurality of instructions for managing data storage in a storage system which includes a first node having a first processor and a plurality of second nodes coupled to the first node, each of the plurality of second nodes having a second processor and one or more second storage devices. The plurality of instructions comprise: instructions that cause the first processor to control to store data and replication of the data in the second storage devices of two or more second nodes; and instructions that, if at least one of the second nodes has failed and a storage capacity of the plurality of second nodes is below a given threshold, cause the second processor of one of the second nodes to receive a first data, which is replication of data stored in a failed second node, from another of the second nodes, and create parity data based on the received first data. 
         [0014]    These and other features and advantages of the present invention will become apparent to those of ordinary skill in the art in view of the following detailed description of the specific embodiments. 
     
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS 
         [0015]      FIG. 1  illustrates an example of a hardware configuration of a computer system in which the method and apparatus of the invention may be applied. 
           [0016]      FIG. 2 a    shows an example of a proxy node. 
           [0017]      FIG. 2 b    shows an example of the proxy node memory. 
           [0018]      FIG. 3 a    shows example of a data store node. 
           [0019]      FIG. 3 b    shows an example of the data store node memory. 
           [0020]      FIG. 4  shows an example of a data store nodes table. 
           [0021]      FIG. 5  shows an example of a replication data layout table. 
           [0022]      FIG. 6  shows an example of a data layout table. 
           [0023]      FIG. 7  shows an example of a replication data node table. 
           [0024]      FIG. 8  is an overview of a normal read operation. 
           [0025]      FIG. 9  is an overview of a normal write operation. 
           [0026]      FIG. 10  is an overview of the process of changing replication mode and generating parity between re-replicated data. 
           [0027]      FIG. 11  is overview of the process of changing replication mode and generating parity between re-replicated data and existing replication data. 
           [0028]      FIG. 12  is an overview of the process of changing replication mode and generating parity between re-replicated data involving a write operation. 
           [0029]      FIG. 13  is an overview of a read operation during a change in the coding mode. 
           [0030]      FIG. 14  shows an example of a flow diagram illustrating a read operation and parity recovery read. 
           [0031]      FIG. 15  shows an example of a flow diagram illustrating a process for host write operation. 
           [0032]      FIG. 16  shows an example of a flow diagram illustrating a process for generating parity between re-replicated data or between re-replicated data and existing replicated data. 
           [0033]      FIG. 17  shows an example of a flow diagram illustrating a process for restoring parity data to normal mode. 
           [0034]      FIG. 18  shows an example of a flow diagram illustrating a process for changing replication mode. 
       
    
    
     DETAILED DESCRIPTION OF THE INVENTION 
       [0035]    In the following detailed description of the invention, reference is made to the accompanying drawings which form a part of the disclosure, and in which are shown by way of illustration, and not of limitation, exemplary embodiments by which the invention may be practiced. In the drawings, like numerals describe substantially similar components throughout the several views. Further, it should be noted that while the detailed description provides various exemplary embodiments, as described below and as illustrated in the drawings, the present invention is not limited to the embodiments described and illustrated herein, but can extend to other embodiments, as would be known or as would become known to those skilled in the art. Reference in the specification to “one embodiment,” “this embodiment,” or “these embodiments” means that a particular feature, structure, or characteristic described in connection with the embodiment is included in at least one embodiment of the invention, and the appearances of these phrases in various places in the specification are not necessarily all referring to the same embodiment. Additionally, in the following detailed description, numerous specific details are set forth in order to provide a thorough understanding of the present invention. However, it will be apparent to one of ordinary skill in the art that these specific details may not all be needed to practice the present invention. In other circumstances, well-known structures, materials, circuits, processes and interfaces have not been described in detail, and/or may be illustrated in block diagram form, so as to not unnecessarily obscure the present invention. 
         [0036]    Furthermore, some portions of the detailed description that follow are presented in terms of algorithms and symbolic representations of operations within a computer. These algorithmic descriptions and symbolic representations are the means used by those skilled in the data processing arts to most effectively convey the essence of their innovations to others skilled in the art. An algorithm is a series of defined steps leading to a desired end state or result. In the present invention, the steps carried out require physical manipulations of tangible quantities for achieving a tangible result. Usually, though not necessarily, these quantities take the form of electrical or magnetic signals or instructions capable of being stored, transferred, combined, compared, and otherwise manipulated. It has proven convenient at times, principally for reasons of common usage, to refer to these signals as bits, values, elements, symbols, characters, terms, numbers, instructions, or the like. It should be borne in mind, however, that all of these and similar terms are to be associated with the appropriate physical quantities and are merely convenient labels applied to these quantities. Unless specifically stated otherwise, as apparent from the following discussion, it is appreciated that throughout the description, discussions utilizing terms such as “processing,” “computing,” “calculating,” “determining,” “displaying,” or the like, can include the actions and processes of a computer system or other information processing device that manipulates and transforms data represented as physical (electronic) quantities within the computer system&#39;s registers and memories into other data similarly represented as physical quantities within the computer system&#39;s memories or registers or other information storage, transmission or display devices. 
         [0037]    The present invention also relates to an apparatus for performing the operations herein. This apparatus may be specially constructed for the required purposes, or it may include one or more general-purpose computers selectively activated or reconfigured by one or more computer programs. Such computer programs may be stored in a computer-readable storage medium including non-transitory medium, such as, but not limited to optical disks, magnetic disks, read-only memories, random access memories, solid state devices and drives, or any other types of media suitable for storing electronic information. The algorithms and displays presented herein are not inherently related to any particular computer or other apparatus. Various general-purpose systems may be used with programs and modules in accordance with the teachings herein, or it may prove convenient to construct a more specialized apparatus to perform desired method steps. In addition, the present invention is not described with reference to any particular programming language. It will be appreciated that a variety of programming languages may be used to implement the teachings of the invention as described herein. The instructions of the programming language(s) may be executed by one or more processing devices, e.g., central processing units (CPUs), processors, or controllers. 
         [0038]    Exemplary embodiments of the invention, as will be described in greater detail below, provide apparatuses, methods and computer programs for managing data recovery in a distributed storage system. 
         [0039]      FIG. 1  illustrates an example of a hardware configuration of a computer system in which the method and apparatus of the invention may be applied. The distributed storage system  1  contains a plurality of proxy nodes  2 , a plurality of data store nodes  3 , and a distributed storage network  5 . A host  4  is connected to the distributed storage system  1  via the proxy nodes  2  using the network  7 . The distributed storage system  1  is connected to an administrator console  6  via a management network. 
         [0040]      FIG. 2 a    shows an example of a proxy node  2 . The proxy node  2  contains a host I/F which connect to the host, a CPU, a memory  22 , a node I/F, and a Disk I/F, which are connected with each other by a bus I/F such as PCI, DDR, SCSI.  FIG. 2 b    shows an example of the proxy node memory  22 . The proxy node memory  22  contains a data layout program  23 , a failure monitor program  24 , a data store nodes table  40  ( FIG. 4 ), and a replication data layout table  50  ( FIG. 5 ). 
         [0041]      FIG. 3 a    shows example of a data store node  3 . The data store node  3  contains a host I/F which connect to the host via a proxy node  2 , a CPU, a memory  33 , a Disk I/F, and HDDs, which are connected with each other by a bus I/F such as PCI, DDR, SCSI.  FIG. 3 b    shows an example of the data store node memory  32 . The data store node memory  32  contains an IO (Input/Output) program  33 , a replication program  34 , a data layout table  60  ( FIG. 6 ), and a replication data node table  70  ( FIG. 7 ). The data layout table  60  and/or replication data node table  70  can be stored to disks in the data store node  3 . 
         [0042]      FIG. 4  shows an example of a data store nodes table  40 . The data store nodes table  40  contains columns of node ID  41 , node condition flag  42 , physical data capacity  43 , stored data capacity  44 , and failure capacity  45 . The node ID  41  is a unique data store node identifier. The physical data capacity  43  is the total usable physical capacity in the data store node. The stored data capacity  44  is the used physical capacity in the data store node which is used for storing data. The failure capacity  45  is the total failure capacity which is related to disk failure, network cable failure, or the node failure in the data store node. The total capacity is equal to the sum of the stored data capacity  44  and the failure capacity  45 . The remaining usable capacity is equal to the physical data capacity  43  minus the stored data capacity  44 . The monitor program  24  in the proxy node monitors the condition status and these capacities. Based on the healthy nodes, the data layout program  23  calculates the object location using consistency hash function. Also, the mirror data location is calculated using the consistency hash function. 
         [0043]      FIG. 5  shows an example of a replication data layout table  50 . The replication data layout table  50  contains columns of object ID  51 , destination node ID  52 , coding mode flag  53 , and replication mode flag  54 . The object ID  51  is a unique object data ID. The destination node ID  52  is a node ID where the replication data is temporarily stored. The coding mode flag  53  contains the coding mode which is changed from the original replication mode such as mirroring or triplication to the coding mode such as parity  2 , parity  3 , parity  4 , and so on. The replication mode flag  54  contains the coding target data which may be data, mirror  1  (i.e., primary copy data for duplication), mirror  2  (i.e., secondary copy data for triplication), etc. If the field contains data, first data is generated parity by other data, so that mirror data can access directory and original data are restored by parity and other data. If the field contains mirror  1 , then mirror data related object ID is generated parity by other data. The replication data layout table  50  contains replication data only when the coding mode is changed due to a lack of physical capacity in the distributed storage system. 
         [0044]      FIG. 6  shows an example of a data layout table  60 . The data layout table  60  contains columns of object ID  61 , HDD ID  62 , and location address  63 . The object ID  61  is a unique object data ID. The object data contains replication mirror data or parity data which is generated from other nodes. The HDD ID  62  and location address  63  contains the location in which the object data is stored. 
         [0045]      FIG. 7  shows an example of a replication data node table  70 . The replication data node table  70  contains columns of object ID  71  and destination node ID  72 . The object ID  61  is a unique object data ID identifying the object data for which the coding mode is changed. The destination node ID  72  contains the node ID identifying the node in which the replication data is stored. 
         [0046]      FIG. 8  is an overview of a normal read operation. In step (1), a host issues a read request to the proxy nodes. In step (2), a proxy node calculates the hash value and searches the data store node in which is stored the primary data or mirror data. This example is based on mirroring; triplication involves the same process flow but with steps (3)-(5) performed repeatedly. In step (3), the proxy node issues a read request to the data store node which has the primary data. If the data store node can access the primary data, the next step is step (5). If the data store node cannot access the primary data or the data store node has failed, the next step is step (4). In step (4), the proxy node issues a read request to the data store node which has the mirror data. If the data store node can access the mirror data, the read request is processed and the next step is step (5). If the proxy node cannot access the mirror data or the data store node has failed, the read request has failed. In step (5), the proxy node returns read data to the host. 
         [0047]      FIG. 9  is an overview of a normal write operation. In step (1), a host issues a write request to the proxy nodes. In step (2), a proxy node calculates the hash value and searches the data store node which is to store the primary data. The example is based on mirroring; triplication involves the same process flow but with step (3) performed repeatedly. The proxy node issues a write request to the data store node which is to store the primary data. If the data store node can store the primary data, the next step is step (3). If the data store node cannot store the primary data or the data store node has failed, then the proxy node recalculates the hash value to remove the failed data store node and retry step (2). In step (3), the data store node calculates the hash value for a replication data location. Then the data store node issues a write request to the data store node which is to store the mirror data. If the data store node cannot store the mirror data or the data store node has failed, then the proxy node recalculates the hash value to remove the failed data store node and retry step (3). If the data store node can store the mirror data, then the data store node which can store the mirror data returns a good response to the data store node which can store the primary data and the proxy node. 
         [0048]      FIG. 10  is an overview of the process of changing replication mode and generating parity between re-replicated data. In step (1), some data store nodes have failed and the failure monitor program in the proxy nodes updates the data store nodes table  40 . In this case, the physical capacity is lower than a threshold which can support storing mirror data to re-replicate data. As a result, the proxy nodes change the coding mode and broadcast the change to all data store nodes in the distributed storage system. In steps (2) and (2′), a data store node checks all mirror data. If the mirror data is stored to failure nodes or failure disk, then the data store node (source) calculates the consistency hash value and determines a specific data store node (destination) and the source data store node re-replicates data to the specific destination data store node which was determined by hash calculation. Then the source data store node updates the replication data node table  70 . In  FIG. 10 , step (2) shows re-replicating Rep Data  3  from data store node  3   a  (first source) to data store node  3   c  (destination) as Data  3  and step (2′) shows re-replicating Data  4  from data store node  3   n  (second source) to data store node  3   c  (destination) as Rep Data  4 . In step (3), if the destination data store node receives some re-replicated data from other data store nodes, then the destination data store node generates parity between the re-replicated data in those other data store nodes. In step (4), the destination data store node notifies the proxy nodes  2  of generating parity between the re-replicated data. Then the proxy nodes  2  update the replicated data layout table  50 . 
         [0049]      FIG. 11  is overview of the process of changing replication mode and generating parity between re-replicated data and existing replication data. In step (1), some data store nodes have failed and the failure monitor program in the proxy nodes updates the data store nodes table  40 . In this case, the physical capacity is lower than another threshold (even lower threshold than that in  FIG. 10 ) which can support storing parity generation for re-replicate data in  FIG. 10 . As a result, the proxy nodes change the coding mode and increase the target of generating parity for existing replicated data, and the proxy nodes broadcast the changing mode to all data store nodes in the distributed storage system. In step (2), a data store node checks all mirror data. If the mirror data is stored to failure nodes or failure disk, then the data store node (source) calculates the consistency hash value and determines a specific data store node (destination) and the source data store node  3   n  re-replicates data to the specific destination data store node  3   c  which was determined by hash calculation. Then the source data store nodes updates the replication data node table  70 . In step (3), if the destination data store node receives the re-replicated data and the destination data store node has existing replication data, then the destination data store node generates parity between the re-replicated data and existing replication data. In step (4), the destination data store node overwrites parity to the existing replication data. Then the destination data store node deletes the received re-replicated data from the source data storage node. In step (5), the destination data store node notifies the proxy nodes  2  of generating parity. The proxy nodes  2  update the replicated data layout table  50 . 
         [0050]      FIG. 12  is an overview of the process of changing replication mode and generating parity between re-replicated data involving a write operation. The process flow is similar to the write flow in  FIG. 9  and the generating parity flow in  FIG. 10 . Step (1) in  FIG. 12  is similar to step (1) in  FIG. 10 . Step (2) in  FIG. 12  is similar to step (1) in  FIG. 9 . Step (3) in  FIG. 12  is similar to step (2) in  FIG. 9 . Step (4) in  FIG. 12  is similar to step (3) in  FIG. 9 . Step (5) in  FIG. 12  is similar to step (2′) in  FIG. 10 . Step (6) in  FIG. 12  is similar to step (3) in  FIG. 10 . Step (7) in  FIG. 12  is similar to step (4) in  FIG. 10 . 
         [0051]      FIG. 13  is an overview of a read operation during a change in the coding mode. In step (1), some data store nodes have failed and the failure monitor program in the proxy nodes updates the data store nodes table  40 . In this case, the physical capacity is lower than a threshold which can support storing mirror data to re-replicate data. As a result, the proxy nodes change the coding mode and broadcast the change to all data store nodes in the distributed storage system. In step (2), a host issues a read request to the proxy nodes. In step (3), a proxy node calculates the hash value and searches the data store node in which is stored the primary data or mirror data. If the proxy node searches replication data layout table  50  and the primary data is generated parity, the proxy node calculates the hash value for replication data since at least one data or replication data is not generated parity. In step (4), the proxy node issues a read request to the data store node  3   a  which stores the data (Data  7 ). If the data store node can access the data, the next step is step (9). If the data store node cannot access the data or the data store node has failed, the next step is step (5). 
         [0052]    In step (5), the proxy node calculates the hash value and searches the data store node in which is stored the parity data. The proxy node issues a read request to the data store node which stores the parity data. In step (6), the data store node which stores the parity data (source) searches the replication data node table  70 , and then the source data store node  3   c  issues a read request to the destination data store node  3   n  which contains the parity restore data (Data  8 ). In step (7), the source data store node which stores the parity data restores recovery data (Recovery Data  7 ) between the received parity restore data from the destination data store node and the parity data (Rep Data  7  xor  8 ). In step (8), the source data store node returns the recovery data. In step (9), the proxy node returns the read data to the host. 
         [0053]      FIG. 14  shows an example of a flow diagram  1400  illustrating a read operation and parity recovery read. In step S 1401 , a host issues a read request to the proxy nodes. In step S 1402 , the proxy node searches the replication data layout table  50 . If the object ID associated with the read request exists in the table (YES), the next step is S 1407 . If No, the next step is S 1403 . The following is related to the normal read operation in  FIG. 8 . In step S 1403 , a proxy node calculates the hash value and searches the data store node in which is stored the primary data or mirror data. In step S 1404 , the proxy node issues a read request to the data store node which stores the data. If the data store node can access the data (YES), then the proxy node returns the read data to the host and the process ends. If the data store node cannot access the data or the data store node has failed (No), the next step is S 1405 . In step S 1405 , the proxy node calculates the hash value and searches the data store nodes in which is stored the replicated data. The proxy node issues a read request to the data store node which stores the replicated data. In step S 1406 , if the data store node can access the replicated data (YES), then the proxy node returns the read data to the host as recovery of failed data in S 1404  and the process ends. If the data store node cannot access the replicated data or the data store node has failed (No), the host read operation has failed and the process ends. 
         [0054]    The following is related to the read operation during a change in the coding mode of  FIG. 13 . In step S 1407 , the proxy node calculates the hash value and searches the data store node in which is stored non-parity data. The proxy node issues a read request to the data store node which stores the non-parity data. In step S 1408 , if the data store node can access the non-parity data (YES), then the proxy node returns the read data and the process ends. If the data store node cannot access the non-parity data or the data store node has failed (No), the next step is S 1409 . In step S 1409 , the proxy node calculates the hash value and searches the data store node in which is stored parity data. The proxy node issues a read request to the data store node which stores the parity data. In step S 1410 , the data store node which stores the parity data (source) searches the replication data node table  70 , and then the source data store node issues a read request to the destination data store node which contains parity restore data. In step S 1411 , the source data store node which stores the parity data restores recovery data between the received parity restore data from the destination data store node and the parity data. In step S 1412 , if the source data store node reads parity and reads restore data and generates recovery data (YES), then the proxy node returns the recovery data to the host as recovery of failed data in S 1408  and the process ends. If the data store node cannot access the recovery data or the data store node has failed (No), the host read operation has failed and the process ends. 
         [0055]      FIG. 15  shows an example of a flow diagram  1500  illustrating a process for host write operation. This is related to the normal write operation of  FIG. 9 . In step S 1501 , a host issues a write request to the proxy nodes. In step S 1502 , a proxy node calculates the hash value and searches the data store node which is to store the primary data. In step S 1503 , the proxy node issues a write request to the data store node which is to store the primary data. If the data store node can store primary data, the next step is S 1504 . If the data store node cannot store the primary data or the data store node has failed, then the proxy node recalculates the hash value to remove the failed data store node and retries step S 1502  or fails the host write operation. In step S 1504 , the data store node calculates the hash value for replication data location. In step S 1505 , the data store node issues a write request to the data store node which is to store the mirror data. If the data store node which is to store the mirror data cannot store the mirror data or it has failed, then the proxy node recalculates the hash value to remove the failed data store node and retries step S 1504 . If the data store node can store the mirror data, then it returns a good response to the data store node which is to store the primary data and the proxy node, and the process ends. 
         [0056]      FIG. 16  shows an example of a flow diagram  1600  illustrating a process for generating parity between re-replicated data or between re-replicated data and existing replicated data. In step S 1601 , a data store node checks all mirror data. If the mirror data is stored to failure nodes or failure disk, the data store node calculates the consistency hash value and determines a specific data store node and the data store node (source) re-replicates data to the specific data store node which was determined by hash calculation. Then the source data store node updates the replication data node table  70 . If the coding mode is parity (YES), the next step is S 1602 . If NO, then all data store nodes re-replicate data to other nodes and the process ends. In step S 1602 , the proxy node checks the capacity threshold to determine the coding mode. If the coding mode is parity with existing replication data mode (YES), the next step is S 1605 . If NO, the next step is S 1603 . 
         [0057]    The following is related to the process of changing replication mode and generating parity between re-replicated data of  FIG. 10 . In step S 1603 , the destination data store node receives some re-replicated data which is for other data store nodes. In step S 1604 , the destination data store node generates parity between the re-replicated data in other data store nodes. The next step is S 1609 . 
         [0058]    The following is related to the process of changing replication mode and generating parity between re-replicated data and existing replication data of  FIG. 11 . In step S 1605 , the destination data store node receives re-replicated data from other data store node(s). In step S 1606 , it is determined whether the destination data store node has existing replication data. If YES, the next step is S 1607 . If NO, the next step is S 1603 . In step S 1607 , the destination data store node generates parity between the re-replicated data and existing replication data. In step S 1608 , the destination data store node overwrites parity to the existing replication data. Then the destination data store node deletes the received re-replicated data from the source data storage node. In step S 1609 , the destination data store node notifies the proxy nodes of generating parity. The proxy nodes update the replicated data layout table  50 . 
         [0059]      FIG. 17  shows an example of a flow diagram  1700  illustrating a process for restoring parity data to normal mode. In step S 1701 , the failed data store node recovers or is replaced. The physical capacity in the distributed storage system exceeds a first threshold such that the coding mode can be changed to normal, with replication or triplication. In step S 1702 , all data store nodes read parity recovery data from other data store nodes based on the replication data node table  70 . In step S 1703 , if the data store node succeeds in reading recovery data, the next step is S 1706 . If the data store node failed to read recovery data, the next step is S 1704 . The approach is to read other original data. In step S 1704 , the data store nodes read other parity recovery data from other data store nodes based on the replication data node table  70 . In step S 1705 , if the data store node succeeds in reading other parity recovery data, the next step is S 1707 . If the data store node fails to read recovery data, the two data are lost and the recovery fails, and the process ends. On the other hand, if the data store node succeeds in reading recovery data in step S 1703 , a process of parity recovery is performed as follows. In step S 1706 , the data store node restores data between existing parity and read recovery data. The next step is S 1707 . If step S 1707  follows step S 1705 , the data store node stores all read data. If step S 1707  follows step S 1706 , the data store node store reads data and parity restore data. If the data store node failed to store data, then the operation has failed and the process ends. In step S 1708 , the data store node deletes the parity data. 
         [0060]      FIG. 18  shows an example of a flow diagram  1800  illustrating a process for changing replication mode. In step S 1801 , the failed data store node recovers or is replaced. The physical capacity in the distributed storage system is at or above the first threshold such that the coding mode can be changed to normal, with replication or triplication (see  FIG. 17 ). In another embodiment, the change to normal coding mode may require a physical capacity at or above a fourth threshold that is equal to or higher than the first threshold. In step S 1802 , some data store nodes have failed and the failure monitor program in the proxy nodes update the data store nodes table  40 . The physical capacity is lower than the first threshold which can support storing mirror data to re-replicate data. The proxy nodes change the coding mode and broadcast the change to all data store nodes in the distributed storage system (see  FIG. 10 or 12 ). In step S 1803 , more data store nodes have failed and the failure monitor program in the proxy nodes updates the data store nodes table  40 . The physical capacity is lower than the second threshold which can support storing parity generation for re-replicate data. As a result, the proxy nodes change the coding mode and increase the target of generating parity for existing replicated data, and the proxy nodes broadcast the changing mode to all data store nodes in the distributed storage system. In step S 1804 , more data store nodes have failed and the physical capacity is lower than an even lower threshold (e.g., a third threshold) than the second threshold. As a result, the proxy nodes change the coding mode and increase the target of generating parity to a higher parity. 
         [0061]    Of course, the system configuration illustrated in  FIG. 1  is purely exemplary of information systems in which the present invention may be implemented, and the invention is not limited to a particular hardware configuration. The computers and storage systems implementing the invention can also have known I/O devices (e.g., CD and DVD drives, floppy disk drives, hard drives, etc.) which can store and read the modules, programs and data structures used to implement the above-described invention. These modules, programs and data structures can be encoded on such computer-readable media. For example, the data structures of the invention can be stored on computer-readable media independently of one or more computer-readable media on which reside the programs used in the invention. 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 local area networks, wide area networks, e.g., the Internet, wireless networks, storage area networks, and the like. 
         [0062]    In the description, numerous details are set forth for purposes of explanation in order to provide a thorough understanding of the present invention. However, it will be apparent to one skilled in the art that not all of these specific details are required in order to practice the present invention. It is also noted that the invention may be described as a process, which is usually depicted as a flowchart, a flow diagram, a structure diagram, or a block diagram. Although a flowchart may describe the operations as a sequential process, many of the operations can be performed in parallel or concurrently. In addition, the order of the operations may be re-arranged. 
         [0063]    As is known in the art, the operations described above can be performed by hardware, software, or some combination of software and hardware. Various aspects of embodiments of the invention may be implemented using circuits and logic devices (hardware), while other aspects may be implemented using instructions stored on a machine-readable medium (software), which if executed by a processor, would cause the processor to perform a method to carry out embodiments of the invention. Furthermore, some embodiments of the invention may be performed solely in hardware, whereas other embodiments may be performed solely in software. Moreover, the various functions described can be performed in a single unit, or can be spread across a number of components in any number of ways. When performed by software, the methods may be executed by a processor, such as a general purpose computer, based on instructions stored on a computer-readable medium. If desired, the instructions can be stored on the medium in a compressed and/or encrypted format. 
         [0064]    From the foregoing, it will be apparent that the invention provides methods, apparatuses and programs stored on computer readable media for managing data recovery in a distributed storage system. Additionally, while specific embodiments have been illustrated and described in this specification, those of ordinary skill in the art appreciate that any arrangement that is calculated to achieve the same purpose may be substituted for the specific embodiments disclosed. This disclosure is intended to cover any and all adaptations or variations of the present invention, and it is to be understood that the terms used in the following claims should not be construed to limit the invention to the specific embodiments disclosed in the specification. Rather, the scope of the invention is to be determined entirely by the following claims, which are to be construed in accordance with the established doctrines of claim interpretation, along with the full range of equivalents to which such claims are entitled.