Cross-datacenter extension of grid encoded data storage systems

• US 10,108,819 B1
• Filed: 07/01/2015
• Issued: 10/23/2018
• Est. Priority Date: 07/01/2015
• Status: Expired due to Fees

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Donlan, Bryan James, Lazier, Colin Laird
• Primary Examiner(s): Chase, Shelly A
• Application Number:
  US14/789,789
• Time in Patent Office: 1,210 Days
• CPC Class Codes: G06F 11/1076 Parity data used in redundant arrays of independent storages, e.g. in RAID systems
• CPC Class Codes: G06F 12/16 Protection against loss of memory contents contains no material, see G06F11/00
• CPC Class Codes: G06F 16/00 Information retrieval; Database structures therefor; File system structures therefor
• CPC Class Codes: G06F 16/184 implemented as replicated file system
• CPC Class Codes: G06F 21/64 Protecting data integrity, e.g. using checksums, certificates or signatures
• CPC Class Codes: H03M 13/09 Error detection only, e.g. using cyclic redundancy check [CRC] codes or single parity bit
• CPC Class Codes: H03M 13/136 Reed-Muller [RM] codes
• CPC Class Codes: H03M 13/1505 Golay Codes
• CPC Class Codes: H03M 13/1515 Reed-Solomon codes
• CPC Class Codes: H03M 13/152 Bose-Chaudhuri-Hocquenghem [BCH] codes
• CPC Class Codes: H03M 13/19 Single error correction without using particular properties of the cyclic codes, e.g. Hamming codes, extended or generalised Hamming codes
• CPC Class Codes: H03M 13/373 with erasure correction and erasure determination, e.g. for packet loss recovery or setting of erasures for the decoding of Reed-Solomon codes
• CPC Class Codes: H03M 13/3761 using code combining, i.e. using combining of codeword portions which may have been transmitted separately, e.g. Digital Fountain codes, Raptor codes or Luby Transform [LT] codes
• CPC Class Codes: H04L 67/1095 Replication or mirroring of data, e.g. scheduling or transport for data synchronisation between network nodes
• CPC Class Codes: H04L 67/1097 for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]

Abstract

Techniques for extending a grid encoded data storage system to additional datacenters are described herein. A grid of shards with a first index and a second index is created and a set of null shards is added to the grid of shards. When a data object is received for storage in the grid of shards, a set of shards with the same first index is selected for the storage location with at least one null shard and one or more other shards. The null shard is enabled for data storage by allocating a storage device for the null shard. The grid is then updated by storing at least a portion of the data object in the set of shards, updating derived shards in the set of shards, and updating derived shards with the same second index as the updated shards.

Claims

1. A computer-implemented method, comprising:

generating a grid of shards, the grid of shards forming a partitioned data set, the grid of shards being indexed by row and column, the grid of shards comprising a set of data shards and a set of derived shards, the set of derived shards comprising a set of horizontally derived shards and a set of vertically-derived shards, wherein;

the partitioned data set includes one or more shards containing one or more portions of the partitioned data set and one or more shards generated by applying an erasure code to the one or more shards containing the one or more portions of the partitioned data set; and

each shard of the grid of shards has a corresponding row and corresponding column and is configured such that;

the shard is reproducible from other shards associated with the row and reproducible from other shards associated with the column;

if the shard is a horizontally-derived shard of the set of horizontally-derived shards, the shard is derived based at least in part on a set of data shards associated with the row; and

if the shard is a vertically-derived shard of the set of vertically derived shards, the shard is derived based at least in part on a set of shards associated with the column;

adding a new datacenter location to a set of datacenter locations by adding a set of null shards to the grid of shards, each null shard of the set of null shards having a corresponding row and a corresponding column, the corresponding column based at least in part on the new datacenter location; and

storing a data object at a storage location in the new datacenter location, the storage location selected based at least in part on a null shard of the set of null shards, the null shard having a corresponding first row and first column by at least;

converting the null shard to a data shard by allocating a storage device corresponding to the storage location;

storing the data object in the data shard using the allocated storage device;

updating a first set of derived shards associated with the first row, the first set of derived shards associated with the first row having a shard that is associated with a second column;

converting each null shard of a first set of null shards associated with the first column to a corresponding derived shard by allocating a corresponding set of storage devices;

updating a second set of derived shards associated with the first column; and

updating a third set of derived shards associated with the second column.

2. The computer-implemented method of claim 1, wherein the storage device corresponding to the storage location is selected from a set of storage devices based at least in part on the row and the column of the converted null shard.

3. The computer-implemented method of claim 1, wherein:

the storage device is selected from a set of storage devices based at least in part on a repair operation associated with the grid of shards; and

each storage device of the corresponding set of storage devices is selected from a set of storage devices based at least in part on the repair operation associated with the grid of shards.

4. The computer-implemented method of claim 1, wherein:

the storage device is selected from a set of storage devices based at least in part on an update operation associated with the grid of shards; and

each storage device of the corresponding set of storage devices is selected from a set of storage devices based at least in part on the update operation associated with the grid of shards.

5. A system, comprising at least one computing device configured to implement one or more services, wherein the one or more services are configured to:

generate a grid of shards, the grid of shards indexed at least by a first index and a second index, the grid of shards comprising a set of data shards and a set of derived shards, wherein each shard of the grid of shards has a corresponding first index, a corresponding second index, and a storage device that stores the shard, the storage device located in a datacenter location, each shard configured such that the shard is reproducible from other shards associated with the first index and the shard is reproducible from other shards associated with the second index;

add a set of null shards to the grid of shards, each null shard of the set of null shards having a corresponding first index and a corresponding second index, the corresponding second index based at least in part on a new datacenter location;

convert a first null shard of the set of null shards to a data shard by at least allocating a storage device selected based at least in part on the first null shard;

store at least a part of a data object in the storage device;

convert a subset of the set of null shards to a corresponding first subset of the set of derived shards by at least allocating a corresponding set of storage devices, the subset of the set of null shards selected based at least in part on the second index of the first null shard; and

update a second subset of the set of derived shards, the second subset of the set of derived shards selected based at least in part on the first index of the first null shard and based at least in part on the second index of the first null shard.

6. The system of claim 5, wherein the shard is reproducible from other shards associated with the first index using a first redundancy code and the shard is reproducible from other shards associated with the second index using a second redundancy code.

7. The system of claim 6, wherein the first redundancy code is different from the second redundancy code.

8. The system of claim 6, wherein each shard of the grid of shards has a corresponding set of grid metadata, the set of grid metadata at least including a set of constants associated with the second redundancy code and usable with the second redundancy code to reproduce the shard.

9. The system of claim 5, wherein the one or more services are further configured to at least:

select a shard of the grid of shards;

select a set of other shards based at least in part on the first index of the selected shard; and

reproduce the selected shard from a subset of the set of other shards.

10. The system of claim 5, wherein the one or more services are further configured to at least:

select a shard of the grid of shards;

select a set of other shards based at least in part on the second index of the selected shard; and

reproduce the selected shard from a subset of the set of other shards.

11. The system of claim 5, wherein the storage device is at least one of:

magnetic tape, magnetic disk, optical disk, a memory resistor, or computer memory.

12. The system of claim 5, wherein the storage device is selected from a set of storage devices based at least in part on a number of changes per unit time of the shard.

13. A non-transitory computer-readable storage medium having stored thereon executable instructions that, when executed by one or more processors of a computer system, cause the computer system to at least:

using a shard of a grid of shards, the grid of shards indexed by at least a first index and a second index, wherein each shard of the grid of shards has at least a corresponding first index and a corresponding second index, the grid of shards configured such that each shard of the grid of shards is reproducible from other shards associated with the first index and each shard of the grid of shards is reproducible from other shards associated with the second index, the grid of shards at least including a set of null shards,

receive a data object for storage in the grid of shards;

select a first set of shards, each shard of the first set of shard having the same first index, the first set of shards at least including a first null shard of the set of null shards and one or more other shards of the grid of shards;

convert the first null shard to a non-null shard by at least allocating a storage device selected based at least in part on the first null shard;

update a plurality of shards of the first set of shards based at least in part on storing at least a part of the data object in one or more data shards of the plurality of shards and updating one or more derived shards of the plurality of shards; and

update a subset of a set of derived shards, the set of the set of derived shards selected based at least in part on having a second index corresponding to a second index of one or more of the plurality of shards.

14. The non-transitory computer-readable storage medium of claim 13, wherein the converted null shard is a data shard.

15. The non-transitory computer-readable storage medium of claim 13, wherein the converted null shard is a derived shard.

16. The non-transitory computer-readable storage medium of claim 13, wherein:

the one or more derived shards of the plurality of shards are derived based at least in part on a first linear erasure code; and

each derived shard of the set of derived shards is derived based at least in part on a second linear erasure code.

17. The non-transitory computer-readable storage medium of claim 16, wherein:

the first linear erasure code is a parity code; and

the second linear erasure code is a Reed-Solomon code.

18. The non-transitory computer-readable storage medium of claim 16, wherein each shard of the grid of shards has a corresponding set of grid metadata, the set of grid metadata including a first set of constants associated with the first linear erasure code and a second set of constants associated with the second linear erasure code.

19. The non-transitory computer-readable storage medium of claim 18, wherein the set of grid metadata includes a third set of constants associated with a cyclic redundancy check usable to validate the grid of shards.

20. The non-transitory computer-readable storage medium of claim 13, wherein the corresponding second index of each shard of the grid of shards is based at least in part on a geographical location of the shard.