Confined recovery in a distributed computing system

1. A computer-implemented method for executing a confined recovery in a distributed computing system having a plurality of worker systems, the worker systems executing a computation in a plurality of supersteps, the worker systems having a plurality of partitions executing a defined function during the supersteps that exchange messages with partitions of other worker systems, the method comprising:

identifying a failed worker system from the plurality of worker systems, wherein the failed worker system failed to execute during a current superstep and lost exchanged messages;

determining states of the partitions of the worker systems during supersteps preceding the current superstep;

determining, responsive to the determination of the states, a recovery initiation superstep preceding the current superstep in which all messages for the recovery initiation superstep are available;

determining a recovery set of partitions for which messages in supersteps after the recovery initiation superstep are not available;

assigning each partition in the recovery set of partitions to a recovery worker system from a set of recovery worker systems configured to process the partition, wherein the set of recovery worker systems includes a proper subset of the plurality of worker systems;

determining messaging worker systems responsible for sending the messages in supersteps after the recovery initiation superstep that are not available, the messaging worker systems being a proper subset of the plurality of worker systems and being different than the recovery worker systems;

instructing the recovery worker systems to execute the defined function for the partitions in the recovery set starting at the recovery initiation superstep to recover the lost exchanged messages; and

instructing the messaging worker systems to transmit the messages in supersteps after the recovery initiation superstep that are not available to the recovery work systems for use in processing the recovery set of partitions.

2. The computer-implemented method of claim 1 , wherein the messages transmitted by the messaging worker systems are stored in a persistent storage.

3. The computer-implemented method of claim 1 , further comprising:

assigning partitions of the failed worker system to at least one of the plurality of worker systems that has not failed.

4. The computer-implemented method of claim 1 , wherein determining states of the partitions of the worker systems comprises:

determining whether a worker system has access to a complete set of incoming messages for a partition at a superstep, the complete set of incoming messages to be processed at the superstep.

5. The computer-implemented method of claim 1 , wherein determining states of the partitions of the worker systems comprises:

determining whether a worker system has access to a complete set of outgoing messages for a partition at a superstep, the complete set of outgoing messages to be transmitted at the superstep.

6. The computer-implemented method of claim 1 , further comprising:

receiving a plurality of status messages from the plurality of worker systems, a status message from a worker system indicating whether the worker system successfully transmitted its outgoing messages;

creating a group status message indicating whether each of the plurality of worker systems succeeded in transmitting its outgoing messages; and

transmitting the group status message to each of the plurality of worker systems.

7. A non-transitory computer-readable storage medium storing executable computer program instructions for executing a confined recovery in a distributed computing system having a plurality of worker systems, the worker systems executing a computation in a plurality of supersteps, the worker systems having a plurality of partitions executing a defined function during the supersteps that exchange messages with partitions of other worker systems, the computer program instructions comprising instructions for:

identifying a failed worker system from the plurality of worker systems, wherein the failed worker system failed to execute during a current superstep and lost exchanged messages;

determining states of the partitions of the worker systems during supersteps preceding the current superstep;

determining, responsive to the determination of the states, a recovery initiation superstep preceding the current superstep in which all messages for the recovery initiation superstep are available;

determining a recovery set of partitions for which messages in supersteps after the recovery initiation superstep are not available;

assigning each partition in the recovery set of partitions to a recovery worker system from a set of recovery worker systems configured to process the partition, wherein the set of recovery worker systems includes a proper subset of the plurality of worker systems;

determining messaging worker systems responsible for sending the messages in supersteps after the recovery initiation superstep that are not available, the messaging worker systems being a proper subset of the plurality of worker systems and being different than the recovery worker systems;

instructing the recovery worker systems to execute the defined function for the partitions in the recovery set starting at the recovery initiation superstep to recover the lost exchanged messages; and

instructing the messaging worker systems to transmit the messages in supersteps after the recovery initiation superstep that are not available to the recovery work systems for use in processing the recovery set of partitions.

8. The computer-readable storage medium of claim 7 , wherein the messages transmitted by the messaging worker systems are stored in a persistent storage.

9. The computer-readable storage medium of claim 7 , further comprising instructions for:

assigning partitions of the failed worker system to at least one of the plurality of worker systems that has not failed.

10. The computer-readable storage medium of claim 7 , wherein determining states of the partitions of the worker systems comprises:

determining whether a worker system has access to a complete set of incoming messages for a partition at a superstep, the complete set of incoming messages to be processed at the superstep.

11. The computer-readable storage medium of claim 7 , wherein determining states of the partitions of the worker systems comprises:

determining whether a worker system has access to a complete set of outgoing messages for a partition at a superstep, the complete set of outgoing messages to be transmitted at the superstep.

12. The computer-readable storage medium claim 7 , further comprising instructions for:

receiving a plurality of status messages from the plurality of worker systems, a status message from a worker system indicating whether the worker system successfully transmitted its outgoing messages;

creating a group status message indicating whether each of the plurality of worker systems succeeded in transmitting its outgoing messages; and

transmitting the group status message to each of the plurality of worker systems.

13. A computer system for executing a confined recovery in a distributed computing system having a plurality of worker systems, the worker systems executing a computation in a plurality of supersteps, the worker systems having a plurality of partitions executing a defined function during the supersteps that exchange messages with partitions of other worker systems, the computer system comprising:

a computer-readable medium storing executable program instructions comprising instructions for:

identifying a failed worker system from the plurality of worker systems, wherein the failed worker system failed to execute during a current superstep and lost exchanged messages;

determining states of the partitions of the worker systems during supersteps preceding the current superstep;

determining, responsive to the determination of the states, a recovery initiation superstep preceding the current superstep in which all messages for the recovery initiation superstep are available;

determining a recovery set of partitions for which messages in supersteps after the recovery initiation superstep are not available;

assigning each partition in the recovery set of partitions to a recovery worker system from a set of recovery worker systems configured to process the partition, wherein the set of recovery worker systems includes a proper subset of the plurality of worker systems;

determining messaging worker systems responsible for sending the messages in supersteps after the recovery initiation superstep that are not available, the messaging worker systems being a proper subset of the plurality of worker systems and being different than the recovery worker systems;

instructing the recovery worker systems to execute the defined function for the partitions in the recovery set starting at the recovery initiation superstep to recover the lost exchanged messages; and

instructing the messaging worker systems to transmit the messages in supersteps after the recovery initiation superstep that are not available to the recovery work systems for use in processing the recovery set of partitions.

14. The computer system of claim 13 , further comprising instructions for:

assigning partitions of the failed worker system to at least one of the plurality of worker systems that has not failed.

15. The computer system of claim 13 , wherein determining states of the partitions of the worker systems comprises:

determining whether a worker system has access to a complete set of incoming messages for a partition at a superstep, the complete set of incoming messages to be processed at the superstep.

16. The computer system of claim 13 , wherein determining states of the partitions of the worker systems comprises:

determining whether a worker system has access to a complete set of outgoing messages for a partition at a superstep, the complete set of outgoing messages to be transmitted at the superstep.

17. The computer system of claim 13 , further comprising instructions for:

receiving a plurality of status messages from the plurality of worker systems, a status message from a worker system indicating whether the worker system successfully transmitted its outgoing messages;

creating a group status message indicating whether each of the plurality of worker systems succeeded in transmitting its outgoing messages; and

transmitting the group status message to each of the plurality of worker systems worker systems that has not failed.