Erasure Code Support¶

EC duplication with composite rings¶

It is recommended that EC Duplication is used with Composite Rings in order to disperse duplicate fragments across regions.

When EC duplication is used, it is highly desirable to have one duplicate of each fragment placed in each region. This ensures that a set of ec_num_data_fragments unique fragments (the minimum needed to reconstruct an object) can always be assembled from a single region. This in turn means that objects are robust in the event of an entire region becoming unavailable.

This can be achieved by using a composite ring with the following properties:

• The number of component rings in the composite ring is equal to the ec_duplication_factor for the policy.
• Each component ring has a number of replicas that is equal to the sum of ec_num_data_fragments and ec_num_parity_fragments.
• Each component ring is populated with devices in a unique region.

This arrangement results in each component ring in the composite ring, and therefore each region, having one copy of each fragment.

For example, consider a Swift cluster with two regions, region1 and region2 and a 4+2x2 EC policy schema. This policy should use a composite ring with two component rings, ring1 and ring2, having devices exclusively in regions region1 and region2 respectively. Each component ring should have replicas = 6. As a result, the first 6 fragments for an object will always be placed in ring1 (i.e. in region1) and the second 6 duplicate fragments will always be placed in ring2 (i.e. in region2).

Conversely, a conventional ring spanning the two regions may give a suboptimal distribution of duplicates across the regions; it is possible for duplicates of the same fragment to be placed in the same region, and consequently for another region to have no copies of that fragment. This may make it impossible to assemble a set of ec_num_data_fragments unique fragments from a single region. For example, the conventional ring could have a pathologically sub-optimal placement such as:

r1
  <timestamp>#0#d.data
  <timestamp>#0#d.data
  <timestamp>#2#d.data
  <timestamp>#2#d.data
  <timestamp>#4#d.data
  <timestamp>#4#d.data
r2
  <timestamp>#1#d.data
  <timestamp>#1#d.data
  <timestamp>#3#d.data
  <timestamp>#3#d.data
  <timestamp>#5#d.data
  <timestamp>#5#d.data

In this case, the object cannot be reconstructed from a single region; region1 has only the fragments with index 0, 2, 4 and region2 has the other 3 indexes, but we need 4 unique indexes to be able to rebuild an object.

Node Selection Strategy for Reads¶

Proxy servers require a set of unique fragment indexes to decode the original object when handling a GET request to an EC policy. With a conventional EC policy, this is very likely to be the outcome of reading fragments from a random selection of backend nodes. With an EC Duplication policy it is significantly more likely that responses from a random selection of backend nodes might include some duplicated fragments.

For this reason it is strongly recommended that EC Duplication always be deployed in combination with Composite Rings and proxy server read affinity.

Under normal conditions with the recommended deployment, read affinity will cause a proxy server to first attempt to read fragments from nodes in its local region. These fragments are guaranteed to be unique with respect to each other. Even if there are a small number of local failures, unique local parity fragments will make up the difference. However, should enough local primary storage nodes fail, such that sufficient unique fragments are not available in the local region, a global EC cluster will proceed to read fragments from the other region(s). Random reads from the remote region are not guaranteed to return unique fragments; with EC Duplication there is a significantly high probability that the proxy sever will encounter a fragment that is a duplicate of one it has already found in the local region. The proxy server will ignore these and make additional requests until it accumulates the required set of unique fragments, potentially searching all the primary and handoff locations in the local and remote regions before ultimately failing the read.

A global EC deployment configured as recommended is therefore extremely resilient. However, under extreme failure conditions read handling can be inefficient because nodes in other regions are guaranteed to have some fragments which are duplicates of those the proxy server has already received. Work is in progress to improve the proxy server node selection strategy such that when it is necessary to read from other regions, nodes that are likely to have useful fragments are preferred over those that are likely to return a duplicate.

Efficient Cross Region Rebuild¶

Work is also in progress to improve the object-reconstructor efficiency for Global EC policies. Unlike the proxy server, the reconstructor does not apply any read affinity settings when gathering fragments. It is therefore likely to receive duplicated fragments (i.e. make wasted backend GET requests) while performing every fragment reconstruction.

Additionally, other reconstructor optimisations for Global EC are under investigation:

• Since fragments are duplicated between regions it may in some cases be more attractive to restore failed fragments from their duplicates in another region instead of rebuilding them from other fragments in the local region.
• Conversely, to avoid WAN transfer it may be more attractive to rebuild fragments from local parity.
• During rebalance it will always be more attractive to revert a fragment from it’s old-primary to it’s new primary rather than rebuilding or transferring a duplicate from the remote region.

High Level¶

The Proxy Server handles Erasure Coding in a different manner than replication, therefore there are several code paths unique to EC policies either though sub classing or simple conditionals. Taking a closer look at the PUT and the GET paths will help make this clearer. But first, a high level overview of how an object flows through the system:

Note how:

• Incoming objects are buffered into segments at the proxy.
• Segments are erasure coded into fragments at the proxy.
• The proxy stripes fragments across participating nodes such that the on-disk stored files that we call a fragment archive is appended with each new fragment.

This scheme makes it possible to minimize the number of on-disk files given our segmenting and fragmenting.

Multi_Phase Conversation¶

Multi-part MIME document support is used to allow the proxy to engage in a handshake conversation with the storage node for processing PUT requests. This is required for a few different reasons.

1. From the perspective of the storage node, a fragment archive is really just another object, we need a mechanism to send down the original object etag after all fragment archives have landed.
2. Without introducing strong consistency semantics, the proxy needs a mechanism to know when a quorum of fragment archives have actually made it to disk before it can inform the client of a successful PUT.

MIME supports a conversation between the proxy and the storage nodes for every PUT. This provides us with the ability to handle a PUT in one connection and assure that we have the essence of a 2 phase commit, basically having the proxy communicate back to the storage nodes once it has confirmation that a quorum of fragment archives in the set have been written.

For the first phase of the conversation the proxy requires a quorum of ec_ndata + 1 fragment archives to be successfully put to storage nodes. This ensures that the object could still be reconstructed even if one of the fragment archives becomes unavailable. As described above, each fragment archive file is named:

<ts>#<frag_index>.data

where ts is the timestamp and frag_index is the fragment archive index.

During the second phase of the conversation the proxy communicates a confirmation to storage nodes that the fragment archive quorum has been achieved. This causes each storage node to rename the fragment archive written in the first phase of the conversation to include the substring #d in its name:

<ts>#<frag_index>#d.data

This indicates to the object server that this fragment archive is durable and that there is a set of data files that are durable at timestamp ts.

For the second phase of the conversation the proxy requires a quorum of ec_ndata + 1 successful commits on storage nodes. This ensures that there are sufficient committed fragment archives for the object to be reconstructed even if one becomes unavailable. The reconstructor ensures that the durable state is replicated on storage nodes where it may be missing.

Note that the completion of the commit phase of the conversation is also a signal for the object server to go ahead and immediately delete older timestamp files for this object. This is critical as we do not want to delete the older object until the storage node has confirmation from the proxy, via the multi-phase conversation, that the other nodes have landed enough for a quorum.

The basic flow looks like this:

• The Proxy Server erasure codes and streams the object fragments (ec_ndata + ec_nparity) to the storage nodes.
• The storage nodes store objects as EC archives and upon finishing object data/metadata write, send a 1st-phase response to proxy.
• Upon quorum of storage nodes responses, the proxy initiates 2nd-phase by sending commit confirmations to object servers.
• Upon receipt of commit message, object servers rename .data files to include the #d substring, indicating successful PUT, and send a final response to the proxy server.
• The proxy waits for ec_ndata + 1 object servers to respond with a success (2xx) status before responding to the client with a successful status.

Here is a high level example of what the conversation looks like:

proxy: PUT /p/a/c/o
     Transfer-Encoding': 'chunked'
     Expect': '100-continue'
     X-Backend-Obj-Multiphase-Commit: yes
obj:   100 Continue
     X-Obj-Multiphase-Commit: yes
proxy: --MIMEboundary
     X-Document: object body
     <obj_data>
     --MIMEboundary
     X-Document: object metadata
     Content-MD5: <footer_meta_cksum>
     <footer_meta>
     --MIMEboundary
<object server writes data, metadata to <ts>#<frag_index>.data file>
obj:   100 Continue
<quorum>
proxy: X-Document: put commit
     commit_confirmation
     --MIMEboundary--
<object server renames <ts>#<frag_index>.data to <ts>#<frag_index>#d.data>
obj:   20x
<proxy waits to receive >=2 2xx responses>
proxy: 2xx -> client

A few key points on the durable state of a fragment archive:

• A durable fragment archive means that there exist sufficient other fragment archives elsewhere in the cluster (durable and/or non-durable) to reconstruct the object.
• When a proxy does a GET, it will require at least one object server to respond with a fragment archive is durable before reconstructing and returning the object to the client.

Partial PUT Failures¶

A partial PUT failure has a few different modes. In one scenario the Proxy Server is alive through the entire PUT conversation. This is a very straightforward case. The client will receive a good response if and only if a quorum of fragment archives were successfully landed on their storage nodes. In this case the Reconstructor will discover the missing fragment archives, perform a reconstruction and deliver those fragment archives to their nodes.

The more interesting case is what happens if the proxy dies in the middle of a conversation. If it turns out that a quorum had been met and the commit phase of the conversation finished, its as simple as the previous case in that the reconstructor will repair things. However, if the commit didn’t get a chance to happen then some number of the storage nodes have .data files on them (fragment archives) but none of them knows whether there are enough elsewhere for the entire object to be reconstructed. In this case the client will not have received a 2xx response so there is no issue there, however, it is left to the storage nodes to clean up the stale fragment archives. Work is ongoing in this area to enable the proxy to play a role in reviving these fragment archives, however, for the current release, a proxy failure after the start of a conversation but before the commit message will simply result in a PUT failure.

GET¶

The GET for EC is different enough from replication that subclassing the BaseObjectController to the ECObjectController enables an efficient way to implement the high level steps described earlier:

1. The proxy server makes simultaneous requests to ec_ndata primary object server nodes with goal of finding a set of ec_ndata distinct EC archives at the same timestamp, and an indication from at least one object server that a durable fragment archive exists for that timestamp. If this goal is not achieved with the first ec_ndata requests then the proxy server continues to issue requests to the remaining primary nodes and then handoff nodes.
2. As soon as the proxy server has found a usable set of ec_ndata EC archives, it starts to call PyECLib to decode fragments as they are returned by the object server nodes.
3. The proxy server creates Etag and content length headers for the client response since each EC archive’s metadata is valid only for that archive.
4. The proxy streams the decoded data it has back to the client.

Note that the proxy does not require all objects servers to have a durable fragment archive to return in response to a GET. The proxy will be satisfied if just one object server has a durable fragment archive at the same timestamp as EC archives returned from other object servers. This means that the proxy can successfully GET an object that had missing durable state on some nodes when it was PUT (i.e. a partial PUT failure occurred).

Note also that an object server may inform the proxy server that it has more than one EC archive for different timestamps and/or fragment indexes, which may cause the proxy server to issue multiple requests for distinct EC archives to that object server. (This situation can temporarily occur after a ring rebalance when a handoff node storing an archive has become a primary node and received its primary archive but not yet moved the handoff archive to its primary node.)

The proxy may receive EC archives having different timestamps, and may receive several EC archives having the same index. The proxy therefore ensures that it has sufficient EC archives with the same timestamp and distinct fragment indexes before considering a GET to be successful.

DiskFile¶

Erasure code policies use subclassed ECDiskFile, ECDiskFileWriter, ECDiskFileReader and ECDiskFileManager to implement EC specific handling of on disk files. This includes things like file name manipulation to include the fragment index and durable state in the filename, construction of EC specific hashes.pkl file to include fragment index information, etc.

Job Construction and Processing¶

Because of the nature of the work it has to do as described above, the reconstructor builds jobs for a single job processor. The job itself contains all of the information needed for the processor to execute the job which may be a synchronization or a data reversion. There may be a mix of jobs that perform both of these operations on the same suffix directory.

Jobs are constructed on a per-partition basis and then per-fragment-index basis. That is, there will be one job for every fragment index in a partition. Performing this construction “up front” like this helps minimize the interaction between nodes collecting hashes.pkl information.

Once a set of jobs for a partition has been constructed, those jobs are sent off to threads for execution. The single job processor then performs the necessary actions, working closely with ssync to carry out its instructions. For data reversion, the actual objects themselves are cleaned up via the ssync module and once that partition’s set of jobs is complete, the reconstructor will attempt to remove the relevant directory structures.

Job construction must account for a variety of scenarios, including:

1. A partition directory with all fragment indexes matching the local node index. This is the case where everything is where it belongs and we just need to compare hashes and sync if needed. Here we simply sync with our partners.
2. A partition directory with at least one local fragment index and mix of others. Here we need to sync with our partners where fragment indexes matches the local_id, all others are sync’d with their home nodes and then deleted.
3. A partition directory with no local fragment index and just one or more of others. Here we sync with just the home nodes for the fragment indexes that we have and then all the local archives are deleted. This is the basic handoff reversion case.

Node Communication¶

The replicators talk to all nodes who have a copy of their object, typically just 2 other nodes. For EC, having each reconstructor node talk to all nodes would incur a large amount of overhead as there will typically be a much larger number of nodes participating in the EC scheme. Therefore, the reconstructor is built to talk to its adjacent nodes on the ring only. These nodes are typically referred to as partners.

Reconstruction¶

Reconstruction can be thought of sort of like replication but with an extra step in the middle. The reconstructor is hard-wired to use ssync to determine what is missing and desired by the other side. However, before an object is sent over the wire it needs to be reconstructed from the remaining fragments as the local fragment is just that - a different fragment index than what the other end is asking for.

Thus, there are hooks in ssync for EC based policies. One case would be for basic reconstruction which, at a high level, looks like this:

• Determine which nodes need to be contacted to collect other EC archives needed to perform reconstruction.
• Update the etag and fragment index metadata elements of the newly constructed fragment archive.
• Establish a connection to the target nodes and give ssync a DiskFileLike class from which it can stream data.

The reader in this class gathers fragments from the nodes and uses PyECLib to reconstruct each segment before yielding data back to ssync. Essentially what this means is that data is buffered, in memory, on a per segment basis at the node performing reconstruction and each segment is dynamically reconstructed and delivered to ssync_sender where the send_put() method will ship them on over. The sender is then responsible for deleting the objects as they are sent in the case of data reversion.