[RFC]: Cleanup processing replicas if the transferring client is down

[nickyc975]

Changes proposed

We noticed the issue #974, and we also encountered the problem recently. We have implemented and verified a simple solution:

1. We refactored the client management logic in master. All clients must be registered to the master before sending mount/put_start requests. Therefore, the master can keep a list of valid clients and recognize faulted clients through pinging timeout.
2. Clients send their client_id along with put_start requests, so the master can identify which client is responsible for transfering data of a given object. When a client fault is detected, the master checks all transferring objects, discards processing replicas of objects which are being processed by the faulted client. Note that completed replicas will not be discarded, so the objects can still be used if they have any completed replicas. Otherwise, they will be removed and other clients will be allowed to insert the same keys.
3. There are cases that the faulted client is not terminated and is still writing data to discarded replicas. Therefore, discarded replicas will be put into a staging list instead of being released immediately. After a long-enough and configurable timeout (e.g. 10 or 20 minutes), the discarded replicas can be released safely.

Limitations

1. Currently, there is no reliable way to prevent potential data corruption caused by reusing space of discarded replicas. We can only set a long enough timeout for releasing discarded replicas to minimize the probability.

Before submitting a new issue...

• Make sure you already searched for relevant issues and read the documentation

[ykwd]

Thanks for reporting this and for sharing your proposed solution.

I think a simpler approach could be to directly record the put start time, and during eviction, when all KVs are scanned, we can delete any entries that have timed out (e.g., after 10 or 20 minutes).

All clients must be registered to the master before sending mount/put_start requests

Considering the possible network partition between clients and master, and the clients might be temporarily disconnected from master, it may take some effort to enforce this rule.
For the same reason, the master cannot reliably determine whether a client has actually crashed or is just experiencing a temporary network partition.

Also, setting a timeout of 10–20 minutes should generally be sufficient, since the Transfer Engine already has its own timeout mechanism — if the network is disconnected, it will quickly return an error anyway.

[nickyc975]

Hi, @ykwd. Thanks for responding!

Considering the possible network partition between clients and master, and the clients might be temporarily disconnected from master, it may take some effort to enforce this rule.

The Register operation could simply replace the existing ServerReady RPC (see master_client.cpp:226). If register fails, we should consider the master as unreachable.

For the same reason, the master cannot reliably determine whether a client has actually crashed or is just experiencing a temporary network partition.
Also, setting a timeout of 10–20 minutes should generally be sufficient, since the Transfer Engine already has its own timeout mechanism — if the network is disconnected, it will quickly return an error anyway.

We do not need to distinguish client crashing and temporary network partition. What we need is to allow processing keys to be overwriten by other clients if the processing client is not responding. A timeout of 10–20 minutes is sufficient for safely releasing the space, but is way too long for error recovery.

The key point of the proposed solution is that, the processing keys can be overwriten once we found that the client is not responding, which is typically a timeout of tens of seconds. The shorter the timeout is, the more chances we have to re-insert and reuse the KV cache. And for data integrity, we can just set a long enough timeout for releasing space, like 10-20 minutes or even hours.

[ykwd]

What we need is to allow processing keys to be overwriten by other clients if the processing client is not responding. A timeout of 10–20 minutes is sufficient for safely releasing the space, but is way too long for error recovery.

The key point of the proposed solution is that, the processing keys can be overwriten once we found that the client is not responding, which is typically a timeout of tens of seconds.

I see the point. It'd be undesirable if a key is at the processing status for 10 minutes, during which no one can put this key again. So what about this solution: If a key is processed for tens of seconds, we either remove the key and put the memory to the staging list to wait for release, or we allow a client to put another replica of this key.

[nickyc975]

So what about this solution: If a key is processed for tens of seconds, we either remove the key and put the memory to the staging list to wait for release, or we allow a client to put another replica of this key.

Then it would be tricky to profile that 'tens of seconds'.

[nickyc975]

And another reason for a Register operation is that, we can sync some cluster level configurations from master to clients in Register result, e.g. the fs_dir of DFS. This could be useful in the future to reduce redundant RPC methods.

[ykwd]

Then it would be tricky to profile that 'tens of seconds'.

Perhaps we can discard the key when a new client tries to put this key. If this key has not been completed for 30s, we discard it and let this client do the put.

And another reason for a Register operation is that, we can sync some cluster level configurations from master to clients in Register result, e.g. the fs_dir of DFS. This could be useful in the future to reduce redundant RPC methods.

Currently, when a client connects to the master, it accesses the ServiceReady interface via RPC. We can return the cluster-level information there.

Overall, your proposal looks good, but I think there might be a simpler way to achieve the same effect with fewer changes.

[nickyc975]

Perhaps we can discard the key when a new client tries to put this key. If this key has not been completed for 30s, we discard it and let this client do the put.

So the whole solution should be look like this:

1. Add client_id parameter to put_start/put_end/put_revoke, and save the client_id in object metadata. This is needed because now we may have more than one client that is processing the same key. We must identify the valid client and prevent unexpected put_end/put_revoke.
2. For a successful put_start, save the client_id and the current timestamp in object metadata. When another put_start on the same object arrived, if the object has been in processing state for more than 30s (maybe configurable), we discard existing replicas of the object and renew its client_id and timestamp.
3. Discarded replicas will be put into a staging list, and be released after 10-20 minutes (maybe configurable).
4. In eviction, if an object is found stucking in processing state for more than 10-20 minutes (can be the same as timeout of discarded replicas), we remove it and release the space immediately.

This is indeed simpler and requires fewer changes. But it may introduce two timeout configurations, one for discarding replicas, and the other for releasing space. Is it OK?

[ykwd]

Thank you for the detailed explanation. It looks good to me.

[nickyc975]

Thank you for the detailed explanation. It looks good to me.

Ok, then I'll submit a PR as soon as possible.