Auto load transfer in geographically distributed systems

• US 9,509,617 B1
• Filed: 02/09/2016
• Issued: 11/29/2016
• Est. Priority Date: 02/09/2016
• Status: Active Grant

• Current Assignee: GrubHub Inc.
• Original Assignee: GrubHub Inc.
• Inventors: Malmgren, Clark, Maloney, Matthew
• Primary Examiner(s): Eskandarnia, Arvin
• Application Number:
  US15/019,860
• Time in Patent Office: 294 Days
• Field of Search: 709/223
• US Class Current: 1/1
• CPC Class Codes: H04L 41/12 Discovery or management of network topologies
• CPC Class Codes: H04L 43/0817 by checking functioning
• CPC Class Codes: H04L 43/0882 Utilisation of link capacity
• 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

In an approach, load balancing is performed in a distributed system of nodes representing computational resources such as clients, servers, server clusters, virtual machines, and so forth. Each node within the distributed system is responsible for processing requests associated with one or more districts in a geographical area. In order to perform load balancing each node is programmed or configured to autonomously perform a set of actions on a periodic basis which includes checking neighbors for updates, determining whether to spawning a new node, determining whether to terminate, and balancing load with neighbors. As a result, the number of nodes and the districts for which each node is responsible dynamically changes over time to minimize the load placed on each node and to constrain the number of nodes that need to be active within the distributed system to reduce resource costs.

Claims

1. A computer system comprising:

one or more nodes, wherein each node of the one or more nodes is a computational resource and is programmed to store district data identifying one or more districts of a geographical region for which the node is responsible for processing requests;

one or more non-transitory computer readable storage media that are communicatively coupled to the one or more nodes and store one or more sequences of program instructions which, when executed by a particular node of the one or more nodes, causes the particular node to perform;

receiving one or more requests associated with the one or more districts for which the particular node is responsible and processing the one or more requests;

periodically updating the district data to reflect one or more respective districts for which each of one or more neighboring nodes of the particular node is responsible;

periodically determining whether current load of the particular node has dropped below a lower bound value and in response to determining that the current load has dropped below the lower bound value initiating a termination action;

periodically determining whether the current load has risen above an upper bound value and in response to determining that the current load has risen above the upper bound value initiating a spawning action;

periodically balancing load with the one or more neighboring nodes by exchanging a set of districts with the one or more neighboring nodes.

2. The system of claim 1, wherein periodically updating the district data includes:

sending one or more messages to the one or more neighboring nodes requesting districts for which the one or more neighboring nodes are responsible;

in response to receiving a response from a particular neighboring node of the one or more neighboring nodes that identifies districts for which the particular neighboring node is responsible, updating the district data based on the identified districts;

in response to failing to receive a response from the particular neighboring node for a length of time, assuming responsibility for one or more districts for which the particular neighboring node was formerly responsible and updating the district data to reflect that the particular node has assumed responsibility for the one or more districts for which the particular neighboring node was formerly responsible.

3. The system of claim 1, wherein the termination action includes:

waiting a period of time;

in response to determining that a message is received from a neighboring node that indicates an intention to terminate within the period of time, entering into conflict resolution with the neighboring node;

in response to determining that the current load has risen above the lower bound value within the period of time, aborting the termination action;

in response to determining that the period of time has passed, assigning the one or more districts for which the particular node is responsible to the one or more neighboring nodes and terminating.

4. The system of claim 3, wherein the conflict resolution involves selecting a random period of time to wait before restarting the termination action and in response to determining that the random period of time has passed restarting the termination action.

5. The system of claim 1, wherein the spawning action includes:

waiting a period of time;

in response to determining that a message is received from a neighboring node that indicates an intention to spawn within the period of time, entering into conflict resolution with the neighboring node;

in response to determining that the current load has fallen below the upper bound value within the period of time, aborting the spawning action;

in response to determining that the period of time has passed, spawning a new node and assigning a portion of the one or more districts for which the particular node is responsible to the new node.

6. The system of claim 5, wherein the conflict resolution involves selecting a random period of time to wait before restarting the spawning action and in response to determining that the random period of time has passed restarting the spawning action.

7. The system of claim 1, wherein balancing the load with the one or more neighboring nodes involves:

generating one or more district exchanges between the particular node and the one or more neighboring nodes, wherein each district exchange of the one or more district exchanges is generated based on a random transfer of one or more districts between the particular node and the one or more neighboring nodes;

evaluating fitness of each district exchange of the one or more district exchanges to determine a most fit district exchange;

executing the most fit district exchange.

8. The system of claim 7, wherein evaluating fitness is performed based on one or more of:

contiguity of respective districts for which the particular node and the one or more neighboring nodes will be responsible after the district exchange is executed, difference in loads between the particular node and the one or more neighboring nodes after the district exchange is executed, border to area ratio of the respective districts for which the particular node and the one or more neighboring nodes will be responsible after the district exchange is executed, density on both sides of a border between the particular node and the one or more neighboring nodes after the district exchange is executed.

9. The system of claim 8, wherein the random transfer of the one or more districts between the particular node and the one or more neighboring nodes has a chance to swap responsibility of a district for which the particular node is responsible and a district for which a neighboring node of the particular node is responsible.

10. The system of claim 8, wherein executing the most fit district exchange involves sending a message to each particular neighbor of neighboring nodes which exchange one or more districts under the most fit district exchange, the message identifying the one or more districts that are to be exchanged between the particular neighbor and the particular node.

11. The system of claim 1, wherein each node of the one or more nodes represents a computer.

12. A computer system comprising:

one or more diner clients, wherein each diner client of the one or more diner clients is communicatively coupled to a respective non-transitory computer-readable storage medium storing one or more instructions which, when executed by a particular diner client of the one or more diner clients, causes the particular diner client to perform;

sending a message to an order receiving server that represents an order for food delivery;

wherein the order receiving server is communicatively coupled to a non-transitory computer-readable storage medium storing one or more instructions which, when executed by the order receiving server, causes the order receiving to perform;

receiving the message from the particular diner client, and

storing an entry identifying the order for food delivery in an order database;

one or more virtual machines executing within an order assignment server cluster, wherein each virtual machine of the one or more virtual machines has accessed to respective district data identifying one or more districts of a geographical region for which the virtual machine is responsible for processing requests, wherein the order assignment server cluster is communicatively coupled to one or more non-transitory computer-readable storage mediums that each store one or more instructions which, when executed by a particular virtual machine of the one or more virtual machines, causes the particular virtual machine to perform;

retrieving one or more orders for food delivery from the order database based on a respective delivery address associated with each of the one or more orders and district data associated with the particular virtual machine and assigning each of the one or more orders for food delivery to a respective driver client of one or more driver clients;

periodically updating the district data to reflect one or more respective districts for which each of one or more neighboring virtual machines of the particular virtual machine is responsible;

periodically determining whether current load of the particular virtual machine has dropped below a lower bound value and in response to determining that the current load has dropped below the lower bound value initiating a termination action;

periodically determining whether the current load has risen above an upper bound value and in response to determining that the current load has risen above the upper bound value initiating a spawning action;

periodically balancing load with the one or more neighboring virtual machines by exchanging a set of districts with the one or more neighboring virtual machines;

wherein the one or more driver clients are each communicatively coupled to a respective non-transitory computer-readable storage medium storing one or more instructions which, when executed by a particular driver client of the one or more driver clients causes the particular driver client to perform;

receiving an assignment of an order sent by a virtual machine of the one or more virtual machines and, in response to receiving the assignment, displaying one or more details of the order within a user interface of the particular driver client.

13. A method comprising:

storing, by a particular node, district data identifying one or more districts of a geographical region for which the particular node is responsible for processing requests, wherein the particular node is a member of one or more nodes, wherein each node of the one or more nodes represents a computational resource;

receiving, by the particular node, one or more requests associated with the one or more districts for which the particular node is responsible and processing the one or more requests;

periodically updating, by the particular node, the district data to reflect one or more respective districts for which each of one or more neighboring nodes of the particular node is responsible;

periodically determining, by the particular node, whether current load of the particular node has dropped below a lower bound value and in response to determining that the current load has dropped below the lower bound value initiating a termination action;

periodically determining, by the particular node, whether the current load has risen above an upper bound value and in response to determining that the current load has risen above the upper bound value initiating a spawning action;

periodically balancing load, by the particular node, with the one or more neighboring nodes by exchanging a set of districts with the one or more neighboring nodes.

14. The method of claim 13, wherein periodically updating the district data includes:

sending one or more messages to the one or more neighboring nodes requesting districts for which the one or more neighboring nodes are responsible;

in response to receiving a response from a particular neighboring node of the one or more neighboring nodes that identifies districts for which the particular neighboring node is responsible, updating the district data based on the identified districts;

in response to failing to receive a response from the particular neighboring node for a length of time, assuming responsibility for one or more districts for which the particular neighboring node was formerly responsible and updating the district data to reflect that the particular node has assumed responsibility for the one or more districts for which the particular neighboring node was formerly responsible.

15. The method of claim 13, wherein the termination action includes:

waiting a period of time;

in response to determining that a message is received from a neighboring node that indicates an intention to terminate within the period of time, entering into conflict resolution with the neighboring node;

in response to determining that the current load has risen above the lower bound value within the period of time, aborting the termination action;

in response to determining that the period of time has passed, assigning the one or more districts for which the particular node is responsible to the one or more neighboring nodes and terminating.

16. The method of claim 15, wherein the conflict resolution involves selecting a random period of time to wait before restarting the termination action and in response to determining that the random period of time has passed restarting the termination action.

17. The method of claim 13, wherein the spawning action includes:

waiting a period of time;

in response to determining that a message is received from a neighboring node that indicates an intention to spawn within the period of time, entering into conflict resolution with the neighboring node;

in response to determining that the current load has fallen below the upper bound value within the period of time, aborting the spawning action;

in response to determining that the period of time has passed, spawning a new node and assigning a portion of the one or more districts for which the particular node is responsible to the new node.

18. The method of claim 17, wherein the conflict resolution involves selecting a random period of time to wait before restarting the spawning action and in response to determining that the random period of time has passed restarting the spawning action.

19. The method of claim 13, wherein balancing the load with the one or more neighboring nodes involves:

generating one or more district exchanges between the particular node and the one or more neighboring nodes, wherein each district exchange of the one or more district exchanges is generated based on a random transfer of one or more districts between the particular node and the one or more neighboring nodes;

evaluating fitness of each district exchange of the one or more district exchanges to determine a most fit district exchange;

executing the most fit district exchange.

20. The method of claim 19, wherein evaluating fitness is performed based on one or more of:

contiguity of respective districts for which the particular node and the one or more neighboring nodes will be responsible after the district exchange is executed, difference in loads between the particular node and the one or more neighboring nodes after the district exchange is executed, border to area ratio of the respective districts for which the particular node and the one or more neighboring nodes will be responsible after the district exchange is executed, density on both sides of a border between the particular node and the one or more neighboring nodes after the district exchange is executed.

21. The method of claim 20, wherein the random transfer of the one or more districts between the particular node and the one or more neighboring nodes has a chance to swap responsibility of a district for which the particular node is responsible and a district for which a neighboring node of the particular node is responsible.

22. The method of claim 20, wherein executing the most fit district exchange involves sending a message to each particular neighbor of neighboring nodes which exchange one or more districts under the most fit district exchange, the message identifying the one or more districts that are to be exchanged between the particular neighbor and the particular node.

23. The method of claim 13, wherein each node of the one or more nodes represents a computer.

24. A method comprising:

storing, by a particular diner client, district data identifying one or more districts of a geographical region for which the particular diner client is responsible for processing requests, wherein the particular diner client is a member of a plurality of diner clients;

sending a message to an order receiving server that represents an order for food delivery;

wherein the order receiving server is communicatively coupled to a non-transitory computer-readable storage medium storing one or more instructions which, when executed by the order receiving server, causes the order receiving to perform;

receiving the message from the particular diner client, and

storing an entry identifying the order for food delivery in an order database;

one or more virtual machines executing within an order assignment server cluster, wherein each virtual machine of the one or more virtual machines has accessed to respective district data identifying one or more districts of a geographical region for which the virtual machine is responsible for processing requests, wherein the order assignment server cluster is communicatively coupled to one or more non-transitory computer-readable storage mediums that each store one or more instructions which, when executed by a particular virtual machine of the one or more virtual machines, causes the particular virtual machine to perform;

retrieving one or more orders for food delivery from the order database based on a respective delivery address associated with each of the one or more orders and district data associated with the particular virtual machine and assigning each of the one or more orders for food delivery to a respective driver client of one or more driver clients;

periodically updating the district data to reflect one or more respective districts for which each of one or more neighboring virtual machines of the particular virtual machine is responsible;

periodically determining whether current load of the particular virtual machine has dropped below a lower bound value and in response to determining that the current load has dropped below the lower bound value initiating a termination action;

periodically determining whether the current load has risen above an upper bound value and in response to determining that the current load has risen above the upper bound value initiating a spawning action;

periodically balancing load with the one or more neighboring virtual machines by exchanging a set of districts with the one or more neighboring virtual machines;

wherein the one or more driver clients are each communicatively coupled to a respective non-transitory computer-readable storage medium storing one or more instructions which, when executed by a particular driver client of the one or more driver clients causes the particular driver client to perform;

receiving an assignment of an order sent by a virtual machine of the one or more virtual machines and, in response to receiving the assignment, displaying one or more details of the order within a user interface of the particular driver client.