Reachability-based coordination for cyclic dataflow

• US 9,832,068 B2
• Filed: 12/17/2012
• Issued: 11/28/2017
• Est. Priority Date: 12/17/2012
• Status: Active Grant

• Current Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Original Assignee: Microsoft Technology Licensing LLC (Microsoft Corporation)
• Inventors: McSherry, Frank D., Isaacs, Rebecca, Isard, Michael A., Murray, Derek G.
• Primary Examiner(s): Sall, El Hadji
• Application Number:
  US13/717,488
• Publication Number:
  US 20140172939A1
• Time in Patent Office: 1,807 Days
• Field of Search: 709201, 709200, 709220, 709217
• CPC Class Codes: G06F 16/2465 Query processing support for facilitating data mining operations in structured databases
• CPC Class Codes: G06F 9/4494 data driven
• CPC Class Codes: G06F 9/4881 Scheduling strategies for dispatcher, e.g. round robin, multi-level priority queues
• CPC Class Codes: H04L 67/10 in which an application is distributed across nodes in the network software deployment G06F8/60; multiprogramming arrangements G06F9/46

Abstract

Various embodiments provide techniques for working with large-scale collections of data pertaining to real world systems, such as a social network, a roadmap/GPS system, etc. The techniques perform incremental, iterative, and interactive parallel computation using a coordination clock protocol, which applies to scheduling computations and managing resources such as memory and network resources, etc., in cyclic graphs including those resulting from a differential dataflow model that performs computations on differences in the collections of data.

Claims

1. A computer-readable storage medium storing computer-executable instructions that, when executed by a processor, configure the processor to perform operations comprising:

scheduling a plurality of processes each comprising a plurality of threads to operate independently on discrete partitions of data;

responsive to a first thread of the plurality of threads receiving a first partition of the data, the first thread beginning an operation on the first partition of the data;

responsive to a second thread of the plurality of threads receiving a second partition of the data, the second thread beginning the operation on the second partition of the data;

tracking progress of the operation using a replicated data structure at each of the first thread and the second thread, the replicated data structure comprising a logical time, the logical time being a tuple comprising at least an integer representing an epoch and an integer representing an iteration;

adding a timestamp to a record for processing at each of the first thread and the second thread, the timestamp corresponding to the logical time the record was produced;

storing the timestamp in the replicated data structure;

determining a number of yet to be processed records from the replicated data structure for at least one of the first thread or the second thread; and

when the number of yet to be processed records for the first thread or the second thread reaches zero and the timestamp includes the highest iteration in the replicated data structure for one of the first thread or the second thread, terminating the one of the first thread or the second thread.

2. A computer-readable storage medium as claim 1 recites, wherein when the number of yet to be processed records for the first thread or the second thread is not zero or the timestamp does not include the highest iteration in the replicated data structure for one of the first thread or the second thread, refraining from terminating the one of the first thread or the second thread.

3. A computer-readable storage medium as claim 1 recites, wherein the replicated data structure includes an entry for each thread participating in the operation.

4. A computer-readable storage medium as claim 3 recites, wherein the entry for each thread participating in the operation includes a thread identifier, the timestamp corresponding to the thread of the entry, and a record count.

5. A computer-readable storage medium as claim 1 recites, further comprising storing the number of yet to be processed records in the replicated data structure.

6. A method comprising:

tracking progress of a computation by;

associating each of a plurality of data items to be processed in a computation with a dataflow vertex of a dataflow graph, the dataflow graph representing a single program, and the dataflow vertex comprising the output of a function and the input of a next function;

associating each of the plurality of data items to be processed in the computation with a timestamp, the timestamp corresponding to an order in which data items are created in the computation;

storing a vertex-time pair associated with each data item, the vertex-time pair comprising the dataflow vertex and the timestamp; and

counting a number of yet to be processed data items that are associated with each vertex-time pair; and

communicating by at least one processor asynchronously sending one or more messages including the number of yet to be processed data items that are associated with at least one vertex-time pair.

7. A method as claim 6 recites, wherein a data item of the plurality of data items includes a record and the dataflow vertex of the dataflow graph includes a dataflow operator.

8. A method as claim 6 recites, wherein counting the number of yet to be processed data items includes:

responsive to a data item of the plurality of data items being consumed by a dataflow graph vertex, subtracting one from the number of yet to be processed data items for that vertex; and

responsive to a data item of the plurality of data items being produced for a dataflow graph vertex, adding one to the number of yet to be processed data items for that vertex.

9. A method as claim 8 recites, wherein two or more processors that communicate by passing messages across a network maintain multiple replicas of the vertex-time pairs associated with the number of yet to be processed data items for that dataflow graph.

10. A method as claim 9 recites, wherein each processor of the two or more processors maintains a replica of the vertex-time pairs associated with the number of yet to be processed data items for that dataflow graph that does not under-approximate an actual number of yet to be processed data items for that dataflow graph.

11. A method as claim 10 recites, wherein the vertex-time pairs are ordered according to joint precedence of operators in the dataflow graph and times in a lattice of logical timestamps.

12. A method as claim 11 recites, further comprising determining a next action to take as one of:

execute the vertex corresponding to a version in a clock, or

block until a message is received.

13. A method as claim 6 recites, the timestamp encoding other information including at least one of:

priority information or security information.

14. A system comprising:

a plurality of processors connected to a network for asynchronously sending messages;

a plurality of memories storing data comprising;

a graph of at least one dataflow operator;

a queue of unprocessed records associated with each dataflow operator, each unprocessed record having a timestamp; and

a mapping from a version, wherein the version includes a dataflow operator-timestamp pair, to an over-approximation of a number of unprocessed records for that version in the system, wherein at least one processor of the plurality of processors communicates the number of unprocessed records.

15. A system as claim 14 recites, wherein a causal order of versions is used to determine a next action to take as one of:

execute the dataflow operator corresponding to the version in a clock, or

block until a message is received.

16. A system as claim 15 recites, wherein a set of one or more versions is determined to form a minimal antichain in the causal order, and a next dataflow operator to execute is drawn from the set.

17. A system as claim 16 recites, wherein the minimal antichain includes dataflow operators that have no active predecessor dataflow operators.

18. A system as claim 15 recites, wherein the causal order of versions is used to determine parts of the computation state that will no longer change.

19. A system as claim 18 recites, where the parts of the computation state that will no longer change are used to compact the computation state, by removing the distinction between equivalent logical times.

20. A system as claim 19 recites, wherein removing the distinction between equivalent logical times includes consolidating multiple copies of a processed record into a single processed record with an integer weight, and cancelling additions and subtractions of the processed record.