Hierarchical chunking of objects in a distributed storage system
Summary by NHIP
Journal-based replica placement
The method splits objects into chunks and blocks, storing them in journals matching specific placement policies. Global metadata updates upon journal replication to a second instance, while local metadata remains unchanged.
Claim Score by NHIP
Abstract
Placement of object replicas in a distributed storage system includes, at a first instance, opening a journal for storage of object chunks. Each journal is associated with a single placement policy. An object is received, which comprises a chunk. The object has a placement policy, and the chunk comprises a plurality of storage blocks. The blocks are stored in a journal that matches the placement policy. Global metadata for the object is stored, which includes a list of chunks for the object. Local metadata for the chunk is stored, which includes a block list identifying each block of the plurality of blocks. The local metadata is associated with the journal. The journal is later closed. The journal is subsequently replicated to a second instance according to the placement policy. The global metadata is updated to reflect the replication, whereas the local metadata is unchanged by the replication.

Term
7.5 yearsleft in the term
Expires 9 April 2034, including 103 days of term adjustment.
- Priority
- Filed
- Granted
- Today
- Expires
20 claims: 3 independent, 17 dependent
- 1Broadest claimClaim Score 36, narrow(NHIP)A method for managing placement of object replicas in a distributed storage system, comprising:at a first instance of the distributed storage system, having one or more processors and memory, wherein the memory stores one or more programs for execution by the one or more processors: receiving a first object that is associated with a first placement policy, wherein the first placement policy specifies criteria for where replicas of the first object are stored in the distributed storage system;splitting the object into a plurality of object chunks and splitting a first object chunk of the plurality of object chunks into a plurality of blocks;storing the plurality of blocks in a first journal whose associated placement policy matches the first placement policy;storing global metadata for the first object, wherein the global metadata includes a list of the plurality of object chunks, and wherein the list includes a respective identifier for each of the object chunks;storing local metadata for the first object chunk, wherein the local metadata includes a block list identifying each block of the plurality of blocks, and wherein the local metadata is associated with the first journal;replicating the first journal to a second instance of the distributed storage system in accordance with the first placement policy, wherein the global metadata is updated to reflect the replication, whereas the local metadata is unchanged by the replication.
- 7A method for managing placement of object replicas in a distributed storage system, comprising:at a first instance of the distributed storage system, having one or more processors and memory, wherein the memory stores one or more programs for execution by the one or more processors: opening one or more journals for storage of object chunks, wherein each respective journal is associated with a single respective placement policy;receiving a first object comprising at least a first object chunk, wherein the first object is associated with a first placement policy, and wherein the first object chunk comprises a first plurality of blocks;storing the first plurality of blocks in a first journal whose associated placement policy matches the first placement policy, wherein the first journal stores only blocks for objects whose placement policies match the first placement policy;storing global metadata for the first object, wherein the global metadata includes a first list of object chunks corresponding to the first object, and wherein the first list includes an identifier of the first object chunk;storing local metadata for the first object chunk, wherein the local metadata includes a block list identifying each block of the first plurality of blocks, and wherein the local metadata is associated with the first journal;for the first journal, repeating the receiving and storing operations for a first plurality of objects whose associated placement policies match the first placement policy, until a first termination condition occurs;after the first termination condition occurs, closing the first journal, thereby preventing any additional blocks from being stored in the first journal;and replicating the first journal to a second instance of the distributed storage system in accordance with the first placement policy, wherein the global metadata is updated to reflect the replication, whereas the local metadata is unchanged by the replication.
- 18A computer system for managing placement of object replicas in a distributed storage system having a plurality of instances, each respective instance comprising:one or more processors;memory;and one or more programs stored in the memory, the one or more programs comprising instructions executable by the one or more processors for: opening one or more journals for storage of object chunks, wherein each respective journal is associated with a single respective placement policy;receiving a first object comprising at least a first object chunk, wherein the first object is associated with a first placement policy, and wherein the first object chunk comprises a first plurality of blocks;storing the first plurality of blocks in a first journal whose associated placement policy matches the first placement policy, wherein the first journal stores only blocks for objects whose placement policies match the first placement policy;storing global metadata for the first object, wherein the global metadata includes a first list of object chunks corresponding to the first object, and wherein the first list includes an identifier of the first object chunk;storing local metadata for the first object chunk, wherein the local metadata includes a block list identifying each block of the first plurality of blocks, and wherein the local metadata is associated with the first journal;for the first journal, repeating the receiving and storing operations for a first plurality of objects whose associated placement policies match the first placement policy, until a first termination condition occurs;after the first termination condition occurs, closing the first journal, thereby preventing any additional blocks from being stored in the first journal;and replicating the first journal to a second instance of the distributed storage system in accordance with the first placement policy, wherein the global metadata is updated to reflect the replication, whereas the local metadata is unchanged by the replication.
Independent claims3
109 paragraphs in 6 sections, as filed
CROSS REFERENCE TO RELATED APPLICATIONS
This application is related to U.S. patent application Ser. No. 13/927,010, filed Jun. 25, 2013, entitled “Grouping of Objects in a Distributed Storage System Based on Journals and Placement Policies,” which is hereby incorporated by reference in its entirety.
TECHNICAL FIELD
The disclosed implementations relate generally to distributed storage systems, and more specifically to splitting objects into chunks and storing the chunks hierarchically.
BACKGROUND
Large-scale data storage has shifted from a central-service architecture to distributed storage systems. Distributed storage systems built from commodity computers can deliver high performance, availability, and scalability at a fraction of the cost compared to monolithic disk arrays. Data is replicated across multiple instances of the distributed storage system at different geographical locations, thereby increasing availability and reducing network distance from clients.
In a distributed storage system, objects are dynamically placed in (i.e., created in, deleted from, and/or moved to) various instances of the distributed storage system based on constraints. There are few existing techniques for efficiently placing objects that are subject to constraints in a planet-wide distributed storage system that stores trillions of objects and petabytes of data, and includes dozens of data centers across the planet.
New visualization, multimedia, and other data-intensive applications use very large objects, which may be hundreds of gigabytes or larger. Managing such very large objects create additional complexities for a distributed storage system. First, uploading such an object into a distributed storage system is typically done in a streaming mode, splitting the object into chunks and writing each chunk individually. This can impose long delays for uploading, which is exacerbated by potential client and server failures. Moreover, chunks may be aggregated into larger shards for better operational efficiency. The terms “shard” and “journal” may be used interchangeably herein. Consequently, efficient upload of large objects is becoming increasingly important for the storage industry being driven by the needs of large-scale systems that allow clients to connect to any cluster available at a time. In addition, the volume of metadata for a single object (e.g., 25000 chunks for a 100 gigabyte file, where each chunk is 4 megabytes) makes replication and compaction less efficient.
SUMMARY
Disclosed implementations distribute the upload of large objects to multiple storage locations simultaneously. As used herein, the storage locations are referred to as “shards,” “aggregated shards,” or “journals.” This scheme is implemented by splitting a large object into multiple chunks, each of which can be uploaded to a different storage cluster (which may be in distinct geographic locations). If a shard becomes unavailable during upload, (e.g., because the shard is “full” or the instance where the shard is stored goes down), the client switches to a new shard, which may be in a different cluster. This scheme does not require sticking with the same shard once started. A finalized object is represented by an ordered list of chunk references.
In some schemes, a chunk is the basic unit of storage, and the location of each chunk is stored in global metadata. For very large objects, this scheme results in a significant amount of metadata stored at the global level for a single object. Therefore, some implementations use a hierarchical chunking scheme, which reduces the amount of global metadata stored for each object. Within a hierarchical implementation, the term “chunk” is used to identify a top level split, which has corresponding metadata stored at the global level. In these implementations, the term “block” is used to identify a basic unit of actual storage (e.g., 2 megabytes or 8 megabytes). The blocks are managed locally for each shard. In a non-hierarchical system the single term “chunk” may be used to identify both concepts because the basic unit of storage is the basic unit for which global metadata is stored.
Hierarchical chunking can be implemented in multiple ways. In some implementations, each chunk comprises a list of blocks, even when there is only one block. In these implementations, there is always an additional hierarchical level for lookup of data corresponding to a chunk. Other implementations use a hybrid scheme so that there is a hierarchy only when needed for large chunks. In such a hybrid implementation, small objects may comprise a single chunk, which corresponds to a single block. On the other hand, for larger objects, each chunk is a list of blocks.
The disclosed hierarchical schemes reduce the amount of global metadata, which reduces the cost of managing objects or moving objects from one storage cluster to another. Whereas the object chunks are managed at the global level, the blocks within a chunk are managed at the local shard level so that the object metadata typically contains just one chunk reference per shard.
In some implementations, the upload process follows these steps: (1) find an available shard for upload; (2) write data to the current shard until either the shard is unavailable (e.g., full) or there is no more data; (3) add the current chunk reference to the ordered list of chunks; (4) if the object upload is done, finalize the object; otherwise (5) repeat starting at step (1) for the remainder of the object.
In some implementations, reading an object from storage follows these steps: (1) for the desired object, find the set of chunk references (there is always at least one); (2) find the location of the shard based on the chunk reference; (3) read the data from the shard location(s) using the chunk identifier and the local shard metadata; (4) repeat steps 2 and 3 for each chunk reference.
For example, suppose an object upload started by writing data to shard1, and switched to shard2 when shard1 became full. (The two shards shard1 and shard2 may be at the same or different instances.) The object metadata (which is global) consists of two chunk references, whereas each shard manages an local list of blocks for each chunk. For example, each shard could store a plurality of blocks for the object. In this case, the storage is completely hierarchical: the object is split into chunks, and each chunk is split into blocks. In other implementations, one of the chunks may be split into a plurality of blocks (such a chunk is sometimes referred to as a “superchunk”), whereas another chunk may consist of a single block. In the latter case, the chunk identifier may be a block identifier.
Because shard1 and shard2 are independent of each other, their replicas may be stored at different instances. For example, shard1 may be stored at instance1 and instance2, whereas shard2 may be stored at instance1 and instance3.
This disclosed methodology substantially improves both upload service availability and storage efficiency. This methodology supports resumable uploads (e.g., when an instance goes down during upload of a large object) as well as switching to a new shard in the middle of upload (e.g., when a shard becomes full). In addition, this methodology supports writing to multiple shards simultaneously, which may improve performance significantly for very large objects. In some implementations, data for a single object may be written to two or more distinct shards at distinct instances simultaneously, two or more shards at the same instance simultaneously, and even within a single journal, two or more process threads may write distinct blocks of data to the single journal simultaneously. Of course, a distributed upload is limited by available resources. The distributed storage system has many different clients uploading objects at the same time, so a single very large object from one client is not permitted to consume too much of the available resources.
According to some implementations, a method for managing placement of object replicas in a distributed storage system is performed at a first instance of the distributed storage system. The first instance has one or more servers, each having one or more processors and memory. The memory stores one or more programs for execution by the one or more processors. The first instance receives a first object that is associated with a first placement policy. The first placement policy specifies criteria for where replicas of the first object are stored in the distributed storage system. In some implementations, each placement policy specifies a target number of object replicas and target locations for those replicas. The first instance splits the object into a plurality of object chunks and splits a first object chunk of the plurality of object chunks into a plurality of blocks. The first instance stores the plurality of blocks in a first journal whose associated placement policy matches the first placement policy. The first instance stores global metadata for the first object, which includes a list of the plurality of object chunks. The list includes a respective identifier for each of the object chunks. The first instance stores local metadata for the first object chunk, which includes a block list identifying each block of the plurality of blocks. The local metadata is associated with the first journal. The first journal is subsequently replicated to a second instance of the distributed storage system in accordance with the first placement policy. The global metadata is updated to reflect the replication, whereas the local metadata is unchanged by the replication.
According to some implementations, a method for managing placement of object replicas in a distributed storage system is performed at a first instance of the distributed storage system. The first instance has one or more servers, each having one or more processors and memory. The memory stores one or more programs for execution by the one or more processors. One or more journals are opened for storage of object chunks. Each journal is associated with a single placement policy. In some implementations, each placement policy specifies a target number of object replicas and target locations for those replicas. The first instance receives a first object comprising at least a first object chunk. The first object is associated with a first placement policy. The first object chunk comprises a first plurality of blocks. The first instance stores the first plurality of blocks in a first journal whose associated placement policy matches the first placement policy. The first journal stores only blocks for objects whose placement policies match the first placement policy. The first instance stores global metadata for the first object, which includes a first list of object chunks corresponding to the first object. The first list includes an identifier of the first object chunk. The first instance also stores local metadata for the first object chunk, which includes a block list identifying each block of the first plurality of blocks. The local metadata is associated with the first journal. For the first journal, the receiving and storing operations are repeated for a first plurality of objects whose associated placement policies match the first placement policy, until a first termination condition occurs. In some implementations, the first termination condition occurs after a predefined span of time or after the first journal has exceeded a predefined size threshold. After the first termination condition occurs, the first journal is closed, thereby preventing any additional blocks from being stored in the first journal. Subsequently, the first journal is replicated to a second instance of the distributed storage system in accordance with the first placement policy. The global metadata is updated to reflect the replication, whereas the local metadata is unchanged by the replication.
BRIEF DESCRIPTION OF THE DRAWINGS
<figref idref="DRAWINGS">FIG. 1</figref> is a conceptual illustration of a distributed storage system, according to some implementations.
<figref idref="DRAWINGS">FIG. 2</figref> is a block diagram illustrating the elements of a distributed storage system, according to some implementations.
<figref idref="DRAWINGS">FIG. 3</figref> is a block diagram of a server, according to some implementations.
<figref idref="DRAWINGS">FIG. 4</figref> is a block diagram of an instance server, according to some implementations.
<figref idref="DRAWINGS">FIG. 5</figref> illustrates the use of journals for storage of object chunks in accordance with some implementations.
<figref idref="DRAWINGS">FIG. 6</figref> illustrates how some implementations manage the storage of a new object.
<figref idref="DRAWINGS">FIG. 7</figref> illustrates the structure of an open journal in accordance with some implementations.
<figref idref="DRAWINGS">FIG. 8</figref> illustrates what happens to object metadata and journal metadata when a journal is replicated from one instance to another, in accordance with some implementations.
<figref idref="DRAWINGS">FIGS. 9A-9C</figref> illustrate a method of managing placement of object replicas in a distributed storage system according to some implementations.
<figref idref="DRAWINGS">FIGS. 10A and 10B</figref> illustrate how object chunks may be further split into blocks in accordance with some implementations.
<figref idref="DRAWINGS">FIGS. 11A-11D</figref> illustrate an alternative method of managing placement of object replicas in a distributed storage system according to some implementations.
<figref idref="DRAWINGS">FIG. 12</figref> illustrates storage for chunks in a partially hierarchical distributed storage system in accordance with some implementations.
Like reference numerals refer to corresponding parts throughout the drawings.
DESCRIPTION OF IMPLEMENTATIONS
Before discussing techniques for managing the placement of objects in a distributed storage system, it is instructive to present an exemplary system in which these techniques may be used.
Distributed Storage System Overview
As illustrated in <figref idref="DRAWINGS">FIG. 1</figref>, the disclosed implementations describe a distributed storage system. There are multiple instances <b>102</b>-<b>1</b>, <b>102</b>-<b>2</b>, . . . <b>102</b>-N at various locations on the Earth <b>100</b>, connected by network communication links <b>104</b>-<b>1</b>, <b>104</b>-<b>2</b>, . . . <b>104</b>-M. Note that an “instance” is also referred to as a “storage location” in this specification. Also note that one or more instances (storage locations) may be located at a particular physical location (e.g., a building, a set of buildings within a predetermined distance of each other, etc.). In some implementations, an instance (such as instance <b>102</b>-<b>1</b>) corresponds to a data center. In some implementations, multiple instances are physically located at the same data center. A single implementation may have both individual instances at distinct geographic locations as well as one or more clusters of instances, where each cluster includes a plurality of instances, and the instances within each cluster are at a single geographic location.
Although the conceptual diagram of <figref idref="DRAWINGS">FIG. 1</figref> illustrates a particular number of network communication links <b>104</b>-<b>1</b>, etc., typical implementations may have more or fewer network communication links. In some implementations, there are two or more network communication links between the same pair of instances. For example, the network communication links <b>104</b>-<b>5</b> and <b>104</b>-<b>6</b> provide network connectivity between instance <b>102</b>-<b>2</b> and instance <b>102</b>-<b>6</b>. In some implementations, the network communication links include fiber optic cable. In some implementations, some of the network communication links use wireless technology, such as microwaves. In some implementations, each network communication link has a specified bandwidth and/or a specified cost for the use of that bandwidth. In some implementations, statistics are maintained about the transfer of data across one or more of the network communication links, including throughput rate, times of availability, reliability of the links, etc. Each instance typically has data stores and associated databases, and utilizes a farm of server computers (“instance servers” as illustrated in <figref idref="DRAWINGS">FIG. 4</figref>) to perform all of the tasks. In some implementations, one or more instances of the distribute storage system has limited functionality. For example, the limited functionality may include acting as a repeater for data transmissions between other instances. Note that limited functionality instances may or may not include any of the data stores.
<figref idref="DRAWINGS">FIG. 2</figref> is a block diagram illustrating the elements of a distributed storage system <b>200</b>, according to some implementations. The distributed storage system <b>200</b> includes instances <b>102</b>-<b>1</b>, <b>102</b>-<b>2</b>, <b>102</b>-<b>3</b>, <b>102</b>-<b>4</b>, . . . <b>102</b>-N. A respective instance <b>102</b>-<b>1</b> includes a replication module <b>220</b> that replicates object chunks <b>238</b> between instances. In some implementations, the object chunks <b>238</b> are stored in data stores <b>224</b> of the respective instance <b>102</b>-<b>1</b>. Each object chunk <b>238</b> comprises an object <b>226</b> or a portion of an object <b>226</b>, as illustrated in <figref idref="DRAWINGS">FIG. 6</figref>. The data stores <b>224</b> may include distributed databases, file systems, tape backups, and any other type of storage system or device capable of storing objects. In some implementations, the replication module <b>220</b> uses one or more replication queues <b>222</b>-<b>1</b>, <b>222</b>-<b>2</b>, . . . , <b>222</b>-L to replicate objects <b>226</b> or journals <b>230</b>. Replication requests for objects or journals to be replicated are placed in a replication queue <b>222</b>, and the objects or journals are replicated when resources (e.g., bandwidth) are available. In some implementations, replication requests in a replication queue <b>222</b> have assigned priorities, and the highest priority replication requests are replicated as bandwidth becomes available.
In some implementations, a background replication process creates and deletes copies of objects or journals based on placement policies <b>212</b> and access data <b>210</b> and/or a global state <b>211</b> provided by a statistics server <b>208</b>. The placement policies <b>212</b> specify how many copies of an object are desired, where the copies should reside, and in what types of data stores the data should be saved. Using placement policies <b>212</b>, together with the access data <b>210</b> (e.g., data regarding storage locations at which replicas of objects were accessed, times at which replicas of objects were accessed at storage locations, frequency of the accesses of objects at the storage locations, etc.) and/or the global state <b>211</b> provided by the statistics server <b>208</b>, a location assignment daemon (LAD) <b>206</b> determines where to create new copies of an object or journal and what copies may be deleted. When new copies are to be created, replication requests are inserted into a replication queue <b>222</b>. In some implementations, the LAD <b>206</b> manages replicas of objects or journals globally for the distributed storage system <b>200</b>. In other words, there is only one LAD <b>206</b> in the distributed storage system <b>200</b>. The use of the placement policies <b>212</b> and the operation of a LAD <b>206</b> are described in more detail below.
Note that in general, a respective placement policy <b>212</b> may specify the number of replicas of an object to save, in what types of data stores the replicas should be saved, storage locations where the copies should be saved, etc. In some implementations, a respective placement policy <b>212</b> for an object includes criteria selected from the group consisting of a minimum number of replicas of the object that must be present in the distributed storage system, a maximum number of the replicas of the object that are allowed to be present in the distributed storage system, storage device types on which the replicas of the object are to be stored, locations at which the replicas of the object may be stored, locations at which the replicas of the object may not be stored, and a range of ages for the object during which the placement policy for the object applies. For example, a first placement policy may specify that each object in a webmail application must have a minimum of 2 replicas and a maximum of 5 replicas, wherein the replicas of the objects can be stored in data centers outside of China, and wherein at least 1 replica of each object must be stored on tape. A second placement policy for the webmail application may also specify that for objects older than 30 days, a minimum of 1 replica and a maximum of 3 replicas are stored in the distributed storage system <b>200</b>, wherein the replicas of the objects can be stored in data centers outside of China, and wherein at least 1 replica of each object must be stored on tape.
In some implementations, a user <b>240</b> interacts with a user system <b>242</b>, which may be a computer system or other device that can run a web browser <b>244</b>. A user application <b>246</b> runs in the web browser, and uses functionality provided by database client <b>248</b> to access data stored in the distributed storage system <b>200</b> using a network. The network may be the Internet, a local area network (LAN), a wide area network (WAN), a wireless network (WiFi), a local intranet, or any combination of these. In some implementations, the database client <b>248</b> uses information in a global configuration store <b>204</b> to identify an appropriate instance to respond to the request. In some implementations, user application <b>246</b> runs on the user system <b>242</b> without a web browser <b>244</b>. Exemplary user applications include an email application and an online video application.
In some implementations, each instance stores object metadata <b>228</b> for each of the objects stored in the distributed storage system. Some instances store object metadata <b>228</b> only for the objects that have replicas stored at the instance (referred to as a “local instances”). Some instances store object metadata <b>228</b> for all objects stored anywhere in the distributed storage system (referred to as “global instances”). The object metadata <b>228</b> is described in more detail with respect to <figref idref="DRAWINGS">FIGS. 3</figref>, <b>4</b>, and <b>5</b>.
In some implementations, each instance stores journal metadata <b>236</b> for each of the journals stored in the distributed storage system <b>200</b>. Some instances store journal metadata <b>236</b> only for the journals that have replicas stored at the instance. Some instances store journal metadata for all journals stored anywhere in the distributed storage system. The journal metadata is described in more detail below with respect to <figref idref="DRAWINGS">FIGS. 3</figref>, <b>4</b>, <b>5</b>, and <b>8</b>.
Stored in the data stores <b>224</b> are multiple types of journals. The majority of the journals are closed journals <b>230</b>. Closed journals <b>230</b> do not store any additional object chunks, but can have content deleted and compacted. In some implementations, two or more small closed journals <b>230</b> for the same placement policy <b>212</b> can be stitched together to form a single replacement closed journal <b>230</b>. Because data within a closed journal <b>230</b> can be deleted and compacted, closed journals <b>230</b> can get smaller over time, and thus become candidates for stitching.
In addition to the closed journals <b>230</b>, an instance <b>102</b> can have open journals <b>232</b> and <b>234</b>. As indicated in <figref idref="DRAWINGS">FIG. 2</figref>, open journals are designated as either primary journals <b>232</b> or secondary journals <b>234</b>. Primary journals <b>232</b> and secondary journals <b>234</b> come in pairs, and are located at distinct instances. As described in more detail below, a primary journal <b>232</b> receives a chunk <b>238</b> for storage and transmits a copy of the chunk <b>238</b> to the instance where the corresponding secondary journal <b>234</b> is stored.
<figref idref="DRAWINGS">FIG. 3</figref> is a block diagram of a server <b>300</b>, according to some implementations. The server <b>300</b> typically includes one or more processing units (CPU's) <b>302</b>, a clock <b>303</b> that reports the current date and/or time, one or more network or other communications interfaces <b>304</b>, memory <b>314</b>, and one or more communication buses <b>312</b> for interconnecting these components. The communication buses <b>312</b> may include circuitry (sometimes called a chipset) that interconnects and controls communications between system components. In some implementations, the clock <b>303</b> is a local clock that is periodically synchronized with a clock server (e.g., a quorum clock server or any other clock server on a network, etc.). The server <b>300</b> optionally may include a user interface <b>306</b> comprising a display device <b>308</b> and input devices <b>310</b> (e.g., keyboard, mouse, touch screen, keypads, etc.). Memory <b>314</b> includes high-speed random access memory, such as DRAM, SRAM, DDR RAM or other random access solid state memory devices; and may include non-volatile memory, such as one or more magnetic disk storage devices, optical disk storage devices, flash memory devices, or other non-volatile solid state storage devices. Memory <b>314</b> may optionally include one or more storage devices remotely located from the CPU(s) <b>302</b>. Memory <b>314</b>, or alternately the non-volatile memory device(s) within memory <b>314</b>, comprises a computer readable storage medium. In some implementations, memory <b>314</b> stores the following programs, modules and data structures, or a subset thereof: <ul id="ul0001" list-style="none"><li id="ul0001-0001" num="0000"><ul id="ul0002" list-style="none"><li id="ul0002-0001" num="0042">an operating system <b>316</b> that includes procedures for handling various basic system services and for performing hardware dependent tasks;</li><li id="ul0002-0002" num="0043">a communication module <b>318</b> that is used for connecting the server <b>300</b> to other computers via the one or more communication interfaces <b>304</b> (wired or wireless) and one or more communication networks, such as the Internet, other wide area networks, local area networks, metropolitan area networks, and so on;</li><li id="ul0002-0003" num="0044">an optional user interface module <b>320</b> that receives commands from the user via the input devices <b>310</b> and generates user interface objects in the display device <b>308</b>;</li><li id="ul0002-0004" num="0045">the configuration <b>204</b>, as described herein;</li><li id="ul0002-0005" num="0046">the LAD <b>206</b>, as described herein;</li><li id="ul0002-0006" num="0047">access data <b>210</b>, as described herein;</li><li id="ul0002-0007" num="0048">the global state <b>211</b>, as described herein;</li><li id="ul0002-0008" num="0049">the placement policies <b>212</b>, as described herein;</li><li id="ul0002-0009" num="0050">object metadata <b>228</b> for the objects stored in the distributed storage system. The object metadata <b>228</b> may include an object ID <b>330</b>, which uniquely identifies the object within the distributed storage system. The metadata <b>228</b> may include the author <b>332</b> of the object, which may be a name and/or identifier of a person or entity (e.g., email address). In some implementations, the identifier is unique. The metadata may include a datestamp or timestamp <b>334</b> when the object was created (e.g., uploaded to the distributed storage system). The metadata may include the size <b>336</b> of the object, which is typically measured in bytes or allocation blocks. The metadata includes an assigned placement policy <b>338</b>, which may be assigned individually or based on other criteria (e.g., all videos uploaded from the United States may have the same assigned placement policy <b>338</b>). The usage of placement policies is described in more detail below with respect to <figref idref="DRAWINGS">FIGS. 5-6</figref> and <b>9</b>A-<b>9</b>C. The metadata <b>228</b> includes a set of chunk ID's <b>346</b> that identify the content chunks for each object. In some implementations, a chunk ID is specified as an offset within an object. For example, the first chunk has an offset of 0. In some implementations, the offsets are specified in megabytes. In some implementations, the chunk ID's are unique identifiers (such as a GUID). In some implementations, each chunk ID is formed by concatenating the object ID with the offset of the chunk. In some implementations, the chunk ID if formed using a content hash or content digest. Corresponding to each chunk ID is an assigned journal ID <b>348</b>, which indicates in which journal the corresponding chunk is stored; and</li><li id="ul0002-0010" num="0051">journal metadata <b>236</b> for each journal stored in the distributed storage system <b>200</b>. The journal metadata <b>236</b> includes a journal ID <b>370</b> for each journal and a set of journal locations <b>372</b> where the journal is stored. The journal locations <b>372</b> specify each instance <b>102</b> where the journal is stored any may specify the data store <b>224</b> at the instance <b>102</b> that stores the journal. The journal metadata <b>236</b> also includes the placement policy ID <b>374</b> associated with each journal. The placement policy ID <b>374</b> identifies the unique placement policy <b>212</b> associated with the journal.</li></ul></li></ul>
Each of the above identified elements may be stored in one or more of the previously mentioned memory devices, and corresponds to a set of instructions for performing a function described above. The set of instructions can be executed by one or more processors (e.g., the CPUs <b>302</b>). The above identified modules or programs (i.e., sets of instructions) need not be implemented as separate software programs, procedures or modules, and thus various subsets of these modules may be combined or otherwise re-arranged in various implementations. In some implementations, memory <b>314</b> may store a subset of the modules and data structures identified above. Furthermore, memory <b>314</b> may store additional modules and data structures not described above.
Although <figref idref="DRAWINGS">FIG. 3</figref> shows a “server,” <figref idref="DRAWINGS">FIG. 3</figref> is intended more as functional description of the various features that may be present in a set of servers <b>300</b> than as a structural schematic of the implementations described herein. In practice, and as recognized by those of ordinary skill in the art, items shown separately could be combined and some items could be separated. For example, some items shown separately in <figref idref="DRAWINGS">FIG. 3</figref> could be implemented on single servers and single items could be implemented by one or more servers. The actual number of servers and how features are allocated among them will vary from one implementation to another, and may depend in part on the amount of data traffic that the system must handle during peak usage periods as well as during average usage periods. In some implementations, a subset of the LAD <b>206</b>, the access data <b>210</b>, the global state <b>211</b>, and the placement policies <b>212</b> are located on separate servers. For example, the LAD <b>206</b> may be located at a server (or set of servers), the access data <b>210</b> and the global state <b>211</b> may be located and maintained by a statistics server <b>208</b> (or a set of statistics servers <b>208</b>), and the placement policies <b>212</b> may be located on another server (or a set of other servers).
<figref idref="DRAWINGS">FIG. 4</figref> is a block diagram of an instance server <b>400</b> for an instance <b>102</b>, according to some implementations. The instance server <b>400</b> typically includes one or more processing units (CPU's) <b>402</b> for executing modules, a clock <b>403</b> that reports the current date and/or time, programs and/or instructions stored in memory <b>414</b> and thereby performing processing operations, one or more network or other communications interfaces <b>404</b>, memory <b>414</b>, and one or more communication buses <b>412</b> for interconnecting these components. In some implementations, the clock <b>403</b> is a local clock that is periodically synchronized with a clock server (e.g., a quorum clock server or any other clock server on a network, etc.). In some implementations, the instance server <b>400</b> includes a user interface <b>406</b> comprising a display device <b>408</b> and one or more input devices <b>410</b>. In some implementations, memory <b>414</b> includes high-speed random access memory, such as DRAM, SRAM, DDR RAM or other random access solid state memory devices. In some implementations, memory <b>414</b> includes non-volatile memory, such as one or more magnetic disk storage devices, optical disk storage devices, flash memory devices, or other non-volatile solid state storage devices. In some implementations, memory <b>414</b> includes one or more storage devices remotely located from the CPU(s) <b>402</b>. Memory <b>414</b>, or alternately the non-volatile memory device(s) within memory <b>414</b>, comprises a computer readable storage medium. In some implementations, memory <b>414</b> or the computer readable storage medium of memory <b>414</b> stores the following programs, modules and data structures, or a subset thereof: <ul id="ul0003" list-style="none"><li id="ul0003-0001" num="0000"><ul id="ul0004" list-style="none"><li id="ul0004-0001" num="0055">an operating system <b>416</b> that includes procedures for handling various basic system services and for performing hardware dependent tasks;</li><li id="ul0004-0002" num="0056">a communications module <b>418</b> that is used for connecting the instance server <b>400</b> to other instance servers or computers via the one or more communication network interfaces <b>404</b> (wired or wireless) and one or more communication networks, such as the Internet, other wide area networks, local area networks, metropolitan area networks, and so on;</li><li id="ul0004-0003" num="0057">an optional user interface module <b>420</b> that receives commands from the user via the input devices <b>410</b> and generates user interface objects in the display device <b>408</b>;</li><li id="ul0004-0004" num="0058">a replication module <b>220</b> and replication queues <b>222</b>, as described herein;</li><li id="ul0004-0005" num="0059">data stores <b>224</b> (e.g., distributed databases, file systems, tape stores, Big Tables, etc.) that store the object chunks <b>238</b> in journals <b>230</b>, <b>232</b>, and <b>234</b> as described with respect to <figref idref="DRAWINGS">FIG. 3</figref>;</li><li id="ul0004-0006" num="0060">object metadata <b>228</b> and corresponding metadata elements <b>330</b>-<b>338</b>, <b>346</b>, and <b>348</b> as described in <figref idref="DRAWINGS">FIG. 3</figref> with respect to server <b>300</b>; and</li><li id="ul0004-0007" num="0061">journal metadata <b>236</b> and corresponding journal metadata elements <b>370</b>, <b>372</b>, and <b>374</b> as described in <figref idref="DRAWINGS">FIG. 3</figref> with respect to server <b>300</b>.</li></ul></li></ul>
Each of the above identified elements may be stored in one or more of the previously mentioned memory devices, and corresponds to a set of instructions for performing a function described above. The set of instructions can be executed by one or more processors (e.g., the CPUs <b>402</b>). The above identified modules or programs (i.e., sets of instructions) need not be implemented as separate software programs, procedures or modules, and thus various subsets of these modules may be combined or otherwise re-arranged in various implementations. In some implementations, memory <b>414</b> may store a subset of the modules and data structures identified above. Furthermore, memory <b>414</b> may store additional modules and data structures not described above.
Although <figref idref="DRAWINGS">FIG. 4</figref> shows an “instance server,” <figref idref="DRAWINGS">FIG. 4</figref> is intended more as functional description of the various features that may be present in a set of instance servers <b>400</b> than as a structural schematic of the implementations described herein. In practice, and as recognized by those of ordinary skill in the art, items shown separately could be combined and some items could be separated. For example, some items shown separately in <figref idref="DRAWINGS">FIG. 4</figref> could be implemented on single servers and single items could be implemented by one or more servers. The actual number of servers and how features are allocated among them will vary from one implementation to another, and may depend in part on the amount of data traffic that the server must handle during peak usage periods as well as during average usage periods. For example, at a single instance <b>102</b> there may be a hundred instance servers <b>400</b> or thousands of instance servers <b>400</b>.
In some implementations, to provide faster responses to clients and to provide fault tolerance, each program or process that runs at an instance is distributed among multiple computers. The number of instance servers <b>400</b> assigned to each of the programs or processes can vary, and depends on the workload.
<figref idref="DRAWINGS">FIG. 5</figref> illustrates the use of journals for storage of object chunks in accordance with some implementations. <figref idref="DRAWINGS">FIG. 5</figref> shows a data store <b>224</b>, as well as a portion of the object metadata <b>228</b> and a portion of the journal metadata <b>236</b>, all at an example instance <b>102</b>. There are many journals <b>230</b>, <b>232</b>, and <b>234</b> stored in this data store <b>224</b>, so it is useful to organize them visually in a two dimensional grid. (Of course the visual display is irrelevant to the actual physical storage of journals in a data store.) In the figure, the journals are partitioned into “rows” of journals, where each row corresponds to a single placement policy <b>212</b>. For example, the first row <b>502</b>-P<b>1</b> corresponds to placement policy P<b>1</b> (<b>212</b>), and includes closed journals <b>230</b>, open primary journals <b>232</b>, and open secondary journals <b>234</b>. All of these journals in the first row <b>502</b>-P<b>1</b> are associated with the placement policy P<b>1</b>. The second row <b>502</b>-P<b>2</b> corresponds to placement policy P<b>2</b> (<b>212</b>), and the last row <b>502</b>-PN corresponds to placement policy PN (<b>212</b>). Typically the number of placement policies is small, such as 10, 20, 50, or perhaps <b>100</b>. When the number of placement policies grows, management of object replicas becomes less efficient.
The journals in the data store <b>224</b> are also partitioned visually into two columns in <figref idref="DRAWINGS">FIG. 5</figref>. The first column identifies the closed journals <b>230</b>, which is the majority of the journals. The second column includes the open primary journals <b>232</b> and the open secondary journals <b>234</b>. As illustrated by the various rectangles <b>238</b> in each journal, each journal (whether closed <b>230</b>, open primary <b>232</b>, or open secondary <b>234</b>) contains object chunks <b>238</b>. The object chunks can be various sizes, but implementations typically set a fixed maximum size (e.g., 2 megabytes, 4 megabytes, or 8 megabytes). The illustration of object chunks <b>238</b> within a journal correctly conveys the fact that a journal stores many object chunks of various sizes, but is otherwise not representative of the actual physical storage of object chunks (e.g., there is generally no unused space between object chunks because each new object chunk <b>238</b> is appended at the beginning of the unallocated space).
<figref idref="DRAWINGS">FIG. 5</figref> illustrates that various combinations of open journals <b>232</b> and <b>234</b> are possible for each placement policy. To identify the different journals replicas in the figures and descriptions herein, a three part label is sometimes used, such as “232.P4.7”. The first portion (e.g., “232”) identifies the type of journal (230=closed, 232=open primary, 234=open secondary); the second portion (e.g., “P4”) specifies the placement policy for the journal; and the third portion (e.g., “7”) just specifies a sequential number for the journal (e.g., the “7” in “232.P4.7” specifies the seventh open journal for placement policy P<b>4</b>).
As illustrated in <figref idref="DRAWINGS">FIG. 5</figref>, for placement policy P<b>1</b> there is a single open primary journal <b>232</b>.P<b>1</b>.<b>1</b>, and no open secondary journals. For placement policy P<b>2</b>, there are two open primary journals <b>232</b>.P<b>2</b>.<b>1</b> and <b>232</b>.P<b>2</b>.<b>2</b>. For placement policy PN, there is one open primary journal <b>232</b>.PN.<b>1</b> and one open secondary journal <b>234</b>.PN.<b>1</b>. As these examples illustrate, the number of open primary journals <b>232</b> and open secondary journals <b>234</b> can vary between placement policies, and is typically configured for each policy <b>212</b> based on the expected number of new objects <b>226</b> for each placement policy <b>212</b> and the desired locations for those objects <b>226</b>
Each instance <b>102</b> also stores both object metadata <b>228</b> and journal metadata <b>236</b>, as previously described with respect to <figref idref="DRAWINGS">FIG. 3</figref>. For each object <b>226</b>, the object metadata <b>228</b> includes the object ID <b>330</b> (which uniquely identifies the object), a set of one or more chunk IDs <b>346</b> that identify the object chunks <b>238</b> from the object, and an assigned journal ID <b>348</b> associated with each chunk ID <b>236</b>. When an object has multiple chunks <b>238</b>, the chunks <b>238</b> are not necessarily all stored in the same journal (e.g., for load balancing), so the object metadata <b>228</b> must track the journal ID <b>348</b> assigned to each chunk ID <b>346</b>.
Each instance <b>102</b> also stores journal metadata <b>236</b> for each journal stored at the instance <b>102</b>. The metadata <b>236</b> includes a journal ID <b>370</b> for each journal, as well as a set of locations <b>372</b>. In some implementations, a location ID identifies an instance where the journal is stored. In some implementations, a location ID also identifies a data store at the specified instance. In some implementations, an instance identifier and a data store identifier are stored as separate attributes for each journal. In some implementations, a journal may be stored in two or more data stores at a single instance (e.g., a file system data store and a tape backup data store). The journal metadata <b>236</b> also includes a placement policy ID <b>374</b> that specifies the unique placement policy <b>212</b> corresponding to each journal. Each journal stores only object chunks <b>238</b> whose placement policies <b>338</b> match the placement policy of the journal.
<figref idref="DRAWINGS">FIG. 6</figref> illustrates how some implementations manage the storage of a new object <b>226</b>. As illustrated in <figref idref="DRAWINGS">FIG. 6</figref>, each new object has object content (i.e., the object <b>226</b> itself), as well as an object ID <b>330</b> (e.g., 58440912) and an assigned placement policy <b>330</b> (e.g., P<b>3</b>). The new object <b>226</b> can come from many different applications <b>246</b>, such as an online email application, a video sharing website, and so on. The distributed storage system <b>200</b> receives the new object <b>226</b> and directs (<b>602</b>) the new object <b>226</b> to an appropriate instance, such as the instance <b>102</b>-<b>1</b>. In some implementations, the application <b>246</b> directs the new object <b>226</b> to a specific instance <b>102</b>-<b>1</b>. When the instance <b>102</b>-<b>1</b> selected by the application <b>246</b> is not proper, some implementations forward the object <b>226</b> to an appropriate instance (e.g., if the placement policy <b>212</b> specifies no storage in Europe, and the object <b>226</b> is received at an instance in Europe, the instance can forward the object <b>226</b> to another instance).
Although most objects have moderate size (e.g., less than 300 kilobytes), there are some objects that are large. Some implementations split (<b>604</b>) large objects into multiple chunks <b>238</b>. In general, each implementation sets a chunk size or has a configurable parameter to set the chunk size, which is typically specified in megabytes (e.g., 2, 4, 8, 16, or 32 megabytes). Each object that is larger than the chunk size is split into multiple chunks, and each object that has size equal to or less than the chunk size consists of a single chunk. In the illustration in <figref idref="DRAWINGS">FIG. 6</figref>, there are three chunks C<b>1</b>, C<b>2</b>, and C<b>3</b>. In this illustration, each of the chunks has a 7 character alphanumeric chunk ID <b>346</b>, but many alternative chunk ID formats are possible that uniquely identify the chunks within each object. In some implementations, a chunk ID <b>346</b> is generated using a content hash or content digest.
In some implementations there can be many object duplicates (e.g., an email attachment sent to a group of people, then forwarded to many additional people), so de-duplication can be useful for efficient storage. Thus, in some embodiments, the content of each new chunk <b>238</b> is compared (<b>606</b>) to existing object chunks <b>238</b> (e.g., using a content hash or content digest) to store only (<b>606</b>) “new” chunks <b>238</b> in an open primary journal. As illustrated in <figref idref="DRAWINGS">FIG. 5</figref>, chunk C<b>2</b> is new, and corresponds to placement policy P<b>3</b>, so chunk C<b>2</b> is stored in an open primary journal <b>232</b>.P<b>3</b>.<b>1</b> corresponding to placement policy P<b>3</b>. Of course de-duplication is only within the context of a placement policy. If two chunks are identical, but are assigned to different placement policies, then the two chunks will be saved in distinct journals. Stated differently, when a new chunk is received, it is only compared against chunks for the same placement policy. A chunk is a “duplicate” only when there is already a saved identical chunk for the same placement policy.
Regardless of whether the object chunk C<b>2</b> is new, the instance <b>102</b>-<b>1</b> stores (<b>608</b>) object metadata <b>228</b> for the chunk <b>238</b>. As described previously with respect to <figref idref="DRAWINGS">FIGS. 3-5</figref>, the metadata <b>228</b> includes the object ID <b>330</b>, the chunk ID <b>346</b>, and the journal ID <b>348</b> for the journal where each chunk is stored. In some implementations, the chunk ID <b>346</b> for an object chunk <b>238</b> is just the offset to the start of the chunk <b>238</b> within the object. The object metadata <b>228</b> shown in <figref idref="DRAWINGS">FIG. 6</figref> also illustrates that the chunks for a single object need not be stored in the same journal. The chunks C<b>1</b> and C<b>3</b> (chunk IDs C190056 and C098663) are in the journal <b>232</b>.P<b>3</b>.<b>2</b> with journal ID J77298045, whereas chunk C<b>2</b> (chunk ID C250116) is in the journal <b>232</b>.P<b>3</b>.<b>1</b> with journal ID J82117094.
The chunk C<b>2</b> is transmitted (<b>610</b>) to instance <b>102</b>-<b>2</b> for storage in secondary journal <b>234</b>.P<b>3</b>.<b>1</b>, and chunks C<b>1</b> and C<b>3</b> are transmitted (<b>612</b>) to instance <b>102</b>-<b>2</b> for storage in secondary journal <b>234</b>.P<b>3</b>.<b>2</b>.
<figref idref="DRAWINGS">FIG. 6</figref> also illustrates that a primary journal <b>232</b> need not be physically identical to its corresponding secondary journal. First, we see that chunks C<b>1</b> and C<b>3</b> are stored in that order in the primary journal <b>232</b>.P<b>3</b>.<b>2</b>, whereas these chunks are stored in the reverse order in the secondary journal <b>234</b>.P<b>3</b>.<b>2</b>. While a journal is open, the individual chunks <b>238</b> may be replicated independently, traverse different network paths, or be processed by different processors <b>402</b>, so there is no guarantee that they are loaded into the secondary journal <b>234</b>.P<b>3</b>.<b>2</b> in the same order. The fact that there can be different orders is handled by the chunk index within each journal as described below with respect to <figref idref="DRAWINGS">FIG. 7</figref>. In addition, primary journal <b>232</b>.P<b>3</b>.<b>1</b> indicates the presence of a garbage “chunk” <b>620</b> labeled as “G” in the figure. Sometimes during an upload there can be failure or glitch that consumes space. For example, during an upload, perhaps the space for an object is allocated, but the chunk is not actually appended. The software retries the upload, which allocates new space for the chunk. This can leave holes or garbage within a journal <b>232</b>. In this case the garbage <b>620</b> is not transmitted to the secondary journal, so the primary journal is physically different from the secondary journal.
<figref idref="DRAWINGS">FIG. 7</figref> illustrates the structure of an open journal in accordance with some implementations. Although <figref idref="DRAWINGS">FIG. 7</figref> describes an open primary journal <b>232</b>, the structure or an open secondary journal <b>234</b> would be the same or similar. A journal <b>232</b> has a header <b>702</b> and a block of storage space <b>714</b>. The storage space <b>714</b> includes a filled portion <b>710</b> that is already storing object chunks <b>238</b>, and an unfilled portion <b>712</b> that is currently unused. These descriptors are not completely accurate for a few reasons. First, the “filled” space <b>710</b> may include garbage portions <b>620</b> that have no useful content. Second, the unused space is not necessarily allocated all at the same time. Some implementations do allocate the entire space for the journal at one time, and close the journal when it is filled (potentially leaving a small amount of unused space at the end). But in other implementations, blocks of additional space are allocated as needed, until the journal reaches a certain size limit or a certain amount of time has elapsed (e.g., one day).
The header <b>702</b> for the journal contains important internal information about the journal <b>232</b>. The header <b>702</b> includes a field <b>704</b> that specifies where the unused space <b>712</b> begins in the journal. Each time a new chunk <b>238</b> is appended to the end of the filled space <b>710</b>, the offset <b>704</b> is incremented by the size of the chunk <b>238</b> so that the journal <b>232</b> is prepared to store the next chunk.
The header <b>702</b> also includes a chunk index <b>706</b>. The chunk index <b>706</b> for a journal <b>232</b> specifies where each chunk <b>238</b> is located within the journal <b>232</b> as well as its size, enabling a rapid read of the chunk data (whether from non-volatile storage or from cache). The key for the chunk index <b>706</b> is the chunk ID <b>346</b>, which uniquely identifies the chunk. Note that multiple distinct object ID's <b>330</b> may refer to the same physical chunks. To avoid a huge chunk index <b>704</b> with many entries pointing to the same object chunk <b>238</b>, implementations typically utilize a single chunk ID to refer to the same physical content. For example, the chunk ID <b>346</b> may be a content hash or a content digest (or a combination of these). For each chunk ID <b>346</b>, the chunk index <b>720</b> specifies an offset <b>720</b> and a size <b>722</b> for the chunk <b>238</b> within the storage space <b>714</b>. The offset <b>720</b> may be specified either as an offset from the beginning of the journal <b>232</b> or an offset from the beginning of the filled space <b>710</b>. In some implementations, the chunk index has additional information, such as a deletion marker that is used later when chunks are deleted and the filled space <b>710</b> compacted.
The header <b>702</b> may contain other journal data <b>708</b> as well to address implementation details. For example, the other journal data <b>708</b> may specify the offset from the beginning of the journal to the beginning of the storage space <b>714</b> (i.e., the size of the header). In some implementations, the other journal data includes a “time to live” parameter for journals that are designated to have a short lifespan.
Although the structure of the journal in <figref idref="DRAWINGS">FIG. 7</figref> is for an open primary journal <b>232</b>, the same basic structure applies to open secondary journals <b>234</b> and closed journals <b>230</b> as well.
<figref idref="DRAWINGS">FIG. 8</figref> illustrates what happens to object metadata <b>228</b> and journal metadata <b>236</b> when a journal is replicated from one instance to another, in accordance with some implementations. In this illustration, closed journal <b>230</b> with journal ID J82117094 is replicated (<b>820</b>) from instance <b>102</b>-<b>1</b> (with instance ID=723) to instance <b>102</b>-<b>4</b> (with instance ID <b>428</b>). Because the journal <b>230</b> itself is replicated as a unit, the entire content is replicated exactly. For example, chunk C<b>8</b> (with chunk ID C408335) is in exactly the same position within the journal. Of course after replication, instance <b>102</b>-<b>1</b> and <b>102</b>-<b>4</b> independently handle deletion and compaction, so their physical structures are not guaranteed to stay the same after replication.
<figref idref="DRAWINGS">FIG. 8</figref> also shows a portion of the object metadata <b>228</b> and journal metadata <b>236</b>, both before and after the replication <b>820</b>. As indicated, the records <b>802</b>-<b>814</b> in the object metadata <b>228</b> are unchanged by the replication <b>820</b>. Each object <b>226</b> has the same chunks <b>238</b>, and the chunks <b>238</b> are stored in the same journal <b>230</b>. For example, the chunk with chunk ID C408335 (in row <b>804</b>) is unchanged. On the other hand, the journal metadata <b>236</b> for the journal <b>230</b> with journal ID J82117094 (<b>370</b>-<b>1</b>) does change. The set of journal locations <b>372</b> changes from <b>372</b>-<b>1</b>(A) to <b>372</b>-<b>1</b>(B), which includes the new location <b>428</b> (for instance <b>102</b>-<b>4</b>).
<figref idref="DRAWINGS">FIGS. 9A-9C</figref> illustrate a method <b>900</b> of managing (<b>902</b>) placement of object replicas in a distributed storage system <b>200</b> according to some implementations. The method is performed (<b>904</b>) at a first instance <b>102</b> of the distributed storage system, which has one or more processors and memory. The memory stores (<b>906</b>) a plurality of objects. The memory also stores (<b>908</b>) one or more programs for execution by the one or more processors. In some implementations, all or part of method <b>900</b> is performed by location assignment daemon <b>206</b>. In some implementations, the distributed storage system has (<b>910</b>) a plurality of instances. In some of these implementations, at least a subset of the instances are (<b>910</b>) at distinct geographic locations. In some implementations, each instance corresponds to a data center. In some implementations, each data center comprises one or more instances.
At the first instance, one or more journals <b>232</b> are opened (<b>912</b>) for storage of object chunks. Each journal is associated with (<b>914</b>) a single respective placement policy <b>212</b>. In some implementations, each placement policy specifies (<b>926</b>) a target number of object replicas and a target set of locations for the object replicas. In some implementations, a placement policy <b>212</b> may specify what type of data store <b>224</b> to use at some of the instances (e.g., on disk or on tape). In some implementations, the distributed storage system <b>200</b> includes (<b>918</b>) object metadata <b>228</b> that specifies in which journal each object chunk <b>238</b> is stored. This was described previously with respect to <figref idref="DRAWINGS">FIGS. 3-5</figref>. In some implementations, each respective journal includes (<b>920</b>) a chunk index <b>706</b> that specifies the location of each object stored in the respective journal. This was described in more detail in <figref idref="DRAWINGS">FIG. 7</figref>. In particular, the location of each chunk within a journal is identified relative to the journal itself, and thus the chunk index <b>706</b> is accurate regardless of where the journal is stored. For example, by specifying the location of chunks within a journal as offsets, the chunks can be accessed by relative addressing.
Disclosed implementations typically include (<b>922</b>) journal metadata <b>236</b> that specifies the locations <b>372</b> where each journal is stored. This was described previously in <figref idref="DRAWINGS">FIGS. 3-5</figref> and <b>8</b>.
The distribution of open primary journals <b>232</b> and open secondary journals <b>234</b> depends on many factors, including the available instances <b>102</b>, the placement policies <b>212</b>, the anticipated distribution of new objects <b>226</b> with the placement policies <b>212</b>, where the new objects are loaded from (e.g., Europe, North America, Asia), processing resources at each of the available instances <b>102</b>, and the network bandwidth between the various instances. For example, if many objects will be uploaded with a specific placement policy at a specific instance, then multiple journals are opened (<b>924</b>) for the same placement policy at that instance. In some scenarios, there may be 5, 10, or more open journals for the same placement policy <b>212</b> at a single instance <b>102</b> when required for load balancing.
As described previously with respect to <figref idref="DRAWINGS">FIGS. 5 and 6</figref>, some implementations transmit (<b>916</b>) a message to a third instance of the distributed storage system <b>200</b> to open journals corresponding to journals opened at the first instance. In this scenario, the journals <b>232</b> opened at the first instance are referred to as primary journals and the journals <b>234</b> opened at the third instance are referred to as secondary journals. (Of course the first instance could also have secondary journals and the third instance could have primary journals.)
At the first instance <b>102</b>, a first object <b>226</b> is received (<b>928</b>), which comprises (<b>928</b>) at least a first object chunk. This was described above with respect to <figref idref="DRAWINGS">FIG. 6</figref>. The first object <b>226</b> is associated with a first placement policy <b>212</b>, and thus all of the object chunks <b>238</b> that comprise the object <b>226</b> are associated with the first placement policy <b>212</b>. The first object chunk <b>238</b> is stored (<b>930</b>) in a first journal <b>232</b> whose associated placement policy matches the first placement policy <b>212</b>. The first journal <b>232</b> stores only (<b>932</b>) object chunks for objects whose placement policies match the first placement policy. In some implementations, each object chunk <b>238</b> stored in the first journal <b>232</b> is transmitted (<b>934</b>) to the third instance for storage in a third journal <b>234</b>.
When the received object is larger than the chunk size, the object is split into multiple chunks <b>238</b>. In this case, the first object <b>226</b> comprises (<b>936</b>) two or more object chunks. Typically the second object chunk is distinct from (<b>936</b>) the first object chunk. (Having two identical chunks within a single object is rare, but could happen, for example, if an object had a very large portion of empty space.) In some circumstances, the second object chunk is stored (<b>938</b>) in a second journal <b>232</b>, distinct from the first journal, whose associated placement policy matches the first placement policy. The second journal stores only (<b>938</b>) object chunks for objects whose placement policies match the first placement policy. In this way, a object that comprises many chunks could have the chunks distributed across many different journals.
This process of receiving objects <b>226</b> and storing the chunks <b>238</b> in the first journal <b>232</b> is repeated (<b>940</b>) for a plurality of objects <b>226</b> whose associated placement policies <b>338</b> match the first placement policy <b>212</b>, until a first termination condition occurs. In some implementations, the first termination condition occurs when (<b>942</b>) the size of the first journal exceeds a predefined threshold. In some implementations, the first termination condition occurs when (<b>944</b>) the first journal has been open for a predefined span of time. Some implementations combine size and time in various ways. For example, some implementations specify both a time span and a size limit, and the termination condition is whichever one occurs first.
After the termination condition occurs, the first journal is closed (<b>946</b>), thereby preventing any additional object chunks from being stored in the first journal <b>232</b>. Generally, implementations confirm that other journals <b>232</b> for the same placement policy are still open (or a new one is opened) prior to closing the first journal. Because new objects can arrive at any moment, it is important to have open journals available for storage. When there is a corresponding secondary journal <b>234</b> at another instance, the first instance transmits (<b>948</b>) a message to the other instance to close the corresponding secondary journal when the first termination condition occurs.
After the first journal <b>232</b> is closed, the journal is subject to its placement policy. Satisfying the placement policy <b>212</b> may require moving a journal replica, making a new copy of a journal replica, or deleting a replica of a journal. In some circumstances, the first journal <b>232</b> is replicated (<b>950</b>) to a second instance <b>102</b> of the distributed storage system <b>200</b> in accordance with the placement policy <b>212</b>. (In other circumstances, a replica of the first journal is deleted.) In implementations that have primary and secondary open journals <b>232</b> and <b>234</b>, there will be two equivalent closed journals <b>230</b> once they are closed. Therefore, either of the replicas could be used as the source for the replication <b>950</b>. As the replication <b>950</b> occurs (i.e., as part of the transaction), the journal metadata <b>236</b> for the first journal is updated (<b>952</b>) to indicate that there is a copy of the journal at the second instance. This was described above with respect to <figref idref="DRAWINGS">FIG. 8</figref>.
After a journal <b>230</b> is closed, the object chunks <b>238</b> may be deleted. For example, an object may correspond to an email attachment. If the recipient of the email deletes the email, then the storage for the attachment can be deleted. After a period of time, there are holes within each journal from the deletions, and thus it is useful to compact the journal to remove the wasted space. This is similar to fragmentation of volatile memory and the process of defragmentation to consolidate the unused space into larger contiguous blocks.
Because a stored object chunk may correspond to many distinct objects (e.g., hundreds, thousands, or millions), an object chunk in a journal can only be deleted if there are no more references to it. Therefore, once a first closed journal <b>230</b> is selected (<b>954</b>), the process <b>900</b> identifies (<b>956</b>) one or more object chunks stored in the first closed journal <b>230</b> for which there are no references in the object metadata <b>228</b>. For these identified chunks <b>238</b>, the chunk index <b>706</b> is updated (<b>958</b>) to remove the corresponding records. In some implementations, the space previously allocated to the identified object chunks are overwritten (e.g., each byte set to ASCII 0), but in other implementations the space is just no longer referenced. In some implementations, the deallocated storage space is tracked as part of the other journal data <b>708</b>. For example, some implementations maintain a list of deallocated storage spaces (e.g., offset and size), or track the deallocated spaces as a linked list.
In some implementations, a garbage collection algorithm runs periodically to compact (<b>960</b>) each of the closed journals. The compaction process consolidates (<b>960</b>) the stored object chunks into a contiguous block, thus reducing the size of the journal <b>230</b>. Over time, journals <b>230</b> can become small as more object chunks are deleted. Managing many small journals has overhead similar to managing individual objects, and thus the benefit of the journal storage is diminished. To address this issue, some implementations stitch together (<b>962</b>) two or more closed journals to form a single replacement journal, and update (<b>962</b>) object metadata <b>228</b> to indicate that object chunks previously stored in the two or more journals are now stored in the replacement journal. Because a stitching operation requires forming an entirely new journal and updating the metadata for all of the objects involved, stitching is usually limited to the scenario where the journals have gotten relatively small.
<figref idref="DRAWINGS">FIGS. 10A and 10B</figref> illustrate two implementations for storing objects and associated metadata in a distributed storage system. The implementation illustrated in <figref idref="DRAWINGS">FIG. 10A</figref> is fully hierarchical: every object is split into one or more chunks, and every chunk is split into one or more blocks. Note that the structure is hierarchical even when there is only one chunk or only one block. On the other hand, the implementation illustrated in FIG. <b>10</b>B is only partially hierarchical. In this implementation, some chunks are “superchunks,” which refer to lists of blocks. For example, a superchunk may have a block list with 100 blocks, 1000 blocks, or even more. A chunk that is not a superchunk is just a block. That is, the chunk identifier refers to an actual storage of object data rather than a list of blocks. This hybrid approach can be useful in distributed storage systems that include both small objects (where no hierarchy is needed) and very large objects, where the storage hierarchy is much more efficient.
The global metadata <b>1002</b> includes the object metadata <b>228</b> and journal metadata <b>236</b>, as illustrated in <figref idref="DRAWINGS">FIGS. 2-4</figref>. In some implementations, each chunk ID <b>346</b> is assigned a decryption key <b>1040</b>, which is used to decrypt the data for the chunk. In these implementations, the same decryption key would apply to all blocks in a chunk for those chunks that are split into multiple blocks. Each chunk has its own decryption key <b>1040</b>, which is effectively unique. Some implementations guarantee uniqueness when new keys are generated, but some implementations generate new keys randomly, with repetition of keys being highly improbable. The decryption key <b>1040</b> corresponds to an encryption key that is used to encrypt new objects as they are stored. Because the decryption key <b>1040</b> is needed to access each object chunk, deleting the decryption key can be used as a “soft” delete of an object chunk. When the decryption key is gone, the encrypted storage is “garbage,” and the actual data is inaccessible. This can allow a garbage collection algorithm more time between compactions, and the garbage collection process can recover more storage space when it does run.
Also illustrated in <figref idref="DRAWINGS">FIG. 10A</figref> is a journal <b>232</b> (which is shown as open), and corresponding local metadata <b>1004</b>. In some implementations, the local metadata <b>1004</b> for a journal <b>232</b> is stored in the journal itself, as part of the header <b>702</b>. In other implementations, the local metadata <b>1004</b> for a journal is stored as a separate file (or in a database, etc.) and associated with the journal. The structure of a journal <b>232</b> for non-hierarchical storage was illustrated above with respect to <figref idref="DRAWINGS">FIG. 7</figref>. In this implementation, rather than storing chunks, the basic unit of storage is a block <b>1016</b>, such as blocks <b>1016</b>-<b>1</b>, <b>1016</b>-<b>2</b>, . . . , <b>1016</b>-N. Each implementation typically specifies a maximum block size, such as 2 megabytes, 4 megabytes, 8 megabytes, or 16 megabytes.
As noted above, the local metadata <b>1004</b> may be stored in the header <b>702</b> of the journal <b>232</b>, or may be stored separately. For each chunk identifier <b>346</b>, there is a corresponding block list <b>1006</b> (typically a unique block list), which comprises one or more block identifiers <b>1008</b>. For a small chunk, the block list <b>1006</b> may contain a single block identifier <b>1008</b>. The local metadata <b>1004</b> also includes a block index <b>1010</b>, which specifies where each block is located within the journal <b>232</b>. In some implementations, the location of a block is specified by an offset and a size. The block offset <b>1012</b> in some implementations is the offset from the beginning of the storage space <b>714</b> or the offset for the beginning of the journal file <b>232</b>. Typically the block size <b>1014</b> is specified in bytes, but other implementations use alternative basic units of size (e.g., 2 bytes, 4 bytes, or 8 bytes). One aspect of the local metadata is that it does not change when a journal is moved or replicated to another instance: the block list <b>1006</b> for a chunk remains the same, the block ID's <b>1008</b> remain the same, the block offsets <b>1012</b> within the journal remain the same, and the block sizes remain the same.
<figref idref="DRAWINGS">FIG. 10B</figref> is similar to <figref idref="DRAWINGS">FIG. 10A</figref>, but illustrates a partially hierarchical structure. In the partially hierarchical structure of <figref idref="DRAWINGS">FIG. 10B</figref>, the global metadata <b>1002</b> includes a “superchunk” field <b>1020</b> that indicates whether each chunk is an ordinary block or refers to a list of blocks (i.e., is a superchunk). In some implementations, most objects are small, consisting of a single chunk. In this case, the chunk ID <b>346</b> identifies a block directly in the chunk/block index <b>1024</b>. That is, the chunk ID <b>346</b> is a chunk/block ID <b>1026</b>. Thus, for chunks that are not superchunks, the chunk ID <b>346</b> can be used to look up the appropriate record in the chunk/block index <b>1024</b> to find the offset <b>1028</b> and size <b>1030</b> for the corresponding block <b>1016</b> in the journal <b>232</b>.
For superchunks, the chunk ID <b>346</b> is a (super) chunk ID <b>1022</b>, which can be looked up in the local metadata <b>1004</b>. Corresponding to the super chunk ID <b>1022</b> is a block list <b>1006</b>, which comprises a set of block ID's <b>1008</b>. In this case, each of the block ID's can be looked up in the chunk/block index <b>1024</b> to identify the offset <b>1028</b> and size <b>1030</b> for each of the block ID's <b>1026</b> in the block list <b>1006</b> for the super chunk ID <b>1022</b>. As before, the offset <b>1028</b> and size <b>1030</b> identify the location of the actual block storage in the storage space <b>714</b> of the journal <b>232</b>. Superchunks thus have an extra level of hierarchy, but reduce the amount of chunk metadata stored in the global metadata <b>1002</b>. This makes is easier and more efficient to move a shard from one instance to another.
<figref idref="DRAWINGS">FIGS. 11A-11D</figref> illustrate a method <b>1100</b> of managing (<b>1102</b>) placement of object replicas in a distributed storage system <b>200</b> according to some implementations. The method is performed (<b>1104</b>) at a first instance <b>102</b> of the distributed storage system, which has one or more processors and memory. The memory stores (<b>1106</b>) one or more programs for execution by the one or more processors. In some implementations, all or part of method <b>1100</b> is performed by a location assignment daemon <b>206</b>. In some implementations, the distributed storage system has (<b>1108</b>) a plurality of instances. In some of these implementations, at least a subset of the instances are (<b>1108</b>) at distinct geographic locations. In some implementations, each instance corresponds to a data center. In some implementations, each data center comprises one or more instances.
At the first instance, one or more journals <b>232</b> are opened (<b>1110</b>) for storage of object chunks. Each journal is associated with (<b>1112</b>) a single respective placement policy <b>212</b>. In some implementations, each placement policy specifies (<b>1122</b>) a target number of object replicas and a target set of locations for the object replicas. In some implementations, a placement policy <b>212</b> may specify what type of data store <b>224</b> to use at some of the instances (e.g., on disk or on tape). In some implementations, the distributed storage system <b>200</b> includes (<b>1114</b>) object metadata <b>228</b> (part of the global metadata <b>1002</b>) that specifies in which journal each object chunk <b>238</b> is stored. This was described previously with respect to <figref idref="DRAWINGS">FIGS. 3-5</figref>, <b>10</b>A, and <b>10</b>B. In some implementations, each respective journal includes (<b>1116</b>) a block index <b>1010</b> or <b>1026</b> that specifies the location of each block stored in the respective journal. This was described in more detail in <figref idref="DRAWINGS">FIGS. 7</figref> (non-hierarchical), <b>10</b>A, and <b>10</b>B. In particular, the location of each block <b>1016</b> within a journal <b>232</b> is identified relative to the journal itself, and thus the block index <b>1010</b> or <b>1026</b> is accurate regardless of where the journal <b>232</b> is stored. For example, by specifying the location of blocks <b>1016</b> within a journal <b>232</b> as offsets, the blocks <b>1016</b> can be accessed by relative addressing.
Disclosed implementations typically include (<b>1118</b>) journal metadata <b>236</b> (part of the global metadata <b>1002</b>) that specifies the locations <b>372</b> where each journal is stored. This was described previously in <figref idref="DRAWINGS">FIGS. 3-5</figref> and <b>8</b>.
The distribution of open primary journals <b>232</b> and open secondary journals <b>234</b> depends on many factors, including the available instances <b>102</b>, the placement policies <b>212</b>, the anticipated distribution of new objects <b>226</b> with the placement policies <b>212</b>, where the new objects are loaded from (e.g., Europe, North America, Asia), processing resources at each of the available instances <b>102</b>, and the network bandwidth between the various instances. For example, if many objects will be uploaded with a specific placement policy at a specific instance, then multiple journals are opened (<b>1120</b>) for the same placement policy at that instance. In some scenarios, there may be 5, 10, or more open journals for the same placement policy <b>212</b> at a single instance <b>102</b> when required for load balancing.
At the first instance <b>102</b>, a first object <b>226</b> is received (<b>1124</b>), which comprises (<b>1124</b>) at least a first object chunk. This was described above with respect to <figref idref="DRAWINGS">FIG. 6</figref>. The first object <b>226</b> is associated with (<b>1124</b>) a first placement policy <b>212</b>, and thus all of the object chunks <b>238</b> that comprise the object <b>226</b> are associated with the first placement policy <b>212</b>. The first object chunk comprises (<b>1126</b>) a first plurality of blocks, as described above with respect to <figref idref="DRAWINGS">FIGS. 10A and 10B</figref>. In some implementations, the process <b>1100</b> receives (<b>1124</b>) the object <b>238</b> already partitioned into chunks and blocks. For example, the splitting may be performed by the client device that uploads the object. In other implementations, the process <b>1100</b> receives the object as a stream and splits the object into chunks and blocks according to stored criteria (e.g., target block and chunk size, available open journals, available instances, available bandwidth, etc.) In some implementations, dynamic allocation of chunks if performed while still receiving data for an object, whereas other implementations split an object into chunks and blocks only after the entire object is received.
The hierarchy of chunks and blocks may be formed in various ways and based on various factors, such as the size of the object. In some implementations, the hierarchy is built dynamically during the upload process. For example, a first object chunk is created, and the stream of data is split into blocks that are assigned to the first object chunk until a threshold number of blocks is assigned to the chunk. At that point, a second chunk is created, and the new blocks are added to the second chunk. In another implementation, the stream of data is stored as blocks of storage initially, and when there are no more blocks, the blocks are grouped into chunks.
In some implementations, every object chunk <b>238</b> comprises (<b>1128</b>) one or more blocks. This was illustrated above with respect to <figref idref="DRAWINGS">FIG. 10A</figref>. In some implementations, the global metadata <b>1002</b> includes (<b>1130</b>) a field <b>1020</b> that specifies whether each object chunk <b>238</b> is a block or a list of blocks. This is illustrated above in <figref idref="DRAWINGS">FIG. 10B</figref>. In some instances, the first object chunk is (<b>1132</b>) a list of blocks (i.e., a superchunk), whereas a second chunk is (<b>1132</b>) an ordinary block (not a superchunk).
The first plurality of blocks <b>1016</b> is stored (<b>1134</b>) in a first journal <b>232</b> whose associated placement policy matches the first placement policy <b>212</b>. The first journal <b>232</b> stores only (<b>1136</b>) blocks for objects whose placement policies match the first placement policy.
When the received object is larger than a specified size (e.g., the chunk size or block size), the object is split into multiple chunks <b>238</b> and/or multiple blocks <b>1016</b>. In some instances, the first object <b>226</b> comprises (<b>1138</b>) two or more object chunks. Typically the second object chunk is distinct from (<b>1138</b>) the first object chunk. (Having two identical chunks within a single object is rare, but could happen, for example, if an object had a very large portion of empty space.) In some circumstances, the second object chunk is stored (<b>1140</b>) in a second journal <b>232</b>, distinct from the first journal, whose associated placement policy matches the first placement policy. The second journal stores only (<b>1140</b>) object chunks for objects whose placement policies match the first placement policy. In this way, a object that comprises many chunks could have the chunks distributed across many different journals.
In some implementations, the process encrypts (<b>1142</b>) the data for each object chunk, and stores (<b>1142</b>) a decryption key for each object chunk in the global metadata. This was illustrated above in <figref idref="DRAWINGS">FIGS. 10A and 10B</figref>. In some implementations, when a chunk is split into multiple blocks, each of the blocks within the chunk are encrypted with the same encryption key, and can thus be decrypted with the same decryption key. In other implementations, each block has its own decryption key, which may be stored as part of the block index <b>1010</b> or <b>1026</b>. In implementations that store the decryption key <b>1040</b> in the global metadata <b>1002</b>, a chunk can be de facto deleted simply by deleting (<b>1144</b>) the decryption key. The chunk is inaccessible because there is no way to retrieve the data for the original chunk. This provides some advantages. First, deleting a chunk is quick and effective. Second, because there is no real risk of accessing the deleted data, a more efficient garbage collection process can be implemented. In particular, garbage collection can be scheduled at appropriate intervals, and can batch process the physical deletes of storage from disk. Because compaction is a resource intensive process, the ability to batch together many deletes can increase efficiency dramatically. Third, some implementations do not require physical erasing of storage space because the encrypted “gibberish” cannot be converted back to meaningful content.
The process <b>1100</b> stores (<b>1146</b>) global metadata for the first object. This was illustrated above in <figref idref="DRAWINGS">FIGS. 3-5</figref>, <b>10</b>A, and <b>10</b>B. The global metadata <b>1002</b> includes (<b>1148</b>) a first list of object chunks corresponding to the first object. In particular, the first list includes (<b>1150</b>) an object identifier <b>330</b> for the first object chunk <b>238</b>. The global metadata <b>1002</b> also identifies the journal where each chunk is stored as well the locations for each of the journals.
In addition to the global metadata <b>1002</b>, local metadata <b>1004</b> is stored for each journal <b>232</b>. In some implementations, the local metadata <b>1004</b> for each journal is stored in the header <b>702</b> of the journal <b>232</b> itself. In other implementations, the local metadata <b>1004</b> is stored separately from the journal. When stored separately, the local metadata <b>1004</b> for each journal may be stored separately (e.g., a distinct metadata file corresponding to each journal), or the local metadata may be grouped together (e.g, in a database).
The first instance stores (<b>1152</b>) local metadata <b>1004</b> for the first object chunk <b>238</b>. The local metadata <b>1004</b> includes (<b>1154</b>) a block list identifying each block in the first plurality of blocks. Note that the block list is stored in the local metadata <b>1004</b>, not in the global metadata <b>1002</b>. The block list <b>1006</b> stored in the local metadata <b>1004</b> tracks how the blocks are allocated within each journal. The local metadata for the first journal <b>232</b> is associated with (<b>1156</b>) the first journal <b>232</b>. In some implementations, the association of the local metadata with a journal is performing by storing the local metadata in the journal, which makes a journal more self-contained. In some implementations, the local metadata for a journal <b>232</b> is stored separately (e.g., in a separate file), and associated with the journal (e.g., by including the journal ID <b>370</b> in the name of the journal and in the name of the associated metadata file). In implementations that store local metadata in a database, the journal ID <b>370</b> is typically part of the primary key for the metadata tables.
This process of receiving objects <b>226</b> and storing the chunks <b>238</b> in the first journal <b>232</b> is repeated (<b>1158</b>) for a plurality of objects <b>226</b> whose associated placement policies <b>338</b> match the first placement policy <b>212</b>, until a first termination condition occurs. In some implementations, the first termination condition occurs when (<b>1160</b>) the size of the first journal exceeds a predefined threshold. In some implementations, the first termination condition occurs when (<b>1162</b>) the first journal has been open for a predefined span of time. Some implementations combine size and time in various ways. For example, some implementations specify both a time span and a size limit, and the termination condition is whichever one occurs first.
After the termination condition occurs, the first journal is closed (<b>1164</b>), thereby preventing any additional blocks from being stored in the first journal <b>232</b>. Generally, implementations confirm that other journals <b>232</b> for the same placement policy are still open (or a new one is opened) prior to closing the first journal. Because new objects can arrive at any moment, it is important to have open journals available for storage.
After the first journal <b>232</b> is closed, the journal is subject to its placement policy. Satisfying the placement policy <b>212</b> may require moving a journal replica, making a new copy of a journal replica, or deleting a replica of a journal. In some circumstances, the first journal <b>232</b> is replicated (<b>1166</b>) to a second instance <b>102</b> of the distributed storage system <b>200</b> in accordance with the placement policy <b>212</b>. (In other circumstances, a replica of the first journal is deleted.) In implementations that have primary and secondary open journals <b>232</b> and <b>234</b>, there will be two equivalent closed journals <b>230</b> once they are closed. Therefore, either of the replicas could be used as the source for the replication <b>1166</b>. As the replication <b>1166</b> occurs (e.g., as part of the transaction), the global metadata <b>1002</b> for the first journal is updated (<b>1168</b>) to indicate that there is a copy of the journal at the second instance. On the other hand, the local metadata <b>1004</b> is unchanged (<b>1168</b>) by the replication. This was described above with respect to <figref idref="DRAWINGS">FIGS. 8</figref>, <b>10</b>A, and <b>10</b>B.
After a journal <b>230</b> is closed, the object chunks <b>238</b> may be deleted. For example, an object may correspond to an email attachment. If the recipient of the email deletes the email, then the storage for the attachment can be deleted. After a period of time, there are holes within each journal from the deletions, and thus it is useful to compact the journal to remove the wasted space. This is similar to fragmentation of volatile memory and the process of defragmentation to consolidate the unused space into larger contiguous storage.
Because a stored object chunk may correspond to many distinct objects (e.g., hundreds, thousands, or millions), an object chunk in a journal can only be deleted if there are no more references to it. Therefore, once a first closed journal <b>230</b> is selected (<b>1170</b>), the process <b>1100</b> identifies (<b>1172</b>) one or more object chunks stored in the first closed journal <b>230</b> for which there are no references in the object metadata <b>228</b>. In some implementations, a garbage collection algorithm runs periodically to compact (<b>1174</b>) each of the closed journals. The compaction process consolidates (<b>1174</b>) the stored blocks into contiguous storage, thus reducing the size of the journal <b>230</b>.
Over time, journals <b>230</b> can become small as more object chunks are deleted. Managing many small journals has overhead similar to managing individual objects, and thus the benefit of the journal storage is diminished. To address this issue, some implementations stitch together (<b>1176</b>) two or more closed journals to form a single replacement journal, and update (<b>1176</b>) object metadata <b>228</b> to indicate that object chunks previously stored in the two or more journals are now stored in the replacement journal. Because a stitching operation requires forming an entirely new journal and updating the metadata for all of the objects involved, stitching is usually limited to the scenario where the journals have gotten relatively small.
<figref idref="DRAWINGS">FIG. 12</figref> illustrates an example of storing chunks in a distributed storage system in accordance with some implementations as illustrated previously with respect to <figref idref="DRAWINGS">FIG. 10B</figref>. In this example, two chunks <b>238</b>-<b>1</b> and <b>238</b>-<b>2</b> are shown. Chunk <b>238</b>-<b>1</b> is an ordinary chunk (i.e., not a superchunk), which chunk ID <b>346</b>-<b>1</b>. Because chunk <b>238</b>-<b>1</b> is an ordinary chunk, it can be looked up directly in the chunk/block index <b>1024</b>. In this illustration, the chunk/block index <b>1024</b> is stored in the header <b>702</b> of the journal <b>232</b> where the data is stored. For this chunk/block, the offset <b>1028</b> is y (<b>1028</b>-<i>y</i>). Using this offset, the corresponding block B <b>1016</b>-B can be found in the storage space <b>714</b>.
Chunk <b>238</b>-<b>2</b>, however, is a superchunk with (super) chunk ID <b>346</b>-<b>2</b>. As illustrated here, the superchunk <b>238</b>-<b>2</b> points to an entry in the block list table <b>1006</b>. For each super chunk ID <b>1022</b> there is a plurality of corresponding block ID's <b>1008</b>. <figref idref="DRAWINGS">FIG. 12</figref> illustrates two corresponding blocks <b>1008</b>-<b>1</b> and <b>1008</b>-<b>2</b>, but for very large objects there could be a very large number of blocks for a single chunk. The block ID's <b>1008</b>-<b>1</b> and <b>1008</b>-<b>2</b> are then looked up in the chunk/block index <b>1024</b> to find the offsets <b>1028</b>-<i>x </i>and <b>1028</b>-<i>z </i>for the blocks. Finally, using the offsets <b>1028</b>-<i>x </i>and <b>1028</b>-<i>z</i>, the corresponding blocks <b>1016</b>-A and <b>1016</b>-C are located in the storage space <b>714</b>. In this example, the two blocks are not contiguous, and in fact the block <b>1016</b>-B for chunk <b>238</b>-<b>1</b> separates the two blocks for chunk <b>238</b>-<b>2</b>. Of course the size of each block is also used so that only the proper data for each block is read. This was described above with respect to <figref idref="DRAWINGS">FIG. 10B</figref>.
Implementations that do not allow ordinary chunks (such as chunk <b>238</b>-<b>1</b>) are fully hierarchical. Also note that the allocation between chunks and blocks varies based on implementation or other dynamic factors. For example, the same object could be stored as a single chunk with 100 blocks, or four chunks with 25 blocks each. Some implementations vary the number of chunks based on empirical feedback from actual usage.
The foregoing description, for purpose of explanation, has been described with reference to specific implementations. However, the illustrative discussions above are not intended to be exhaustive or to limit the invention to the precise forms disclosed. Many modifications and variations are possible in view of the above teachings. The implementations were chosen and described in order to best explain the principles of the invention and its practical applications, to thereby enable others skilled in the art to best utilize the invention and various implementations with various modifications as are suited to the particular use contemplated.
Contents6
20 sheets
Sheet 1 Sheet 2 Sheet 3 Sheet 4 Sheet 5 Sheet 6 Sheet 7 Sheet 8 Sheet 9 Sheet 10 Sheet 11 Sheet 12 Sheet 13 Sheet 14 Sheet 15 Sheet 16 Sheet 17 Sheet 18 Sheet 19 Sheet 20
Every citation, both waysCites: the store holds 19 of 20
| Document | Relation | Office | Cited during |
|---|---|---|---|
| US11775491B2 | Cited by | United States of America | Applicant |
| US12253941B2 | Cited by | United States of America | Applicant |
| US10885054B2 | Cited by | United States of America | Search report |
| US11995318B2 | Cited by | United States of America | Applicant |
| US11138082B2 | Cited by | United States of America | Applicant |
| US11593203B2 | Cited by | United States of America | Applicant |
| US12067282B2 | Cited by | United States of America | Applicant |
| US11722455B2 | Cited by | United States of America | Applicant |
| US10209901B2 | Cited by | United States of America | Applicant |
| US11861188B2 | Cited by | United States of America | Applicant |
| US12373340B2 | Cited by | United States of America | Applicant |
| US11704073B2 | Cited by | United States of America | Applicant |
| US11782625B2 | Cited by | United States of America | Applicant |
| US11822444B2 | Cited by | United States of America | Applicant |
| US12141118B2 | Cited by | United States of America | Applicant |
| US12242425B2 | Cited by | United States of America | Applicant |
| US11868309B2 | Cited by | United States of America | Applicant |
| US9400828B2 | Cited by | United States of America | Search report |
| US12341848B2 | Cited by | United States of America | Applicant |
| US12235743B2 | Cited by | United States of America | Applicant |
| US11494498B2 | Cited by | United States of America | Applicant |
| US12135654B2 | Cited by | United States of America | Applicant |
| US11385979B2 | Cited by | United States of America | Applicant |
| US12475041B2 | Cited by | United States of America | Applicant |
| US12212624B2 | Cited by | United States of America | Applicant |
| US11886288B2 | Cited by | United States of America | Applicant |
| US11281394B2 | Cited by | United States of America | Applicant |
| US12050774B2 | Cited by | United States of America | Applicant |
| US11188476B1 | Cited by | United States of America | Applicant |
| US12069133B2 | Cited by | United States of America | Applicant |
| US12093545B2 | Cited by | United States of America | Applicant |
| US12061814B2 | Cited by | United States of America | Applicant |
| US11656939B2 | Cited by | United States of America | Applicant |
| US10817431B2 | Cited by | United States of America | Applicant |
| US12032724B2 | Cited by | United States of America | Applicant |
| US11971828B2 | Cited by | United States of America | Applicant |
| US11614880B2 | Cited by | United States of America | Applicant |
| US11656768B2 | Cited by | United States of America | Applicant |
| US12067032B2 | Cited by | United States of America | Applicant |
| US12481442B2 | Cited by | United States of America | Applicant |
| US11604585B2 | Cited by | United States of America | Applicant |
| US12373289B2 | Cited by | United States of America | Applicant |
| US12147715B2 | Cited by | United States of America | Applicant |
| US11204701B2 | Cited by | United States of America | Applicant |
| US10838633B2 | Cited by | United States of America | Applicant |
| US12277106B2 | Cited by | United States of America | Applicant |
| US11671496B2 | Cited by | United States of America | Applicant |
| US11507313B2 | Cited by | United States of America | Search report |
| US12105620B2 | Cited by | United States of America | Applicant |
| US11734186B2 | Cited by | United States of America | Applicant |
| US12086029B2 | Cited by | United States of America | Applicant |
| US11310317B1 | Cited by | United States of America | Applicant |
| US11899582B2 | Cited by | United States of America | Applicant |
| US11500552B2 | Cited by | United States of America | Applicant |
| US12056386B2 | Cited by | United States of America | Applicant |
| US11442625B2 | Cited by | United States of America | Applicant |
| US11567917B2 | Cited by | United States of America | Applicant |
| US12158814B2 | Cited by | United States of America | Applicant |
| US11592985B2 | Cited by | United States of America | Applicant |
| US12117900B2 | Cited by | United States of America | Applicant |
| US11550752B2 | Cited by | United States of America | Applicant |
| US12340107B2 | Cited by | United States of America | Applicant |
| US11582046B2 | Cited by | United States of America | Applicant |
| US12066895B2 | Cited by | United States of America | Applicant |
| US12105584B2 | Cited by | United States of America | Applicant |
| US11704192B2 | Cited by | United States of America | Applicant |
| US11190580B2 | Cited by | United States of America | Applicant |
| US11354058B2 | Cited by | United States of America | Applicant |
| US12293111B2 | Cited by | United States of America | Applicant |
| US12430059B2 | Cited by | United States of America | Applicant |
| US11204830B2 | Cited by | United States of America | Applicant |
| US11210330B2 | Cited by | United States of America | Search report |
| US12137140B2 | Cited by | United States of America | Applicant |
| US12135878B2 | Cited by | United States of America | Applicant |
| US12067260B2 | Cited by | United States of America | Applicant |
| US12079494B2 | Cited by | United States of America | Applicant |
| US12282799B2 | Cited by | United States of America | Applicant |
| US12204788B1 | Cited by | United States of America | Applicant |
| US12216903B2 | Cited by | United States of America | Applicant |
| US11079962B2 | Cited by | United States of America | Applicant |
| US12314183B2 | Cited by | United States of America | Applicant |
| US11289169B2 | Cited by | United States of America | Applicant |
| US12204768B2 | Cited by | United States of America | Applicant |
| US12393340B2 | Cited by | United States of America | Applicant |
| US12366972B2 | Cited by | United States of America | Applicant |
| US11960371B2 | Cited by | United States of America | Applicant |
| US11604690B2 | Cited by | United States of America | Applicant |
| US12001700B2 | Cited by | United States of America | Applicant |
| US11775428B2 | Cited by | United States of America | Applicant |
| US11869583B2 | Cited by | United States of America | Applicant |
| US11489668B2 | Cited by | United States of America | Applicant |
| US11392522B2 | Cited by | United States of America | Applicant |
| US11922070B2 | Cited by | United States of America | Applicant |
| US11507597B2 | Cited by | United States of America | Applicant |
| US11689610B2 | Cited by | United States of America | Applicant |
| US12271359B2 | Cited by | United States of America | Applicant |
| US2019179542A1 | Cited by | United States of America | Search report |
| US11409437B2 | Cited by | United States of America | Applicant |
| US12439544B2 | Cited by | United States of America | Applicant |
| US11416144B2 | Cited by | United States of America | Applicant |
23 members in 8 offices
Priority claims6
| Document | Office | Kind | Date |
|---|---|---|---|
| 201313927010 | United States of America | A | |
| 201313927010 | United States of America | A | |
| 201314142706 | United States of America | A | |
| 13927010 | – | – | – |
| US201313927010 | – | – | – |
| US201314142706 | – | – | – |
Members23
| Document | Office | Kind | |
|---|---|---|---|
| US2014379715A1 | United States of America | A1 | |
| WO2014209911A1 | World Intellectual Property Organization (WIPO) | A1 | |
| CA2935215A1 | Canada | A1 | |
| US2015186043A1 | United States of America | A1 | |
| WO2015100416A1 | World Intellectual Property Organization (WIPO) | A1 | |
| US9158472B2This record | United States of America | B2 | |
| US2016034549A1 | United States of America | A1 | |
| EP3014487A1 | European Patent Office (EPO) | A1 | |
| AU2014369830A1 | Australia | A1 | |
| US9400828B2 | United States of America | B2 | |
| CN105940396A | China | A | |
| EP3087513A1 | European Patent Office (EPO) | A1 | |
| JP2017500670A | Japan | A | |
| DE202014010898U1 | Germany | U1 | |
| DE202014010953U1 | Germany | U1 | |
| US9600558B2 | United States of America | B2 | |
| EP3087513A4 | European Patent Office (EPO) | A4 | |
| AU2014369830B2 | Australia | B2 | |
| JP6479020B2 | Japan | B2 | |
| CN105940396B | China | B | |
| CA2935215C | Canada | C | |
| EP3014487B1 | European Patent Office (EPO) | B1 | |
| EP3087513B1 | European Patent Office (EPO) | B1 |
66 transactions on the USPTO file
Allowed without a rejection on record.
- Non-final rejections
- 0
- Final rejections
- 0
- RCEs
- 0
- Appeals
- 0
Over time
Point at a mark for the transactionTransactions
| Event | Code | |
|---|---|---|
| Payment of Maintenance Fee, 8th Year, Large EntityM1552 | M1552 | |
| Payment of Maintenance Fee, 4th Year, Large EntityM1551 | M1551 | |
| Email NotificationEML_NTR | EML_NTR | |
| Change in Power of Attorney (May Include Associate POA)PA.. | PA.. | |
| Correspondence Address ChangeC.AD | C.AD | |
| Recordation of Patent Grant MailedPGM/ | PGM/ | |
| Patent Issue Date Used in PTA CalculationAllowedPTAC | PTAC | |
| Email NotificationEML_NTR | EML_NTR | |
| Issue Notification MailedAllowedWPIR | WPIR | |
| Dispatch to FDCD1935 | D1935 | |
| Application Is Considered Ready for IssuePILS | PILS | |
| Issue Fee Payment VerifiedN084 | N084 | |
| Issue Fee Payment ReceivedIFEE | IFEE | |
| Application ready for PDX access by participating foreign officesCCRDY | CCRDY | |
| Email NotificationEML_NTR | EML_NTR | |
| Printer Rush- No mailingTCPB | TCPB | |
| Mailing Corrected Notice of AllowabilityMCNOA | MCNOA | |
| Email NotificationEML_NTR | EML_NTR | |
| PG-Pub Issue NotificationPG-ISSUE | PG-ISSUE | |
| Reasons for AllowanceEX.R | EX.R | |
| Corrected Notice of AllowabilityCNOA | CNOA | |
| Information Disclosure Statement consideredIDSC | IDSC | |
| Pubs Case Remand to TCPUBTC | PUBTC | |
| Electronic ReviewELC_RVW | ELC_RVW | |
| Email NotificationEML_NTF | EML_NTF | |
| Mail Notice of AllowanceAllowedMN/=. | MN/=. | |
| Notice of Allowance Data Verification CompletedAllowedN/=. | N/=. | |
| Miscellaneous Incoming LetterLET. | LET. | |
| Reference capture on IDSRCAP | RCAP | |
| Information Disclosure Statement (IDS) FiledM844 | M844 | |
| Information Disclosure Statement (IDS) FiledWIDS | WIDS | |
| Reasons for AllowanceEX.R | EX.R | |
| Information Disclosure Statement consideredIDSC | IDSC | |
| Information Disclosure Statement consideredIDSC | IDSC | |
| Case Docketed to Examiner in GAUDOCK | DOCK | |
| Reference capture on IDSRCAP | RCAP | |
| Information Disclosure Statement (IDS) FiledM844 | M844 | |
| Information Disclosure Statement (IDS) FiledWIDS | WIDS | |
| Electronic ReviewELC_RVW | ELC_RVW | |
| Email NotificationEML_NTF | EML_NTF | |
| PG-Pub RequestPG-RQST | PG-RQST | |
| Rescind Nonpublication Request for Pre Grant PublicationRESC | RESC | |
| Case Docketed to Examiner in GAUDOCK | DOCK | |
| FITF set to YES - revise initial settingFTFS | FTFS | |
| Application Dispatched from OIPEOIPE | OIPE | |
| Email NotificationEML_NTR | EML_NTR | |
| Email NotificationEML_NTR | EML_NTR | |
| Application Is Now CompleteCOMP | COMP | |
| Filing Receipt - UpdatedFLRCPT.U | FLRCPT.U | |
| Change in Power of Attorney (May Include Associate POA)PA.. | PA.. | |
| Sent to Classification ContractorPGPC | PGPC | |
| Patent Term Adjustment - Ready for ExaminationPTA.RFE | PTA.RFE | |
| Additional Application Filing FeesADDFLFEE | ADDFLFEE | |
| Electronic ReviewELC_RVW | ELC_RVW | |
| Email NotificationEML_NTF | EML_NTF | |
| Email NotificationEML_NTR | EML_NTR | |
| Notice Mailed--Application Incomplete--Filing Date AssignedINCD | INCD | |
| Filing ReceiptFLRCPT.O | FLRCPT.O | |
| Cleared by OIPE CSRL194 | L194 | |
| IFW Scan & PACR Auto Security ReviewSCAN | SCAN | |
| PGPubs nonPub RequestNPRQ | NPRQ | |
| Reference capture on IDSRCAP | RCAP | |
| Information Disclosure Statement (IDS) FiledM844 | M844 | |
| Information Disclosure Statement (IDS) FiledWIDS | WIDS | |
| Entity status set to undiscounted (initial default setting or status change)BIG. | BIG. | |
| Initial Exam Team nnIEXX | IEXX |
5 legal events, as the office reported them to INPADOC
Over the term
Point at a mark for the eventEvents
| Event | Code | |
|---|---|---|
| Maintenance fee paymentMAFP | MAFP | |
| Maintenance fee paymentMAFP | MAFP | |
| AssignmentAS | AS | |
| Information on status: patent grantGrantedPATENTED CASESTCF | STCF | |
| AssignmentAS | AS |
Numbers
- Publication
- 09158472
- Publication, DOCDB
- 9158472
- Publication, EPODOC
- US9158472
- Application
- 14142706
- Application, DOCDB
- 201314142706
- Application, EPODOC
- US201314142706
Titles
- English
- Hierarchical chunking of objects in a distributed storage system
Patent term adjustment
- A delay
- +103 daysthe office missed an examination deadline
- Net adjustment
- 103 days
Classification
- CPC, 7
- G06F16/27
- G06F3/065
- G06F16/278
- G06F3/067
- G06F16/2228
- G06F3/0614
- G06F16/2291
- IPC, 2
- G06F12 00
- G06F3 06
- USPC, 1
- 001001000