Map-Reduce Ready Distributed File System

• US 20120101991A1
• Filed: 12/29/2011
• Published: 04/26/2012
• Est. Priority Date: 06/19/2010
• Status: Active Grant

• Current Assignee: Hewlett Packard Enterprise Development LP (Hewlett-Packard Enterprise Company)
• Original Assignee: MapR Technologies, Inc. (Hewlett-Packard Enterprise Company)
• Inventors: SRIVAS, Mandayam C., Ravindra, Pindikura, Saradhi, Uppaluri Vijaya, Pande, Arvind Arun, Sanapala, Chandra Guru Kiran Babu, Renu, Lohit Vijaya, Vellanki, Vivekanand, Kavacheri, Sathya, Hadke, Amit
• Granted Patent: US 9,207,930 B2
• Time in Patent Office: Days
• US Class Current: 707/623
• CPC Class Codes: G06F 16/10 File systems; File servers
• CPC Class Codes: G06F 16/128 Details of file system snapshots on the file-level, e.g. snapshot creation, administration, deletion use of snapshots for error detection or correction G06F11/14, G06F11/16
• CPC Class Codes: G06F 16/178 Techniques for file synchronisation in file systems
• CPC Class Codes: G06F 16/182 Distributed file systems
• CPC Class Codes: G06F 16/1844 Management specifically adapted to replicated file systems
• CPC Class Codes: G06F 16/1865 Transactional file systems
• CPC Class Codes: G06F 16/22 Indexing; Data structures therefor; Storage structures
• CPC Class Codes: G06F 16/2246 Trees, e.g. B+trees
• CPC Class Codes: G06F 16/23 Updating
• CPC Class Codes: G06F 16/235 Update request formulation
• CPC Class Codes: G06F 16/2365 Ensuring data consistency and integrity
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
• CPC Class Codes: G06F 16/273 Asynchronous replication or reconciliation
• CPC Class Codes: G06F 16/275 Synchronous replication
• CPC Class Codes: G06F 8/658 Incremental updates; Differential updates
• CPC Class Codes: H04L 65/102 Gateways arrangements for connecting between networks having differing types of switching systems, e.g. gateways, H04L12/66

Abstract

A map-reduce compatible distributed file system that consists of successive component layers that each provide the basis on which the next layer is built provides transactional read-write-update semantics with file chunk replication and huge file-create rates. Containers provide the fundamental basis for data replication, relocation, and transactional updates. A container location database allows containers to be found among all file servers, as well as defining precedence among replicas of containers to organize transactional updates of container contents. Volumes facilitate control of data placement, creation of snapshots and mirrors, and retention of a variety of control and policy information. Also addressed is the use of distributed transactions in a map-reduce system; the use of local and distributed snapshots; replication, including techniques for reconciling the divergence of replicated data after a crash; and mirroring.

Claims

1. A map-reduce compatible distributed file system, comprising:

a container location database (CLDB) configured to maintain information about where each of a plurality of containers is located;

a plurality of cluster nodes, each cluster node containing one or more storage pools, each storage pool containing zero or more containers; and

a plurality of inodes for structuring data within said containers;

wherein said containers are replicated to other cluster nodes with one container designated as master for each replication chain;

wherein data in the CLDB is itself stored as inodes in well known containers;

wherein said CLDB nodes are configured to maintain a database that contains at least the following information about all of said containers;

nodes that have replicas of a container;

an ordering of a replication chain for each container;

wherein updates to a container are sent to a master for said updated container; and

wherein changes to content of a container are propagated to the replicas of the container by said master.

2. The system of claim 1, wherein said propagation is arranged in any of a linear pattern replication, a star pattern replication, or any combination of said replication patterns.

3. The system of claim 1, wherein said propagation is arranged in a linear replication pattern, in which a node holding said master of a container propagates updates to a first slave node that contains another replica;

and

wherein said first slave node, in turn, propagates updates to a second slave node that contains a third replica of the container.

4. The system of claim 1, wherein said propagation is arranged in a star replication pattern, in which a node containing said master of a container propagates updates directly and simultaneously to all other nodes containing replicas of said container.

5. The system of claim 1, wherein when an update is received by a node containing a master of a container, a lock is taken on a particular portion of a file being updated to guarantee that updates that change a same portion of said file are transactionally serialized;

and

wherein updates are made consistently on all replicas of said container.

6. The system of claim 1, wherein a number of update transactions that are allowed to be pending for a single file is limited to increase the amount of parallelism available for updates.

7. The system of claim 1, wherein updates to containers are acknowledged once enough of the nodes containing replicas have acknowledged that they have received the update;

8. The system of claim 1, wherein multiple updates are pending at a same time to increase potential parallelism, but all updates to a container are assigned a serial update identifier by the container master.

9. The system of claim 1, wherein updates are applied on different replicas in different orders, but only updates that have been acknowledged by all replicas are reported back to a client as committed.

10. The system of claim 1, wherein, before reporting an update as completed, all updates with smaller update identifiers are also completed.

11. The system of claim 1, wherein said CLDB stores a location of all replicas of a container, a structure of a replication for the container, and an epoch number for each container, wherein said epoch number is incremented each time the replication structure for a container is changed.

12. The system of claim 1, wherein potentially out-of-date replicas are resynchronized to a current state of the master of a container.

13. The system of claim 1, wherein the replication structure for a container is changed if a node holding any replica fails or is taken out of service, or if a node that previously contained a replica returns to service.

14. The system of claim 1, wherein when a node containing an out-of-date replica joins an existing replica chain, the node containing the out-of-date replica contacts the CLDB;

wherein the CLDB may decide that enough replicas are already known, in which case the node is instructed to discard the replica; and

wherein the CLDB may also assign the out-of-date replica to the replication structure, at which point a resynchronization process is executed to bring the out-of-date container up to a current state.

15. The system of claim 1, wherein when non-empty sets of nodes containing replicas of a container fail and a set of failing nodes includes a node containing the master, the CLDB detects the loss of contact with the failing nodes and notices that the failing nodes contain a previous master;

wherein a surviving node containing a replica, if any such exists, is designated as a new master and other nodes are assigned positions in the replication structure;

wherein an epoch number for the container is incremented at the same time that the new master is designated;

wherein the new master increments the transaction identifier to guarantee that a gap occurs in a sequence of transaction identifiers;

wherein the new master then records an end of a set of transactions that were handled by the previous master in a previous epoch and records a starting point for transaction identifiers in said new epoch; and

wherein all other replicas are considered to be out-of-date.

16. The system of claim 1, wherein when a set of failing nodes does not include a node containing the master, failing nodes are removed from the replication structure, but the master survives;

and

wherein all other replicas are considered to be out-of-date.

17. The system of claim 1, wherein when a number of nodes containing replicas of a container come into service at about the same time, said nodes contact the CLDB as they come up and report a last epoch that they have seen, as well as a last update identifier that they have recorded;

wherein until a configurable minimum number of nodes comes up, no updates are allowed and a master is not designated, although read operations may be allowed;

wherein once a minimum number of nodes have reported to the CLDB and a configurable delay has passed to allow any slow nodes to reappear, the CLDB selects one of the nodes as master and arranges the others in a replication structure;

wherein the node selected as master is the node that reports that it has seen the most recent epoch and transaction identifier; and

wherein all other replicas are considered to be out-of-date.

18. The system of claim 11, wherein an epoch'"'"'s changes are noted in a transaction history for each version of the container and gaps are inserted whenever the master is noted in a replication chain;

and

wherein said CLDB traces back through transactions that have been applied to each copy when examining the master and target copies of a same container to determine a point in a history of the two containers when they were identical.

19. The system of claim 18, wherein a set of committed transactions are run-length encoded;

wherein an on-disk state is updated only when a new epoch is begun as a new master is designated; and

wherein transactions are identified by the epoch in which they were performed, as well as by a transaction counter in the container.

20. The system of claim 18, wherein a target container is brought up to date relative to the master by undoing any changes made to a secondary version of the container because changes on the target may not all have been applied to the master.

21. The system of claim 20, wherein once a common ancestor transaction is identified, a secondary version of the container is scanned to find all blocks that have been modified since the common ancestor transaction;

wherein as these blocks are found, the target over-writes them by requesting these same blocks from the master;

wherein all blocks in the target container are left either in a state that they had as of the common ancestor state or in a later state present on the master version of the container; and

wherein all blocks are updated to a current state of the master version of the container.

22. The system of claim 21, wherein after a target has been brought back to a common ancestor state, the master is scanned to find all blocks that have been updated since the common ancestor transaction;

and

wherein said blocks are copied to the target copy of the container.

23. The system of claim 22, wherein a snapshot of the master container that is used as a source of blocks and said update are each repeated until the target copy is less than a configurable number of transactions behind the master.

24. The system of claim 23, wherein once a secondary copy of the container is a predetermined distance from the master, updates to the master are frozen while a final update is done to the secondary.

25. The system of claim 24, wherein when a copy process converges, a container identifier table in a storage pool containing the target is modified to swap identifiers for the target container and a clone of the target container is used for the copy operation;

and

wherein an original container is then discarded.

26. The system of claim 1, wherein chunked files are used to support a distributed file system in which some files are larger than a single container and in which some files are spread over a larger number of nodes than a set represented by a replication chain of a single container.

27. The system of claim 26, wherein a file chunk and an original inode are updated in a coordinated fashion.

28. The system of claim 26, wherein an original inode and multiple file chunks are updated together.

29. The system of claim 1, wherein a transaction is performed in a manner that is unaffected by failures:

wherein a master node writes data on a subsidiary node B to a transaction log;

wherein a transaction is written into an orphanage that reverses the effect of a write operation if a reference on a subsidiary node A is not found at a later time;

wherein said subsidiary node B returns a reference to the written data to the master node;

wherein said reference is sent to said subsidiary node A;

wherein when said data was created on node B, a background thread was started or a cleanup event was scheduled causing said node B to inspect said orphanage at a time substantially after the original write occurs;

wherein said orphanage entry causes node B to inspect node A or one of the replicas of node A to see if a reference to the data written on node B exists;

wherein if a reference does exist, then no action is taken;

wherein if a reference does not exist, said orphanage entry created in said transaction on node B is executed, which reverses the effect of the original writing of the data;

wherein if a reference on node A is never created, then new data on node B is never accessible, such that said reference and said data appear atomically or not at all.

30. The system of claim 1, further comprising:

a distributed transaction in the form of a snapshot of a file system volume, consisting of directories and files spread over a number of containers;

wherein all data and meta-data for a volume is organized into a single name container and zero or more data containers; and

wherein all cross-container references to data are segregated into the name container while keeping all of the data in data containers.

31. The system of claim 30, wherein all references from one data container to another data container are mediated by data structures in the name volume;

wherein a volume snapshot proceeds by creating a snapshot of the name container and then creating snapshots of the data containers.

32. The system of claim 31, wherein data structures that are inserted into the data containers are limited to references to them from data structures that name the container.

33. The system of claim 32, wherein a container location database locks out addition of new containers for a volume during creation of the snapshot.

34. The system of claim 32, wherein once snapshots of all containers in a volume are created, a table is created that maps identifiers of the original containers to identifiers of corresponding snapshots;

wherein said table is injected into at least the name container snapshot and, optionally, into all of the snapshot containers;

wherein said table is used to translate references to any of the original containers into references to the corresponding snapshot container; and

wherein as modifications are made to data in the original volume, changed disk blocks are written to fresh storage and modified data structures are copied to new locations because the copy on write bit is set, leaving data in the snapshot intact.

35. The system of claim 1, wherein disks in the distributed file system are combined by mirroring to provide higher level block addressable objects;

wherein in concatenation of two disks, all updates and accesses to blocks with addresses less than the size of a first component disk are addressed to that first disk while all other updates and accesses are directed to a second disk;

wherein with said mirroring updates go to all component disks and accesses go to any component disk chosen at random; and

wherein a result of any such combination is considered to be a disk in its own right.