Dynamic multicast routing facility for a distributed computing environment

A Dynamic Multicast Routing (DMR) facility is provided for a distributed computing environment having a plurality of networks of computing nodes. The DMR facility automatically creates virtual interfaces between selected computing nodes of the networks to ensure multicast message reachability to all functional computing nodes within the distributed computing environment. The DMR facility employs a group of group leader nodes (GL_group) among which virtual interfaces for multicast messaging are established. Upon failure of one of the group leader nodes, another computing node of the respective network having the failing group leader node is assigned group leader status for re-establishing virtual interfaces. Virtual interfaces are established between the group leader nodes such that redundancy in message routing is avoided.

BEST MODE FOR CARRYING OUT THE INVENTION In one embodiment, the techniques of the present invention are used in distributed computing environments in order to provide multi-computer applications that are highly-available. Applications that are highly-available are able to continue to execute after a failure. That is, the application is fault-tolerant and the integrity of customer data is preserved. It is important in highly-available systems to be able to coordinate, manage and monitor changes to subsystems (for example, process groups) running on processing nodes within the distributed computing environment. In accordance with the principles of the present invention, a facility is provided for dynamically or automatically accomplishing this in a distributed computing environment employing multicast routing of data messages between nodes. The Dynamic Multicast Routing facility (herein referred to as the “DMR facility”) of the present invention employs, in one example, the concepts referred to as “Group Services” in the above-incorporated U.S. patent applications and Letters Patent. As used herein, a “host” comprises a computer which is capable of supporting network protocols, and a “node” is a processing unit, such as a host, in a computer network. “Multicast” refers to an internet protocol (IP) multicast as the term is used in the above-incorporated Addison/Wesley publication entitled “TCP/IP Illustrated”. A “daemon” is persistent software which runs detached from a controlling terminal. “Distributed subsystem” is a group of daemons which run in different hosts. “Group Services” is software present on International Business Machines Corporation's Parallel System Support Programs (PSSP) Software Suite (i.e., operating system of the Scalable Parallel (SP)), and IBM's High Availability Cluster Multi-Processing/Enhanced Scalability (HACMP/ES) Software Suite. Group Services is a system-wide, fault-tolerant and highly-available service that provides a facility for coordinating, managing and monitoring changes to a subsystem running on one or more processors of a distributed computing environment. Group Services provides an integrated framework for designing and implementing fault-tolerant subsystems and for providing consistent recovery of multiple subsystems. Group Services offers a simple programming model based on a small number of core concepts. These concepts include a cluster-wide process group membership and synchronization service that maintains application specific information with each process group. Although as noted above, in one example, the mechanisms of the present invention are implemented employing the Group Services facility, the mechanisms of this invention could be used in or with various other facilities, and thus, Group Services is only one example. The use of the term “Group Services” in explaining one embodiment of the present invention is for convenience only. In one embodiment, the mechanisms of the present invention are incorporated and used in a distributed computing environment, such as the one depicted in FIG. 1 . Distributed computing environment 100 includes, for instance, a plurality of frames 102 coupled to one another via a plurality of LAN gates 104 . Frames 102 and LAN gates 104 are described in detail below. In the example shown, distributed computing environment 100 includes eight (8) frames, each of which includes a plurality of processing or computing nodes 106 . In one instance, each frame includes sixteen (16) processing nodes (a.k.a., processors). Each processing node is, for instance, a RISC/6000 computer running AIX, i.e., a UNIX based operating system. Each processing node within a frame is coupled to the other processing nodes of the frame via, for example, an internal LAN connection. Additionally, each frame is coupled to the other frames via LAN gates 104 . As examples, each LAN gate 104 includes either a RISC/6000 computer, any computer network connection to the LAN, or a network router. However, these are only examples. It will be apparent to those skilled in the relevant art that there are other types of LAN gates, and that other mechanisms can be used to couple the frames to one another. In addition to the above, the distributed computing environment of FIG. 1 is only one example. It is possible to have more or less than eight frames, or more or less than sixteen nodes per frame. Further, the processing nodes do not have to be RISC/6000 computers running AIX. Some or all of the processing nodes can include different types of computers and/or different operating systems. All of these variations are considered a part of the claimed invention. In one embodiment, a Group Services subsystem incorporating the mechanisms of the present invention is distributed across a plurality of processing nodes of distributed computing environment 100 . In particular, in one example, a Group Services daemon 200 ( FIG. 2 ) is located within one or more of processing nodes 106 . The Group Services daemons 200 are accessed by each process via an application programming interface 204 . The Group Services daemons are collectively referred to as “Group Services”. Group Services facilitates, for instance, communication and synchronization between multiple processes of a process group, and can be used in a variety of situations, including, for example, providing a distributed recovery synchronization mechanism. A process 202 ( FIG. 2 ) desirous of using the facilities of Group Services is coupled to a Group Services daemon 200 . In particular, the process is coupled to Group Services by linking at least a part of the code associated with Group Services (e.g., the library code) into its own code. In one embodiment, Group Services 200 includes an internal layer 302 ( FIG. 3 ) and an external layer 304 . Internal layer 302 provides a limited set of functions for external layer 304 . The limited set of functions of the internal layer can be used to build a richer and broader set of functions, which are implemented by the external layer and exported to the processes via the application programming interface. The internal layer of Group Services (also referred to as a “metagroup layer”) is concerned with the Group Services daemons, and not the processes (i.e., the client processes) coupled to the daemons. That is, the internal layer focuses its efforts on the processors, which include the daemons. In one example, there is only one Group Services daemon on a processing node; however, a subset or all of the processing nodes within the distributed computing environment can include Group Services daemons. The internal layer of Group Services implements functions on a per processor group basis. There may be a plurality of processor groups in the distributed computing environment. Each processor group includes one or more processors having a Group Services daemon executing thereon. The processors of a particular group are related in that they are executing related processes. (In one example, processes that are related provide a common function.) For example, referring to FIG. 4 , processor group X ( 400 ) includes processing node 1 and processing node 2 , since each of these nodes is executing a process X, but it does not include processing node 3 . Thus, processing nodes 1 and 2 are members of processor group X. A processing node can be a member of none or any number of processor groups, and processor groups can have one or more members in common. In order to become a member of a processor group, the processor needs to request to be a member of that group. A processor requests to become a member of a particular processor group (e.g., processor group X) when a process related to that group (e.g., process X) requests to join a corresponding process group (e.g., process group X) and the processor is not aware of that corresponding process group. Since the Group Services daemon on the processor handling the request to join a particular process group is not aware of the process group, it knows that it is not a member of the corresponding processor group. Thus, the processor asks to become a member, so that the process can become a member of the process group. Internal layer 302 ( FIG. 3 ) implements a number of functions on a per processor group basis. These functions include, for example, maintenance of group leaders. A group leader is selected for each processor group of the network. In one example, the group leader is the first processor requesting to join a particular group. As described herein, the group leader is responsible for controlling activities associated with its processor group(s). For example, if a processing node, node 2 ( FIG. 4 ), is the first node to request to join processor group X, then processing node 2 is the group leader and is responsible for managing the activities of processor group X. It is possible for processing node 2 to be the group leader of multiple processor groups. If the group leader is removed from the processor group for any reason, including, for instance, the processor requests to leave the group, the processor fails or the Group Services daemon on the processor fails, then group leader recovery must take place. In one example, in order to select a new group leader, a membership list for the processor group, which is ordered in sequence of processors joining that group, is scanned, by one or more processors of the group, for the next processor in the list. The membership list is preferably stored in memory in each of the processing nodes of the processor group. Once the group leader is selected, the new group leader informs, in one embodiment, a name server that it is the new group leader. A name server might be one of the processing nodes within the distributed computing environment designated to be the name server. The name server serves as a central location for storing certain information, including, for instance, a list of all processor groups of the network and a list of the group leaders for all the processor groups. This information is stored in the memory of the name server processing node. The name server can be a processing node within the processor group or a processing node independent of the processor group. In large clustered systems, multicast messages require routing in order to reach multiple networks. As noted initially, the problem of maintaining multicast message reachability is often complicated by the fact that certain older routers do not support multicast routing. This routing problem is conventionally solved by manually configuring the selected hosts and routers in the distributed system as “routing points”. Such host routing points are capable of running host discovery protocols that enable them to configure their routing tables in such a way that all nodes in the system become reachable via multicast. In certain cases, multicast messages have to be routed through IP routers which do not support multicast routing. In such cases, a virtual interface or “tunnel” has to be configured, such that two nodes in different networks can interface and act as routing points for multicast messages. Tunneling is described in greater detail in the above-incorporated publication by Gary Wright and Richard Stevens entitled “TCP/IP Illustrated”. Again, tunneling end-points are traditionally configured manually by a network administrator. In accordance with the principles of the present invention, a Dynamic Multicast Routing (DMR) facility is provided which utilizes the above-described Group Services software. As noted, the Group Services software provides facilities for other distributed processes to form “groups”. A group is a distributed facility which monitors the health of its members and is capable of executing protocols for them. The DMR facility of the present invention utilizes, in one example, Group Services to monitor the health of the routing nodes, and in executing its election protocols which ultimately determine which node of a plurality of nodes in the group should act as an end-point for a tunnel for multicast routing. A DMR facility in accordance with the present invention also employs the mrouted daemon (again, as specified in the above-incorporated publication by Gary Wright and Richard Stevens entitled “TCP/IP Illustrated”) to establish tunneling end-points. The DMR facility of this invention utilizes the mrouted daemon in such a way that it does not require any of a node's host discovery mechanisms to be deployed; and does not alter the established IP routing tables of the node. This behavior is desirable because the IP routes are configured separately. The DMR thus supports any configuration of IP routing, i.e., whether dynamic, static or custom made. FIG. 5 depicts a further example of a distributed computing environment, denoted 500 , having a plurality of nodes 510 distributed among multiple networks (Network A, Network B, Network C). The DMR facility described herein implements a routing topology, in which exactly one point in each network of a plurality of interconnected networks acts as a routing or tunneling agent. For reasons described below, a network of nodes is synonymous herein with a group of nodes. The input information to the DMR facility is a collection of nodes with an arbitrary network configuration, where every node is reachable via conventional IP datagrams. Output is a dynamic set of router nodes which are configured to route multicast datagrams, either via a real interface or a virtual interface (i.e., a tunnel). This dynamic set of nodes ensures that all functional nodes within the distributed computing environment are reachable via multicast. In the example of FIG. 5 , solid lines represent actual network connections. These physical connections define three physical networks within the distributed computing environment. Namely, Network A comprising node 1 , node 2 , node 3 and a router 520 , Network B including node 4 , node 5 , node 6 and router 520 , and Network C having nodes 1 & 4 . Router 520 in FIG. 5 is assumed to comprise a specialized hardware element which is used only for network routing. Router 520 does not comprise a computing node in the sense that it can only execute a pre-determined number of protocols. In contrast, nodes 1 - 6 comprise processing or computing nodes as described above and execute the DMR facility of the present invention. The circles around nodes 2 , 4 & 6 identify these nodes as multicast routing points or routing nodes selected by the DMR facility for multicast message forwarding as described further herein. One aspect of FIG. 5 is that any two computing nodes could actually be used as routing points to tunnel across the router. The DMR facility of this invention runs a special group protocol described below that ensures that only two nodes between two groups will be chosen. This DMR facility monitors the health of these chosen routing points, and does immediate, automatic reconfiguration in the case of failure. Because reconfiguration is automatic, the routing facility is referred to herein as “dynamic”. One detailed embodiment of a technique in accordance with the principles of the present invention to dynamically determine tunneling end-points for the forwarding of multicast datagrams can be summarized as follows: The DMR process runs in every node of the system, i.e., every node of the distributed computing environment could potentially be selected as a multicast routing node. At initialization time, the DMR process reads the IP address and subnet mask for each communication interface (i.e., adapter) which is configured in the machine (i.e., node) that the DMR process is running on. Every node has to have at least one communication interface in order for the node to be within one network of the multiple networks in the distributed computing environment. The DMR process makes a logical (Boolean) AND operation of the IP address and subnet mask of each communication interface, obtaining in this way a network ID. Specifically, networkID&equals;IP_address & subnet_mask. The DMR process then uses the network ID as a group identifier in accordance with this invention. Each DMR process will join as many groups as there are communication adapters in the node where it runs, again using the network IDs as the group identifiers. Once the node joins a group, the DMR processes of the group act as a distributed subsystem. This means that the DMR processes are now aware of the existence of each other, and they run synchronized protocols. When a DMR process joins a group, the process receives a membership list from Group Services. The Group Services subsystem guarantees that the first element in the list is the process that has first successfully joined the group. The DMR utilizes the ordering within the membership list to determine the group leader for each group. After joining a group, the DMR process checks to see if it is the group leader of any group; that is, the process checks if it is the first member on any of the group membership lists. The processes which are appointed group leaders will then join another group, which consists only of group leaders. This special group is referred to herein as the “group leaders group” or “GL_group”. The members of the GL_group utilize the same technique described above to elect a group leader; that is, they pick the first member identified on the GL_group membership list. The leader of the GL_group is referred to herein as the “system leader”. Once a DMR process is appointed a system leader, the tunneling end-points are created. The system leader's DMR will start an mrouted process and configure it for tunneling using a configuration file and a refresh signal. The system leader DMR will configure its local mrouted daemon to tunnel multicast datagrams from all of its configured communication interfaces to each of the group leaders of the various network groups, i.e., the groups which were first formed and which utilize the networkID as group name. The other members of the GL_group, which are leaders of some network group, will in turn also start an mrouted process configured to route multicast datagrams from the communication interface that they are the leader of to all communication interfaces of the system leader. The resulting network topology is that the system leader acts as a routing point for all the leaders of all the network groups. Applying the above procedure to the distributed computing environment 500 of FIG. 5 , results in the topology shown in FIG. 6 . This topology is arrived at by assuming that node 2 is the first listed node of the membership list of the group comprising the nodes of network A, node 6 is the first listed node in the membership list of the nodes comprising network B, and node 4 is the first listed node in the membership list of the nodes comprising network C. Further, the topology is obtained by assuming that node 2 is the first listed group leader in the membership list for the GL_group comprising nodes 2 , 4 & 6 . Again, mrouted daemons at nodes 2 , 4 and 6 are employed to establish the multicast tunnel connections or interfaces between these nodes. Node 2 operates to forward multicast messages to any node within Network A, node 4 forwards multicast messages to any node within Network B and node 6 forwards multicast messages to any node within Network C. Note that although not shown, the same node could operate as group leader for multiple groups of nodes. For example, node 6 could have been group leader for both network B and network C. FIG. 7 depicts a flowchart of the above-described initialization processing in accordance with the present invention. The DMR facility 700 is started on each node of the distributed computing environment, and for each communication interface 710 the DMR facility reads its corresponding IP address and subnet mask to determine a networkID 720 . The networkID, which is defined as IP_address & subnet_mask, is employed herein as a “group identifier”, or “groupID”. After determining a groupID, the node joins a Group Services group using the groupID 730 and determines whether a groupID has been determined for each of its communication interfaces 740 . If no, the process repeats until each interface has a groupID determined for it, and the node has joined the corresponding Group Services group identified by that groupID. When the DMR process joins a group, it receives a membership list from the Group Services. This membership list is then employed as described above to determine whether the node is a group leader of any group 750 . Again, in one example, a node is a group leader if it is the first member on any of the membership lists of a group to which it belongs. If the node is a leader of a group, then the node joins the group of group leaders, i.e., the GL_group 760 . If the node is other than a group leader, or after the node has joined the GL_group, initialization continues as noted in the recovery processing of FIG. 8 770 . In the example of FIG. 8 , recovery processing starts 800 with the DMR process inquiring whether it is the leader of a network group 810 . If the DMR process is not a group leader, then the node simply waits for a Group Services notification of a membership change 870 , such as the addition or deletion of a node to the group. Upon notification of a membership change, the recovery process repeats as shown. Assuming that the DMR process is running on a node that is a group leader, then the node joins the GL_group if not already a member 820 , and determines whether it is the GL_group leader 830 . If so, then the node builds a configuration file for the mrouted daemon for tunneling from all of this node's interfaces to all other members of the GL_group 840 . Once the configuration file is established, the mrouted daemon is started or signaled if it is already running 860 . If the DMR process is on a node which is not the GL_group leader, then the node builds the configuration file for mrouted to tunnel from the network that the process is a leader of to the GL_group leader 850 , and again starts the mrouted daemon or signals it if it is already running 860 . After completing tunneling, the node waits for the Group Services to notify it of a membership change 870 , after which processing repeats as indicated. In accordance with the present invention, the DMR processes recover automatically from the failure of any node in the distributed computing environment by employing the group formation protocols of FIGS. 7 & 8 . Any failure within the environment is immediately detected by the Group Services subsystem, which will then inform the DMR processes which belong to any of the groups that the failed node used to belong to. The surviving nodes will perform the same election mechanisms as described above. If the failed node was the group leader for a network group, a new leader is elected. Again, in one example, the new leader comprises the first listed node in the membership list of the effected group. If the failed node was the leader of the GL_group, a new leader is similarly chosen for that group. Whenever a group leader is elected, it re-establishes the tunnel end-points as described above. The operational loop of the DMR process depicted in FIG. 8 is based on membership within the several groups employed. After initialization, the process joins the appropriate groups, and configures the mrouted daemon as indicated. When another process joins the group, or leaves the group due to a failure, all processes get notified by the Group Services of a membership change, and all processes will make another pass at the recovery loop of FIG. 8 , updating the configuration as appropriate. The present invention can be included, for example, in an article of manufacture (e.g., one or more computer program products) having, for instance, computer usable media. This media has embodied therein, for instance, computer readable program code means for providing and facilitating the capabilities of the present invention. The articles of manufacture can be included as part of the computer system or sold separately. Additionally, at least one program storage device readable by machine, tangibly embodying at least one program of instructions executable by the machine, to perform the capabilities of the present invention, can be provided. The flow diagrams depicted herein are provided by way of example. There may be variations to these diagrams or the steps (or operations) described herein without departing from the spirit of the invention. For instance, in certain cases, the steps may be performed in differing order, or steps may be added, deleted or modified. All of these variations are considered to comprise part of the present invention as recited in the appended claims. While the invention has been described in detail herein in accordance with certain preferred embodiments thereof, many modifications and changes therein may be effected by those skilled in the art. Accordingly, it is intended by the appended claims to cover all such modifications and changes as fall within the true spirit and scope of the invention.