Coordinated garbage collection in distributed systems

• US 10,642,663 B2
• Filed: 05/27/2015
• Issued: 05/05/2020
• Est. Priority Date: 09/10/2014
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Harris, Timothy L., Maas, Martin C.
• Primary Examiner(s): Wu, Benjamin C
• Application Number:
  US14/723,425
• Publication Number:
  US 20160070593A1
• Time in Patent Office: 1,805 Days
• CPC Class Codes: G06F 11/301 where the computing system is a virtual computing platform, e.g. logically partitioned systems virtual machines G06F9/45533; logical partitioning of resources G06F9/5077
• CPC Class Codes: G06F 11/34 Recording or statistical evaluation of computer activity, e.g. of down time, of input/output operation ; Recording or statistical evaluation of user activity, e.g. usability assessment
• CPC Class Codes: G06F 11/3409 for performance assessment
• CPC Class Codes: G06F 12/0253 Garbage collection, i.e. reclamation of unreferenced memory
• CPC Class Codes: G06F 12/0276 Generational garbage collection
• CPC Class Codes: G06F 2009/45583 Memory management, e.g. access or allocation
• CPC Class Codes: G06F 2212/1024 Latency reduction
• CPC Class Codes: G06F 2212/152 Virtualized environment, e.g. logically partitioned system
• CPC Class Codes: G06F 2212/154 Networked environment
• CPC Class Codes: G06F 9/45558 Hypervisor-specific management and integration aspects
• CPC Class Codes: G06F 9/522 Barrier synchronisation

Abstract

Fast modern interconnects may be exploited to control when garbage collection is performed on the nodes (e.g., virtual machines, such as JVMs) of a distributed system in which the individual processes communicate with each other and in which the heap memory is not shared. A garbage collection coordination mechanism (a coordinator implemented by a dedicated process on a single node or distributed across the nodes) may obtain or receive state information from each of the nodes and apply one of multiple supported garbage collection coordination policies to reduce the impact of garbage collection pauses, dependent on that information. For example, if the information indicates that a node is about to collect, the coordinator may trigger a collection on all of the other nodes (e.g., synchronizing collection pauses for batch-mode applications where throughput is important) or may steer requests to other nodes (e.g., for interactive applications where request latencies are important).

Claims

1. A system, comprising:

a plurality of hardware computing nodes interconnected via a network, each of which comprises at least one processor and a memory, and each of which hosts one or more virtual machine instances, wherein each virtual machine instance has its own heap memory and garbage collector that are not shared with other virtual machine instances; and

a garbage collection coordinator;

wherein each of the virtual machine instances executes a respective process of a distributed application that communicates over the network with one or more other ones of the processes of the distributed application executing on respective other virtual machine instances, and wherein performing a garbage collection on one of the virtual machine instances hosted on a particular node of the plurality of computing nodes delays execution of one or more other of the virtual machine instances hosted on respective nodes of the plurality of computing nodes other than the particular node;

wherein the garbage collection coordinator is configured to;

receive information from at least one node of the plurality of hardware computing nodes indicating that the at least one node is not ready to receive communications from other ones of the processes of the distributed application;

determine that a particular garbage collection should be performed on one of the virtual machine instances dependent, at least in part, on the received information; and

initiate garbage collection on the one of the virtual machine instances.

2. The system of claim 1, wherein the determination that the particular garbage collection should be performed is further dependent on a pre-defined garbage collection policy.

3. The system of claim 2, wherein the pre-defined garbage collection policy specifies that all of the virtual machine instances hosted on the plurality of computing nodes should perform garbage collection at substantially the same time in response to a garbage collection trigger condition being met on at least one of the virtual machine instances.

4. The system of claim 2,

wherein the pre-defined garbage collection policy specifies a predetermined fixed time interval between garbage collections or specifies a predetermined threshold amount or a predetermined threshold percentage for heap occupancy;

and

wherein to determine that a particular garbage collection should be performed on the one of the virtual machine instances, the garbage collection coordinator is configured to;

determine that at least one of the virtual machine instances hosted on the plurality of computing nodes has not performed garbage collection in at least an amount of time equal to the predetermined fixed time interval;

or

determine that heap occupancy for at least one of the virtual machine instances hosted on the plurality of computing nodes meets or exceeds the predetermined threshold amount or predetermined threshold percentage for heap occupancy.

5. The system of claim 2, wherein the pre-defined garbage collection policy specifies or enforces a limit on the number of virtual machine instances that perform garbage collection at substantially the same time.

6. The system of claim 1, wherein the garbage collection coordinator comprises a single garbage collection coordinator component on one of the plurality of computing nodes.

7. The system of claim 1,

wherein the garbage collection coordinator comprises a plurality of garbage collection coordinator components, each of which is associated with a respective virtual machine instance;

and

wherein each of the garbage collection coordinator components is a component of the one of the plurality of computing nodes on which the associated virtual machine instance is hosted.

8. The system of claim 1,

wherein the system further comprises one or more monitor components;

wherein each of the one or more monitor components is configured to send information to the garbage collection coordinator indicating the state of a process of the distributed application with respect to its readiness to receive communications from other ones of the processes of the distributed application executing on other virtual machine instances, wherein the state of a process of the distributed application indicates that it is not ready to receive communications from other ones of the processes of the distributed application when it is performing a garbage collection or has determined a need to perform a garbage collection; and

wherein the determination that the particular garbage collection should be performed on the one of the virtual machine instances is further dependent on the state information of the respective processes of the distributed application that are executing on one of more of the virtual machine instances.

9. The system of claim 1,

wherein the system further comprises one or more monitor components;

and

wherein each of the one or more monitor components is configured to send information to the garbage collection coordinator indicating a heap occupancy for a virtual machine instance hosted on one of the plurality of computing nodes; and

wherein the determination that the particular garbage collection should be performed on the one of the virtual machine instances is further dependent on the heap occupancy information.

10. A method, comprising:

performing, by a plurality of hardware computing nodes interconnected via a network;

beginning execution of a distributed application, wherein respective processes of the distributed application executing on the plurality of hardware computing nodes communicate over the network with other processes of the distributed application executing on other ones of the plurality of hardware computing nodes, and wherein executing the distributed application comprises performing one or more particular operations that when performed on one of the plurality of computing nodes delays operations on one or more other ones of the plurality of computing nodes until it is complete;

determining, during execution of the distributed application, that a given one of the plurality of computing nodes should perform one of the particular operations;

receiving information from the given one of the plurality of computing nodes indicating that the given one of the plurality of computing nodes is not ready to receive communications from the other processes of the distributed application; and

performing, by the given one of the plurality of computing nodes, the one of the particular operations;

wherein said determining comprises applying a coordination policy that is dependent, at least in part, on the received information.

11. The method of claim 10, wherein the coordination policy is one of a plurality of application-dependent or workload-dependent policies from which it was selected for use in coordinating performance of the one or more particular operations on the plurality of computing nodes during execution of the distributed application.

12. The method of claim 10,

wherein said determining comprises determining that execution of the distributed application on one or more of the plurality of computing nodes should be paused or stopped to perform an activity initiated by a process of the distributed application itself or should be paused or stopped to perform an activity initiated by a process that is executing on one or more of the plurality of computing nodes other than a process of the distributed application.

13. The method of claim 10,

wherein the distributed application is an application that was written in a garbage collected programming language;

and

wherein said determining comprises determining, by a garbage collection coordinator, that a major or minor garbage collection should be performed on the given one of the plurality of computing nodes and that execution of the distributed application on the given computing node should be paused or stopped while the garbage collection is performed.

14. The method of claim 13,

wherein executing the distributed application comprises a process of the distributed application executing on one of the plurality of computing nodes communicating with one or more processes of the distributed application that are executing on one or more other ones of the plurality of computing nodes;

wherein said communicating comprises;

sending, to the one or more processes of the distributed application that are executing on one or more other ones of the plurality of computing nodes, information indicating that the computing node on which the process of the distributed application is performing, or is a determination has been made to perform, a garbage collection;

or

sending, to the one or more processes of the distributed application that are executing on one or more other ones of the plurality of computing nodes, information indicating a heap occupancy for one of the plurality of computing nodes; and

wherein said determining that a major or minor garbage collection should be performed on the given one of the plurality of computing nodes is dependent on said communicating.

15. The method of claim 10,

wherein the coordination policy specifies that all of the plurality of computing nodes should perform the one of the particular operations at substantially the same time in response to a trigger condition being met on one of the computing nodes;

wherein said determining that the given one of the plurality of computing nodes should perform the one of the particular operations comprises;

a coordinator component determining that the trigger condition has been met on one of the plurality of computing nodes other than the given one;

the coordinator component notifying one or more other ones of the plurality of computing nodes, including the given one of the computing nodes, that the computing node on which the trigger condition has been met is to perform the one of the particular operations; and

wherein the method further comprises, in response to receiving the notification, the given one of the plurality of computing nodes performing the one of the particular operations at substantially the same time as the computing node on which the trigger condition has been met.

16. The method of claim 10,

wherein the coordination policy enforces an upper bound on the number of computing nodes that can perform the one of the particular operations at substantially the same time;

wherein applying the coordination policy comprises;

meeting, by the given one of the plurality of computing nodes, a trigger condition for performing the one of the particular operations on the given computing node;

requesting, by the given computing node in response to meeting the trigger condition, a token from a pool of tokens managed by a distributed or centralized coordinator process, wherein the number of tokens in the pool is dependent on the number of computing nodes that can perform the one of the particular operations at substantially the same time according to the coordination policy;

receiving, by the given computing node, an available token from the pool of tokens; and

performing the one of the particular operations on the given computing node;

meeting, by another one of the plurality of computing nodes while no tokens are available in the pool of tokens, a trigger condition for performing the one of the particular operations on the other computing node;

requesting, by the other computing node in response to meeting the trigger condition, a token from the pool of tokens;

waiting, by the other computing node, for a token to be returned the pool of tokens; and

in response to a token being returned to the pool of tokens;

receiving, by the other computing node a token from the pool of tokens; and

performing the one of the particular operations on the other computing node.

17. A non-transitory, computer-readable storage medium storing program instructions that when executed on one or more computing nodes cause the one or more computing nodes to implement a garbage collection coordinator and an application programming interface for communication with the garbage collection coordinator;

wherein the garbage collection coordinator is configured to;

receive information from each of one or more of a plurality of computing nodes on which a respective process of a distributed application is executing, wherein the respective process of the distributed application executing on each of the plurality of computing nodes communicates with one or more other ones of the respective processes executing on other ones of the computing nodes, and wherein the respective processes of the distributed application executing on each of the plurality of computing nodes do not share heap memory; and

initiate a garbage collection on at least one of the plurality of computing nodes dependent on the information;

wherein the information is usable to determine whether or not the computing node is ready to receive communications from other ones of the processes of the distributed application;

wherein the information is received via an operation defined by the application programming interface; and

wherein the garbage collection coordinator is configured to initiate the garbage collection via an operation defined by the application programming interface.

18. The non-transitory, computer-readable storage medium of claim 17,

wherein the information is received via an operation defined by the application programming interface that allows the plurality of computing nodes to share heap occupancy information with the garbage collection coordinator, an operation defined by the application programming interface that allows a computing node to publish state information indicating its readiness to perform garbage collection, an operation defined by the application programming interface that allows a computing node to publish state information indicating its readiness to receive communications from other ones of the plurality of computing nodes;

or an operation defined by the application programming interface through which the garbage collection coordinator queries the one or more of the plurality of computing nodes for the information; and

wherein the application programming interface further defines an operation through which the garbage collection coordinator initiates an expansion of heap memory on one of the plurality of computing nodes, or an operation through which the garbage collection coordinator initiates a contraction of heap memory on one of the plurality of computing nodes.

19. The non-transitory, computer-readable storage medium of claim 17,

wherein the program instructions that when executed on the one or more computing nodes cause the one or more computing nodes to implement a communication library, operations of which are invoked by the respective processes of the distributed application executing on each of the plurality of computing nodes to communicate with each other;

wherein the communication library encapsulates communications that are directed to the respective processes of the distributed application executing on each of the plurality of computing nodes in a manner that steers them to computing nodes that are ready to receive communications from other ones of the plurality of computing nodes or that steers them away from computing nodes that are performing, or have determined a need to perform, a garbage collection.

20. The non-transitory, computer-readable storage medium of claim 17,

wherein the garbage collection coordinator is configured to initiate a garbage collection on all of the plurality of computing nodes in response to a determination that a garbage collection trigger condition has been met for at least one of the plurality of computing nodes;

wherein to determine that a garbage collection trigger condition has been met, the garbage collection coordinator is configured to;

determine an amount of time that has elapsed since the at least one of the computing nodes has performed a garbage collection and that the elapsed time is greater than a predetermined threshold amount of time;

or

determine that the at least one of the computing nodes has a current percentage of heap occupancy that exceeds a predetermined threshold for heap occupancy percentage; and

wherein to initiate a garbage collection on all of the plurality of computing nodes, the garbage collection coordinator is configured to broadcast a message to all of the plurality of computing nodes indicating that a garbage collection is to be performed using an operation defined by the application programming interface.