Distributed-leader-election service for a distributed computer system

TECHNICAL FIELD OF THE INVENTION

0001The present invention is related to distributed computing and, in particular, to a method and system for efficiently and robustly allocating leadership within a distributed computer system.

BACKGROUND OF THE INVENTION

0002In the early days of computing, computer systems were stand-alone devices accessed by computer users via input/output (“I/O”) peripheral components, including control-panel toggle switches, Hollerith-card readers, line printers, and eventually cathode-ray-tube (“CRT”) 24-line terminals and keyboards. When a user wished to carry out a computational task on more than one computer system, the user would manually transfer data between the computer systems via Hollerith cards, magnetic tape, and, later, removable magnetic-disk packs.

0003With the advent of multi-tasking operating systems, computer scientists discovered and addressed the need for synchronizing access by multiple, concurrently executing tasks to individual resources, including peripheral devices, memory, and other resources, and developed tools for synchronizing and coordinating concurrent computation of decomposable problems by independent, concurrently executing processes. With the advent of computer networking, formerly independent computer systems were able to be electronically interconnected, allowing computer systems to be linked together to form distributed computer systems. Although initial distributed computer systems were relatively loosely coupled, far more complex, tightly coupled distributed computer systems based on distributed operating systems and efficient, distributed computation models, have since been developed.

0004There are many different models for, and types of, distributed computing. In some models, relatively independent, asynchronous, peer computational entities execute relatively autonomously on one or more distributed computer systems, with sufficient coordination to produce reliable, deterministic solutions to computational problems and deterministic behavior with respect to external inputs. In other distributed systems, tightly controlled computational entities execute according to pre-determined schedules on distributed computer systems, closely synchronized by various protocols and computational tools. In many fault-tolerant and highly available distributed computer systems, computational tasks are distributed among individual nodes, or computers, of the distributed computer system in order to fairly distribute the computational load across the nodes. In the event of failure of one or more nodes, surviving nodes can assume, or be assigned, tasks originally distributed to failed nodes so that the overall distributed computational system is robust and resilient with respect to individual node failure. However, even in distributed systems of relatively independent peer nodes, it is frequently the case that, for certain tasks, a single node needs to be chosen to be responsible for the task, rather than simply allowing any of the peer nodes to contend for the task, or for subtasks that together compose the task. In other words, a single node is assigned to be, or elected to be, the leader with respect one or more tasks that require investing responsibility for the one or more tasks in a single node. Tasks for which leaders need to be assigned are generally tasks that are not efficiently decomposed, iterative tasks with high, initial-iteration computational overheads, and tasks that require assembling complex sets of privileges and controls over resources. Examples of such tasks include coordinator-type tasks in which a single node needs to be responsible for distributed state changes related to distributed-system management, distributed-system-updating tasks, including installation of software or software updates on nodes within the distributed system, system-state-reporting tasks, in which a single node needs be responsible for accessing and reporting the distributed state of a distributed computer system, and, in certain systems, scheduling, distribution, and control tasks for the distributed system.

0005A leadership-role allocation can be hard wired, or statically assigned at distributed-system initialization, for all, a subset of, or individual tasks needing a leader. However, relatively static leader assignment may lead to time-consuming and difficult leader-reassignment problems when a leader node fails or becomes incapable of carrying out those tasks required of the leader node. Alternatively, all nodes can constantly contend for leader roles for tasks requiring a leader on an on-demand basis, but constant leader-role contention may be inefficient and may even lead to thrashing. Strong-leader self-election based on a distributed consensus service is a useful model for certain of these distributed computer systems and distributed computing tasks. The strong-leader-election method based on a distributed consensus service can be extended to provide strong-leader election for multiple roles within a distributed computer system. However, in more complex distributed computer systems, leadership may need to be allocated for multiple roles on a continuing basis, and leadership may need to be distributed among individual processes running on nodes within a distributed computer system. For these environments, researchers, developers, manufacturers, and users of distributed computer systems have recognized the need for a practical and efficient means for continuous, dynamic allocation of leadership among processes within nodes of a multi-node distributed computer system.

SUMMARY OF THE INVENTION

0006Embodiments of the present invention provide methods and systems for leadership allocation in a distributed computer system. In certain embodiments of the present invention, a leader-election-service process runs within each node of a distributed computer system, together cooperatively providing a distributed-leader-election service. The distributed-leader-election service employs a distributed consensus service to manage distributed state information related to roles and leadership allocation within a distributed computer system. Client processes within each node interface with the leader-election-service process of the node in order to assume leadership of particular roles within the distributed computer system. Leadership-allocation management is thus centralized, within each node. In alternative embodiments, the distributed-leader-election service may be implemented as a collection of library routines that run in the context of client processes.

BRIEF DESCRIPTION OF THE DRAWINGS

0007<figref idref="DRAWINGS">FIGS. 1A-G</figref> illustrate the Paxos distributed consensus service.

0008<figref idref="DRAWINGS">FIGS. 2-8</figref> illustrate the basic operation of a distributed storage register.

0009<figref idref="DRAWINGS">FIG. 9</figref> shows the components used by a process or processing entity P_ithat implements, along with a number of other processes and/or processing entities, P_j≠i, a distributed storage register.

0010<figref idref="DRAWINGS">FIG. 10</figref> illustrates determination of the current value of a distributed storage register by means of a quorum.

0011<figref idref="DRAWINGS">FIG. 11</figref> shows pseudocode implementations for the routine handlers and operational routines shown diagrammatically in <figref idref="DRAWINGS">FIG. 9</figref>.

0012<figref idref="DRAWINGS">FIG. 12</figref> illustrates a disk-Paxos or active-disk-Paxos distributed computer system.

0013<figref idref="DRAWINGS">FIG. 13</figref> illustrates an exemplary distributed computer system in which strong-leader election may be practiced.

0014<figref idref="DRAWINGS">FIG. 14</figref> is a control-flow diagram illustrating general node operation and strong-leader election.

0015<figref idref="DRAWINGS">FIG. 15</figref> is a control-flow diagram illustrating the routine “elect self.”

0016<figref idref="DRAWINGS">FIGS. 16A-G</figref> illustrate operation of a strong-leader election method.

0017<figref idref="DRAWINGS">FIG. 17</figref> illustrates, in similar fashion to <figref idref="DRAWINGS">FIG. 13</figref>, an alternative distributed computer system in which strong-leader election may be practiced.

0018<figref idref="DRAWINGS">FIGS. 18A-D</figref> illustrate operation of delay-timer-based strong-leader election.

0019<figref idref="DRAWINGS">FIG. 19</figref> is a control-flow diagram illustrating an alternative strong-leader election method.

0020<figref idref="DRAWINGS">FIGS. 20A-C</figref> illustrate operation of the alternative strong-leader-election method, described above.

0021<figref idref="DRAWINGS">FIG. 21</figref> illustrates a leader-election-service-based distributed computer system that represents one embodiment of the present invention.

0022<figref idref="DRAWINGS">FIG. 22</figref> illustrates the basic components of the leadership-election service that represents one embodiment of the present invention.

0023<figref idref="DRAWINGS">FIG. 23</figref> illustrates a portion of the distributed state information maintained cooperatively by leader-election services within a distributed computer system according to one embodiment of the present invention.

0024<figref idref="DRAWINGS">FIG. 24</figref> illustrates the local state information maintained by the leader-election-service process within a node of a distributed computer system, according to one embodiment of the present invention.

0025<figref idref="DRAWINGS">FIG. 25</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Register.”

0026<figref idref="DRAWINGS">FIG. 26</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Unregister.”

0027<figref idref="DRAWINGS">FIG. 27</figref> is a control-flow diagram illustrating the client/leader election-service-interface routine “Assert Leadership.”

0028<figref idref="DRAWINGS">FIG. 28</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Forward Request.”

0029<figref idref="DRAWINGS">FIG. 29</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Renounce.”

0030<figref idref="DRAWINGS">FIG. 30</figref> is a control-flow diagram for the client/leader-election-service-interface routine “Forward Role.”

0031<figref idref="DRAWINGS">FIG. 31</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Info Request.”

0032<figref idref="DRAWINGS">FIG. 32</figref> is a control-flow diagram illustrating callback of a client by the leader-election service.

DETAILED DESCRIPTION OF THE INVENTION

0033The present invention is related to distributed computing. Certain embodiments of the present invention rely on previously developed techniques for distributing state information among the nodes of a distributed system. One such technique is the Paxos distributed consensus service, described below in a first subsection. Another, related technique provides a distributed storage register to multiple nodes in a distributed system, described in a second subsection, below. Using the Paxos distributed consensus service, or related disk-Paxos services, described below in a third subsection, a robust and efficient leader-election method can be devised. The efficient leader-election method is described in a fourth subsection, below. For more complex distributed computer systems, a higher-level, distributed-leader-election service may be provided, according to method and system embodiments of the present invention, described below in a fifth subsection.

Paxos Distributed Computing Model

0034<figref idref="DRAWINGS">FIGS. 1A-G</figref> illustrate the Paxos distributed consensus service. <figref idref="DRAWINGS">FIGS. 1A-G</figref> employ the same illustration conventions, next described with reference to <figref idref="DRAWINGS">FIG. 1A</figref>. In <figref idref="DRAWINGS">FIG. 1A</figref>, five nodes <b>102</b>-<b>106</b> are interconnected by a communications medium <b>108</b>. Each node can send messages to, and receive messages from, each of the remaining nodes. Each node includes an ordered, sequential list of state changes, such as local state change list <b>110</b> within node <b>102</b>. Together, the nodes constitute a distributed computer system that manages a global, master, ordered and sequential list of state changes <b>112</b>. The global list of state changes <b>112</b> is shown as a dashed-line rectangle, to indicate that the distributed system may not contain a single, full copy of the global list of state changes, but may instead maintain the global list of state changes in various pieces distributed across the nodes of the distributed computer system.

0035Any node can request a state change. For example, as shown in <figref idref="DRAWINGS">FIG. 1B</figref>, node <b>105</b> has formulated the state change request: m=“ready.”<b>114</b>. This state change is directed to setting the contents of variable m to the string “ready.” Node <b>105</b> asserts the state-change request by sending a state-change request message <b>116</b> to all active nodes in the distributed system. Either of two outcomes is possible. In one outcome, the state-change request is successful, and all or a portion of the active nodes in the distributed system update their local state change lists <b>118</b>-<b>122</b> to indicate the state change. The global distributed state-change list <b>112</b> is always updated as a product of committing a state-change request. In a second outcome, the state-change request is denied, or is unsuccessful, and no local state-change list is updated, and, of course, the global distributed state-change list <b>112</b> is also not updated.

0036At any point in time, certain of the nodes may become inactive, due to node failure, communications medium failure, or other failures within the distributed computer system. For example, as shown in <figref idref="DRAWINGS">FIG. 1D</figref> by “X” symbols <b>124</b>-<b>125</b>, the communications links to nodes <b>103</b> and <b>105</b> may fail, leaving only nodes <b>102</b>, <b>104</b>, and <b>105</b> active within the distributed computer system. Following failure of nodes <b>103</b> and <b>105</b>, an active node may wish to request a subsequent state change. In <figref idref="DRAWINGS">FIG. 1D</figref>, node <b>102</b> has formulated the state change request: n=“stop” <b>126</b>. Node <b>102</b> requests this state change by sending a state-change request message <b>128</b> to the active nodes <b>104</b> and <b>105</b> within a distributed computer system. When there are sufficient active nodes to constitute a quorum of nodes, where a quorum is, in many situations, at least a majority of the nodes in a distributed system, a state-change request may succeed. In the case shown in <figref idref="DRAWINGS">FIG. 1D</figref>, the state-change request made by node <b>102</b> succeeds, resulting in updates to the local state-change lists of the active nodes <b>129</b>-<b>131</b>, as shown in <figref idref="DRAWINGS">FIG. 1E</figref>. Inactive nodes <b>103</b> and <b>105</b> do not reflect the most recently successful state-change request, since they are not in communication with the active nodes. The distributed global state-change list <b>112</b> necessarily reflects the most recent state change. Should a formerly failed, or inactive, node be rehabilitated, and rejoin the distributed computer system, as in the case of node <b>103</b> in <figref idref="DRAWINGS">FIG. 1F</figref>, the reactivated node can update its local state-change list by issuing a no-operation (“NOP”) request <b>132</b> to the distributed computer system. A by-product of issuing any request, including a state-change request, is that the local state-change list of the node issuing the request is brought up to date with respect to the global, master state-change list <b>112</b> prior to the request being issued. This guarantees that there is a global ordering of issued and executed requests, memorialized in the distributed global state-change list <b>112</b>. Following execution of the NOP request, reactivated node <b>103</b> has an updated local state-change list, shown in <figref idref="DRAWINGS">FIG. 1G</figref>.

0037When a sufficient number of nodes have failed that a quorum of nodes is not active, no state-change request can succeed. The Paxos distributed consensus services employs a communications protocol to achieve a distributed global state-change list and to manage state-change requests and state-change-request execution. Two or more nodes may simultaneously issue state-change requests, or issue state-change requests in a sufficiently short period of time that the state-change requests cannot be distinguished from one another in time-precedence order. In such cases, the Paxos protocol chooses one of the contending state-change requests for execution, and fails the remaining, simultaneous state-change requests. Depending on the particular Paxos implementation, a local state-change list may not be updated as a result of commitment of a next state-change request, if the containing node is not a member of the quorum for commitment of the next state-change request. However, when the node itself next makes a state=change request or issues a NOP request, the node's local state-change list is guaranteed to include all previous, committed state-change requests.

0038In summary, the Paxos distributed consensus service is a protocol that provides for a global ordering of committed state-change requests requested by individual nodes of a distributed computer system. Each node has a local state-change list that the node can access at any time, locally, to review all committed requests up through the latest committed request within the local state-change list, or a pruned subset of such all committed requests. When a node remains active, its local state-change list generally accurately reflects a global, distributed, master state-change list maintained via the Paxos protocol within the distributed computer system, with a possible lag in updates due to not being involved in recent quorums. If a node loses communications contact with the remaining nodes of the distributed computer system, the node may still use the local state-change list for stand-alone computation. When node rejoins the distributed computer system, the node can update its local state-change list by issuing a state-change request, including a NOP request. Thus, a node learns of any committed requests not yet known to the node no later than the point in time at which the node makes a next state-change request.

0039In the next subsection, a distributed storage register implementation is discussed. A distributed storage register is less complex than a global state-change list, and may be used as the basis for more complex, globally consistent data sets. The distributed storage register implementation is illustrative of the types of techniques used to implement quorum-based distributed-computing service, such as Paxos.

Storage Register Model

0040As discussed in the previous section, a distributed storage register is a relatively simple distributed-computing entity that can be implemented by quorum-based techniques similar to those employed in the Paxos protocol. A distributed storage register is a globally shared data entity that is distributed across the nodes of a distributed computer system and that can be updated by any of the nodes according to a Paxos-like protocol.

0041<figref idref="DRAWINGS">FIGS. 2-8</figref> illustrate the basic operation of a distributed storage register. As shown in <figref idref="DRAWINGS">FIG. 2</figref>, the distributed storage register <b>202</b> is preferably an abstract, or virtual, register, rather than a physical register implemented in the hardware of one particular electronic device. Each process running on a processor or computer system <b>204</b>-<b>208</b> employs a small number of values stored in dynamic memory, and optionally backed up in non-volatile memory, along with a small number of distributed-storage-register-related routines, to collectively implement the distributed storage register <b>202</b>. At the very least, one set of stored values and routines is associated with each processing entity that accesses the distributed storage register. In some implementations, each process running on a physical processor or multi-processor system may manage its own stored values and routines and, in other implementations, processes running on a particular processor or multi-processor system may share the stored values and routines, providing that the sharing is locally coordinated to prevent concurrent access problems by multiple processes running on the processor.

0042In <figref idref="DRAWINGS">FIG. 2</figref>, each computer system maintains a local value <b>210</b>-<b>214</b> for the distributed storage register. In general, the local values stored by the different computer systems are normally identical, and equal to the value of the distributed storage register <b>202</b>. However, occasionally the local values may not all be identical, as in the example shown in <figref idref="DRAWINGS">FIG. 2</figref>, in which case, if a majority of the computer systems currently maintain a single locally stored value, then the value of the distributed storage register is the majority-held value.

0043A distributed storage register provides two fundamental high-level functions to a number of intercommunicating processes that collectively implement the distributed storage register. As shown in <figref idref="DRAWINGS">FIG. 3</figref>, a process can direct a READ request <b>302</b> to the distributed storage register <b>202</b>. If the distributed storage register currently holds a valid value, as shown in <figref idref="DRAWINGS">FIG. 4</figref> by the value “B” within the distributed storage register <b>202</b>, the current, valid value is returned <b>402</b> to the requesting process. However, as shown in <figref idref="DRAWINGS">FIG. 5</figref>, if the distributed storage register <b>202</b> does not currently contain a valid value, then the value NIL <b>502</b> is returned to the requesting process. The value NIL is a value that cannot be a valid value stored within the distributed storage register.

0044A process may also write a value to the distributed storage register. In <figref idref="DRAWINGS">FIG. 6</figref>, a process directs a WRITE message <b>602</b> to the distributed storage register <b>202</b>, the WRITE message <b>602</b> including a new value “X” to be written to the distributed storage register <b>202</b>. If the value transmitted to the distributed storage register successfully overwrites whatever value is currently stored in the distributed storage register, as shown in <figref idref="DRAWINGS">FIG. 7</figref>, then a Boolean value “TRUE” is returned <b>702</b> to the process that directed the WRITE request to the distributed storage register. Otherwise, as shown in <figref idref="DRAWINGS">FIG. 8</figref>, the WRITE request fails, and a Boolean value “FALSE” is returned <b>802</b> to the process that directed the WRITE request to the distributed storage register, the value stored in the distributed storage register unchanged by the WRITE request. In certain implementations, the distributed storage register returns binary values “OK” and “NOK,” with OK indicating successful execution of the WRITE request and NOK indicating that the contents of the distributed storage register are indefinite, or, in other words, that the WRITE may or may not have succeeded.

0045<figref idref="DRAWINGS">FIG. 9</figref> shows the components used by a process or processing entity P_ithat implements, along with a number of other processes and/or processing entities, P_j≠i, a distributed storage register. A processor or processing entity uses three low level primitives: a timer mechanism <b>902</b>, a unique ID <b>904</b>, and a clock <b>906</b>. The processor or processing entity P_iuses a local timer mechanism <b>902</b> that allows P_ito set a timer for a specified period of time, and to then wait for that timer to expire, with P_inotified on expiration of the timer in order to continue some operation. A process can set a timer and continue execution, checking or polling the timer for expiration, or a process can set a timer, suspend execution, and be re-awakened when the timer expires. In either case, the timer allows the process to logically suspend an operation, and subsequently resume the operation after a specified period of time, or to perform some operation for a specified period of time, until the timer expires. The process or processing entity P_ialso has a reliably stored and reliably retrievable local process ID (“PID”) <b>904</b>. Each processor or processing entity has a local PID that is unique with respect to all other processes and/or processing entities that together implement the distributed storage register. Finally, the processor processing entity P_ihas a real-time clock <b>906</b> that is roughly coordinated with some absolute time. The real-time clocks of all the processes and/or processing entities that together collectively implement a distributed storage register need not be precisely synchronized, but should be reasonably reflective of some shared conception of absolute time. Most computers, including personal computers, include a battery-powered system clock that reflects a current, universal time value. For most purposes, including implementation of a distributed storage register, these system clocks need not be precisely synchronized, but only approximately reflective of a current universal time.

0046Each processor or processing entity P_iincludes a volatile memory <b>908</b> and, in some embodiments, a non-volatile memory <b>910</b>. The volatile memory <b>908</b> is used for storing instructions for execution and local values of a number of variables used for the distributed-storage-register protocol. The non-volatile memory <b>910</b> is used for persistently storing the variables used, in some embodiments, for the distributed-storage-register protocol. Persistent storage of variable values provides a relatively straightforward resumption of a process's participation in the collective implementation of a distributed storage register following a crash or communications interruption. However, persistent storage is not required for resumption of a crashed or temporally isolated processor's participation in the collective implementation of the distributed storage register. Instead, provided that the variable values stored in dynamic memory, in non-persistent-storage embodiments, if lost, are all lost together, provided that lost variables are properly re-initialized, and provided that a quorum of processors remains functional and interconnected at all times, the distributed storage register protocol correctly operates, and progress of processes and processing entities using the distributed storage register is maintained. Each process P_istores three variables: (1) val <b>934</b>, which holds the current, local value for the distributed storage register; (2) val-ts <b>936</b>, which indicates the time-stamp value associated with the current local value for the distributed storage register; and (3) ord-ts <b>938</b>, which indicates the most recent timestamp associated with a WRITE operation. The variable val is initialized, particularly in non-persistent-storage embodiments, to a value NIL that is different from any value written to the distributed storage register by processes or processing entities, and that is, therefore, distinguishable from all other distributed-storage-register values. Similarly, the values of variables val-ts and ord-ts are initialized to the value “initialTS,” a value less than any time-stamp value returned by a routine “newTS” used to generate time-stamp values. Providing that val, val-ts, and ord-ts are together re-initialized to these values, the collectively implemented distributed storage register tolerates communications interruptions and process and processing entity crashes, provided that at least a majority of processes and processing entities recover and resume correction operation.

0047Each processor or processing entity P_imay be interconnected to the other processes and processing entities P_j≠ivia a message-based network in order to receive <b>912</b> and send <b>914</b> messages to the other processes and processing entities P_j≠i. Each processor or processing entity P_iincludes a routine “newTS” <b>916</b> that returns a timestamp TS_iwhen called, the timestamp TS_igreater than some initial value “initialTS.” Each time the routine “newTS” is called, it returns a timestamp TS_igreater than any timestamp previously returned. Also, any timestamp value TS_ireturned by the newTS called by a processor or processing entity P_ishould be different from any timestamp TS_jreturned by newTS called by any other processor processing entity P_j. One practical method for implementing newTS is for newTS to return a timestamp TS comprising the concatenation of the local PID <b>904</b> with the current time reported by the system clock <b>906</b>. Each processor or processing entity P_ithat implements the distributed storage register includes four different handler routines: (1) a READ handler <b>918</b>; (2) an ORDER handler <b>920</b>; (3) a WRITE handler <b>922</b>; and (4) an ORDER&amp;READ handler <b>924</b>. It is important to note that handler routines may need to employ critical sections, or code sections single-threaded by locks, to prevent race conditions in testing and setting of various local data values. Each processor or processing entity P_ialso has four operational routines: (1) READ <b>926</b>; (2) WRITE <b>928</b>; (3) RECOVER <b>930</b>; and (4) MAJORITY <b>932</b>. Both the four handler routines and the four operational routines are discussed in detail, below.

0048Correct operation of a distributed storage register, and liveness, or progress, of processes and processing entities using a distributed storage register depends on a number of assumptions. Each process or processing entity P_iis assumed to not behave maliciously. In other words, each processor or processing entity P_ifaithfully adheres to the distributed-storage-register protocol. Another assumption is that a majority of the processes and/or processing entities P_ithat collectively implement a distributed storage register either never crash or eventually stop crashing and execute reliably. As discussed above, a distributed storage register implementation is tolerant to lost messages, communications interruptions, and process and processing-entity crashes. When a number of processes or processing entities are crashed or isolated that is less than sufficient to break the quorum of processes or processing entities, the distributed storage register remains correct and live. When a sufficient number of processes or processing entities are crashed or isolated to break the quorum of processes or processing entities, the system remains correct, but not live. As mentioned above, all of the processes and/or processing entities are fully interconnected by a message-based network. The message-based network may be asynchronous, with no bounds on message-transmission times. However, a fair-loss property for the network is assumed, which essentially guarantees that if P_ireceives a message m from P_j, then P_jsent the message m, and also essentially guarantees that if P_irepeatedly transmits the message m to P_j, P_jwill eventually receive message m, if P_jis a correct process or processing entity. Again, as discussed above, it is assumed that the system clocks for all processes or processing entities are all reasonably reflective of some shared time standard, but need not be precisely synchronized.

0049These assumptions are useful to prove correctness of the distributed-storage-register protocol and to guarantee progress. However, in certain practical implementations, one or more of the assumptions may be violated, and a reasonably functional distributed storage register obtained. In addition, additional safeguards may be built into the handler routines and operational routines in order to overcome particular deficiencies in the hardware platforms and processing entities.

0050Operation of the distributed storage register is based on the concept of a quorum. <figref idref="DRAWINGS">FIG. 10</figref> illustrates determination of the current value of a distributed storage register by means of a quorum. <figref idref="DRAWINGS">FIG. 10</figref> uses similar illustration conventions as used in <figref idref="DRAWINGS">FIGS. 2-8</figref>. In <figref idref="DRAWINGS">FIG. 10</figref>, each of the processes or processing entities <b>1002</b>-<b>1006</b> maintains the local variable, val-ts, such as local variable <b>1007</b> maintained by process or processing entity <b>1002</b>, that holds a local time-stamp value for the distributed storage register. If, as in <figref idref="DRAWINGS">FIG. 6</figref>, a majority of the local values maintained by the various processes and/or processing entities that collectively implement the distributed storage register currently agree on a time-stamp value val-ts, associated with the distributed storage register, then the current value of the distributed storage register <b>1008</b> is considered to be the value of the variable val held by the majority of the processes or processing entities. If a majority of the processes and processing entities cannot agree on a time-stamp value val-ts, or there is no single majority-held value, then the contents of the distributed storage register are undefined. However, a minority-held value can be then selected and agreed upon by a majority of processes and/or processing entities, in order to recover the distributed storage register.

0051<figref idref="DRAWINGS">FIG. 11</figref> shows pseudocode implementations for the routine handlers and operational routines shown diagrammatically in <figref idref="DRAWINGS">FIG. 9</figref>. It should be noted that these pseudocode implementations omit detailed error handling and specific details of low-level communications primitives, local locking, and other details that are well understood and straightforwardly implemented by those skilled in the art of computer programming. The routine “majority” <b>1102</b> sends a message, on line 2, from a process or processing entity P_ito itself and to all other processes or processing entities P_j≠ithat, together with P_i, collectively implement a distributed storage register. The message is periodically resent, until an adequate number of replies are received, and, in many implementations, a timer is set to place a finite time and execution limit on this step. Then, on lines 3-4, the routine “majority” waits to receive replies to the message, and then returns the received replies on line 5. The assumption that a majority of processes are correct, discussed above, essentially guarantees that the routine “majority” will eventually return, whether or not a timer is used. In practical implementations, a timer facilitates handling error occurrences in a timely manner. Note that each message is uniquely identified, generally with a timestamp or other unique number, so that replies received by process P_ican be correlated with a previously sent message.

0052The routine “read” <b>1104</b> reads a value from the distributed storage register. On line 2, the routine “read” calls the routine “majority” to send a READ message to itself and to each of the other processes or processing entities P_j≠i. The READ message includes an indication that the message is a READ message, as well as the time-stamp value associated with the local, current distributed storage register value held by process P_i, val-ts. If the routine “majority” returns a set of replies, all containing the Boolean value “TRUE,” as determined on line 3, then the routine “read” returns the local current distributed-storage-register value, val. Otherwise, on line 4, the routine “read” calls the routine “recover.”

0053The routine “recover” <b>1106</b> seeks to determine a current value of the distributed storage register by a quorum technique. First, on line 2, a new timestamp ts is obtained by calling the routine “newTS.” Then, on line 3, the routine “majority” is called to send ORDER&amp;READ messages to all of the processes and/or processing entities. If any status in the replies returned by the routine “majority” are “FALSE,” then “recover” returns the value NIL, on line 4. Otherwise, on line 5, the local current value of the distributed storage register, val, is set to the value associated with the highest value timestamp in the set of replies returned by routine “majority.” Next, on line 6, the routine “majority” is again called to send a WRITE message that includes the new timestamp ts, obtained on line 2, and the new local current value of the distributed storage register, val. If the status in all the replies has the Boolean value “TRUE,” then the WRITE operation has succeeded, and a majority of the processes and/or processing entities now concur with that new value, stored in the local copy val on line 5. Otherwise, the routine “recover” returns the value NIL.

0054The routine “write” <b>1108</b> writes a new value to the distributed storage register. A new timestamp, ts, is obtained on line 2. The routine “majority” is called, on line 3, to send an ORDER message, including the new timestamp, to all of the processes and/or processing entities. If any of the status values returned in reply messages returned by the routine “majority” are “FALSE,” then the value “NOK” is returned by the routine “write,” on line 4. Otherwise, the value val is written to the other processes and/or processing entities, on line 5, by sending a WRITE message via the routine “majority.” If all the status vales in replies returned by the routine “majority” are “TRUE,” as determined on line 6, then the routine “write” returns the value “OK.” Otherwise, on line 7, the routine “write” returns the value “NOK.” Note that, in both the case of the routine “recover” <b>1106</b> and the routine “write,” the local copy of the distributed-storage-register value val and the local copy of the timestamp value val-ts are both updated by local handler routines, discussed below.

0055Next, the handler routines are discussed. At the onset, it should be noted that the handler routines compare received values to local-variable values, and then set local variable values according to the outcome of the comparisons. These types of operations may need to be strictly serialized, and protected against race conditions within each process and/or processing entity for data structures that store multiple values. Local serialization is easily accomplished using critical sections or local locks based on atomic test-and-set instructions. The READ handler routine <b>1110</b> receives a READ message, and replies to the READ message with a status value that indicates whether or not the local copy of the timestamp val-ts in the receiving process or entity is equal to the timestamp received in the READ message, and whether or not the timestamp ts received in the READ message is greater than or equal to the current value of a local variable ord-ts. The WRITE handler routine <b>1112</b> receives a WRITE message determines a value for a local variable status, on line 2, that indicates whether or not the local copy of the timestamp val-ts in the receiving process or entity is greater than the timestamp received in the WRITE message, and whether or not the timestamp ts received in the WRITE message is greater than or equal to the current value of a local variable ord-ts. If the value of the status local variable is “TRUE,” determined on line 3, then the WRITE handler routine updates the locally stored value and timestamp, val and val-ts, on lines 4-5, both in dynamic memory and in persistent memory, with the value and timestamp received in the WRITE message. Finally, on line 6, the value held in the local variable status is returned to the process or processing entity that sent the WRITE message handled by the WRITE handler routine <b>1112</b>.

0056The ORDER&amp;READ handler <b>1114</b> computes a value for the local variable status, on line 2, and returns that value to the process or processing entity from which an ORDER&amp;READ message was received. The computed value of status is a Boolean value indicating whether or not the timestamp received in the ORDER&amp;READ message is greater than both the values stored in local variables val-ts and ord-ts. If the computed value of status is “TRUE,” then the received timestamp ts is stored into both dynamic memory and persistent memory in the variable ord-ts.

0057Similarly, the ORDER handler <b>1116</b> computes a value for a local variable status, on line 2, and returns that status to the process or processing entity from which an ORDER message was received. The status reflects whether or not the received timestamp is greater than the values held in local variables val-ts and ord-ts. If the computed value of status is “TRUE,” then the received timestamp ts is stored into both dynamic memory and persistent memory in the variable ord-ts.

0058Using the distributed storage register method and protocol, discussed above, shared state information that is continuously consistently maintained in a distributed data-storage system can be stored in a set of distributed storage registers, one unit of shared state information per register. The size of a register may vary to accommodate different natural sizes of units of shared state information. The granularity of state information units can be determined by performance monitoring, or by analysis of expected exchange rates of units of state information within a particular distributed system. Larger units incur less overhead for protocol variables and other data maintained for a distributed storage register, but may result in increased communications overhead if different portions of the units are accessed at different times. It should also be noted that, while the above pseudocode and illustrations are directed to implementation of a single distributed storage register, these pseudocode routines can be generalized by adding parameters identifying a particular distributed storage register, of unit of state information, to which operations are directed, and by maintaining arrays of variables, such as val-ts, val, and ord-ts, indexed by the identifying parameters.

Disk Paxos

0059Disk Paxos and active-disk Paxos are two additional distributed-computing techniques similar to Paxos. However, while Paxos distributes a global, master list of committed state-change requests over the nodes of a distributed computer system, disk Paxos and active-disk Paxos distribute a global list of state changes over a number of mass-storage devices. <figref idref="DRAWINGS">FIG. 12</figref> illustrates a disk-Paxos or active-disk-Paxos distributed computer system. As shown in <figref idref="DRAWINGS">FIG. 12</figref>, each computing node <b>1202</b>-<b>1206</b> includes a view of the global state-change list, such as the view <b>1208</b> in node <b>1202</b>. A global state-change list is distributed across mass-storage devices <b>1210</b>-<b>1213</b>. In disk Paxos and active disk Paxos, a quorum of mass-storage devices, rather than computing nodes, is required for committing state-change requests. Thus, state-change requests may succeed in a disk Paxos or active disk Paxos system when even a single computing node is active. Active-disk Paxos involves different classes of disk nodes, with the disk nodes of only one class involved in quorums, and with better scalability to large systems. Both disk Paxos and active-disk Paxos provide functionality equivalent to Paxos. All three protocols allow for maintaining a shared, ordered, sequential, global list of committed state changes across the entire distributed computer system, allow individual nodes of a distributed computer system to request state changes, and resolve request-submission contention.

Strong-Leader Self Election

0060A robust and reliable strong-leader election method within a distributed computer system can be implemented by using a quorum-based distributed consensus system, such as Paxos, along with two additional functionalities included in each node of the distributed system. A strong leader is a leader that, once acquiring a leadership role, continues in the leadership role for an extended period of time or over an extended computational process. By contrast, a weak leader assumes a leadership role only for a particular, well-bounded task or period of time, after which all or a large number of nodes contend for the leadership role for a subsequent, well-bounded task or period of time.

0061<figref idref="DRAWINGS">FIG. 13</figref> illustrates an exemplary distributed computer system in which strong-leader election may be practiced. In <figref idref="DRAWINGS">FIG. 13</figref>, five computing nodes <b>1302</b>-<b>1306</b> are linked together by a communications medium <b>1308</b>. Each node includes a local state-change list or view of a global state-change list, such as state-change list <b>1309</b> in node <b>1302</b>, provided according to a Paxos, disk-Paxos, or Paxos-like distributed consensus service or a distributed-storage-register-based consensus service. Each node also includes timing functionality, such as timing functionality <b>1310</b> in node <b>1302</b>, a fail-stop mechanism, such as fail-stop mechanism <b>1312</b> in node <b>1302</b>, and leader-election functionality, generally implemented in software or firmware, such as leader-election functionality <b>1314</b> in node <b>1302</b>.

0062The timing functionality includes a node clock that indicates regular intervals in time, such as milliseconds, generally with monotonically increasing values based on an arbitrary starting point. The timing functionalities of all of the nodes are synchronized at some level of precision. In other words, any disparities between times indicated by timing functionalities of the different nodes of a distributed computer system, at any given instant in time, are less than a maximum disparity value. Associated with the timing functionality are software and/or hardware timers that can be set to expire, and to provide notice of expiration, after an arbitrary interval of time.

0063The fail-stop functionality provides a means to signal a fail-stop condition and discontinue computation related to the signaled fail-stop condition. It is desirable that the fail-stop functionality be implemented at least partially in hardware, to ensure that time lags between recognition of a fail-stop condition and fail-stop signaling are minimized. For example, a fail-stop device may involve high-priority hardware interrupts and non-blocking interrupt handlers.

0064The leader-election functionality, generally implemented in software, or firmware, or a combination of software and firmware, implements the strong-leader election method of the present invention. The leader-election functionality is described below, using several control-flow diagrams. In general, the strong-leader election methods of the present invention provide for self election by nodes. Nodes issue Paxos or Paxos-like state-change requests to request that they become the leader for a next lease period. The current leader, if still active, is provided an advantage of requesting re-election at time part-way through the current lease period as which the current leader's request for re-election is unopposed.

0065It is assumed that, upon node initialization or upon re-initialization of a node following failure and recovery of the node, the timing functionality of the node is synchronized with the timing functionalities of the other, active nodes within the distributing computing system. Furthermore, all nodes are initialized or re-initialized to include a constant time value LEASE, which represents the length, in time increments, of a lease period for holding a leadership role, and to include a value n that is used to generate, by division, a fraction of the lease time LEASE at which current leaders request re-election. The leader re-election request period, LEASE/n, needs to be sufficiently less than the lease period that a current leader succeeds in re-election despite network delays, scheduling delays, intra-node timer misalignment, and other such potential sources of delay.

0066<figref idref="DRAWINGS">FIG. 14</figref> is a control-flow diagram illustrating general node operation and strong-leader election. Node operation can be considered, at a high level, to be an endless loop in which the node recognizes and handles events. It is assumed that, when events occur simultaneously, event processing hardware and software sequentially order and prioritize the events, such as the interrupt ordering and prioritization that occurs within operating systems. In step <b>1402</b> of the endless loop, the node waits for a next event, while continuing to process any computational tasks currently being executed by one or more processes or threads within the node. Upon occurrence of an event, the node first determines, in step <b>1404</b>, whether a leader-election fail-stop event has occurred. If so, then in step <b>1406</b>, the node sets a node global variable fail_stop to TRUE, and may additionally actively shut down any current processing activities related to a leadership role previously assumed by the node. In alternative embodiments, in which it is important that a leader that has failed to re-elect itself immediately halt any leader-role-related processing, a hardware-reset may instead be generated by a hardware timer, to immediately halt leader operation. If a leadership-election fail-stop event has not occurred, then in step <b>1408</b>, the node checks whether a leader-election timer has expired. If a leader-election timer has expired, then, in step <b>1410</b>, the node determines whether the global node variable fail_stop is TRUE. If so, then no action is taken. Otherwise, the node executes the “elect self” routine, in step <b>1412</b>, to be discussed below. If a leader-election timer expiration has not occurred, then, in step <b>1414</b>, the node determines whether a leader-election fail-stop-reset event has occurred. If so, then the global node variable fail_stop is set to FALSE, in step <b>1416</b>. If a leader-election fail-stop-reset event has not occurred then, in step <b>1418</b>, another event has been detected, then that event is handled in step <b>1420</b>. In other words, step <b>1420</b> represents handling of myriad non-leader-election-related events that occur within a node during node operation. In summary, a node carries out computational tasks while, at the same time, monitoring for the occurrence of leader-election fail-stop events, leader-election fail-stop-reset events, and expiration of leader-election timers. When a leader-election fail-stop event occurs, the node discontinues processing any leader-related tasks. When a leader-election timer expires, the node calls the routine “elect self,” described below, to attempt to elect itself to a leadership role.

0067<figref idref="DRAWINGS">FIG. 15</figref> is a control-flow diagram illustrating the routine “elect self.” In step <b>1502</b>, the routine determines the value for a next lease interval next_interval. In the embodiment shown in <figref idref="DRAWINGS">FIG. 15</figref>, a next-interval value is determined by integer division of the current time by the LEASE period, with the value incremented when the current time is closer to onset of a subsequent lease interval than to onset of the current lease interval. Next, in step <b>1504</b>, the routine issues a Paxos state-change request requesting that the node be designated leader for the lease period indicated by the value of the variable next_interval. If the request succeeds, as determined in step <b>1506</b>, then the node has become the leader for the next lease period, and sets the leader-election timer to expire at a fraction LEASE/n of the next lease interval in step <b>1508</b>. The node then sets the fail-stop functionality to expire at the end of the next lease period in step <b>1510</b>. The node then assumes a leadership role for the next lease time period. Otherwise, if the Paxos request does not succeed, then, in step <b>1512</b>, the node sets the leader-election timer to expire at the end of the next lease period. By setting the leader-election timers to a fraction of the next lease period, in step <b>1508</b>, when the node is the leader, the node ensures that, should the node continue to be active, the node will always request re-election unopposed by other nodes within the distributed computer system.

0068<figref idref="DRAWINGS">FIGS. 16A-G</figref> illustrate the strong-leader election method. <figref idref="DRAWINGS">FIGS. 16A-G</figref> all use the same illustration conventions, next described with reference to <figref idref="DRAWINGS">FIG. 16A</figref>. In <figref idref="DRAWINGS">FIG. 16A</figref>, there are five nodes (nodes <b>1</b>-<b>5</b>) <b>1602</b>-<b>1606</b>. Events are discussed with respect to a time line <b>1608</b> that is divided into discrete, contiguous lease periods, such as lease period <b>1610</b>. In <figref idref="DRAWINGS">FIG. 16A</figref>, no node is currently leader, and all nodes have been initialized to contend for a leadership role as soon as the lease period k+1 <b>1612</b> begins. The request for a leadership role is represented for each node by an arrow pointing to the time line, such as arrow <b>1614</b> representing a request for the leadership role made by node <b>1602</b>.

0069According to the Paxos protocol, only one of the simultaneous or close-to-simultaneous requests for the leadership role, shown in <figref idref="DRAWINGS">FIG. 16A</figref>, succeeds. Assuming that the request issued by node <b>3</b><b>1604</b> succeeds, then, as shown in <figref idref="DRAWINGS">FIG. 16B</figref>, node <b>3</b> is designated as the leader for lease period k+1 <b>1612</b>, and immediately begins to execute leader-related processing.

0070As shown in <figref idref="DRAWINGS">FIG. 16C</figref>, assuming n is equal to 2, node <b>3</b> requests the leadership role for the next lease period k+2 <b>1610</b> halfway through the current lease period <b>1612</b>, as represented by arrow <b>1616</b>. Because all other, non-leader nodes wait until the start of the next lease period to request a leadership role, the request by node <b>3</b> proceeds unopposed, and therefore succeeds. Node <b>3</b> is designated the leader both for the current lease period k+1 and for the subsequent lease period k+2. At the beginning of lease period k+2 <b>1610</b>, as shown in <figref idref="DRAWINGS">FIG. 16E</figref>, the remaining nodes issue leadership-role requests, following the expiration of their leader-election timers. However, node <b>3</b> has already succeeded in obtaining leadership for lease period n+2, so these requests fail. Although not shown in a control-flow diagram, a Paxos, Paxos-like, or distributed-storage-register protocol is supplemented so refuse a second and any additional, subsequent requests for the leadership role for a particular lease period. Disk Paxos and active-disk Paxos do not need to be so supplemented, since these distributed consensus services fail a request made by a node or process that is not locally updated to see all recent state changes made by other nodes or processes. As long as node <b>3</b> continues to operate, node <b>3</b> remains the leader.

0071If, as shown in <figref idref="DRAWINGS">FIG. 16F</figref>, node <b>3</b> fails prior to the point LEASE/n within lease period m, then node <b>3</b> will fail to request a leadership role for lease period m+1 <b>1622</b>. Therefore, when the remaining non-leader nodes request a leadership role at the beginning of lease period m+1, one of the other nodes will assume leadership for lease period m+1 as shown in <figref idref="DRAWINGS">FIG. 16G</figref>.

0072The strong-leader-election method therefore ensures that, even when a leader node fails, the leadership role resumes at most after a period equal to the lease period plus the fractional lease period at which leader nodes request re-election. Furthermore, once elected to a leadership role, a node can retain the leadership role as long as the node remains active and desires the leadership role. At any point in time, the leader node may surrender the leadership role by failing to request the leadership role for the subsequent lease period and optionally disabling fail-safe functionality.

0073While the above-described embodiment employs roughly synchronized, absolute-time-reflecting node clocks, alternative implementations use unsynchronized delay timers in each node. <figref idref="DRAWINGS">FIG. 17</figref> illustrates, in similar fashion to <figref idref="DRAWINGS">FIG. 13</figref>, an alternative distributed computer system in which strong-leader election may be practiced. In <figref idref="DRAWINGS">FIG. 17</figref>, six computing nodes <b>1702</b>-<b>1707</b> are linked together by a communications medium <b>1708</b>. Each node includes a local view of a global variable indicating the current leader node, such as local view <b>1709</b> in node <b>1702</b>, provided according to a Paxos, disk-Paxos, or Paxos-like distributed consensus service or a distributed-storage-register-based consensus service. Each node also includes a local delay timer, such as local delay timer <b>1710</b> in node <b>1702</b>, a fail-stop mechanism, such as fail-stop mechanism <b>1712</b> in node <b>1702</b>, and leader-election functionality, generally implemented in software or firmware, such as leader-election functionality <b>1714</b> in node <b>1702</b>. The delay timers of the nodes are not synchronized with one another, unlike the node clocks of the above, first-described embodiment of the present invention. By contrast, the fail-stop mechanisms and leader-election functionalities are similar to, and play similar roles, as the fail-stop mechanisms and leader-election functionalities in the above, first-described embodiment of the present invention. As with the above, first-described embodiment of the present invention, each node includes, or has access to, a lease period defined by a constant LEASE and a fractional lease period defined by a value n used to compute the fractional lease period as LEASE/n. The control-flow diagram illustrating general node operation shown in <figref idref="DRAWINGS">FIG. 14</figref> is applicable to the alternative embodiments of the present invention, with the exception that a different routine is called in step <b>1412</b>.

0074<figref idref="DRAWINGS">FIGS. 18A-D</figref> illustrate operation of delay-timer-based strong-leader election. In <figref idref="DRAWINGS">FIG. 18A</figref>, four computing nodes <b>1802</b>-<b>1805</b> that together employ a distributed consensus system to implement a distributed state variable or state variables <b>1806</b> that store a value indicative of the current leader node, and, in certain embodiments, other related information. At power-on, after failure of a leader node, and at other such points, some or all of the computing nodes may attempt to change the state variable in order to acquire leadership, as shown in <figref idref="DRAWINGS">FIG. 18B</figref>. As shown in <figref idref="DRAWINGS">FIG. 18C</figref>, one of the competing computing nodes is guaranteed, by the distributed consensus service, to acquire leadership, while the others' requests for leadership fail. In <figref idref="DRAWINGS">FIG. 18C</figref>, node <b>1</b> (<b>1802</b>) successfully acquired leadership, and has established a re-election cycle <b>1810</b> in which the node seeks re-election following each period of time LEASE/n. The remaining nodes <b>1803</b>-<b>1805</b> failed to acquire leadership, as shown in <figref idref="DRAWINGS">FIG. 18D</figref>, and therefore establish election cycles <b>1812</b>-<b>1814</b> in which each non-leader node seeks election following each period of time LEASE. In <figref idref="DRAWINGS">FIG. 18D</figref>, and in subsequent figures, the longer non-leader election cycles are shown as circles with larger diameters than the re-election cycle <b>1810</b> of the current leader node. In the alternative embodiments, the absolute time of each node's actions is irrelevant. Because, in the alternative embodiments, a nodes reference a local delay timer to decide when to seek election or re-election, rather than a node clock, each node may seek election at any time with respect to the actions taken by the remaining nodes, rather than within some range of times about an absolute time recognized by all nodes.

0075<figref idref="DRAWINGS">FIG. 19</figref> is a control-flow diagram illustrating an alternative strong-leader election method. In step <b>1902</b>, the routine issues a Paxos or Paxos-like state-change request requesting that the node be designated leader. If the request succeeds, as determined in step <b>1906</b>, then the node has become the leader for the next lease period, and sets the local delay timer to expire at a fraction LEASE/n of the next lease period in step <b>1906</b>. The node then sets the fail-stop functionality to expire at the end of the next lease period in step <b>1908</b>. Otherwise, if the Paxos or Paxos-like request does not succeed, then, in step <b>1910</b>, the node updates the node's local view of the distributed state variable. Finally, in step <b>1912</b>, the current node sets the current node's delay timer to the lease period LEASE, in order to again seek election for a next lease period. Otherwise, the current node has acquired leadership, and executes steps <b>1906</b> and <b>1908</b>, as described above. In this embodiment of the present invention, a new leader is elected within a time of (2*LEASE)−LEASE/n.

0076<figref idref="DRAWINGS">FIGS. 20A-C</figref> illustrate operation of the alternative strong-leader-election method, described above. In <figref idref="DRAWINGS">FIG. 20A</figref>, the current leader node <b>2002</b> has just been re-elected, and therefore has a fully updated local view of the distributed state variable. Node <b>2004</b> is just about to seek election, but does not have an updated local view of the distributed state variable. Nodes <b>2006</b> and <b>2008</b> are in different points in their respective election cycles, and do not have an updated local view of the distributed state variable. <figref idref="DRAWINGS">FIG. 20B</figref> shows the four-node system of <figref idref="DRAWINGS">FIG. 20A</figref> following additional lapse of time. The delay timers for all four nodes have advanced, with node <b>2004</b> having unsuccessfully sought election, and, as a result, having an updated local view of the distributed state variable. In <figref idref="DRAWINGS">FIG. 20C</figref>, the delay timers have further advanced, and the current leader node <b>2002</b> has failed, as indicated by the “X” symbol <b>2012</b> in the re-election cycle. As shown in <figref idref="DRAWINGS">FIG. 20C</figref> by dashed lines <b>2014</b>-<b>2016</b>, the various non-leader nodes need different periods of time to be elected to the leadership role. Node <b>2004</b> may become the new leader in the shortest period of time <b>2014</b>, since node <b>2004</b> has an updated local view of the distributed state variable. Nodes <b>2006</b> and <b>2008</b> both need to first update their local views of the distributed state variable before waiting for a lease period in order to seek election.

Embodiments of the Present Invention

0077The strong-leader-election method and system discussed in the previous subsection may be replicated in order to provide strong-leader election for multiple, different roles within a distributed computer system. However, in more complex distributed computer systems, there may be a great number of different roles for which leaders may need to be elected on a relatively continuous basis, with roles allocated to processes within nodes of the distributed computer system, rather than assigned to nodes. In such cases, a higher-level leader-election service may be desirable.

0078<figref idref="DRAWINGS">FIG. 21</figref> illustrates a leader-election-service-based distributed computer system that represents one embodiment of the present invention. In <figref idref="DRAWINGS">FIG. 21</figref>, seven nodes <b>2102</b>-<b>2108</b> are shown together composing a distributed computer system. Each node includes a leader-election-service process, such as leader-election-service process <b>2110</b> within node <b>2102</b>. Each leader-election-service process maintains local state information <b>2112</b> within a node, and cooperates with the leader-election-service processes of the remaining nodes in order to maintain distributed state information <b>2114</b> via a distributed consensus service, such as Paxos or Disk Paxos. Client processes, such as client process <b>2116</b> within node <b>2102</b>, interface to the leader-election-service process <b>2110</b> within the same node to request election to leadership roles for particular roles within the distributed computer system, as well as to renounce leadership roles assigned to them, request leadership and role information, and to carry out other leader-election-service-related activities, discussed in detail below. The leader-election-service process within each node therefore represents a centralized leader-election service to which various processes in the node may interface. The leader-election-service processes of all of the nodes of the distributed computer system cooperate to maintain distributed state information and to collect and distribute client requests to current leaders for each role within the distributed computer system.

0079In the described embodiment, each leader-election-service process is a distinct, continuously running process within each node. However, in alternative embodiments, the leader-election service may be implemented by library routines that run in the context of client processes. In the current discussion, the leader-election-service processes manage election of client processes within nodes to leadership of distributed-computer-system-wide roles. In alternative embodiments, the leader-election service may also manage allocation of leadership to client processes within a node with respect to intra-node roles. For intra-node roles, the leader-election service needs to access only local state information.

0080<figref idref="DRAWINGS">FIG. 22</figref> illustrates the basic components of the leadership-election service that represents one embodiment of the present invention. As discussed above, each node <b>2202</b> includes a leader-election service <b>2204</b>, in the described embodiment a continuously running process that services leader-election-service-related requests from client processes within the node and, in certain situations, receives and forwards requests forwarded to the leader-election service from leader-election-service processes in external nodes. A leader-election-service identifier (“ID”) <b>2206</b> is associated with, and identifies, a leader-election-service process. Each node is identified by a node ID <b>2208</b>. Each client process, such as client process <b>2210</b>, is identified by a client ID <b>2212</b> and may currently be a leader for one or more roles identified by role IDs <b>2214</b> and <b>2216</b>. In general, the leader-election service <b>2204</b> cooperates with leader-election-service processes in external nodes to distribute leadership for system-wide roles among client processes within the distributed computer system.

0081<figref idref="DRAWINGS">FIG. 23</figref> illustrates a portion of the distributed state information maintained cooperatively by leader-election services within a distributed computer system according to one embodiment of the present invention. The distributed state information is shown as two relational-database tables, and relational-database tables are also used, below, to illustrate local state information locally maintained by each leader-election service within a node. In alternative embodiments, any number of different types of databases, file systems, or other information-storage systems and methods may be employed to store both global, distributed state information and local state information by leader-election services.

0082The information stored in the distributed state information maintained within the distributed computer system includes specific leadership allocations to client processes running on nodes within the distributed computer system. The specific leadership allocations are stored, in the described embodiment, in the table “Global Role Assignments” <b>2302</b>. Each row in the table, such as row <b>2304</b>, represents a particular leadership assignment. For example, row <b>2304</b> indicates that the role identified by the role ID “A” has been allocated to a process within the node identified by node ID “X.” The leader-election-service process within node identified by node ID “X” is identified by the leader-election-service ID “PROC3204.” The time which the leadership role was assumed by the current leader is represented by a large integer <b>2306</b>. Thus, in the described embodiment, a leadership allocation is identified by a four-element tuple: {role ID, node ID, leader-election-service ID, time assumed}. In alternative embodiments, any number of different tuples with different element types having different element-type value ranges may be used to represent leadership-role allocations.

0083In certain embodiments, the distributed state information may include constant information, such as descriptions of the various roles within the distributed computer system stored in the table “Roles” <b>2310</b> shown in <figref idref="DRAWINGS">FIG. 23</figref>. In this table, each row, such as row <b>2312</b>, represents each role that can be allocated to a client process within the distributed computer system as the association of a role ID with a role description. For example, in <figref idref="DRAWINGS">FIG. 23</figref>, the role identified by role ID “A” is described by the text string “software installation manager.” Additional dynamic and constant information may be stored within the distributed state information managed by the leader-election services of a distributed computer system. Such additional state information may include information used for synchronizing role transfer between client processes and nodes, synchronization of request collection and request distribution on behalf of client leaders, and other such information.

0084<figref idref="DRAWINGS">FIG. 24</figref> illustrates the local state information maintained by the leader-election-service process within a node of a distributed computer system, according to one embodiment of the present invention. In one embodiment, the information includes: (1) identifications of all client processes, currently running within a node, that are able to accept a leadership role, stored in the table “Clients” <b>2402</b>; (2) indications of the roles currently allocated to each client, stored in the table “Accepted Roles” <b>2404</b>; and (3) indications of the roles that each client process is willing to accept, stored in the table “Registrations” <b>2406</b>. The table “Clients” stores associations between client IDs and a process identifier identifying the client within a node. For example, in row <b>2408</b> in the table “Clients,” the client process identified by client ID “J” corresponds to a process running within a node identified by the process number “PROC 6222.” Row <b>2410</b> in the table “Accepted Roles” indicates that the client process identified by client ID “J” is currently the leader for the role identified by role ID “A.” Row <b>2412</b> in table “Registrations” indicates that the client identified by client ID “J” registered a willingness to accept leadership of the role identified by role ID “A,” and has furnished a callback routine identified by an alphanumeric string <b>2414</b> for receiving requests and callbacks from the leader-election service related to the client's willingness to accept the role.

0085Next, the distributed-leader-election service that represents one embodiment of the present invention is described in a series of control-flow diagrams representing the client/leader-election-service interface. These control-flow diagrams are divided into two parts: (1) a client portion, shown on the left side of the diagram; and (2) a leader-election-service portion, on the right side of the diagram. These control-flow diagrams cover the basic aspects of the client/leader-election-service interface, but, in most embodiments, additional types of interactions are generally provided for. Moreover, the leader-election-service processes in a distributed computer system interact, primarily through a distributed consensus service, such as Paxos, to provide a consistent allocation of leadership among client processes within the distributed computer system. Leader-election-service processes avail themselves of any of a number of different types of distributed system management systems in order to detect node and process failures within a distributed computer system in order to update the distributed state information and, in certain embodiments, to anticipate the need for electing new leaders for roles, the leaders for which are terminated or were running on nodes that failed. In alternative embodiments, node and process failure may be discovered by client processes, which can then use the client/leader-election service interface to restore leadership for roles abandoned due to process, node, or system-component failures on an as-detected-and-as-needed basis.

0086<figref idref="DRAWINGS">FIG. 25</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Register.” In a first step <b>2502</b>, the client transmits to the leader-election service a role ID for the role that the client is willing to accept leadership of as well as a callback routine by which the client can receive requests related to the role, once leadership of the role is allocated to the client, and by which the leader-election service can communicate with the client process with regard to the allocation of leadership for the role. The leader-election service receives the transmitted information <b>2504</b>, including the client ID for the client that transmitted the message. In certain embodiments, the client ID is automatically transferred as part of the messaging system, while in alternative embodiments, the leader-election service uses information received from the messaging service to look up the client ID in the table “Clients” in the local state information maintained by the leader-election service. In still additional embodiments, the client ID may be explicitly furnished by the client. If the client ID is known to the leader-election service or, in other words, an entry for the client ID exists in the table “Clients,” then the leader-election service can proceed to register the client for the particular role. Otherwise, in step <b>2508</b>, the leader-election service inserts a client ID/client handle tuple into the table “Clients,” in step <b>2508</b>. If the role identified by the role ID transmitted by the client is valid, as determined in step <b>2510</b> by accessing the table “Roles” in the distributed state information, and if the callback routine furnished by the client is valid, as determined in step <b>2512</b>, then registration proceeds. Otherwise, in step <b>2514</b>, a failure is returned to the client. In step <b>2516</b>, the leader-election service enters a new client ID/role ID/callback tuple into the table “Registrations” and then, in step <b>2518</b>, returns a success status to the client. In step <b>2520</b>, the client receives the return value from the leader-election service, and undertakes any additional processing based on that return value. Thus, successful execution of the register-role interface routine results in a new entry in the table “Registrations” (<b>2406</b> in <figref idref="DRAWINGS">FIG. 24</figref>) indicating that the client is willing to accept a leadership role for the role identified by the role ID transmitted to the leader-election service by the client. In alternative embodiments, the client may specify multiple role IDs in a single call to the register-role routine. In yet additional embodiments, the client may additionally specify a length of time for which the client wishes to assume a role, specify various role-termination conditions, specify a time following which the client is willing to accept leadership roles, and other such parameters, constraints, and options.

0087<figref idref="DRAWINGS">FIG. 26</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Unregister.” This routine allows a client to indicate that the client is no longer willing to accept a leadership role with respect to a role identified by a role ID. In step <b>2602</b>, the client sends the role ID to the leader-election service. The leader-election service receives the information, in step <b>2604</b>, and then, in steps <b>2606</b>-<b>2610</b>, and verifies that the unregister request is valid, returning a failure status or failure code when the request is not valid. When the request is valid, the leader-election service deletes the client ID/role ID/callback tuple corresponding to the unregister request from the table “Registrations” in step <b>2612</b> and returns a success status in step <b>2614</b>. An unregister request may be invalid for a variety of reasons, including lack of an entry corresponding to the requesting client in the table “Clients” or lack of a corresponding entry in the table “Registrations.” As determined in step <b>2606</b>, when the leader-election service has asserted allocation to the client of leadership for the role identified by the supplied role ID, or the client is already allocated leadership for that role, then the leader-election service returns a failure status so that the client first relinquishes the leadership role or waits until the assertion has failed before unregistering with the leader-election service.

0088<figref idref="DRAWINGS">FIG. 27</figref> is a control-flow diagram illustrating the client/leader election-service-interface routine “Assert Leadership.” In step <b>2702</b>, the client transmits a role ID and timeout to the leader-election service. In step <b>2704</b>, the leader-election service receives the transmitted role ID and timeout. The routine “Assert Leadership” allows a client to request leadership for a specified role, providing a timeout value indicating an amount of time for which the leader-election service should attempt to assert leadership of the specified role on behalf of the client before failing. If an entry for the client and specified role is not present in the table “Registrations” as detected in step <b>2706</b>, then the leader-election service returns a failure in step <b>2708</b>. Otherwise, in the while-loop of steps <b>2710</b>-<b>2716</b>, the leader-election service continues to assert allocation of a leadership role for the specified role to the requesting client, in step <b>2711</b>, until either the client is allocated the role, as determined in step <b>2712</b>, and a success status is returned in step <b>2713</b>, or until another client is allocated the role or the timeout period is exceeded, as determined in steps <b>2714</b> and <b>2716</b>, in which case a failure status is returned in step <b>2715</b>. The leader-election service asserts allocation of the leadership role to the client by attempting to add an entry to the table “Global Role Assignments.” The attempt fails when an entry for the specified role already exists in the table “Global Role Assignments.” The assertion is made through a distributed consensus service, such as Paxos or Disk Paxos, as discussed above in previous subsections.

0089<figref idref="DRAWINGS">FIG. 28</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Forward Request.” This routine allows a client to forward a request or condition related to a particular role to the current leader for that role. In step <b>2802</b>, the client forwards the request, a role ID, and a callback routine to the leader-election service. In step <b>2804</b>, the leader-election service receives the transmitted information, and, in addition, the client ID of the requesting client, as discussed above. In step <b>2806</b>, the leader-election service accesses the table “Global Role Assignments” to determine the process currently allocated leadership for the specified role. If no such leader can be found, as determined in step <b>2808</b>, then failure is returned in step <b>2810</b>. Otherwise, in step <b>2812</b>, the leader-election service transmits the request, the leader-election-service ID for the leader-election service and a node ID to the leader-election service to which the leader process interfaces, in the case of a leader process executing on an eternal node, or transmits the request to a local client via the local client's callback routine, in the case that leadership for the specified role has been allocated to a local client. If the leader-election service receives an acknowledgement for the transmitted request, as determined in step <b>2814</b>, then the leader-election service routine returns a success status, in step <b>2816</b>. Otherwise, the leader-election service returns failure in step <b>2818</b>. Once the request has been successfully forwarded to the process to which leadership for the role has been allocated, that process executes the request and, if necessary, communicates with the requesting client via the requesting client's call back furnished by the requesting client in step <b>2802</b>. The details of request processing and communications between the leader process and the requesting client is beyond the scope of the current application, and is dependent on the specific type of role within a specific type of distributed computer system.

0090<figref idref="DRAWINGS">FIG. 29</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Renounce.” This routine allows a client process to renounce a leadership role previously assumed by the client process. In step <b>2902</b>, the client transmits to the leader-election service a role ID and timeout value. In step <b>2904</b>, the leader-election service receives the transmitted role ID and timeout value. In steps <b>2906</b>-<b>2908</b>, the leader-election service determines whether the renounce request is valid, returning a no-role failure status, in step <b>2908</b>, when the client is unknown or has not been allocated leadership for the specified role. In step <b>2910</b>, the leader-election service asserts a NULL entry for the role in the table “Global Role Assignments” via the distributed consensus service. In other words, the leader-election service attempts to delete the entry for the client ID/role ID in order to deallocate the leadership role. If the leader-election service is successful in deleting the role from the table “Global Role Assignments,” as determined in step <b>2912</b>, then a success status is returned in step <b>2914</b>. Otherwise, the leader-election service continues to wait for indication of a successful assertion until the timeout is exceeded, as determined in step <b>2916</b>, before returning a failure in step <b>2918</b>. In certain embodiments, as represented by dashed line <b>2920</b>, the leader-election service may need to continuously reassert deletion of the entry from the table “Global Role Assignments” until deletion occurs. Of course, a successful deletion followed by entry of a new leadership allocation by another leader-selection service may also need to be detected, and when detected, elicit return of a success status, in the case that a new entry can be added by the distributed consensus service prior to detection of successful deletion.

0091<figref idref="DRAWINGS">FIG. 30</figref> is a control-flow diagram for the client/leader-election-service-interface routine “Forward Role.” The routine “forwardRole” allows a client to forward leadership of a role to another client. In various embodiments, the routine “Forward Role” may allow a client to forward a role to another client process within the same node, to a leader-election service running on an external node, or both. In step <b>3002</b>, the client forwards a role ID, timeout value, and client ID or leader-election-service ID and node ID to the leader service. In step <b>3004</b>, the leader service receives the transmitted information, as well as the ID of the client that transmitted the request. If the request is invalid, as determined in step <b>3006</b> in similar fashion to the more detailed request-validation diagrams described above, then the leader-election service returns failure in step <b>3008</b>. Otherwise, in step <b>3010</b>, the leader-election service calls a callback routine of the client to which the role is to be forwarded, if the specified client is local, or sends a request to the specified leader-election service, in those embodiments in which a role can be forwarded between nodes, to determine whether the client to which the role is being requested to be forwarded is willing to accept the role. When the client is not willing to accept leadership of the role, as determined in step <b>3012</b>, then a failure status is returned in step <b>3014</b>. Otherwise, in step <b>3016</b>, the leader-election service executes a renounce request on behalf of the requesting client and, when the renounce request does not succeed, as determined in step <b>3018</b>, returns a failure status to the requesting client in step <b>3020</b>. Otherwise, in step <b>3022</b>, the leader-election service returns success to the client and then carries out an assert-leadership request on behalf of the client to which the role is to be forwarded, in step <b>3024</b>. In alternative embodiments, once a client has indicated a willingness to accept a role from another client, as detected in step <b>3012</b>, then the leadership-accepting client issues an assert-leadership request to that client's local leader-election service.

0092<figref idref="DRAWINGS">FIG. 31</figref> is a control-flow diagram illustrating the client/leader-election-service-interface routine “Info Request.” This routine allows a client to obtain various types of information related to allocation of leadership roles within the distributed computer system. In step <b>3102</b>, the client transmits to the leader-election service an indication of the information which the client wishes to receive. In step <b>3104</b>, the leader-election service receives the indication of the information which the client wishes to receive. If the request is invalid, as determined in step <b>3106</b>, then a failure status is returned in step <b>3108</b>. Otherwise, in step <b>3110</b>, the leader-election service extracts the desired information from one or both of the local state information and the global, distributed state information and returns the information to the client in step <b>3112</b>. An information request may be invalid if the client is unknown to the leader-election service, if the information requested is privileged and inaccessible to the client, and for other reasons.

0093<figref idref="DRAWINGS">FIG. 32</figref> is a control-flow diagram illustrating callback of a client by the leader-election service. <figref idref="DRAWINGS">FIG. 32</figref> illustrates one particular type of callback, although the client processes and leader-election service employ callbacks for various types of communication between a leader-election service and the client processes, particularly in the case that the communication is asynchronous. In step <b>3202</b>, the leader-election service receives a request from a local or remote client that is to be directed to the leader of the role related to the request. If the request is invalid, as determined in step <b>3204</b>, then the leader-election service returns failure in step <b>3206</b>. Otherwise, in step <b>3210</b>, the leader-election service determines the client ID of the leader for the specified role. The leader-election service then transmits the request to the leader via the leader's previously submitted callback mechanism, in step <b>3212</b>. The leader receives the request forwarded from the leader-election service, in step <b>3214</b>, and processes the request in step <b>3216</b>. Additional error checking may occur by the leader-election service. For example, leader-election service may check for an invalid request related to a role for which no leader is currently allocated.

0094Thus, the distributed-leader-election service that represents an embodiment of the present invention allows for allocation and distribution of leadership roles among processes running on nodes of a distributed computer system in a distributed, but relatively centralized manner. The leadership-election service avoids obligating each process to cooperate with all other processes within a distributed computer system to maintain distributed state information, and may lead to a vast economy of communications overhead, synchronization overhead, and other computational and time overheads associated with maintaining and managing a distributed-leader-election service.

0095Although the present invention has been described in terms of particular embodiments, it is not intended that the invention be limited to these embodiments. Modifications within the spirit of the invention will be apparent to those skilled in the art. For example, strong-leader-election functionality can be implemented in any number of different programming languages, using any number of different modularizations, routines, control structures, variables, data structures, and by varying other such programming parameters. As discussed above, the leader-election service may be implemented as a separate, continuously running process within each node of the distributed computer system, or may alternatively be implemented as library routines that run in the context of client processes. A distributed-leader-election service may employ any of a variety of different types of distributed consensus services, including Paxos, Disk Paxos, Active Disk Paxos, or a global data service such as the distributed storage register described above. The distributed-leader-election service may manage allocation of leadership roles for any of an essentially limitless number of types of roles that may be designated for distribution within a distributed computer system.

0096The foregoing description, for purposes of explanation, used specific nomenclature to provide a thorough understanding of the invention. However, it will be apparent to one skilled in the art that the specific details are not required in order to practice the invention. The foregoing descriptions of specific embodiments of the present invention are presented for purpose of illustration and description. They are not intended to be exhaustive or to limit the invention to the precise forms disclosed. Obviously many modifications and variations are possible in view of the above teachings. The embodiments are shown and described in order to best explain the principles of the invention and its practical applications, to thereby enable others skilled in the art to best utilize the invention and various embodiments with various modifications as are suited to the particular use contemplated. It is intended that the scope of the invention be defined by the following claims and their equivalents: