This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 155501673886427f5eb28b0eb82be2aecbc4dc7b (commit) via 8308e7466aaa9dbd45bf981ae4c522d22fd4bbf1 (commit) from 69395960a5db99a0607dbc4e88ca97d868848490 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 155501673886427f5eb28b0eb82be2aecbc4dc7b Author: Phil Carns <[email protected]> Date: Tue Feb 28 10:18:53 2012 -0500 edits and comments from RobR commit 8308e7466aaa9dbd45bf981ae4c522d22fd4bbf1 Author: Phil Carns <[email protected]> Date: Tue Feb 28 10:18:26 2012 -0500 edits and comments from RobR ----------------------------------------------------------------------- Summary of changes: code/doc/{resilience => }/fault-injection.txt | 0 code/doc/module.mk.in | 3 +- code/doc/resilience/client-fault-handling.txt | 10 ++- code/doc/resilience/failure-detection.txt | 43 ++++++++++--- code/doc/resilience/module.mk.in | 1 - code/doc/resilience/non-idempotent-retry.txt | 14 +++-- code/doc/resilience/object-placement.txt | 26 +++++++- code/doc/resilience/object-rebuild.txt | 54 +++++++++++------ code/doc/resilience/resilience-book.txt.in | 10 +-- code/doc/resilience/resiliency-overview.txt | 7 +-- code/doc/resilience/rosd-replication.txt | 78 +++++++++++++++++-------- 11 files changed, 167 insertions(+), 79 deletions(-) rename code/doc/{resilience => }/fault-injection.txt (100%) Diff of changes: diff --git a/code/doc/resilience/fault-injection.txt b/code/doc/fault-injection.txt similarity index 100% rename from code/doc/resilience/fault-injection.txt rename to code/doc/fault-injection.txt diff --git a/code/doc/module.mk.in b/code/doc/module.mk.in index 49b4a92..3f80ac9 100644 --- a/code/doc/module.mk.in +++ b/code/doc/module.mk.in @@ -4,4 +4,5 @@ DOCSRC += $(DIR)/test.txt \ $(DIR)/pipelining.txt \ $(DIR)/compiling-triton.txt \ $(DIR)/aesop.txt \ - $(DIR)/aesop-pbranch-cancel.txt + $(DIR)/aesop-pbranch-cancel.txt \ + $(DIR)/fault-injection.txt diff --git a/code/doc/resilience/client-fault-handling.txt b/code/doc/resilience/client-fault-handling.txt index da22561..cfddd56 100644 --- a/code/doc/resilience/client-fault-handling.txt +++ b/code/doc/resilience/client-fault-handling.txt @@ -27,11 +27,12 @@ recovery mechanisms. The client can query any server at any time to retrieve a snapshot of the current state of the system that indicates which servers are available. The client will use this information, in conjunction with an algorithmic -placement function, to map object IDs to servers. See object-placement.txt +placement function, to map object IDs to servers and to make decisions on +object placement. See object-placement.txt for more details. A client will assume that it's snapshot of the system state is correct -until it either encounters a failure (i.e., a timeout communicating with a +unless it either encounters a failure (i.e., a timeout communicating with a specific server) or else receives an error code from a server indicating that it has contacted the wrong server. The client will then refresh its state information by requesting an update from a functioning server. @@ -151,8 +152,9 @@ into each RPC) to correlate high level client operations to RPCs. Note that if the client keeps receiving the same system state information each time that it requests a copy from the servers, then the system -has reached a quiescent state (ie, it is detecting no new failures). +has reached a quiescent state (i.e., it is detecting no new failures). If operations continue to fail with retryable errors despite the fact that the system is in a quiescent state, then the client library must eventually -report a failure to the application. +report a failure to the application. This would be indicative of a software +failure. diff --git a/code/doc/resilience/failure-detection.txt b/code/doc/resilience/failure-detection.txt index 9763a12..5faf6e6 100644 --- a/code/doc/resilience/failure-detection.txt +++ b/code/doc/resilience/failure-detection.txt @@ -24,16 +24,17 @@ stateless. There is therefore no need for the servers to pro-actively detect cl failures. There is also no need for clients to detect the failure of other clients. -* Failure detection must scale to thousands of server nodes without a -significant increase in either detection or dissemination time. +* Failure detection must scale to large numbers of nodes without a +significant increase in either detection or dissemination time. Linear +scaling of detection time is not acceptable. * If failure detection protocol requires periodic exchange of messages between servers, then we will attach extra information to these messages -(ie, piggy-backing) in order to publish and disseminate +(i.e., piggy-backing) in order to publish and disseminate system-specific state information among servers that extends beyond simple "up" and "down" status values. This will give each server the ability to construct a relatively recent snapshot of the overall system state in -terms of several parameters such as load or capacity. +terms of several parameters, such as load or capacity. * We must be able to calculate an upper bound on failure detection and dissemination for any given number of servers. This information will be used to @@ -42,11 +43,24 @@ timeout values. * We must handle the case of multiple concurrent failures. +* We must gracefully handle network partitions, in which no servers have +failed but a subset of servers have lost connectivity to the other servers +in the system. In this scenario the system must at least maintain +availability for the majority partition while preventing the minority partition +from servicing requests. + * We _do not_ need strong ordering of failure detection. In other words, if server A fails before server B then it is still acceptable for server C to find out -about those failures in the reverse order as long as both notifications +about those failures in the reverse order, as long as both notifications occur within a bounded amount of time. +It is possible that future systems or system software will provide some form +of "resilience backplane" capable of reporting error conditions. The Triton +design does not rely on this functionality, but the presence of such a +component would simplify failure detection significantly. The remainder of +this document operates on the assumption that there is no resilience +backplane available in the system. + === Failure Detection Background and Terminology Failure detectors generally consist of two parts: @@ -59,7 +73,7 @@ failure detection communication among a collection of nodes The traditional mechanism for failure detection is to send _heartbeat_ messages. Heartbeat relies on regularly _telling_ other nodes that you're still alive. If a heartbeat message is not received from a node -within a given time period the the node is assumed to have failed. An alternative is to +within a given time period, then the the node is assumed to have failed. An alternative is to _ping_ other nodes, by explicitly asking them if they are still alive, and if they don't acknowledge, they are assumed to have failed. Intuitively, heartbeat only requires a single message sent to indicate liveness, @@ -175,7 +189,7 @@ system state information with minimal overhead. The *system state* component of each Triton server will be responsible for detecting failures, maintaining group membership, and disseminating general system state information. It will automatically communicate -with other servers to maintain an up-to-date view of the system as +with other servers to maintain an relatively up-to-date view of the system as a whole on each server. The data maintained on a given server can be queried by other components at any time without triggering any additional explicit communication. @@ -184,7 +198,9 @@ The interface to the system state component will allow callers to obtain a list of servers in the system along with their current status. The status will include simple "up" and "down" states, as well as finer-grained states to indicate when a server is rebuilding or attempting to join -the system. Published state information for a given server can be +the system. If the number of servers is large, then the interface will also +allow callers to retrieve subsets of servers based on ID ranges in order to +limit the scope of the data. Published state information for a given server can be queried from the system state component using a key/value interface. See the placement document for information on how data provided by the system state component is used by the placement algorithm. @@ -268,7 +284,7 @@ automatically derive appropriate timeout values for various scenarios. The system will start with conservative estimates as initial conditions and update those values over time as more information is gathered about the -storage system. +storage system, allowing faster convergence to consensus. === Failure actions @@ -301,7 +317,7 @@ else request updated system state information from another server. ==== Rebuilding: -The peer servers for a failed server (ie, servers within N distance of the +The peer servers for a failed server (i.e., servers within N distance of the failed server according to the placement algorithm, where N is the maximum replication factor) must take action to rebuild missing objects following the failure of a server. However, once a peer has detected a failure, it must wait _at @@ -319,6 +335,13 @@ code to clients). Triton will not allow writes to proceed until there is a consensus on the ownership of a given object. However, it can allow read access to those objects. +Note that the behavior described above (blocking write access for a period +of time until consensus is achieved) will only impact a subset of servers in +the system. Servers that are not participating in a rebuild will be +unaffected. Our expectation is that higher-level softare layers could work +around affected servers by selecting different OID regions when writing new +objects. + == Related work in file system failure detection === Ceph diff --git a/code/doc/resilience/module.mk.in b/code/doc/resilience/module.mk.in index 0ca79dd..a312df1 100644 --- a/code/doc/resilience/module.mk.in +++ b/code/doc/resilience/module.mk.in @@ -2,7 +2,6 @@ DIR := doc/resilience DOCSRC += \ $(DIR)/resiliency-overview.txt \ - $(DIR)/fault-injection.txt \ $(DIR)/rosd-replication.txt \ $(DIR)/object-rebuild.txt \ $(DIR)/client-fault-handling.txt \ diff --git a/code/doc/resilience/non-idempotent-retry.txt b/code/doc/resilience/non-idempotent-retry.txt index db3264a..3b3db63 100644 --- a/code/doc/resilience/non-idempotent-retry.txt +++ b/code/doc/resilience/non-idempotent-retry.txt @@ -113,8 +113,8 @@ used in DAFS <<dafs>>. === Lustre A Lustre <<Schwan>> file system consists of a set of object storage servers -and a metadata server. All object storage server operations are -idempotent, while the metadata server provides a mechanism to safely retry NI +and a metadata server. The metadata server is authoritative in terms of the +state of user-accessible files and directories provides a mechanism to safely retry NI operations. Clients assign unique transaction identifier to each metadata request. The metadata server stores this identifier as well as relevant data from the response persistently on disk @@ -134,7 +134,7 @@ client is actually a proxy for multiple independent processes. This scenario may arise when multiple cores or processes on a node share a single client instance or if I/O function calls are forwarded from lightweight compute node kernels to intermediate gateway nodes. As an example of the latter -case, the current generation IBM BlueGene/P systems use I/O forwarding +case, the IBM BlueGene/P systems use I/O forwarding nodes that each service I/O operations from many compute nodes and compute cores through a single file system client interface. @@ -203,7 +203,7 @@ non-idempotent remote storage operations with the following properties: The triton storage system is built atop a local object storage abstraction known as the Transactional Object Storage Device (TOSD). We propose that this object storage model be augmented to associate an additional parameter, known as the non-idempotent identifier, with each -object operation. This non-idempotent identifier (NIID) is an opaque 64-bit +non-idempotent object operation. This non-idempotent identifier (NIID) is an opaque 64-bit integer that uniquely identifies the client request that triggered the operation. The NIID may be set to zero for operations that are considered to be idempotent by both the file system and the @@ -250,7 +250,7 @@ These idle periods can be used for any automatic reclamation activities. === Protocol The basic protocol for client RPCs is similar to that employed by -existing systems. Each request generated by the client is assigned a new +existing systems. Each non-idempotent request generated by the client is assigned a new transaction ID. If the request fails, then the same request is resent using the same transaction ID. The retry policy and retry timeout is up to the system implementer and has no impact on how NI operations are @@ -271,7 +271,9 @@ latexmath:[sizeof(OID/NIID values) * num\_servers * max\_concurrent]. This approach is similar to that taken by Ceph, but it relaxes the requirement that transaction IDs be assigned sequentially. Requests can be stalled for an arbitrarily long period of time without impeding the reclamation of -subsequent transaction IDs. +subsequent transaction IDs. This aggressive approach to reclamation will +most likely be relevant primarily for object delete operations which have the +potential to leave shadow objects on the system. == Bibliography diff --git a/code/doc/resilience/object-placement.txt b/code/doc/resilience/object-placement.txt index 4354bd7..8fe776a 100644 --- a/code/doc/resilience/object-placement.txt +++ b/code/doc/resilience/object-placement.txt @@ -84,12 +84,22 @@ retain authoritative ability to overrule a client. The trade-off for any deterministic placement system is that once an object has been assigned an identifier and stored on the system, there is no -particularly practical way to move it. We will attempt to make intelligent +particularly practical way to move it from a storage system point of view. We will attempt to make intelligent placement decisions at object creation time. However, if an object (or set of objects) is found to be in a suboptimal location later due to changing system parameters or changing workloads, then we will explicitly reorganize the objects by selecting new object identifiers and transferring data as -needed. This approach will be explored as a long term research avenue. +needed via support in higher-level software. +This approach will be explored as a long term research avenue. The +prototype will focus on the common case of static object mapping. + +To summarize the object placement strategy of Triton, any entity in the +system (client or server) will be capable of mapping arbitrary objects to +appropriate servers using a known algorithm with no explicit communication. +This algorithm will also allow any client or server to deterministically +idenitify affected servers following a failure. Failure events +will therefore have a deterministic, well-known impact on the system in that +regard. === Implications for rebuild and replication @@ -126,7 +136,17 @@ latexmath:[$ {\cal F}(O, \{N_1,N_2,N_3,...\}) \rightarrow \{M=N_i, R_1=N_j, R_2= R_3=N_l, ...\} $] This allows us to also define a _replication group_, consisting -of nodes that a node latexmath:[$N_i$] shares objects with through replication. +of nodes that a node latexmath:[$N_i$] shares objects with through +replication. Clients can select any node in a replica group to read data +from. We expect that higher level software may leverage this functionality +to increase bandwidth in the read case by leveraging as many copies of data +as possible. For consistency reasons (to be defined in +rosd-replication.txt) only the first server in the replication group will accept +write operations, however. + +Servers will use the same mapping function as clients in order to confirm +the placement consistency for incoming requests and to identify peers for +replication purposes. == Bibliography diff --git a/code/doc/resilience/object-rebuild.txt b/code/doc/resilience/object-rebuild.txt index 649537d..85afba6 100644 --- a/code/doc/resilience/object-rebuild.txt +++ b/code/doc/resilience/object-rebuild.txt @@ -12,10 +12,10 @@ case. == Overview -Each Triton server will monitor its N*2 nearest neighbor servers according -to the placement algorithm, where N is the maximum replication factor -supported by the storage system. Because the system is using consistent -hashing, these are the only peers that may share objects with a server +Each Triton server will monitor its N*2 nearest neighbor servers using data +obtained via the SWIM protocol described in failure-detection.txt, where N is the maximum replication factor +supported by the storage system. Because of the placement scheme used by +Triton, these are the only peers that may share objects with a server following a failure. Note that the "nearest" servers in this context refer to servers that are virtually adjacent according to the object placement algorithm. There is no implication of physical locality. @@ -76,6 +76,13 @@ digraph rosd_rec_reb { === Principles and terminology +* Note from rosd-replication.txt that the default mode of replication relays +data from the primary server to its replicas in a chain configuration. The +master relays to replica 1, which then relays to replica 2, and so on. In +this chain configuration we will refer to the replica (or master) +immediately preceding a local object as its _replication parent_, while we +refer to a replica following a local object as its _replication child_. + * Rebuilding servers will _pull_ data from their peers. That is, each server individually determines which objects it is missing after a failure (by querying potential parent servers) and then copies data from @@ -127,9 +134,10 @@ incoming requests from clients or nacks with "try again in N seconds" . query the two adjacent servers in the ID space to get a list of missing object IDs that need to be created locally * those two servers can identify all missing local replicas because they are the -only servers that could be replication parents for local objects +only servers that can immediately precede the local server in a replica +chain -. create a local placeholder (ie, a <<stub,stub>>) for each missing object +. create a local placeholder (i.e., a <<stub,stub>>) for each missing object . scan local objects in order to prepare to answer queries from adjacent servers for missing objects @@ -161,7 +169,7 @@ From a server point of view, writes (and other modifications) will be forwarded to stubs in the same manner as they would be forwarded to a normal object. The updates will be applied to stubs just they would be applied to a normal object. This behavior allows stubs to stay in sync -even while rebuilds are occurring at the same time (ie, avoiding the problem +even while rebuilds are occurring at the same time (i.e., avoiding the problem of "chasing a moving target" when trying to rebuild). The "rebuild" server state is equivalent to the "nominal" server state in that client @@ -218,9 +226,18 @@ the safest general approach will be to simply restart the recovery process on all affected servers. The number of affected servers will be expanded to include those within maximum replication distance of both failed nodes. -The recovery process will be restarted from scratch to make sure that we -don't accidentally assume that we have finished creating stubs for an object -already if it actually requires additional stubs due to the second failure. +The recovery process will be restarted from scratch in this scenario rather +than attempting to account for objects that have already been processed in +response to the previous failure. For example, consider a server that (as +part of recovery) has created stubs for half of the failed objects that it +must reconstruct. If another failure occurs at this point, we cannot make +any assumptions about the integrity of the ID space that the server has +previously processed, because the new failure may have impacted portions of +that same ID range. It is also possible that another immediately adjacent +server has failed. The local server will therefore stop its current +recovery procedure and restart it based on the new state of the system to +insure that all objects are accounted for and that there are no race conditions +between competing recovery procedures. This approach requires that stub creation be idempotent in case a server calculates that it needs to create a new stub when there is one already in place from a @@ -238,7 +255,7 @@ reconstruction. [[rebuild]] The Triton rebuild process is driven by servers that hold secondary objects -that are out of sync with their primaries (ie, stubs). These servers "pull" data +that are out of sync with their primaries (i.e., stubs). These servers "pull" data from primary (or parent) objects. Once they have pulled all necessary data for a given object, then the stub designation will be removed from that object and clients will be allowed to interact with it as usual. @@ -249,14 +266,14 @@ will want to batch these operations together and generate a schedule for an optimal order to perform rebuilds. . for each stub object: -.. retrieve update state (ie, what updates have been applied to the stub +.. retrieve update state (i.e., what updates have been applied to the stub already- see the replication documentation for details) ... may be "0" if it this is a new stub with no data .. contact parent to request rebuild data .. on parent: ... checkpoint the object in question -... stream a _pickled_ (ie, serialized) copy of any object data that is newer than the replica's +... stream a _pickled_ (i.e,, serialized) copy of any object data that is newer than the replica's update state to the replica ... delete checkpoint .. on local server: @@ -301,7 +318,7 @@ At a global level, there will be a subset of servers that each need to pull replica data for some objects from some peers. In the long run a research contribution here would be to determine an optimal global schedule for the data transfers so that we utilize the network and disks effectively -and priorities critical objects. +and prioritize critical objects. We won't do this in the prototype, but here are some thing that we will want to consider in future work: @@ -341,7 +358,7 @@ These operations will probably be implemented inefficiently at first until we implement the first cut of recovery and rebuild to make sure that we have the right API and use cases. Then we can go back and optimize these as needed within the TOSD. The TOSD already utilizes an internal database that -can be leverage for fast range queries and bulk operations. +can be leveraged for fast range queries and bulk operations. == Interacting with servers that are rebuilding @@ -355,13 +372,14 @@ server activities lead to which client responses: * attempt to access a server that is in "recovery state": ** receive a nack saying "try again in N seconds" -* attempted to access an object stub (ie, an object that needs to be rebuilt +* attempted to access an object stub (i.e., an object that needs to be rebuilt or resynced): ** receive a nack saying "try primary instead" * attempt to access a server that is down ** timeout, perceived transient network error, or retryable error -** retryable errors includes server responses that say "I'm dead" +** retryable errors includes server nacks indicating that the server has +failed and is not valid for any storage operations === Possible optimization: allowing client access to rebuilding replicas @@ -380,7 +398,7 @@ as long as they have at least started or queued up the rebuild process for that object. Servers could even initiate a rebuild on demand if a client tried to access a rebuilding replica. -This optimization will be explored in future work. It is not necessary +These optimization will be explored in future work. It is not necessary for prototype correctness. === Explicitly failing a server diff --git a/code/doc/resilience/resilience-book.txt.in b/code/doc/resilience/resilience-book.txt.in index 6708bef..d988970 100644 --- a/code/doc/resilience/resilience-book.txt.in +++ b/code/doc/resilience/resilience-book.txt.in @@ -1,21 +1,19 @@ -Triton server-driven response to storage failures -================================================= +Triton: Data Organization, Replication, and Resilience +====================================================== include::@abs_srcdir@/resiliency-overview.txt[] -include::@abs_srcdir@/failure-detection.txt[] - include::@abs_srcdir@/object-placement.txt[] include::@abs_srcdir@/rosd-replication.txt[] +include::@abs_srcdir@/failure-detection.txt[] + include::@abs_srcdir@/object-rebuild.txt[] include::@abs_srcdir@/client-fault-handling.txt[] include::@abs_srcdir@/non-idempotent-retry.txt[] -include::@abs_srcdir@/fault-injection.txt[] - include::@abs_srcdir@/glossary.txt[] diff --git a/code/doc/resilience/resiliency-overview.txt b/code/doc/resilience/resiliency-overview.txt index c5f2601..35a5bf2 100644 --- a/code/doc/resilience/resiliency-overview.txt +++ b/code/doc/resilience/resiliency-overview.txt @@ -10,7 +10,7 @@ resiliency. This document outlines the Triton object server approach to resiliency. Triton employs a server-driven response to storage failures, in which servers actively detect failures and autonomously repair the storage system in response to those failures. Some of the most -notable design features of Triton from a resiliency perspective include: +notable design features of Triton, from a resiliency perspective, include: * integrated replication mechanisms that can leverage commodity storage hardware @@ -18,7 +18,7 @@ hardware affected objects * scalable algorithms for failure detection, object placement, and replication -* tunable semantics +* tunable consistency semantics Note that the Triton object storage system does not directly provide a file system interface or a traditional POSIX name space. It simply provides access to @@ -49,7 +49,4 @@ respond to various fault scenarios. transient failures, especially in the context of non-idempotent storage operations -* fault-injection.txt: how to inject various faults for -prototyping/testing/validation purposes - * glossary.txt: definitions for common terms that appear in the above documents diff --git a/code/doc/resilience/rosd-replication.txt b/code/doc/resilience/rosd-replication.txt index d9ec7f0..43479a6 100644 --- a/code/doc/resilience/rosd-replication.txt +++ b/code/doc/resilience/rosd-replication.txt @@ -25,35 +25,46 @@ simultaneously to shared objects or unique objects. Aggregate read performance is a high priority for the ASG project. I/O access patterns may or may not be bursty. -We anticipate that HPC workloads often involve highly coordinated clients -and high level I/O libraries. Such workloads do not typically require POSIX -read/write semantics (notably, MPI-IO, NFS, and PVFS have proved successful -in many deployments despite a lack of strict POSIX conformance). The ROSD -will support a range of possible semantics at run time. The semantics will -therefore be tunable per-application to reflect the needs of each -application rather than enforcing POSIX semantics for all applications. +The triton servers provide a distributed object storage service rather +than a conventional file system interface, but we expect that file +system interfaces will be built atop the object storage interface. +The presence of a file system layer atop the system does not imply +that POSIX read/write consistency semantics are required at the object +level, however. HPC workloads often involve highly coordinated clients +and high level I/O libraries. Such workloads do not require POSIX +read/write semantics (notably, MPI-IO, NFS, +and PVFS have proved successful in many deployments despite a lack of +strict POSIX conformance). The ROSD semantics will therefore be tunable +per-application to reflect the needs of each application, regardless of the +application interface that it is using. ==== Hardware An exascale storage system will likely rely on commodity storate components in order to achieve performance targets in a cost-effective manner. In this deployment model, there is no shared storage between -servers. All redundancy must be supplied in software by the +servers. Server-level redundancy must be supplied in software by the storage system. ==== Redundancy model This document describes the replication of data, in which all data is -fully copied to one or more redundant locations in order to tolerate +fully copied to one or more redundant servers in order to tolerate failure. An alternative to this approach is to use RAID or erasure coding -algorithms to tolerate failures while consuming less storage capacity. +algorithms across servers to tolerate failures while consuming less storage capacity. Distributed erasure coding algorithms impose significant challenges to preserving consistency in the presence of concurrent writes, however. Our prototype will therefore focus on object replication in the short -term and explore erasure coding as a longer range option. We believe -that in the future the system is likely to support a model in which replicated +term and explore erasure coding as a longer range option. In the future, the system is likely to support a model in which replicated data is automatically migrated into an erasure coded format in order to -reclaim storage capacity once the data is quiescent. +reclaim storage capacity, once the data is quiescent. + +We also anticipate that the local storage devices attached to servers +will support RAID configurations as well. This functionality will allow +servers to tolerate local disk failures without perturbing other servers. In future work we will use mathematical models in conjunction +with discrete event simulation to determine the appropriate balance +of local redundancy and inter-server redundancy based on the expected +component failure rate and the rebuild time of inter-server replication. ==== Rebuild @@ -68,7 +79,7 @@ sets that are not fully synchronized. More specifically, the system should not block access to objects while they are in the process of rebuilding. The replication protocol should be capable of accepting new write operations while simultaneously rebuilding failed objects. This is similar to the -behavior or traditional RAID disk arrays, in which availability is +behavior of traditional RAID disk arrays, in which availability is maintained even while the array is in a degraded mode. [[sec-overview]] @@ -144,10 +155,10 @@ database entries. * Triton object replication deals only with _replicated_ data, and not _partitioned_ data. We are replicating -at an object granularity, not at striped file granularity. Any semantics +at an object granularity, not at striped file or striped object granularity. Any semantics required across striped sets will be enforced at a higher level of the storage hierarchy. This means that a distributed object storage system need -not implement locking or transactional semantics _across stripe boundaries_. +not implement locking or transactional semantics _across objects_. * Reads that overlap with concurrent writes are allowed to return undefined results. We expect HPC applications to be coordinated at a higher level so @@ -183,7 +194,12 @@ independently. In addition, many distributed Internet service storage systems need only support write-once semantics. For example, it may store key/value pairs or images that will not be updated once written. In contrast, HPC systems must at least support appendable objects in order to -handle the volume of data stored by large scale simulations. +handle the volume of data stored by large scale simulations, which often cannot be +transferred in a single write operation. The Triton +storage system is intended to persistently and resiliently store all +classes of persistent objects. That includes both bulk data (which is often +written once) and metadata (including indices which might be updated +frequently). === Challenges @@ -192,9 +208,15 @@ must address. First of all, it must avoid serializing I/O operations to shared objects. Concurrent read and write operations should be able to proceed immediately in order to maximize bandwidth. Secondly, the replication must preserve consistency across replicas even when multiple -write operations are in flight concurrently. Finally, each replica should -be self-describing, in that the system can query replicas to determine if -they are synchronized, and if not, exactly how they diverge. +write operations are in flight concurrently. In this scenario, +each replica may be applying interleaved or even overlapping write +operations simultaneously without synchronization from other replicas. Even +in this scenario, the storage system must honor well-defined read and write +semantics at all times and preserve replica consistency even if failures +occur while writing. +Finally, each replica should be self-describing, in that the system +can query replicas to determine if they are synchronized, and if not, +exactly how they diverge. == The Triton ROSD replication protocol @@ -300,7 +322,7 @@ increasing numbers rather than time stamps, and write ordering will be resolved before reads occur. The default rules for replica divergence in the ROSD protocol will be defined as follows: Triton object replicas can -diverge as long as concurrent client I/O operations are in flight. Once a +diverge as long as concurrent client write operations are in flight. Once a write completes from a client's perspective, it is guaranteed to be visible at all replicas and in the correct order relative to any _subsequent_ reads or writes. @@ -362,13 +384,15 @@ approach. The first is that the server is unaware of the scope of the overall write operation and therefore cannot schedule accordingly. It only sees a small part of the overall operation at a time. The second drawback is that it makes it difficult to preserve atomicity (and POSIX semantics) -across a write operation that has been broken into distinct segments. +across a write operation that has been broken into distinct segments, when +this is required. The approach taken by Triton instead is to establish a "pipeline" operation, in which the client informs the server of the entire desired transfer, and the server then chooses its own plan for dividing the transfer into smaller components. The server is aware of the entire update and can schedule or -overlap operations in an optimal manner. +overlap operations in an optimal manner, similar to previous reseach in +server-directed I/O <<Seamons>>. In order to implement this, the TOSD local storage abstraction provides a transaction construct that allows update numbers to refer to an arbitrary number of write @@ -394,7 +418,7 @@ alternative model would be to accept the write operation at the master as usual, but then have the master fan out updates to a set of replicas simultaneously. This approach may be more appropriate for small, latency bound operations, while the chain approach is likely more appropriate for -large bandwidth-bound operations. We evaluate the fan-out replication +large bandwidth-bound operations. We will evaluate the fan-out replication model and develop heuristics to automatically select the appropriate protocol for each type of operation. @@ -434,7 +458,7 @@ other things. === Discussion -A ROSD server has to handle four failure classes: +The ROSD protocol has to handle four failure classes: . Network failure while communicating with a remote replica * Retry implicitly via RPC unless/until the Traffic Cop (which in turn @@ -499,3 +523,7 @@ WA, November 2006. - [[[Thomas]]] Robert H. Thomas. "A majority consensus approach to concurrency control for multiple copy databases". ACM Transactions on Database Systems 4 (2): 180–209. +- [[[Searmons]]] Seamons, K. E., Chen, Y., Jones, P., Jozwiak, J., and Winslett, M. 1995. + Server-directed collective I/O in Panda. In Proceedings of Supercomputing ’95. IEEE + Computer Society + Press, Los Alamitos, Calif. hooks/post-receive -- Triton Repository