%\documentclass{cpeauth}

\documentclass{article}
\usepackage{epsfig}

\newtheorem{requirement}{Requirement}[section]
\newtheorem{conjecture}{Conjecture}[section]
%\CPE{1}{7}{00}{00}{2000}

%\runningheads{S. B. Pallickara and G.C. Fox}{The Grid Event Service}

\title{An Event Service to Support Grid Computational Environments}
\date{}
\author{Geoffrey Fox\thanks{Department of Computer Science, Indiana University. gcf@indiana.edu} \and  Shrideep Pallickara\thanks{Department of Electrical Engineering \&Computer Science, Syracuse University. shrideep@cat.syr.edu} }

\addtolength{\textwidth}{3.8cm}
\addtolength{\hoffset}{-2.0cm}
\addtolength{\voffset}{-0.55cm}
\addtolength{\textheight}{3.7cm}

%\longauthor{Shrideep Pallickara\affil{1} and Geoffrey Fox\affil{2}} 

%\longaddress{\affilnum{1}\ Department of Electrical Engineering \& Computer Science, \\Syracuse University \\
%\affilnum{2} Department of Computer Science, \\Indiana University}


%\corraddr{3-211 CST, 111 College Place, Syracuse University, Syracuse NY-13244, USA}
 
%\cgsn{Publishing Arts Research Council}{98--1846389}

%\received{10 July 2001}
%\revised{11 July 2001}
%\noaccepted{}
\begin{document}

%\begin{titlepage}
%\large{Grid Event Service: Research Issues and Directions}
%
%\end{titlepage}

\maketitle
\begin{abstract}
%The Grid Event Service (GES) is a distributed event service designed to run on a very large network of broker nodes. Clients interested in using this service can attach themselves to one of the broker nodes. Clients specify an interest in the type of events that they are interested in and the service routes events, which satisfy the constraints specified by the clients. Clients can have prolonged disconnects from the broker network and can also roam the network (in response to failure suspicions or for better response times) and attach themselves to any other node in the broker node network. Events published during the intervening period, of prolonged disconnects and roams, must still be delivered to clients that originally had an interest in these events. The delivery constraints must be satisfied even in the presence of broker failures. Broker nodes can fail and remain failed forever. Clients need not wait for the failed broker nodes to recover. Affected clients can then roam to a new location and thus not experience any denial of service. 
%GES provides a hierarchical dissemination scheme for the delivery of events to relevant clients. The system provides for an efficient calculation of routes to reach relevant destinations. This paper also outlines the GES publish/subscribe model, the natural extension of GES to support dynamic topologies and also the security framework that needs to be in place. 
We believe that it is interesting to study the system and software architecture of environments which integrate the evolving ideas of computational grids, distributed objects, web services, peer-to-peer networks and message oriented middleware. Such peer-to-peer (P2P) Grids should seamlessly integrate users to themselves and to resources which are also linked to each other. We can abstract such environments as a distributed system of "clients" which consist either of "users" or "resources" or proxies thereto. These clients must be linked together in a flexible fault tolerant efficient high performance fashion. In this paper, we study the messaging or event system - termed GES or the Grid Event Service -- that is appropriate to link the clients (both users and resources of course) together. For our purposes (registering, transporting and discovering information), events are just messages - typically with time stamps. The messaging system GES must scale over a wide variety of devices - from hand held computers at one end to high performance computers and sensors at the other extreme. We have analyzed the requirements of several Grid services that could be built with this model, including computing and education and incorporated constraints of collaboration with a shared event model. We suggest that generalizing the well-known publish-subscribe model is an attractive approach and here we study some of the issues to be addressed if this model is used in the GES. 
\end{abstract}

%\keywords{distributed messaging, publish subscribe, guaranteed delivery, grid systems and event distribution systems}
\section{Introduction}
The web in recent years has experienced an explosion in the number of devices users employ to access services. A single user may access a certain service using multiple devices. Most services allow clients to access the service through a broker. The client is then forced to interact with the service via this broker throughout the duration that it is using the service. If the broker fails, the client is denied servicing till such that the failed broker recovers. In the event that this service is running on a fixed set of brokers the client, since it knows about this set of brokers, could then connect to one of these brokers and continue using the service. Whether the client missed any servicing and whether the service would notify the client of this missed servicing depends on the implementation of the service. In all these implementations the identity of the broker that the client connects to is just as important as the service itself. Clients do not always maintain an online presence, and when they are online they may the access the service using a different device with different computing and content-handling capabilities. The communication channels employed during every such service interaction may have different bandwidth constraints and communication latencies. Besides this a client accesses services from different geographic locations. 

A truly distributed service, would allow a client to use services by connecting to a broker nearest to the client's geographical location. By having such local broker, a client does not have to re-connect all the way back to the broker that it was last attached to. If the client is not satisfied with the response times that it experiences or if the broker that it has connected to fails, the client could very well choose to connect to some other local broker. Concentration of clients from a specific location accessing a remote broker, leads to very poor bandwidth utilization and affects latencies associated with other services too. Also it should not be assumed that a failed broker node would recover within a finite amount of time. Stalling operations for certain sections of the network, and denying service to clients while waiting for failed processes to recover could result in prolonged, probably interminable waits. Such a model potentially forces every broker to be up and running throughout the duration that this service is being provided. Models that require brokers to recover within a finite amount of time generally imply that each broker has some state. Recovery for brokers that maintain state involves state reconstruction, usually involving a calculation of state from the neighboring brokers. This model runs into problems when there are multiple neighboring broker failures. Invariably brokers get overloaded, and act as black holes where messages are received but no processing is performed. By ensuring that the individual brokers are stateless (as far as the servicing is concerned), we can allow these brokers to fail and not recover. A failure model that does not require a failed node to recover within a finite amount of time, allows us to purge such slow processes and still provide the service while eliminating a bottleneck.

What is indispensable is the service that is being provided and not the brokers which are cooperating to provide the service. Brokers can be continuously added or fail and the broker network can undulate with these additions and failures of brokers. The service should still be available for clients to use. Brokers thus do not have an identity - any one broker should be just as good as the other. Clients however have an identity, and their service needs are very specific and vary from client to client. Any of these brokers should be able to service the needs of every one of these millions and millions of clients. It's the system as a whole, which should be able to reconstruct the service nuggets that a client missed during the time that it was inactive. Clients just specify the type of events that they are interested in, and the content that the event should at least contain. Clients do not need to maintain an active presence during the time these interesting events are taking place. Once it registers an interest it should be able to recover the missed event from any of the broker nodes in the system. Removing the restriction of clients reconnecting back to the same broker that it was last attached to and the departure from the time-bound recovery failure model, leads to a situation where brokers could be dynamically instantiated based on the concentration of clients at certain geographic locations. Clients could then be induced to roam to such dynamically created brokers for optimizing bandwidth utilization. The network can thus undulate with the addition and failure/purging of broker node processes.

The system we are considering needs to support communications for $10^9$ devices. The users using these devices would be interested in peer-to-peer (P2P) style of communication, business-to-business (B2B) interaction or a system comprising of agents where discoveries are initiated for services from any of these devices. Finally, some of these devices could also be used as part of a computation. The devices are thus part of a complex distributed system. Communication in the system is through events, which are encapsulated within messages. Events form the basis of our design and are the most fundamental units that entities need to communicate with each other. Events are \emph{anything} transmitted including updates, objects themselves (file uploads), database updates and audio/video  streams. 
These events encapsulate expressiveness at various levels of abstractions - content, dependencies and routing. Where, when and how these events reveal their expressive power is what constitutes information flow within our system. Clients provide services to other clients using events. These events are routed by the system based on the service advertisements that are contained in the messages published by the client. Events routed to a broker are queued and routing decisions are made based on the service advertisements contained in these events and also based on the state of the network fabric. 

We believe that it is interesting to study the system and software architecture of environments which integrate the evolving ideas of computational grids, distributed objects, web services, peer-to-peer networks and message oriented middleware. Such peer-to-peer (P2P) Grids should seamlessly integrate users to themselves and to resources which are also linked to each other. We can abstract such environments as a distributed system of "clients" which consist either of "users" or "resources" or proxies thereto. These clients must be linked together in a flexible fault tolerant efficient high performance fashion. In this paper, we study the messaging or event system - termed GES or the Grid Event Service -- that is appropriate to link the clients (both users and resources of course) together. For our purposes (registering, transporting and discovering information), events are just messages - typically with time stamps. The messaging system GES must scale over a wide variety of devices - from hand held computers at one end to high performance computers and sensors at the other extreme. We have analyzed the requirements of several Grid services that could be built with this model, including computing and education and incorporated constraints of collaboration with a shared event model. We suggest that generalizing the well-known publish-subscribe model is an attractive approach and here we study some of the issues to be addressed if this model is used in the GES. 


\subsection{Messaging Oriented Middleware}
Messaging systems based on queuing include products such as Microsoft's MSMQ \cite{msmq}and IBM's MQSeries \cite{ibm-mq}. The queuing model with their store-and-forward mechanisms come into play where the sender of the message expects someone to handle the message while imposing asynchronous communication and guaranteed delivery constraints. A widely used standard in messaging is the Message Passing Interface Standard (MPI) \cite{mpi}. 
MPI is designed for high performance on both massively parallel machines and workstation clusters. Messaging systems based on the classical remote procedure calls include CORBA \cite{omg:services}, Java RMI \cite{rmi-spec} and DCOM \cite{dcom}. In publish/subscribe systems the routing of messages from the publisher to the subscriber is within the purview of the message oriented middleware (MOM), which is responsible for routing the right content to the right consumers. Industrial strength products in the publish subscribe domain include solutions like \emph{TIB/Rendezvous} \cite{tibco} from TIBCO and \emph{SmartSockets} \cite{smart-sockets} from Talarian. Other related effort in the research community include  \emph{Gryphon} \cite{gryphon-multicast, matching-events}, \emph{Elvin} \cite{Elvin:pub/sub} and \emph{Sienna} \cite{Sienna:Scaling}.  The push by Java to include publish subscribe features into its messaging middleware include efforts like JMS \cite{jms-spec} and JINI \cite{jini-spec}. One of the goals of JMS is to offer a unified API across publish subscribe implementations. Various JMS implementations include solutions like \emph{SonicMQ} \cite{sonicMQ} from Progress, \emph{JMQ} \cite{jmq} from iPlanet, \emph{iBus} \cite{iBus} from Softwired  and \emph{FioranoMQ} \cite{firanoMQ} from Fiorano. 


\subsection{Service provided}
	We have built a "production" system and an advanced research prototype. The production system uses the commercial Java Message Service (SonicMQ) and has been used very successfully to build a synchronous collaboration environment applied to distance education. The publish/subscribe mechanism is powerful but this comes at some performance cost and so it is important that it satisfies the reasonably stringent constraints of synchronous collaboration. We are not advocating replacing all messaging with such a mechanism - this would be quite inappropriate for linking high performance devices such as nodes of a parallel machine linked today by messaging systems like MPI or PVM. Rather we have recommended using a hybrid approach in such cases. Transport of messages concerning the control of such HPCC resources would be the responsibility of the GES but the data transport would be handled by high performance subsystems like MPI. This approach was successfully used by the Gateway computing portal.

	Here we study an advanced publish/subscribe mechanism for GES which goes beyond JMS and other operational publish/subscribe systems in many ways. A basic JMS environment has a single server (although by linking multiple JMS invocations you can build a multi-server environment and you can also implement the function of a JMS server on a cluster). We propose that GES be implemented on a network of brokers where we avoid the use of the term servers for two reasons; the publish/subscribe broker service could be implemented on any computer - including a users desktop machine. Secondly we have included the many application servers needed in a P2P Grid as clients in our abstraction for they are the publishers and subscribers to many of the events to be serviced by the GES. Brokers can run on either on separate machines or on clients whether these are associated with users or resources. This network of brokers will need to be dynamic for we need to service the needs of dynamic clients. For example suppose one started a distance education session with six distributed classrooms each with around 20 students; then the natural network of brokers would have one for each classroom (created dynamically to service these clusters of clients) combined with static or dynamic brokers associated with the virtual university and perhaps the particular teacher in charge.

	Here we study the architecture and characteristics of the broker network. We are using a particular internal structure for the events (defined in XML but currently implemented as a Java object). Further we assume a sophisticated matching of publishers and subscribers defined as general topic objects (defined by an XML Schema that we have designed). However these are not the central issues to be discussed here. Our study should be useful whether events are defined and transported in Java/RMI or XML/SOAP or other mechanisms; it does not depend on the details of matching publishers and subscribers. Rather, we are interested in the capabilities needed in any implementation a GES in order to abstract the broker system in a scalable hierarchical fashion (section \ref{events-clients-broker-topology}); the delivery mechanism (section \ref{The-event-delivery-problem}); the guarantees of reliable delivery whether brokers crash or disappear or whether clients leave or (re)join the system (section \ref{reliable-delivery}). This section also discusses persistent archiving of the event streams. We have emphasized the importance of dynamic creation of brokers but this was not implemented in our initial prototype. However by looking at the performance of our system with different static broker topologies we can study the impact of dynamic creation and termination of broker services.
 
%In this paper we propose the Grid Event Service (GES) \cite{ges, ges-framework, ges-routing} where we have taken a system model that encompasses Internet/Grid messages. GES incorporates an inbuilt publish/subscribe mechanism for the efficient dissemination of content with the additional feature of location transparency for clients. GES is designed to include JMS as a special case. However, GES provides a far richer set of interactions and selectivity between clients than the JMS model. GES is not restricted to Java of course, this is our initial implementation. We envision a system with hundereds of thousands of broker nodes providing a distributed event service in a federated fashion. In GES a client can attach itself to any of the broker nodes comprising the system. This client specifies the type of events it is interested in through its profile. A client such as this could then fail, leave or \emph{roam} (in response to failure suspicions, system induced roams etc.) the system. When such a client reconnects back into the system, this client should receive all the events that it was supposed to receive between the time it left the system and the time it reconnects back into the system.


\subsection{Status}
There exists a prototype implementation of GES. This prototype includes implementation of the routing scheme, propagation of profiles, content matching, replication and guaranteed delivery of events. This implementation uses TCP as the transport protocol for communication within the system. The implementation is developed using Java. Support for XML is currently being added to the system. Future work would include work on support for dynamic topologies and security frameworks for authentication, authorization and dissemination of content. The results from our prototype implementation are presented in this paper. 



\section{Clients and the Broker Topology}
\label{events-clients-broker-topology}
In this section we outline the destinations that are associated with an event. We discuss the connection semantics for any client within the system, and also present our rationale for a distributed model in implementing the solution. We then present our scheme for the organization of the broker network, and the nomenclature that we would be referring to in the remainder of this paper. 

\subsection{Destination lists and the generation of unique identifiers}
\label{event-anatomy:destination-lists}
Clients in the system specify an interest in the type of events that they are interested in. Some examples of interests specified by clients could be sports events or events sent to a certain discussion group. It is the system which computes the clients that should receive a certain event. 
A particular event may thus be consumed by zero or more clients registered with the system. Events have explicit or implicit information pertaining to the clients which are interested in the event. In the former case we say that the destination list is \emph{internal} to the event, while in the latter case the destination list is \emph{external} to the event.

An example of an internal destination list is ``Mail'' where the recipients are clearly stated. Examples of external destination lists include sports score, stock quotes etc. where there is no way for the issuing client to be aware of the destination lists. External destination lists are a function of the system and the types of events that the clients, of the system, have registered their interest in.

\subsection{Client}
Clients can generate and consume events in
the system.  Events in the system are continuously generated and consumed within the system.
 Clients on the other hand have intermittent connection semantics. Clients can be present in the system for a certain duration and  be disconnected later on.
Clients reconnect at a later time and receive events, which they were supposed to receive as well as events that they are supposed to receive during their present incarnations. Clients can issue/create events while in disconnected mode, these events would be held in a local queue to be released to the system during a reconnect. 

Associated with every clients is its profile, which keeps track of information pertinent to the client. This includes the application type., events the client is interested in and the broker node the client was attached to in its previous incarnation.

\subsection{The Broker Node Topology}
\label{broker-topology:node}
One of the reasons why one would use a distributed model is high availability. Having a centralized model would imply a single broker hosting multiple clients. 
While, this is a simple model, the inherent simplicity is more than offset by the fact that it constitutes a single point of failure. 
Thus all the clients present in the system would be unable to use any of the services provided by the system till a recovery mechanism kicks in. 
A highly available distributed solution would have data replication at various broker nodes in the network. Solving issues of consistency while executing operations, in the presence of replication, leads to a model where other broker nodes can service a client despite certain broker node failures.
The smallest unit of the system is a \emph{broker node} and constitutes a unit at level-0 of the system. Broker nodes grouped together form a \emph{cluster}, the level-1 unit of the system.
 Clusters could be clusters in the traditional sense, groups of broker nodes connected together by high speed links.
 A single broker node could decide to be part of such traditional clusters, or along with other such broker nodes form a cluster connected together by geographical proximity but not necessarily high speed links.
 
%\begin{figure}[htp]
%\centering
%\epsfig{file=images/ClusterConnections.eps,width=264.00pt, height=203.00pt}
%\caption{\label{fig:super-cluster} A Super Cluster -  Cluster Connections}
%\end{figure}

Several such clusters grouped together as an entity comprises a level-2 unit of our network and is referred to as a \emph{super-cluster}.
 Clusters within a super-cluster have one or more links with at least one of the other clusters within that super-cluster.
When we refer to the links between two clusters, we are referring to the links connecting the nodes in those individual clusters. 
%Referring to figure \ref{fig:super-cluster} Cluster-A has links to Clusters B, C and D while Cluster-B has links to Clusters A and C. For two clusters with at least one link between them, any node in either of the clusters can communicate with any other node of the other cluster.
In general there would be multiple links connecting a single cluster to several other clusters. This approach provides us with a greater degree of fault-tolerance, by providing us with multiple \emph{routes} to reach nodes within other clusters.
%\begin{figure}[htp]
%\centering
%\epsfig{file=images/SuperClusterConnections.eps,width=331.00pt, height=229.00pt}
%\caption{\label{fig:super-super-cluster} A Super-Super-Cluster - Super Cluster Connections}
%\end{figure}
This topology could be extended in a similar fashion to comprise of \emph{super-super-clusters} (level-3 units), \emph{super-super-super-clusters} (level-4 units) and so on. A client thus connects to a broker node, which is part of a cluster, which in turn is part of a super-cluster and so on and so forth.
 We limit the number of super-clusters within a super-super-cluster, the number of clusters within a super cluster and the number of nodes within a cluster. This limit, the \emph{block-limit}, is set at 64.
In an $N$-level system  this scheme allows for $2^6_{N} \times 2^6_{N-1} \times \cdots 2^6_0 $ i.e $2^{6*(N+1)}$ broker nodes to be present in the system.
% comprising of one super-super-super-cluster which encompasses all the nodes within the system.
%What we essentially have here is a set of strongly connected broker nodes  comprising a cluster and a set of links connecting a cluster to other clusters. We are interested in the delays that would be involved in connecting from one node in the network to another node in the network. This is proportional to the broker node hops that need to be taken en route to the final destination. 

We now delve into the \emph{small world graphs} introduced in \cite{small-world} and employed for the analysis of real world peer-to-peer systems in \cite[pages 207 -- 241]{p2p-book}. In a graph comprising several nodes, \emph{pathlength} signifies the average number of hops that need to be taken to reach from one node to the other. \emph{Clustering coefficient} is the ratio of the number of connections that exist between neighbors of node and the number of connections that are actually possible between these nodes. For a regular graph consisting of  $n$ nodes, each of which is connected to its nearest $k$ neighbors -- for cases where $n \gg  k \gg 1$, the pathlength is approximately $n/2k$. As the number of vertices increases to a large value the clustering coefficient in this case approaches a constant value of 0.75. 


At the other end of the spectrum of graphs is the \emph{random graph}, which is the opposite of a regular graph. In the random graph  case the pathlength is approximately $\log n / \log k$, with a clustering coefficient of  $k/n$. The authors in \cite{small-world} explore graphs where the clustering coefficient is high, and with \emph{long connections} (inter-cluster links in our case). They go on to describe how these graphs have pathlengths approaching that of the random graph, though the clustering coefficient looks essentially like a regular graph. The authors refer to such graphs as \emph{small world graphs}. This result is consistent with our conjecture that for our broker node network, the pathlengths will be logarithmic too. Thus in the topology that we have the cluster controllers provide control to local classrooms etc, while the links provide us with \emph{logarithmic} pathlengths and the multiple links, connecting clusters and the nodes within the clusters, provide us with robustness.  

\subsubsection{GES Contexts}
\label{broker-topology:contexts}
Every unit within the system, has a unique Grid Event Service (GES) context associated with it.
In an $N$-level system, a broker exists within the GES context $C^1_i$ of a cluster, which in turn exists within the GES context $C^2_j$ of a super-cluster and so on.
 In general a GES context $C^\ell_i$ at level $\ell$ exists within the GES context $C^{\ell+1}_j$ of a level $(\ell+1)$ unit. 
%In an $N$-level system the following hold ---
%\begin{eqnarray}	
%C^0_i & = & (C^1_j, i) \\
%C^1_j & = & (C^2_k, j) \\
%\vdots \nonumber \\
%C^{N-2}_p & = & (C^{N-1}, p)\\
%C^{N-1}_q & = & q
%\end{eqnarray}
In an $N$-level system, a unit at level $\ell$ can be uniquely identified by $(N-\ell)$ GES context identifiers of each of the higher levels.
 Of course, the units at any level $\ell$ within a GES context $C^{\ell+1}_i$ should be able to reach any other unit within that same level.
If this condition is not satisfied we have a \emph{network partition}.

\subsubsection{Gatekeepers}
\label{broker-topology:gatekeepers}
Within the GES context $C^2_i$ of a super-cluster, clusters have broker nodes at least one of which is  connected to at least one of the nodes existing within some other cluster. 
%  In some cases there would be multiple links from a cluster to some other cluster within the same super-cluster  $C^2_i$. This architecture provides a greater degree of fault tolerance by providing multiple routes to reach the same cluster. 
Some of the nodes in the cluster thus maintain connections to the nodes in other clusters. Similarly, some nodes in a cluster could be connected to nodes in some other super-cluster. We refer to such nodes as \emph{gatekeepers}. 
%Nodes, which maintain connections to other nodes in the system, have different GES contexts. 
Depending on the highest level at which there is a difference in the GES contexts of these nodes, the nodes that maintain this active connection are referred to as gatekeepers at the corresponding level. 
Nodes, which are part of a given cluster, have GES contexts that differ at level-0. 
Every node in a cluster is connected to at least one other node within that cluster. 
Thus, every node in a cluster is a gatekeeper at level-0. 

Let us consider a connection, which exists between nodes in a different cluster, but within the same super-cluster. 
In this case the nodes that maintain this connection have different GES cluster contexts i.e. their contexts at level-1 are different. 
These nodes are thus referred to as gatekeepers at level-1. 
Similarly, we would have connections existing between different super-clusters within a super-super-cluster GES context $C^3_i$. 
 In an $N$-level system gatekeepers would exist at every level within a higher GES context. 
The link connecting two gatekeepers is referred to as the \emph{gateway}, which the gatekeepers provide, to the unit that the other gatekeeper is a part of. 
% A gatekeeper at level $\ell$ within a higher GES context $C^{\ell+1}_j$, denoted $g^\ell_i(C^{\ell+1}_j)$, comprises of --
%\begin{itemize}
%\item The higher level GES Context $C^{\ell+1}_j$
%\item The gatekeeper identifier $i$
%\item The list of gatekeepers at level $\ell$ that it is connected to, within the GES context  $C^{\ell+1}_j$.
%\end{itemize}
\begin{figure}[ht]
\centering
\epsfig{file=images/GatekeepersandEventRoutings.eps,width=429.00pt, height=398.00pt}
\caption{\label{fig:gatekeepers-and-links} Gatekeepers and the organization of the system}
\end{figure} 
%It should be noted that a gatekeeper at level $\ell$ can be a gatekeeper at any other level. In fact, every node within the system is a gatekeeper at level-0. 
Figure \ref{fig:gatekeepers-and-links} shows a system comprising of 78 nodes organized into a system of 4 super-super-clusters, 11 super-clusters and 26 clusters. 
%When a node establishes a link to another node in some other cluster, it provides a gateway for the dissemination of events. If the node it connects to is in a different cluster within the same super-cluster GES context $C^2_i$ both the nodes are designated as cluster gatekeepers.
In general, if a node connects to another node, and the nodes are such that they share the same GES context $C^{\ell+1}_i$ but have differing GES contexts $C^\ell_j$, $C^\ell_k$, the nodes are designated as gatekeepers at $level-\ell$ i.e. $g^\ell(C^{\ell+1})$. Thus, in figure \ref{fig:gatekeepers-and-links} we have 12 super-super-cluster gatekeepers, 18 super-cluster gatekeepers (6 each in \textbf{SSC-A} and \textbf{SSC-C}, 4 in \textbf{SSC-B} and 2 in \textbf{SSC-D}) and 4 cluster-gatekeepers in super-cluster \textbf{SC-1}.


%\subsubsection{The addressing scheme}
%\label{broker-topology:addressing-scheme}
%The addressing scheme provides us with a way to uniquely identify each broker node within the system. 
%This scheme plays a crucial role in the delivery and dissemination of events to nodes in the system(discussed in Section \ref{sec:routing-events-to-reconnected-clients}). As discussed earlier, units at each level are defined within the GES context of a unit at the next higher level.
%In an $N$-level system the GES context $C^\ell_j$ is
%$C^\ell_i = \overbrace{C^N_j (C^{N-1}_k (\cdots(C^{\ell+1}_m(C^\ell_i)) \cdots) )}^{N-l}$. Thus in a 4-level system, to identify a broker node, the addressing scheme specifies the super-super-cluster $C^3_i$, super-cluster $C^2_j$ and cluster $C^1_k$ that the node is a part of, along with the node-identifier within $C^1_k$. Thus for broker node \textbf{a}, within cluster \textbf{B}, within super-cluster \textbf{C} and super-super-cluster \textbf{D} the logical address within the system is \textbf{D.C.B.a}. This addressing scheme is very similar to the IP addressing scheme. 







\section{The problem of event delivery}
\label{The-event-delivery-problem}
The event delivery problem is one of routing events to clients based on the type of events that clients are interested in. Events need to be relayed through the broker network prior to being delivered to clients. 
The dissemination process should efficiently deliver events to the destinations, which could be internal or external to the event. 
 In the latter case the system needs to compute the destination lists pertaining to the event. 
 The system merely acts as a conduit to efficiently route the events from the issuing client to the interested clients.  A simple approach would be to route all events to all clients, and have the clients discard those events that they are not interested in. 
This approach would however place a strain on network resources. 
Under conditions of high load and increasing selectivity by the clients, the number of events that a client discards would far exceed the number of events it is actually interested in. 
This scheme also affects the latency associated with the reception of real time events at the client. The increase in latency is due to the cumulation of queuing delays associated with the \emph{uninteresting/flooded} events. 
The system thus needs to be very selective of the kinds of events that it routes to a client.

\subsubsection*{Prior Art}
Different systems address the problem of event delivery to relevant clients in different ways. In \cite{Elvin:matching} each subscription is converted into a deterministic finite state automaton. This conversion and the matching solutions nevertheless can lead to an explosion in the number of states.  In \cite{Elvin:pub/sub} network traffic reduction is accomplished through the use of \emph{quench} expressions. Quenching prevents clients from sending notifications for which there are no consumers. Approaches to content based routing in \emph{Elvin} are discussed in \cite{Elvin:CBR}. 
In \cite{Sienna:Scaling, Sienna:CBR} optimization strategies include assembling patterns of notifications as close as possible to the publishers, while multicasting notifications as close as possible to the subscribers. 
In \cite{gryphon-multicast} each broker  maintains a list of all subscriptions within the system in a parallel search tree (PST). The PST is annotated with a trit vector encoding link routing information. These annotations are then used at matching time by a broker to determine which of its neighbors should receive that event.
\cite{gryphon:group-multicast} describes approaches for exploiting group based multicast for event delivery. These approaches exploit universally available multicast techniques.  

 The approach adopted by the OMG \cite{omg:services} is one of establishing channels and registering suppliers and consumers to those event channels. The channel approach in the event service \cite{omg:event-service} approach could entail clients (consumers) to be aware of a large number of event channels. 
The two serious limitations of event channels are the lack of event filtering capability and the inability to configure support for different qualities of service. 
These are sought to be addressed in the Notification Service \cite{omg:notification-service} design. 
However the Notification service attempts to preserve all the semantics specified in the OMG event service, allowing for interoperability between Event service clients and Notification service clients. 
Thus even in this case a client needs to subscribe to more than one event channel. In \emph{TAO} \cite{tao}, a real-time event service that extends the CORBA event service is available. This provides for rate-based event processing, and efficient filtering and correlation. However even in this case the drawback is the number of channels that a client needs to keep track of. 

In some commercial JMS implementations, events that conform to a certain topic are routed to the interested clients. Refinement in subtopics is made at the receiving client. For a topic with several subtopics, a client interested in a specific subtopic could continuously discard uninteresting events addressed to a different subtopic. This approach could thus expend network cycles for routing events to clients where it would ultimately be discarded. Under conditions where the number of subtopics is far greater than the number of topics, the situation of client \emph{discards} could approach the flooding case. 

In the case of servers that route static content to clients such as Web pages, software downloads etc. some of these servers have their content mirrored on servers at different geographic locations. Clients then access one of these mirrored sites and retrieve information. This can lead to problems pertaining to bandwidth utilization and servicing of requests, if large concentrations of clients access the wrong mirrored-site. In an approach sometimes referred to as \emph{active mirroring}, websites powered by \emph{EdgeSuite} \cite{akamai} from Akamai, redirect their users to specialized Akamized URLs. \emph{EdgeSuite} then accurately identifies the geographic location from which the clients have accessed the website. This identification is done based on the IP addresses associated with the clients. Each client is then directed to the server farm that is closest to the client's network point of origin. As the network load and server loads change clients could be redirected to other servers. 

\subsubsection{GES Solution Highlights}
Our solution to the problem of event delivery handles the dissemination problem in a near optimal fashion. An event is routed only to those units that have at least one client that is interested in the event. Further, the links employed during the routing ensures the fastest dissemination since each broker makes routing decisions, which ensure that the path from that broker to the intended recipients is the fastest and the shortest path. The routing decisions are made based on the current state of the network. A broker or islands of brokers could fail and the routes computed would avoid these failed sections of the broker network while routing to recipients. Solutions to the delivery problem, involve a matching step being performed at every broker. In our solution for a broker network, organized as an N-level system, the matching step is not performed at every broker as the event is being relayed through the broker network to its intended recipients. In fact this matching step is performed at most (N+1) times prior to delivery at a given recipient. Our solution to the event delivery problem handles dense and sparse interests in events equally well. The solution for delivery of events to clients experiencing service interruptions due to single/multiple broker failures is discussed in the next section. 

\subsection{The gateway propagation protocol - GPP}
\label{gateway-propogation-protocol}
The gateway propagation protocol (GPP) accounts for the process of adding gateways and  is responsible for the dissemination of connection information within relevant parts of the sub system to facilitate creation of abbreviated system interconnection graphs.   
However, GPP should also account for failure suspicions/confirmations of nodes and links, and provide information for alternative routing schemes. 
The organization of gateways reflects the connectivities, which exist between various units within the system. 
Using this information, a node should be able to communicate with any other node within the system. 
%Any given node within the system is connected to one or more other nodes within the system. We refer to these direct links from a given node to any other node as \textit{hops}. 
%The routing information associated with an event specifies the units, which should receive the event. At each $g^{\ell+1}(C^{\ell+1}_i)$ finer grained disseminations targeted for units $u^\ell$ within $C^{\ell+1}_i$ are computed. 
%When presented with such a list of destinations, based on the gateway information the best hops to take to reach the destinations needs to be computed. 
%A node is required to route the event in such a way that it can service both the coarser grained disseminations and the finer grained ones. 
%Thus, a node should be able to compute the hops that need to be taken to reach units at different levels. 
%A node is a level-0 unit, however it computes the hops to take to reach level-$\ell$ units within its GES context $C^{\ell+1}$ (where $\ell=0,1, \cdots, N$ --  $N$ being the system level). 
\begin{figure}[ht]
\centering
\epsfig{file=images/ConnectivityGraph-Connections.eps,width=429.00pt, height=363.00pt}
\caption{\label{fig:connections} Connectivities between units}
\end{figure}	
%What is required is an abstract notion of the connectivities that exist between various units (sub-units and super-units alike) within the system. 
This constitutes the \emph{connectivity graph} of the system. 
At each node the connectivity graph is different while providing a consistent overall view of the system. The view that is provided by the connectivity graph at a node should be of those connectivities that are relevant to the node in question. 
Figure \ref{fig:connections} depicts the connections that exist between the various units of the system that we would be using as an example in further discussions. 
%\subsubsection{Constructing the connectivity graph}
%The organization of gateways should be one which provides an abstract notion of the connectivity between units $u^\ell$ within the GES context $C^{\ell+1}$ of the node. This interconnection can span multiple levels, where, if the gateway level is $\ell$, a unit $u^x_i$ ($x < \ell$) within the GES context $C^{x+1}$ is connected to $u^\ell_j$ within $C^{\ell+1}$. Units $u^x_i$ and $u^{\ell}_j$ share the same $C^{\ell+1}$ GES context.  
%For any given node within the system, the connectivity graph captures the connections that exist between units $u^\ell$'s within the GES context $C^{\ell+1}_i$ that it is a part of. Thus every node is aware of all the connections that exist between the nodes within a cluster, and also of the connections that exist between clusters within a super cluster and so on. 
The connectivity graph is constructed based on the information routed by the system in response to the addition or removal of gateways within the system. 
This information is contained within the \emph{connection}. 
Not all gateway additions or removals/failures affect the connectivity graph at a given node. This is dictated by the restrictions imposed on the dissemination of connection information to specific sub-systems within the system. 
%The connectivity graph should also provide us with information regarding the best hop to take to reach any unit within the system. 
%The link cost matrix maintains the cost associated with traversal over any edge of the connectivity graph. 
%The connectivity graph depicts the connections that exist between units at different levels. 
%Depending on the node that serves as a level-$\ell$ gatekeeper, the cluster that the node is a part of is depicted as a level-1 unit having a level-$\ell$ connection to a level-$\ell$ unit, by all the other clusters within the super cluster that the gatekeeper node is a part of.

\subsubsection{The connection}
\label{sec:connectivity-graph-connection}
A connection depicts the interconnection between units of the system, and defines an edge in the connectivity graph. 
Interconnections between the units snapshot the kind of gatekeepers that exist within that unit. 
A connection exists between two gatekeepers. 
%A level-$\ell$ \emph{node} denoted $n^\ell_i$ in the connectivity graph, is the level-$\ell$ GES context of the gatekeeper in question and is the tuple $<u^\ell_i, \ell>$. 
%A level$-\ell$ connection is the tuple $<n^{x}_i, n^{y}_j, \ell> $ where $x \mid y = \ell$ and $x,y \leq \ell$. Units $u^{x}_i$ and $u^{y}_j$ share the same level-$(\ell +1)$ GES context $C^{\ell+1}_k$.  For any given node $n^\ell_i$ in the connectivity graph we are interested only in the level $\ell, \ell+1, \cdots, N$ connections that exist within the unit and not the $\ell-1, \ell-2, \cdots, 0$ connections that exist within that unit. 
If  a  level-$\ell$ connection is established, the connection information is disseminated only within the higher level GES context $C^{\ell+1}_i$ of the sub-system that the gatekeepers are a part of. Thus, connections established between broker nodes in a cluster are \emph{not} disseminated outside that cluster.
%This is ensured by never sending a level-$\ell$ gateway addition information across any gateway $g^{\ell+1}$. 
%Thus, in figure \ref{fig:connections} for a super-cluster gateway established within \textbf{SSC-A}, the connection information is disseminated only within the super-clusters \textbf{SC-1}, \textbf{SC-2} and \textbf{SC-3}, and subsequently the nodes in super-super-cluster \textbf{SSC-A}. 


%When a connection information is sent over a gateway $g^{\ell}$ only the contextual information from level-$\ell$ onwards are sent over the gateway. 
%For a gateway established between units $u^\ell_i$ and $u^\ell_j$ within a context $C^\ell_i$ the information disseminated within $u^\ell_i$ is the level-l context of the gatekeeper in  $u^\ell_i$. 

%When a level-$\ell$ connection is established between two units, the gatekeepers at each end create the connection information in the following manner --- 
%\begin{itemize}
%\item[(a)]For the gatekeeper at the far end of the connection, the node information in the connection is constructed using its level-$\ell$ GES context.
%\item[(b)]The other node of the connection is constructed as level-0 node using its level-0 GES context.  
%\item[(c)]The last element of the connection tuple, is the connection level $\ell_c$.
%\end{itemize}
When the connection information is being disseminated throughout the GES context $C^{\ell+1}_i$, it arrives at gatekeepers at various levels. 
Depending on the kind of link this information is being sent over, the information contained in the \emph{connection} is modified. 
Details regarding the information encapsulated in a connection, the update of this information during disseminations and the enforcement of dissemination constraints can be found in \cite{ges, ges-routing, ges-framework}.
%Every gatekeeper $g^p \ni p\leq \ell_c$,  at which the connection information is received, checks to see if any of the node information depicts a node $n^x$ where $x < \ell_c$. If this is the case the next check is to see if $p > x$. 
%If $p >x$ the node information is updated to reflect the node as level-$p$ node by including the level-$p$ GES contextual information of $g^p$. 
%If $p \not > x$ the connection information is disseminated \emph{as is}. 
Thus, in figure \ref{fig:connections} the connection between \textbf{SC-2} and \textbf{SC-1} in \textbf{SSC-A}, is disseminated as one between node \textbf{5} and \textbf{SC-2}. 
When this information is received at \textbf{4}, it is sent over as a connection between the cluster \textbf{c} and \textbf{SC-2}. 
When the connection between cluster \textbf{c} and \textbf{SC-2} is sent over the cluster gateway to cluster \textbf{b}, the information is not updated. As was previously mentioned, the super cluster connection \textbf{(SC-1,SC-2)} information is disseminated only within the super-super-cluster \textbf{SSC-A} and is not sent over the super-super-cluster gateway available within the cluster \textbf{a} in \textbf{SC-1} and cluster \textbf{g} in \textbf{SC-3}. 

%\subsubsection{Link count}
%For every connection that is created there is a unique identifier associated with that connection. All connections relevant for a node are maintained in a connection table. This scheme allows us to detect if the connection table already contains a certain connection. There could be multiple connections between two specific units, this feature provides for greater fault tolerance. However, what is maintained in the connectivity  graph is simply the connection, which exists between the two units. 
Every edge created due to the dissemination of connection information also has a link count associated with it, which is incremented by one every time a new connection is established between two units that were already connected. This scheme also plays an important role in determining if a connection loss would lead to partitions. 
%\subsubsection{The link cost matrix}
%\label{sec:connectivity-graph-cost-matrix}
Further, associated with every edge is the cost of traversal. In general the cost associated with traversing a level-$\ell$ link from a unit $u^x$ increases with increasing values of both $x$ and $\ell$. 
%The link cost matrix specifies the cost associated with traversing a link. The cost associated with traversing a level-$\ell$ link from a unit $u^x$ increases with increasing values of both $x$ and $\ell$. Thus the cost of communication between nodes within a cluster is the cheapest, and progressively increases as the level of the unit that it is connected to increases. The cost associated with communication between units at different levels increases as the levels of the units increases. One of the reasons why we have this cost scheme is that the dissemination scheme employed by the system is selective about the links employed for finer grained dissemination. In general a higher level gateway is more overloaded than a lower level gateway. Table \ref{table:link-cost-matrix} depicts the cost associated with communication between units at different levels. 
%\begin{table}[htp]
%\centering
%\begin{tabular}{|c|c|c|c|c|c|c|} \hline
%$level$ & \textbf{0} & \textbf{1}  & \textbf{2}  &  \textbf{3}   & $\mathbf{\ell_i}$ & $\mathbf{\ell_j}$ \\ \hline
%\textbf{0}       & 0 & 1  & 2  &  3   & $\ell_i$ & $\ell_j$ \\ \hline
%\textbf{1}       & 1 & 2  & 3  &  4   & $\ell_i+1$ & $\ell_j+1$ \\ \hline
%\textbf{2}       & 2 & 3  & 4  &  5   & $\ell_i+2$ & $\ell_j+2$ \\ \hline
%\textbf{3}       & 3 & 4  & 5  &  6   & $\ell_i+3$ & $\ell_j+3$ \\ \hline
% $\mathbf{\ell_i}$ &$\ell_i$  & $\ell_i+1$  & $\ell_i+2$  &  $\ell_i+3$   & $2\times \ell_i$ & $\ell_i+\ell_j$ \\ \hline
% $\mathbf{\ell_j}$ & $\ell_j$  & $\ell_j+1$  & $\ell_j+2$  &  $\ell_j+3$   & $\ell_j+ \ell_i$ & $2\times \ell_j$ \\ \hline
%\end{tabular}
%\caption{\label{table:link-cost-matrix}The Link Cost Matrix}
%\end{table}
This cost scheme is encapsulated in the link cost matrix, which can be dynamically updated to reflect changes in link behavior. Thus, if a certain link is overloaded, we could increase the cost associated with traversal along that link. This check for updating the link cost could be done every few seconds.

\subsubsection{Organizing the nodes}
%The connectivity graph is different at every node, while providing a consistent view of the connections that exist within the system. 
%This section describes the organization of the information contained in connections (section \ref{sec:connectivity-graph-connection}) and super-imposing costs as specified by the link cost matrix (section \ref{sec:connectivity-graph-cost-matrix}) resulting in the creation of a weighted graph. The connectivity graph constructed at the node imposes directional constraints on \emph{certain} edges in the graph.  

The first node in the connectivity graph is the \emph{vertex node}, which is the level-0 broker node hosting the connectivity graph. 
The nodes within the connectivity graph are organized as nodes at various levels. A graph node $k$ at level $\ell$ in the connectivity graph is denoted as $n^\ell_k$. 
Associated with every level-$\ell$ node in the graph are two sets of links, the set $L_{UL}$, which comprises of connections to nodes $n^a_i \ni a \leq \ell$ and $L_D$ with connections to nodes $n^b_i \ni b > \ell$. 
When a connection is received at a node, the node checks to see if either of the graph nodes (representing the corresponding units at different levels) is present in the connectivity graph. 
If any of the units within the connection is not present in the connectivity graph, the corresponding graph node is added to the connectivity graph. 
%For every connection,  $<n^{x}_i, n^{y}_j, \ell> \; where \; x \mid y = \ell\; and \; x,y \leq \ell$, that is received; if $y \leq x$ then --  
%\begin{itemize}
%\item Graph node $n^{y}_j$ is added to the set $L_{UL}$ associated with node  $n^{x}_i$ 
%\item Graph node $n^{x}_i$ is added to the set $L_D$ associated with node  $n^{y}_j$.
%\end{itemize} 
%The process is reversed if  $x \leq y$. For the edge created between nodes $n^{x}_i$ and $n^{y}_j$, the weight is given by the element $(x,y)$ in the link cost matrix. 


\begin{figure}[htp]
\centering
\epsfig{file=images/ConnectivityGraph.eps,width=285.00pt, height=219.00pt}
\caption{\label{fig:connectivity-graph} The connectivity graph at node 6.}
\end{figure}	
Figure \ref{fig:connectivity-graph} depicts the connectivity graph that is constructed at the node \textbf{SSC-A.SC-1.c.6} in figure \ref{fig:connections}. The set  $L_{UL}$ at the node \textbf{SC-3} in the figure comprises of node \textbf{SC-2} at level-2 and node \textbf{b} at level-1. The set  $L_D$ at \textbf{SC-3} comprises of the node \textbf{SSC-B} at level-3.
The cost associated with traversal over a level-3 gateway between a level-2 unit \textbf{b} and a level-3 unit \textbf{SC-3} as computed from the linkcost matrix is 3, and is the weight of the connection edge. There are two connections between the super-super-cluster units \textbf{SSC-B} and \textbf{SSC-D}, this is reflected in the link count associated with the edge connecting the corresponding graph nodes. 
%The directional issues associated with certain edges are imposed by the algorithm for computing the shortest path to reach a node. 
%\subsubsection{Computing the shortest path}

To reach the vertex from any given node, a set of links need to be traversed. This set of links constitutes a \emph{path} to the vertex node. 
In the connectivity graph, the best hop to take to reach a certain unit is computed based on the shortest path that exists between the unit and the vertex. 
This process of calculating the shortest path, from the node to the vertex, starts at the node in question. 
The directional arrows indicate the links, which comprise a valid path from the node in question to the vertex node. 
Edges with no imposed directional constraints are bi-directional. 
For any given node, the only links that come into the picture for computing the shortest path are those that are in the set $L_{UL}$ associated with any of the nodes in a valid path. 

%The algorithm proceeds by recursively computing the shortest paths to reach the vertex node, along every valid link ($L_{UL}$) originating at every node that falls within the valid path. 
%Each fork of the recursion keeps track of the nodes that were visited and the total cost associated with the path traversed. 
%This has two useful features -
%\begin{itemize}
%\item[(a)] It allows us to determine if a recursive fork needs to be sent along a certain edge. If we do not keep track of the nodes that were visited, we could end up in an infinite recursion  where we revisit the same node over and over again.  
%\item[(b)] It helps us decide on the best edge that could have been taken at the end of every recursive fork.
%\end{itemize}
%For example in the connectivity graph of figure \ref{fig:connectivity-graph} we are interested in computing the shortest path to \textbf{SSC-B} from the vertex. This process would start at the node \textbf{SSC-B}. The set of valid links from \textbf{SSC-B} include edges to reach nodes \textbf{a}, \textbf{SC-3} and \textbf{SSC-D}. At each of these three recursions the paths are reflected to indicate the node traversed (\textbf{SSC-B}) and the cost so far i.e 4,5 and 6 to reach \textbf{a}, \textbf{SC-3} and \textbf{SSC-B} respectively. Each recursion at every node returns with the shortest path to the vertex. Thus the recursions from \textbf{a}, \textbf{SC-3} and \textbf{SSC-D} return with the shortest paths to the vertex. This along with the shortest path to reach those nodes, provides us with the means to decide on the shortest path to reach the vertex. 

\subsubsection{Building and updating the routing cache}
The best hop to take to reach a certain unit is the last node that was reached prior to reaching the vertex, when traversing the shortest path from the corresponding unit graph node to the vertex.  
This information is collected within the \emph{routing cache}, so that events can be disseminated faster throughout the system. The routing cache should be used in tandem with the routing information contained within a routed event to decide on the next best hop to take to ensure efficient dissemination. 
Certain portions of the cache can be invalidated in response to the addition or failures of certain edges in the connectivity graph.

%In general when a level-$\ell$ node is added to the connectivity graph, connectivities pertaining to units at level $\ell, \ell+1, \cdots, N$ are effected. 
%For a level-$N$ system if a gateway $g^\ell$ within $u^{\ell+1}_i$ is established, the information contained in the routing cache to reach units at level $\ell, \ell+1, \cdots N$ needs to be updated for all the units within $u^{\ell+1}_i$.  
%The cases of gateway failures, node failures, detection of partitions and the updating of the routing cache in response to these failures are dealt with in a later section. 

%\subsubsection{Exchanging information between super-units}
%When a subsystem $u^{\ell}_i$ is added to an existing system $u^{\ell+j+1}$; information regarding $g^{\ell+j}, g^{\ell+j-1}, \cdots , g^{\ell}$ connections are exchanged between the system and the newly added sub system. Thus when a super cluster is added to an existing system comprising of super-super-clusters, the existing system routes information regarding super-cluster and super-super-cluster connections to the newly added super-cluster. The way the set of connections, comprising the connectivity graph, is sent over the newly established link is consistent with the rules, which we had set up for sending a connection information over a gateway as discussed in section \ref{sec:connectivity-graph-connection}. Thus, if a new super cluster \textbf{SC-4} is added to the \textbf{SSC-A} sub-system and a super cluster gateway is established between \textbf{SC-4} and node \textbf{SC-1.c.6}, then, the connectivity graphs at node \textbf{6} would be as depicted in figure \ref{fig:connectivity-graph-nua}.(a) while the connectivity graph at the gatekeeper in \textbf{SC-4} would comprise of the connections that were sent over the newly established gateway by node \textbf{6}. 

%\begin{figure}[htp]
%\centering
%\epsfig{file=images/ConnectivityGraph-newUnitAdded.eps,width=426.00pt, height=229.00pt}
%\caption{\label{fig:connectivity-graph-nua} Connectivity graphs after the addition of a new super cluster SC-4.}
%\end{figure}

%Figure \ref{fig:connectivity-graph-nua}.(b) depicts only the connections which describe the connections involving level-2 gateways and upwards at node \textbf{99} in SC-4. There would be clusters comprising of strongly connected broker nodes in \textbf{SC-4}, we however do not need to depict these, in figure \ref{fig:connectivity-graph-nua}.(b), for the present discussion regarding connection information exchange. 

\subsection{Organization of Profiles and the calculation of destinations}
Every event conforms to a signature which comprises of an ordered set of attributes $\{ a_1, a_2, \cdots, a_n\}$. 
The values these attributes can take are dictated and constrained by the \emph{type} of the attribute. 
Clients within the system that issue these events, assign values to these attributes. 
The values these attributes take comprise the content of the event. 
All clients are not interested in all the content, and are allowed to specify a filter on the content that is being disseminated within the system. 
Thus a filter allows a client to register its interest in a certain type of content. 
Of course one can employ multiple filters to signify interest in different types of content. These filters specified by the client constitutes its \emph{profile}. 
The organization of these profiles, dictates the efficiency of matching content. 
%A level-$\ell$ gatekeeper snapshots the profiles of all the level-{($\ell$-1)} units that share the same GES context $C^{\ell}_i$ with it. 

%\subsubsection{The problem of computing destinations}
%Clients express interest in certain types of content, and events which conform to that content need to be routed to the client. A simple approach would be to route all events to all clients, and have the clients discard the content that they are not interested in.  This approach would however place a strain on network resources. Under conditions of high load and increasing selectivity by the clients, the number of events a client discards would far exceed the number of events it is actually interested in. This scheme also affects the latency associated with the reception of real time events at the client. The system thus needs to be very selective of the kinds of events that it routes to a client. In other words the system should be able to efficiently compute destination lists associated with the event. Depending on the event this destination list could be \emph{internal} to the event or \emph{external} to the event. In the case of events with external destination lists, the system relies on information contained within the client's profile and also the content of the event to arrive at the set of destinations that need to receive the event. These destinations should be computed in such a way that it exploits the network topology in place, as also the routing algorithms that make use of abbreviated views of inter-connections existing within the system. Profiles need to be organized so that they lend themselves to very efficient calculation of destinations upon receiving a \emph{relevant} event. In our approach a level-$\ell$ gatekeeper maintains the profiles of all the level-($\ell$-1) units that share the same GES context $C^{\ell}_i$ with it. This scheme fits very well with our routing algorithms, since the destinations contained within the event are those that are consistent with the node's abbreviated view of the system. To allow for a node to maintain profiles contained at different units (clusters, brokers, clients etc.) we need to be able to be able to propagate profile additions and changes to nodes responsible for the generation of destination lists. The problem of computing destinations for a certain event comprises of the following -- \begin{itemize} 
%\item[(a)] Organization of profiles in a profile graph
%\item[(b)] Propagation of profiles to the nodes that are responsible for the calculation of hierarchical destination lists.
%\item[(c)] Navigation of the profile graph to compute the destinations associated with the content. 
%\end{itemize}
%A given node can compute destinations only at certain level. 
%Thus the computation of destinations is itself a distributed process in our model.



\subsubsection{Constructing a profile graph}
Events encapsulate content in an ordered set of $<attribute,$ $value>$ tuples. The constraints specified in the profiles should maintain this order contained within the event's signature.  Thus to specify a constraint on the second attribute ($a_2$) a constraint should have been specified on the first attribute ($a_1$). What we mean by constraints, is the specification of the value that a particular attribute can take. We however also allow for the weakest constraint, denoted \textbf{$*$}, on any of the attributes. The $*$ signifies that the filtered events can take any of the valid values within the range permitted by the attribute's type.  
By successively specifying constraints on the event's attributes, a client narrows the content type that it is interested in. 
%It is not necessary that a constraint be specified on all the attributes  $\{ a_1, a_2, \cdots, a_n\}$. What is necessary is that if a constraint is imposed on an attribute $a_i$ constraints for attributes $a_1, a_2, \cdots, a_{i-1}$ must be in place, even if some or all of these constraints is the weakest constraint $*$. Thus if a constraint is specified till attribute $a_i$ and no constraints are imposed on some of the attributes $a_1, a_2, \cdots, a_{i-1}$, the system assigns these attributes the weakest constraint $*$. It makes  more sense imposing the constraint $*$  on higher order attributes $a_{i+1} \cdots a_n$ than on the lower order attributes $a_1, a_2, \cdots a_{i-1}$. Such a scheme has the effect of narrowing content down to the ones which are very closely related to each other. For every event type we maintain a profile chain. Different profile chains when added up constitute the profile graph. 


We use the general matching algorithm, presented in \cite{matching-events}, of the Gryphon system to organize profiles and match the events. 
Constraints from multiple profiles are organized in the \emph{profile graph}.  
Every attribute on which a constraint is specified constitutes a node in the profile graph. 	
When a constraint is specified on an attribute $a_i$, the attributes $a_1, a_2, \cdots, a_{i-1}$ appear in the profile graph. 
A profile comprises of constraints on successive attributes in an event's signature. 
The nodes in the profile graph are linked in the order that the constraints have been specified. 
Any two successive constraints in a profile result in an edge connecting the nodes in the profile graph. 
Depending on the kinds of profiles that have been specified by clients, there could be multiple edges, \emph{originating from} a node. 
%Following the scheme in \cite{matching-events} we do not allow multiple edges \emph{terminating at} a node since it results in a situation where the event matching results in an invalid destination, due to that event having satisfied partial constraints of different profiles from within the same unit.   
%\begin{figure}[htp]
%\centering
%\epsfig{file=images/Profiles-Example.eps,width=182.00pt, height=216.00pt}
%\caption{\label{fig:profile-graph-example} The profile graph - An example.}
%\end{figure}
Figure \ref{fig:profile-graph-edgeInfo} depicts the profile graph constructed from three different profiles. The example depicts how some of the profiles share partial constraints between them, some of which result in profiles sharing edges in the profile graph.  
%A certain edge is marked as traversed by an event if the two successive constraints that created the edge, have been satisfied by that event. The presence of an edge signifies the existence of at least one client, which is interested in the content satisfying at least two of the constraints contained in that edge. An event's traversal along an edge simply indicates that the event's content has satisfied some partial constraint imposed by one or more of the clients. As we traverse further down the profile chain, the events we are looking for get more fine grained. The final constraint on an attribute leads to the creation of a destination edge. The edges arising out of node \textbf{C} in figure \ref{fig:profile-graph-example} are destination edges.

%\subsubsection{Information along the edges}
%To support hierarchical disseminations and also to keep track of the addition and removal of edges, besides the basic organization of constraints, the graph needs to maintain additional information along its edges. This additional information also plays a very important role in the reliable delivery of events to clients (we discuss this in a later section).  
Along every edge we maintain information regarding the units that are interested in its traversal. For each of these units we also maintain the number of predicates $\delta\omega$ within that unit that are interested in the traversal of that edge. 
The first time an edge is created between two constraints as a result of the profile specified by a unit, we add the unit to the \emph{route information} maintained along the edge.  
For a  new profile $\omega_{new}$ added by a unit, if two of its successive constraints already exist in the profile graph, we simply add the unit to the existing routing information associated with the edge. If the unit already exists in the routing information, we increment the predicate count associated with that destination.

\begin{figure}[htp]
\centering
\epsfig{file=images/Profiles-edgeInformation.eps,width=247.00pt, height=171.00pt}
\caption{\label{fig:profile-graph-edgeInfo} The complete profile graph with information along edges.}
\end{figure}
The information regarding the number of predicates $\delta\omega$ per unit that are interested in two successive constraints allows us to remove certain edges and nodes from the profile graph, when no clients are interested in the constraints any more. Figure \ref{fig:profile-graph-edgeInfo} provides a simple example of the information maintained along the edges. 
%We discuss how the profiles are propagated, where they are propagated and how this information along the edges is modified and updated in section \ref{sec:propagation-of-profile-changes}.
%\subsubsection{Computing destinations from the profile graph}
%Once the profile graph has been constructed, we can compute the destinations that are associated with an event. 
%Traversal along an edge is said to be complete if two successive constraints at end points of the edge have been satisfied by the content in question. 
When an event comes in we first check to see if the profile graph contains the first attribute contained in the event. If that is the case we can proceed with the matching process. 
When an event's content is being matched, the traversal is allowed to proceed only if  --
\begin{itemize}
\item[(a)] There exists a wildcard ($*$) edge connecting the two successive attributes in the event. 
\item[(b)] The event satisfies the constraint on the first attribute in the edge, and the attribute node that this edge leads into is based on the next attribute contained in the event.

\end{itemize}
As an event traverses the profile graph, for each destination edge that is encountered if the event satisfies the destination edge constraint, that destination is added to the destination list associated with the event.

\subsubsection{The profile propagation protocol - Propagation of $\pm \delta\omega$ changes}
\label{profile-propogation-protocol}
\label{sec:propagation-of-profile-changes}
In the hierarchical dissemination scheme that we have, gatekeepers $g^{\ell+1}$ compute destination lists for the $u^{\ell}$ units that it serves as a $g^{\ell+1}$ for. 
A cluster gatekeeper would maintain profiles of all the broker nodes contained with that cluster. 
A gatekeeper $g^{\ell+1}$ should thus maintain information regarding the profile graphs at each of the $u^{\ell}$ units. 
Profile graph ${\mathcal{P}}^{\ell+1}_i$ maintains information contained in profiles $\mathcal{P}^{\ell}$ at all the $u^\ell$ units within $u^{\ell+1}_i$. 
This should be done so that when an event arrives over a $g^{\ell+1}$ in $u^{\ell+1}_i$  --
\begin{itemize}
\item[(a)] The events that are routed to destination $u^\ell$'s, are those with content such that at least one destination exists for those events within the sub-units that comprise the profile for $u^\ell$.
\item[(b)] There are no events, that were not routed to  $u^\ell_i$, with content such that $u^\ell_i$ would have had a destination within the sub-units whose profile it maintains.
\end{itemize}
Properties (a) and (b) ensure that the events routed to a unit, are those that have at least one client interested in the content contained in the event. 
%When an event is received over a cluster gateway, there would be at least one client attached to one of the nodes in the cluster which is interested in that event.

%When we send the profile graph information over to the higher level gatekeeper $g^\ell$, the information contained along the edges in the graph needs to be updated to reflect the nodes logical address at that level. Thus when a node propagates the clients profile to the cluster gatekeeper, it propagates the edges created/removed with the broker as the destination associated with the profile predicate. Similarly, when this is being propagated to a super-cluster gatekeeper the profile change is sent across as a profile change in the cluster. Any change in the client's profile is propagated to gatekeepers at higher levels, that the broker node in its abbreviated view of the system is aware of. What we are trying to do is to maintain information in the profile graph, in a manner which is consistent with the dissemination constraints imposed by properties (a) and (b). The reason we maintain destination information the way we do is that this model ties in very well with our topology and the routing algorithms that are in place. The connectivity graph provides us with an overall view of the interconnections between units at different levels. The organization and calculation of destinations from the profiles comprising the profile graph, feeds right into our routing algorithms that compute the shortest path to reach the units (destinations) where an event needs to be routed.  In general for a level-$N$ system, if there is a subscribing client with GES context $C^{N}_j$ and the issuing client has GES context $C^{N}_i$ the destinations are computed (N+1) times. Thus, in a system comprising of super-super-clusters, the destinations are computed four times prior to reception at the client. 


For profile changes that result in a profile change of the unit, the changes need to be propagated to relevant nodes, that maintain profiles for different levels. 
A cluster gateway snapshots the profile of all clients attached to any of the broker nodes that are a part of that cluster. 
%Thus a change in the profile of a client needs to be propagated to its broker node. 
The change in profile of the  broker node should in turn be propagated to the cluster gateway(s) within the cluster that the node is a part of. A profile change in broker (as a result of a change in an attached client's profile) needs to be propagated to the unit (cluster, super-cluster, etc) gatekeeper within the unit that the broker is a part of. 
%Similarly a super-cluster gateway snapshots the profiles of all the clusters contained in the super-cluster.When a profile change occurs at any level, the updates need to be routed to relevant destinations. The connectivity graph provides us with this information. From the connectivity graph, it can be seen that node \textbf{4} is the cluster gateway. Thus, changes in profiles at level-0 (i.e. $\delta\omega^0$) at any of the nodes in cluster \textbf{SSC-A.SC-1.c} are routed to node \textbf{4}.  $\delta\omega^1$ changes need to be routed to level-2 gateways within \textbf{SSC-A}.  
In general the gatekeepers to which the profile changes need to be propagated are computed as follows ---
\begin{itemize}
\item[(a)] Locate the level-($\ell$) node in the connectivity graph.
\item[(b)] The uplink from this node of the connectivity graph to any other node in the graph, indicates the presence of a level-$\ell$ gateway at the unit corresponding to the graph node. 
\end{itemize}
This scheme provides us with information regarding the level-$\ell$ gateway, within the part of the system that we are interested in. 
%We are not interested in the lateral links since they provide us with information regarding all the  level-$\ell$ gateways within the next higher level GES context $C^{\ell+1}$.
%\begin{figure}[htp]
%\centering
%\epsfig{file=images/ConnectivityGraph.eps,width=285.00pt, height=219.00pt}
%\caption{\label{fig:connectivity-graph-destinations} The connectivity graph at node 6.}
%\end{figure}
In the connectivity graph depicted in figure \ref{fig:connectivity-graph} any  $\delta\omega^0$ changes at any of the nodes within cluster \textbf{c}, need to be routed to node \textbf{4}. Any $\delta\omega^1$ changes at node \textbf{4} need to be routed to node \textbf{5}, and also to a node in cluster \textbf{b}. Similarly $\delta\omega^2$ changes at node \textbf{5} needs to be routed to the level-3 gatekeeper in cluster \textbf{a} and superclusters  \textbf{SC-3}, \textbf{SC-2}. When such propagations reach any unit/super-unit the process is repeated till such time that the gateway that the node seeks to reach is reached. Every profile change has a unique-id associated it, which aids in ensuring that the \emph{reference count scheme} does not fail due to delivery of the same profile change multiple times within the same unit.

%\subsubsection{Summary}
%Summarizing the discussion so far, the profile graph snapshots the profiles of units at a certain level, and as such can compute destinations only for this set of units. The profile snapshot that is created ensures that there is at least one sub-unit attached to one of the units within the super unit under consideration which should receive the event. Thus the profile matching scheme ensures that there is at least one client which will receive the event when it is received within a unit. If we do not have a scheme which snapshots profiles in the following manner, we could end up in a scenario where none of the events received in a unit have any clients which are interested in that event.  

%\subsubsection{Unit additions and the propagation of profiles}
%When a unit (with publishing and subscribing clients) is being added to a larger existing broker network, besides the sequence of actions pertaining to the generation/update of logical addresses and the exchange of system inter connectivities, profiles would need to be propagated in exactly the same way that we described. Thus when a cluster is added to the system, the broker nodes within the cluster route their profiles to the newly created cluster gatekeeper. This gatekeeper is in turn responsible for the propagation of profiles to the super-cluster gatekeepers in the newly merged system. 

%\subsubsection{Active profiles}
%\label{active-profiles}
%The profile propagation protocol aids in the creation of destination lists at units within different levels. These destination lists are then employed at each level for finer grained disseminations. Since the profile add/change propagates through the system to higher level gateways, it is possible that a gateway at a higher level has not yet been notified about the profile add/change.  Thus though it may receive an event which would match the profile change, the destination list may not include the lower level unit. It is possible that a client may receive events issued by clients within a certain unit, though it may not receive similar events from clients published by units within a different GES context. 

%What interests us is the precise instant of time from which point on we can say that all events that satisfy the client's profile will be delivered to the client. To address this issue we introduce the concept of \emph{active profiles}, which provides guarantees in the routing of events within a unit. The active profile approach provides us with a unit-based incremental approach towards achieving system guarantees during a profile add/change. If a profile is \emph{super-cluster active} all events issued by clients attached to any of the broker nodes within a super-cluster $C^2_i$ will be routed to the interested client. Thus the first event that is received by the client is an indication that all subsequent events routed to that unit, matching the same profile would also be received by the client. When we say that a profile is \emph{unit-active}\footnote{The unit we are referring to in this case are the clusters, super-clusters, super-super-clusters etc. Of course these units are assumed to be within some higher level GES context of the broker node to which the interested client  is attached to or was last attached to} what we mean is that for every event that is being routed within that unit the destination lists calculated would include information to facilitate routing to the client. Since a client profile is unit active, all events, issued within the unit, will be routed to the client if it satisfies the client profile.

%Events contain routing information in them, which indicate the units where these events were disseminated. The routing information contained in an event thus includes the unit in which the event was issued. Since the dissemination is hierarchical, an event will not be routed to a client till such time that the client's profile change has been propagated to higher level gatekeepers. If a profile change issued by a client $c_A$ is routed to a super-cluster gatekeeper, all events issued by clients attached to any of the nodes within this super-cluster, will be routed to the client $c_A$ if these events match the corresponding profile change. The routing information, for events issued by clients in this super-cluster, indicate the dissemination within the units in that super-cluster. If this event matches the profile change initiated by one of the attached clients, and if this event is routed to such a client then the profile change associated with that client is said to be super-cluster active. In an $N$-level system if the routing information depicts the dissemination of the event within another level-$N$ unit within the system, the profile change issued by the client is said to be \emph{system active}. When a profile change initiated by a client is system active, events issued by any other clients within the system will be routed to this client, if those events match the system active profile change that was initiated by this client.


\subsection{The event routing protocol - ERP}
\label{the-event-routing-protocol}
Event routing is the process of disseminating events to relevant clients. This includes matching the content, computing the destinations and routing the content along to its relevant destinations by determining the next broker node that the event must be relayed to. 
Events have routing information associated with them, which indicate its dissemination within various parts of the broker network. The dissemination information at each level can be accessed to verify disseminations in various sections of the broker network. Routing decisions and the brokers that an event must be relayed to are made on the basis of this information. 
This routing information is not added by the client issuing this event but by the system to ensure faster dissemination and recovery from failures. 
When an event is first issued by the client the broker node that the client is attached to adds the routing information to the event. 
This routing information is the GES contextual information (see section \ref{broker-topology:contexts}) pertaining to this particular node in the system.
As the event flows through the system, via gateways, the routing
information is modified to snapshot its dissemination within the
system. 
%This information is then used to avoid routing the event to the same unit twice. What a node also needs to decide is when it would be futile to try and find a higher order gateway, and also when all the higher level units that could possibly be covered have been covered. Of course it should also know if there is a higher order gateway that needs to be reached.  This decision is based on the event routing information and the information pertaining to gateways that's available at a node. If there are no such units that need to be reached, the event routing would proceed with lower order disseminations. However if there is a unit that needs to be reached, gateways would have to be employed to reach this unit as fast as possible. The event routing information contained with an event simply indicates the units, which were present en route to reception at the node.

%\begin{figure}[htp]
%\centering
%\epsfig{file=images/GatekeepersandEventRoutings-example.eps,width=427.00pt, height=369.00pt}
%\caption{\label{fig:routing-events} Routing events}
%\end{figure}

%A gateway $g^{\ell +1}$ in $u^{\ell+1}_i$ is responsible for the dissemination of events throughout the relevant $u^\ell$ units within $u^{\ell+1}_i$.  This is a recursive process and the gateway $g^{\ell+1}$ delegates this dissemination process to the lower level gateways $g^{\ell}, g^{\ell-1}, \cdots, g^1$ to aid in finer grained disseminations. Thus a super-super-cluster gateway is responsible for disseminating the event to all the super-clusters which comprise the super-super-cluster that it is a part of. A gateway $g^\ell$ is concerned with the routing information from level-$\ell$ to level-$N$. When an event has been routed to a gatekeeper $g^{\ell}$ the routing information associated with the event is modified to reflect the fact that the event was received at that particular unit. It is the gatekeeper $g^{\ell}$'s responsibility to ensure that the event is routed to all the relevant nodes within the level-$\ell$ unit, using the delegation mechanism described earlier.
 Prior to routing an event across the gateway a level-$\ell$ gatekeeper takes the following sequence of actions --
\begin{itemize}
\item Check the level-$\ell$ routing information for the event to determine if the event has already been consumed by the unit at level-$\ell$. If this is the case the event will not be sent over the gateway to that unit.
%There could be multiple links connecting a unit to some other unit. This scheme provides us with a greater degree of fault-tolerance. This also leads to the situation\footnote{One of the reasons that this situation arises is a fork in the event's routing which send it to two gateways within the same unit } where the event could be routed to the same unit over multiple links. In this case the duplicate detection algorithm detects this duplicate event and halts any further routing for this event.

\item In case the gateway decides to send the event over the gateway, all routing information pertaining to lower level disseminations are stripped from the event routing information.

%This is because the routing information pertaining to the lower level disseminations is within the GES context of a specific level-$\ell$ unit and will not be valid within other level-$\ell$ units. Also, in general a higher order gateway would be more overloaded\footnote{This is because a lower order gateway is primarily employed for finer grained dissemination of events, and only rarely if at all would be used to get to a higher order gateway. Besides this a higher order gateway $g^{\ell}_i(C^{\ell +1}_i)$ is the one responsible for deciding if the event needs to be routed to any of the lower units comprising the level-$\ell$.} compared to a lower order gateway. Reducing the amount of information being transferred over the gateway helps conserve bandwidth.
\end{itemize}
%Figure \ref{fig:routing-events} depicts the routing scheme which we have discussed so far. The routings depicted in the figure outline how routing information is updated to reflect the traversal at units in different levels.

%In addition to the information regarding where the event has been delivered already, events also need to contain information regarding the units which an event should be routed to.
%Gatekeepers $g^\ell(C^{\ell +1})$ decide the level-$(\ell-1)$ units which are supposed to the receive the event.
%This decision is based on the profiles available at the gatekeeper as outlined in the profile propagation protocol.
%This calculation of the targeted units is a recursive process with the lower order disseminations being handled by the corresponding lower order gatekeepers. Thus two levels of routing information are contained within an event ---
%\begin{itemize}
%\item[(a)] Units where an event should be routed within a unit.
%\item[(b)] Units which have already received the event.
%\end{itemize}
%This routing scheme plays a crucial role in determining which events need to be stored to a stable storage during failures and partitions.

When a gatekeeper $g^\ell$ with GES context $C^\ell_i$ is presented with an event it computes the $u^{\ell-1}$'s within $C^\ell_i$ that the event must be routed to. A cluster gatekeeper, when it receives an event, computes the broker destinations associated with that event. This calculation is based on the profiles available at the gatekeeper as outlined in the profile propagation protocol. At every node the best hops to reach the destinations are computed. 
Thus, at every node the best decision is taken. 
Nodes and links that have not been failure suspected  are the only entities that can be part of the shortest path. 
The event routing protocol, along with the profile propagation protocol and the gateway information ensure the optimal routing scheme for the dissemination of events in the existing topology.

%\subsection{Routing real-time events}
%\label{routing-real-time-events}
%Real time events can have destination lists (see section \ref{event-anatomy:destination-lists}) which are internal or external to the event. In each case the routing differs, in the case of internal lists the destination's location needs to be precisely located by the system. Routing events with external destination lists involves the system calculating the destinations for delivery.

%\subsubsection{Events with External Destination lists}
%\label{routing:events-with-external-destination-lists}
%When an event arrives at a gatekeeper $g^\ell$, the gatekeeper checks to see if the event satisfies its profile. The profile maintained at  $g^\ell$ snapshots the profile of the level-$\ell$ unit that the gatekeeper belongs to.  This check is necessary to confirm if the event needs to be disseminated within the level-$\ell$ unit. Routing events based on the gatekeeper profile is the process which calculates the destination lists.  This is a recursive process in which each higher order gatekeeper performs this check before disseminating the event to lower order gatekeepers.

% When an event doesn't match the gatekeeper $g^\ell$'s profile, $g^\ell$ decides upon the next route that event would take based on the routing information encoded into the event by the event routing protocol.
%\begin{itemize}
%\item The gatekeeper $g^\ell_j(C^{\ell +1}_i)$ checks the routing information provided by ERP to see if it needs to relay the event to other gatekeepers $g^\ell$  within the GES context $C^{\ell +1}_i$.

%\item The gatekeeper also uses the information provided by ERP to check if it could route the event to a higher order gateway which has not received the event.
%\end{itemize}

%In the event that these steps lead to no actions on part of the gatekeeper $g^\ell$ the gatekeeper takes no further actions to route this event. If the gatekeeper decides to route this event to other level-$\ell$ and higher order gatekeepers, the system can employ lower order gateways within the GES context  $C^{\ell +1}_i$ to relay this event.

%\subsubsection{Events with Internal Destination lists}
%\label{routing:events-with-internal-destination-lists}
%These are events which require the system to be able to route the event to a specific client in the system. Clients which are interested in receiving point-to-point events thus need to include their identifier in their profile. The sequence of steps that are needed to route the event are similar to the steps we take to route events with external destination lists as discussed in section \ref{routing:events-with-external-destination-lists}.

%\subsection{Interaction between the protocols and performance gains}
%The broker topology comprising of strongly connected broker nodes in clusters connected by \emph{long links} mirrors \emph{small world} \cite{small-world} networks, which ensure that the pathlength increases logarithmically for geometric increases in the size of the broker node network. The feature of having multiple links between two units/super-units ensures a greater degree of fault tolerance. Links could fail, and the routing to those units could still be performed using the alternate links. The organization of connection information ensures that connection losses (or additions) are incorporated into the connectivity graph hosted at relevant nodes. Certain sections of the routing cache are invalidated in response to this addition (or loss) of connections. This invalidation and subsequent calculation of best \emph{hops} to reach units (at different levels) ensure that the paths computed are consistent with the state of the network, and include only valid/active links. The ability to compute routes to reach destinations at different levels lends the scheme very useful for hierarchical disseminations. 

%In our scheme for the organization of profiles we employ an approach where profiles of sub-units are maintained at the unit gatekeeper. Events almost always arrive at the unit gatekeepers first, since they provide a gateway to the unit. The only exception is in the cluster where a client issues an event. Having this unit gatekeeper intelligently decide on the sub-units, which should receive an event helps eliminate redundant routing of events. By maintaining sub-unit profiles at the unit gatekeeper we ensure that the only events that are routed to a unit are those for which there is at least one client, attached to one of the broker nodes in that unit, which is interested in the specific event. We obtain information regarding the nodes/units to route profile changes based on the information contained in the connectivity graph. We then employ hops (at every broker node en route) obtained from the routing cache to ensure that this profile dissemination is the fastest. The information maintained in the profile graph is consistent with the dissemination scheme and can be used to compute destinations at different levels. In an N-level system, an event is matched (N+1) times prior to routing the event to a client. 

%The event routing protocol uses the profile information available at the unit gatekeepers to compute the sub-units that the event should be routed to. To reach these destinations every node, at which this event is received, employs the \emph{best hops} to reach the destinations.  This \emph{best hop} is computed based on the cost of traversal as also the number of links connecting the different units. Thus in our system, based on the organization of profiles and subsequent matching of events, the only units to which an event is routed are those that have clients interested in that event. Further, based on the connectivity graph and the associated routing cache we compute the fastest/reliable hops to take to reach the relevant destinations.  The routing information encoded into the event along with the duplicate detection scheme ensures that we eliminate \emph{continuous event echoing}, where the event is routed to the same unit over and over again.  

%These approaches result in only the relevant links and functioning nodes being employed for disseminations. The \emph{small world} behavior that would exist in broker network, when appropriately organized, ensures that the pathlengths for these disseminations would only increase logarithmically with the number of broker nodes. 







\section{The Reliable Delivery Of Events}
\label{reliable-delivery}
Reliable delivery involves the guaranteed delivery of events to intended recipients. The delivery guarantees need to be satisfied even in the presence of single or multiple broker failures, link failures and network partitions. In GES clients need not maintain an active online presence and can also roam the network attaching themselves to any of the nodes in the broker network. Events missed by clients in the interim need to be delivered to these clients irrespective of the failures that have or are currently present in the system.

\subsubsection*{Prior Art}
The problem of reliable delivery \cite{toueg:fault-tolerant-broadcasts, birman:process-reliable} and ordering \cite{birman:order, birman:Response-CATOCS} in traditional group based systems  with process crashes has been extensively studied.
The approaches normally have employed the \emph{primary partition} model \cite{schiper:partitions}, 
which allows the system to partition under the assumption that there would be a unique partition, which could make decisions on behalf of the system as a whole, without the risk of contradictions arising in the other partitions and also during partition mergers. 
However the delivery requirements are met only within the primary partition \cite{kathy:message-stability}. 
Recipients that are slow or temporarily disconnected may be treated as if they had left the group. 
This model, adopted in Isis \cite{birman:replication-in-ISIS}, works well for problems such as propagating updates to replicated sites. 
This approach does not work well in situations where the client connectivity is intermittent, and where the clients can roam around the network. 
Systems such as Horus \cite{horus} and  Transis \cite{transis} manage \emph{minority partitions}, and can handle concurrent views in different partitions. The overheads to guarantee consistency are however too strong for our case. DACE \cite{dace} introduces a failure model, for the strongly decoupled nature of pub/sub systems. This model tolerates crash failures and partitioning, while not relying on consistent views being shared by the members. DACE achieves its goal through a self-stabilizing exchange of views through the Topic Membership protocol. In \cite{toueg:lossy-links}, the effect of link failures on the solvability of problems  (which are solved with reliable links) in asynchronous systems, has been rigorously studied. \cite{schneider} describes approaches to building fault-tolerant services using the state machine approach. 

Systems such as \emph{Sienna} \cite{Sienna:CBR, Sienna:Scaling} and \emph{Elvin} \cite{Elvin:CBR, Elvin:matching, Elvin:pub/sub} focus on efficiently disseminating events, and do not sufficiently address the reliable delivery problem in the presence of failures. 
In \emph{Gryphon} the approach to dealing with broker failures is one of reconstructing the broker state from its neighboring brokers. This approach requires a failed broker to recover within a finite amount of time, and recover its state from the brokers that it was attached to prior to its failure.  
\emph{SmartSockets} \cite{smart-sockets} provides high availability/reliability through the use of software redundancies. Mirror processes receiving the same data and performing the same sequence of actions as the primary process, allows for the mirror process to take over in the case of process failures. The \emph{mirror process} approach runs into scaling problems as the number of processes increase, since each process needs to have a mirror process. Since there is an entire server network that would be mirrored in this approach the network cycles expended for dissemination also increases as the number of server nodes increases. \emph{SmartSockets} also allows for routing tables to be updated in real time in response to link failures and process failures.  
What is not clear though, is how the system is affected if both the process and its mirror counterpart fail.  
\emph{TIB/Rendezvous} \cite{tibco} integrates fault tolerance through delegation to another software \emph{TIB/Hawk} which provides it with immediate recovery from unexpected failures or application outages. This is achieved through the distributed \emph{TIB/Hawk} micro-agents, which support autonomous network behavior, while continuing to perform local tasks even in the event of network 
failures. 

Message queuing products such as IBM's MQSeries \cite{ibm-mq}  and Microsoft's MSMQ \cite{msmq} are statically pre-configured to forward messages from one queue to another. This leads to the situation where they  generally do not handle changes to the network (node/link failures) very well. They also require these queues to recover within a finite amount of time to resume operations. 
To achieve guaranteed delivery, JMS provides two modes: persistent for sender and durable for subscriber.
When messages are marked persistent, it is the responsibility of the JMS provider \cite{sonicMQ, jmq, iBus, firanoMQ} to utilize a store-and-forward mechanism to fulfill its contract with the sender (producer). 

\subsubsection{GES Solution Highlights}
Our solution to the reliable delivery problem eliminates the constraint that a failed broker recovers within a finite amount of time. Also, we do not rely on state reconstructions of broker since these solutions lead to problems during multiple broker failures. In our solution we allow for a broker to fail and remain failed forever. The reliable delivery scheme also incorporates a replication scheme that provides for different replication strategies to exist at different parts of the system. The scheme also allows for this replication strategy to change at different parts of the system as time progresses. Changes to the replication scheme at a certain section of the broker network results in updates at the corresponding subsystems of the broker network. Roaming, disconnected and connected clients are not affected by these changes in replication schemes within the broker network. Clients reconnecting after a failure or prolonged disconnect make a complete recovery of the events that were published in the interim. The recovery is very precise and only those events that were missed by a reconnecting client are routed through the system en route to the recovering client as recovery events. The persistence model in most systems involves the persistent store subscribing to all events that are published in the system. This could lead to exponential rise in storage requirements at different stable storages in the system. With increasing selectivity in the type of events a client is interested in and also in the number of clients that are present in the system, the events stored in storages servicing subsystems far exceeds the number of events that clients in those subsystems are actually interested in. In the GES replication scheme a stable storage servicing a subsystem subscribes to only those events that the brokers (and in turn the clients) that it servicing, is interested in. We also have a garbage collection scheme which ensures that the storage space does not increase exponentially with time. The epochs and reference counting scheme ensures that no client is starved of events that it was supposed to receive while ensuring that the event is garbage collected once all clients interested in receiving those events have received that event.

In our failure model a unit can fail and remain failed forever. The broker nodes involved in disseminations compute paths based on the active nodes and traversal times within the system. The routing scheme is thus based on the state of the network at any given time. Brokers could be dynamically created, connections established or removed, and the events would still be routed to the relevant clients.  Any given node in the system would thus see the broker network undulate, as the brokers are being added and removed. Connections could also be instantiated dynamically based on the average pathlength for communication with any other node within the system. The connectivity graph maintains abbreviated system views; each node in this graph could also maintain information regarding the average pathlengths for communication with any other node within the unit, which the graph node represents. Connections could be dynamically instantiated to vary clustering coefficients and also to reduce average pathlengths for communications. The routing algorithms and the failure model allow support for dynamic reconfiguration of networks. 


%\subsubsection{Detection of partitions}
%\label{sec:detection-of-partitions}
%Partitions arise due to node failures or link failures. There are two different kinds of partitions that can arise in our system due to a connection failure -- unit partitions and system partitions. The way the system deals with each case is different. Dealing with partitions is through \emph{delegation} where each super-unit of the system deals with the partitions that arise within its units. Detection of partitions  is an extremely desirable feature since, in our  system, a client can roam in response to the partition. Thus, clients hosted within  partitioned units can roam to nodes that are in the majority partition. This calculation of the nodes to roam to during partitions could also be based on some system defined rule.  Healing of the partitions could result in the affected units being able to deal with clients in a consistent manner and share the client load of the system. 

%\begin{figure}[ht]
%\centering
%\epsfig{file=images/ConnectivityGraph-Connections.eps,width=429.00pt, height=363.00pt}
%\caption{\label{fig:connections-2} Connectivities between units and the detection of partitions}
%\end{figure}	
%Figure \ref{fig:connections-2} depicts the connections that exist between various units of the 3 level system which we would use as an example in our discussions. 
%The nodes within the connectivity graph are organized as nodes at various levels. 
%Associated with every level-$\ell$ node in the graph are two sets of links -- the set $L_{UL}$ that comprises of connections to nodes $n^a_i \ni a \leq \ell$ and the set $L_D$ comprising of connections to nodes $n^b_i \ni b > \ell$. 

%\begin{figure}[htp]
%\centering
%\epsfig{file=images/ConnectivityGraph.eps,width=285.00pt, height=219.00pt}
%\caption{\label{fig:connectivity-graph-2} The connectivity graph at node 6 and the detection of partitions.}
%\end{figure}	
%Figure \ref{fig:connectivity-graph-2} depicts the connectivity graph that is constructed at the node SSC-A.SC-1.c.6 in Figure \ref{fig:connections-2}. The set  $L_{UL}$ at the node \textbf{SC-3} in the figure comprises of node \textbf{SC-2} at level-2 and node \textbf{b} at level-1. The set  $L_D$ at \textbf{SC-3} comprises of the node \textbf{SSC-B} at level-3. The information contained in the loss of connections is identical to that contained in the addition of a new connection. Also the dissemination of this loss of connection is dealt with in exactly the same way as additions are as described in section \ref{sec:connectivity-graph-connection}. 

%When a connection is lost we remove the  connection from the connection table maintained at every node. This is  based on the unique identifier associated with every connection. Next we check to see if there are other connections that exist between the corresponding nodes in the connectivity graph. If the link count associated with the connection edge is greater than one, then we decrement the link count associated with the connection and conclude that the connection loss is compensated by other existing connections between the two corresponding units. A situation where the link count is reduced to zero results in the removal of link information from the sets $L_{UL}$ and $L_D$ associated with the node.  If the connection that was lost is the connection  $<n^{x}_i, n^{y}_j, \ell>$ (where $x \mid y = \ell$ and  $x,y \leq \ell$), then if $y \leq x$ node  $n^{y}_j$ is removed from the set $L_{UL}$ associated with node  $n^{x}_i$ and $n^{x}_i$ is removed from the set $L_D$ associated with the node  $n^{y}_j$. The process is reversed if  $x \leq y$. The detection of partitions is very simple. At the node whose $L_{UL}$ is updated to reflect the connection loss. If $\#L_{UL} =0 \bigcap \#L_D=0$ (where $\#$ is the cardinality of the individual sets), then the unit corresponding to the node is system partitioned.  If $\#L_{UL} =0$, then the unit corresponding to the node is unit partitioned. 

%Referring to the connectivity graphs in Figure \ref{fig:connectivity-graph-2}, in the case of node \textbf{6}, the loss of the connection between clusters \textbf{a} and \textbf{b}, results in the cluster \textbf{b} being unit partitioned within \textbf{SC-1} though it is connected to \textbf{SC-3}. If however the only link that failed is the one connecting \textbf{SC-1} and \textbf{SC-3}, no units are partitioned. In the last case, if the link connecting  clusters \textbf{a} and \textbf{b} fails and the link connecting cluster \textbf{b} and super-cluster \textbf{SC-3} also fails, node \textbf{6} in Figure \ref{fig:connectivity-graph-2} concludes that cluster \textbf{b} has been system partitioned. 

%For nodes with $\#L_{UL} =0$ the cost associated with reaching the vertex node approaches $\infty$. All units which have their shortest path to the vertex resulting in a cost, which approaches $\infty$ are unit partitioned.


%\subsubsection{Ensuring progress in concurrent partitions}
%Concurrent partitions may contain clients which issue events and also other clients which are interested in those events. The interested clients should thus be able to receive events which are currently being issued within that partition. All these events would, of course, need to be stored onto a stable storage, for rerouting during partition mergers.

%\subsubsection{Partition Mergers}
%Each partition keeps track of the last events that were received by the gatekeepers in individual partitions. Based on this information appropriate events are routed. Of course prior to this we need to also account for the profile reconstruction since there could be clients which have initiated a roam. Similarly, events issued by clients, either during disconnected mode operations or broker node failures, and subsequently held in the client's local queue would be fed back in to the system.

\subsection{Stable Storage Issues}
Storages exist en route to destinations but decisions need to be made regarding when and where to store an event and also on the number of replications that we intend to have for any given event.
Events can be forwarded to clients only after they  have been written to stable storage.
The greater the number of stable storage hops en route to delivery to a client, the greater the latency in delivering the event to that client. 
In section \ref{sec:replication-granularity} we discuss the replication scheme for our system, and the process of adding stable storages within a sub-system. Section \ref{sec:need-for-epochs} describes the need for epochs, the assigning of epochs and the storage scheme for events. Section \ref{sec:guarnateed-delivery-of-events} describes the guaranteed delivery of events to all units within the subsystem. Finally in section \ref{sec:routing-events-to-reconnected-clients} we describe the recovery scheme for roaming clients or clients connecting back after a prolonged disconnect.

\subsubsection{Replication Granularity}
\label{sec:replication-granularity}
In our storage scheme, data can be replicated a few times, the exact number being proportional to the number of units within a super unit and also on the \emph{replication granularity} that exists within a specific unit. 
If there is a stable storage set up for servicing all the broker nodes within a level-$\ell$ unit, then we denote the replication granularity for nodes within that part of the sub system as $r_\ell$. 
%Thus if the replication strategy is one of replicating within every cluster in case of a 3-level system with $M$ units at each level, a certain event that would be received by all the clients within the system would be replicated $M \times M \times M$ times. Of course what we are considering here is the extreme case, but nevertheless, it is an exemple of how the replication strategy is a crucial element within the system. We also need a garbage collection scheme, which ensures that the storage space does not increase exponentially. 
Stable storages exist within the context of a certain unit, with
the possibility of multiple stable storages at different levels within the 
same unit.
We do not impose a homogeneous replication granularity throughout the system. 
Instead, we impose a constraint on the minimum replication scheme for the system. 
In an $N$-level system, comprising of level-$N$ units, we require that every node have a replication granularity of at least $r_N$. 
Thus, in a system comprising of super-super-clusters we require that every broker node within every super-super-cluster have a replication granularity of at least $r_3$. 
This is, of course, the coarsest grained replication scheme. There could be units present within the system that have a replication strategy, which is more finely grained. 
The other constraint, which we impose is that within a level-$\ell$ unit $u^\ell_i$ there can be only one stable storage at level $\ell$. 

\begin{figure}[ht]
\centering
\epsfig{file=images/Replication-StableStorageIssues.eps,width=334.00pt, height=323.00pt}
\caption{\label{fig:replication-strategy} The replication scheme}
\end{figure}

%The interaction between the stable storages of a unit and the stable storages within the sub units needs to address both the redundancy and garbage collection issues. Stable storages store events that the unit it is servicing, is interested in. This is ensured by the ERP, which would ensure the routing of only the \emph{interesting} events. The node which best serves this purpose is the gatekeeper node. As discussed earlier (section \ref{profile-propogation-protocol}) PPP ensures that a  gatekeeper $g^\ell_i(C^{\ell +1}_j)$ snapshots the profile of every level-$\ell-1$ unit within its level-$\ell$ GES context $C_i$. Thus, if we fix the replication granularity at $\ell$, one of the gatekeepers $g^\ell(C^{\ell+1}_j)$ within the GES context $C^{\ell+1}_j$ is responsible for the event storage. One of the advantages of this scheme is that we store only those events that we are interested in; since a unit gatekeeper is aware of the unit's profile.  

Figure \ref{fig:replication-strategy} depicts the different
replication granularities that can exist within different parts of a sub system. 
As can be seen super-super-cluster \textbf{SSC-B} has a replication granularity
$r_3$, while super-cluster \textbf{SC-4} within \textbf{SSC-B} has a replication granularity $r_2$. Cluster \textbf{l} has a replication granularity of
$r_1$. 
Also, in the depicted replication scheme there could be no other node in \textbf{SSC-B} that serves as a stable storage to provide the nodes in \textbf{SSC-B} with a replication granularity of $r_3$. 
Similarly, there could be no other stable storages, which try to service units \textbf{SC-4} and \textbf{SC-6} with a replication granularity of $r_2$. 
Table \ref{table:replication-granularity} lists the
replication granularities available at different nodes within the
sub system depicted in figure  \ref{fig:replication-strategy}.
\begin{table}[htp]
\centering
\begin{tabular}{|c|c|c|} \hline
Nodes             & Granularity $r_\ell$ & Servicing Storage   \\\hline
10,11,12          & $r_3$                & $\mathbf{1}$ \\ \hline
1,2,3,4,5,6,7,8,9 & $r_2$                & $\mathbf{9}$ \\ \hline
16,17,18,19,20,21 & $r_2$                & $\mathbf{19}$ \\ \hline
13,14,15          & $r_1$                & $\mathbf{14}$ \\ \hline
\end{tabular}
\caption{\label{table:replication-granularity} Replication granularity at different nodes within a sub system}
\end{table}

%Requirements (\ref{requirement:atleast-one-stable-storage}), (\ref{requirement:gatekeeper-stable-storage}) and (\ref{requirement:gatekeeper-stable-storage-number}) snapshot the various constraints that we impose on our replication strategy. 
%\begin{requirement} \label{requirement:atleast-one-stable-storage}In an $N$-level system, the replication granularity at each and every node in the system must be at least $r_N$. There should be at least one stable storage present in the system.\end{requirement}

%\begin{requirement}\label{requirement:gatekeeper-stable-storage}A level-$\ell$ stable storage can only be set up at a node, which serves as a level-$\ell$ gatekeeper. \end{requirement}

%\begin{requirement}\label{requirement:gatekeeper-stable-storage-number}For a level-$\ell$ unit $u^\ell_i$ only one of the gatekeepers $g^\ell$ can be configured as a level-$\ell$ stable store.  \end{requirement}

\subsubsection{Adding stable storages and updates of replication granularity}
When a stable storage is configured as a level-$\ell$ storage, we try to update the replication granularities associated with the nodes within the level-$\ell$ unit $u^\ell_i$ that the stable storage is a part of. For a node $x$ if the node's replication granularity is $r^x_m$, there are two possible outcomes. If $m > \ell$ the node's replication granularity is updated to $\ell$ i.e.  $r^x_\ell$. On the other hand if  $m < \ell$ the replication granularity for the node is left unchanged. 
Thus for example if the unit had a granularity of $r^x_3$ and $r_2$ has been added, the granularity is changed to $r^x_2$.  
A condition where $m = \ell$ is an error condition since it depicts the presence of multiple stable storages at the same level, a situation which should not arise because of the constraint that we have imposed. 
%Every node also keeps track of $r_\ell(min)$ and $r_\ell(next)$, which refer to the minimum replication granularity and the next highest one respectively. This comes into the picture during the guaranteed delivery of events, and is used to retrieve data from other stable storages when a finer grained store is added (discussed in section \ref{sec:adding-stable-storages-and-epochs}).  
%We then proceed to set up a stable storage at node \textbf{SSC-B.SC-6.m.18}. Further, we configure this stable store as a cluster storage with a replication granularity of 1. The replication granularities at nodes \textbf{16,17} and \textbf{18} are then updated to $r_1$ from $r_2$. 
In the topology in figure \ref{fig:replication-strategy} if we were to set up a level-2 stable storage at node \textbf{10}, the replication granularities at nodes \textbf{10,11} and \textbf{12} would be updated from $r_3$ to $r_2$. The replication granularity for every broker node in the cluster \textbf{SC-5.l} remains unchanged at $r_1$. 


\subsubsection{Stability }
\label{stable-storage:stability} 
For an $N$-level system  with a minimum replication granularity of $r_N$, the responsibility for ensuring stability of messages is delegated to the finer grained stable storages for those sub systems where the replication granularity is less than N. In the case of multiple stable storages, at different levels within a single super-unit, the stability requirements for individual nodes are delegated to the finest grained store servicing the corresponding node. Thus, in figure \ref{fig:replication-strategy}  stability for nodes \textbf{13,14} and \textbf{15} is handled by the cluster storage at node \textbf{14} while that for nodes \textbf{10,11} and \textbf{12} is handled by the level-3 stable storage at node \textbf{1}. 
%The presence of a stable storage within a unit at level-($\ell$ -1) for a system with replication granularity $\ell$ delegates the stability responsibilities for events within the level-($\ell$ -1) to the stable storage at the lower level. 
Every event in the system should be stable since we should be able to retrieve the event in case of failures or roams initiated by clients. 
%Stable storages need to wait for notifications prior to the garbage collection of events. 
%To aid in this process of garbage collection of events from stable storages, we make a small change to the way an event's destinations are computed at a storage node. 
When a node is hosting a stable storage with $r_\ell$, this node is responsible for computing destinations, comprising of units at level-($\ell$-1), within the level-$\ell$ unit that the node belongs to. 
Along with these destinations the node also computes the \emph{predicate count per destination} that are interested in receiving the event, this feature allows us to garbage collect events upon receiving acknowledgements. 
%The  allows us to garbage collect events upon receiving acknowledgements from the destinations associated with a given event. 
%The acknowledgements include the number of predicates that were serviced at the destinations and have system imposed dissemination constraints. 
%The destination associated with the acknowledgement is updated depending on the gateway that the destination is being transmitted over. 
%For acknowledgements issued by a broker node that has a replication granularity of $r_\ell$ the acknowledgement is never sent over a level-$\ell$ gateway $g^\ell$. Thus, acknowledgements to decrement the predicate count for a cluster storage should never be allowed to leave the cluster. 

If finer grained stable storages are present within the subsystem with $r_\ell$, the receipt notification is slightly different. As soon as the event is stored to the finer grained stable storage, this stable storage sends a notification to the coarser grained storage indicating the receipt of the event and also the predicate count that can be decremented for the sub-unit that this storage is servicing. Thus, in figure \ref{fig:replication-strategy}, when an event stored at node \textbf{1} is received at node \textbf{19}, we can assume that all nodes in unit \textbf{SC-6} can be serviced and decrement the reference counts at the level-3 stable storage at node \textbf{1} accordingly. 


\subsection{Epochs}
\label{sec:need-for-epochs}
%We digress here to discuss the need for \emph{epochs}.
%When a node is hosting a stable storage with $r_\ell$, the node is responsible for computing destinations at level-$(\ell-1)$ within the level-$\ell$ unit that the node belongs to. Along with these destinations the node also computes the number of predicates per destination that are interested in receiving the event. The predicate count per destination allows us to garbage collect an event upon receiving acknowledgements from the event's destinations. 
%Consider the following scenario where the predicate count equals the client count for the destination associated with an event. Unit $s_A$ has a total of 156 clients attached to it and unit $s_A$ fails. Clients which detect this failure would initiate a roam. Local queues could be constructed for each client that has initiated a roam in response to this failure. For each queue constructed and sent across the system to its new hosting unit, the reference count associated with every event contained within the queue is decremented by one. However, it is conceivable that a client could have been attached to $s_A$ and that this client had joined the system for the first time prior to the unit's failure. This client is thus \emph{not} the intended recipient of any of the local queues that would be constructed in response to the servicing of roaming  clients. If this client is one of the first clients to initiate a roam, local queues would be constructed for it and the reference counts of the events contained within this local queue would be decremented by one. This operation would lead to the \emph{starvation} of at least one client, if any of the 156 clients contained a profile which partially matched that of the new client. 

%The second scenario is for a client $c_A$, which has received events $e_0 \cdots e_{25}$ in its incarnations (past and present) prior to a disconnect in its present incarnation. During the time that $c_A$ was disconnected the only event targeted to it was $e_{26}$. When $c_A$ reconnects back the only event that should be routed to it should be $e_{26}$ and not the events that it has already received in its previous incarnations.  

Epochs are used to aid the reconnected clients and also to recover from failures. We use epochs to ensure that the recovery queues constructed for clients would not comprise of events that a client was not originally interested in. Failure to ensure this could lead to starvation of some of the clients. We also need epochs to provide us with a precise indication of the time from which point on a client should receive events. Not having this precise indication (during recoveries) leads to client starvations and also would also cause the system to expend precious network cycles in routing these events. We also have an epoch associated with every profile change and require that the client to waits till it receives the epoch notification, before it can disconnect from the system.

%The reason why we can not delegate the event queue generation scheme to the individual units is that a unit can fail and remain failed forever. It is best that the event queue generation is handled by the system as there could be stable storages that could be added within the system and the storage could be delegated to the stable storages that exist at different levels within the same GES context.

\subsubsection{Epoch generation}
Epochs, denoted $\xi$, are truly determined by the replication granularities that exist in different parts of the system. 
%In the case of a client, it is the GES context of the broker node that the client is attached to, which determines the epoch. A client could be operating in disconnected mode.  Such a client is nevertheless still serviced based on its profile, the destination for delivery being the node or unit (in case the node fails) at which the client was last present. This profile along with its last logical address serves as a proxy for the client in its absence. 
Some of the details pertaining to epoch generation are listed below --
\begin{itemize}
\item[(a)] Epochs should monotonically increase.
\item[(b)] Epochs for clients exist within the context of the finest grained stable storage that the broker node (that it is attached to) is a part of. 
Thus, if the broker node has a replication granularity of $r_2$, valid epochs for events received by the client, would be those that have been assigned by the corresponding level-2 storage.  
\item[(c)] For every client with a profile $\omega$ there is a epoch $\xi^{\omega}$ associated with it.
%\subitem The fact that there is only one epoch associated with every $\omega$,  follows from property (b) and also from the constraint that there can be only  one stable storage configured for servicing a unit $u^\ell_i$ with a granularity of $r_\ell$.
\end{itemize}

%\begin{requirement}
%\label{axiom:events-and-epochs}
%A persistent client will not receive an event $e$ unless there is an epoch, $\xi_e$, associated with the event. Also, this epoch should be assigned by the stable storage servicing the broker node that this client is attached to. 
%\end{requirement}
For a profile $\omega$ associated with a client, we denote individual profile predicates  as $\delta\omega$. Events are routed to a client based on the
$\delta\omega$ that exist within a profile $\omega$. 
%For every $\delta\omega$ associated with a profile $\omega$ there is a epoch $\xi^{\delta\omega}$ associated with it. 
However, every event received at a client needs to have an epoch associated with it to aid in the recovery from failures and also to service events that have not been received by the client. 
The arrival of such an event results in an update of the corresponding epoch associated with the client's profile. 
%Profile changes initiated by a client also have an epoch associated with it. 
%This is discussed in a later section. The reason why we do not need an epoch for every $\delta\omega$ is that the epochs are assigned to a client by the stable store, and irrespective of the addition of stable stores at different levels these epochs monotonically advance, and the reception of an epoch easily allows us to conjecture about the events that should be (or have been) received.
The replication granularity within the system could be different in different sub systems. Within a subsystem having a replication granularity $r_\ell$, it is possible that there are subsystems  with replication granularity $r_{\ell-1},
r_{\ell-2}, \cdots, r_0$. In such cases the epochs assigning process is delegated to the corresponding replicators. 
%If a node within $u^\ell_i$ has a granularity of $r_\ell$, it needs to await the receipt of an epoch assigned by the level-$\ell$ storage at $u^\ell_i$, before forwarding this event to the relevant clients attached to this node. Thus the epoch associated with the same event could be different at different clients in the system. In figure \ref{fig:replication-strategy} it is possible that by the time an event arrives at node \textbf{15} there will be two different epochs associated with it, only one of which is valid for clients attached to node \textbf{15}. Also epochs associated with the same event could be the same at different parts in the system. Thus clients attached to any of the nodes in \textbf{SC-6}, in figure \ref{fig:replication-strategy}, have the same epochs (assigned by the store at node \textbf{19}) for events that they would all receive.  



\subsubsection{The storage format}
\label{sec:storage-format}
When an event is written to a stable storage, there is an epoch number associated with it. Since all events are not routed to all destinations we maintain the destinations associated with the event. For each destination we also store the predicate count associated with that destination. 
%Besides the destinations associated with the event, the matching operation at the stable storage nodes also return the predicate count associated with the event. This information is used to service roaming clients or clients rejoining after a prolonged disconnect. Using this information these reconnecting clients are still able to ensure that the events they consume can be scheduled for garbage collection from the stable storages, once the reference count for these consumed events reduces to zero. 
We also maintain information pertaining to the \emph{type} of the event and the length of the serialized representation of the event. Finally we maintain the serialized representation of the event. Thus, the storage format is the tuple  --  $< \xi^e, (d^e_0,d^e_1, \cdots, d^e_n), (p^e_0,p^e_1, \cdots, p^e_n), e.type, e.length, e.serialize >$. 


%\subsubsection{Epochs and profile changes}
%Whenever a profile change is made, there needs to be an epoch associated with the profile change. The epoch, assigned by a replicator, depending on the subsystems granularity is an indicator of the time from which point on, that change  would be serviced by the system. If an epoch is not associated with profile changes, it is conceivable that starvation of some client would occur. Consider the following scenario, a  client receives a sequence of events $e_1, e_2, \cdots , e_n$. For an extended duration this client does not receive any events. The last epoch that it received was $\xi_n$. This client then proceeds to make a profile change, and leaves the system. When the client rejoins the system at a later time, this client would expect to receive all the events that it missed. This set of missed events would include events, which satisfy the profile change the client last made, starting with its last known epoch $\xi_n$. We thus have an epoch associated with every profile change and require that the client to waits till it receives the epoch notification, before it can disconnect from the system.

\subsubsection{Epochs and the addition of stable storages}
\label{sec:adding-stable-storages-and-epochs}
In this section we describe the process of adding  stable storages. Consider a scenario where a new store is being added within a unit $u^n_i$. The present replication granularity of this unit is $r_m$ and the new storage for this unit is at level-$n$. The addition of a stable storage at level $n$ is disseminated only within the unit $u^n_i$ that the hosting node belongs to. 

If $n < m$, the new \emph{finer grained} stable storage should access the \emph{coarser grained} storage with $r_m$ and retrieve the events, which were meant to be disseminated within the unit $u^n_i$. 
The predicate count associated with the destinations for each individual event needs to be updated accordingly to reflect the predicate counts associated with the sub-units in $u^n_i$. 
The epochs associated with these \emph{retrieved} events should however remain unchanged. 
This is especially crucial since there are clients, attached to (disconnected from) broker nodes in the unit $u^n_i$, which have epoch numbers associated with their profiles based on the ones assigned by storage hosting $r_m$. 
The epochs associated with the client profiles should remain consistent even if a new stable storage is added. 
Once this event retrieval process is complete, the newly added stable storage is ready to assign epoch numbers to the events. 
%The first event that this newly added storage is ready to store, after the retrieval process is complete, is the epoch number from which point on the epoch numbers assigned by the old store $r_m$ and the new store $r_n$ can deviate. If $n > m$ at any of the sub-units within $u^n_i$ the replication granularity for nodes in those sub-units will not be updated.


%\begin{figure}[ht]
%\centering
%\epsfig{file=images/AddingStableStorage.eps,width=395.00pt, height=323.00pt}
%\caption{\label{fig:adding-stablestorage} Adding stable stores}
%\end{figure}

%Figure \ref{fig:adding-stablestorage} depicts the replication scheme that exists in different parts of the system and also the addition of a new super-cluster storage at node \textbf{10}. Prior to the addition of the stable storage at node \textbf{10}, the replication granularity at the broker nodes in cluster \textbf{k} is $r_3$ while that in cluster \textbf{l} is $r_1$.  When the new level-2 storage is added, this information is disseminated only within the super-cluster \textbf{SC-5}. After the dissemination of the storage information, node \textbf{10} needs to communicate with the $r_3$ storage at node \textbf{1} and retrieve the events that have \textbf{SC-5} in the destination lists. These \emph{retrieved} events when they are stored at node \textbf{10} have their epoch numbers unchanged. This is because clients attached to the nodes in cluster \textbf{k} have their profile epochs updated by the events with epochs assigned by the $r_3$ storage at \textbf{1}.  These epochs thus need to be consistent across the addition of stable storages. Let us say that the epoch associated with the last such retrieved event by node \textbf{10} is $\xi^{node(1)}_{250}$, the next time an event arrives at node \textbf{10} the epoch assigned to that event would be $\xi^{node(10)}_{251}$. Thus clients which were attached to cluster \textbf{SC-5.k} in an earlier incarnation, when they reconnect back, can use their epoch to recover completely.  For the stable storage at node \textbf{10} the destination list and the corresponding predicate counts associated with the \emph{retrieved} and \emph{new} events should be with respect to the clusters \textbf{k} and \textbf{l}. 

%The addition of the level-2 stable storage at node \textbf{10} has no effect on the replication granularity of broker nodes in cluster \textbf{SC-5.l} since for the broker nodes in this cluster, the storage that is added is a coarser grained stable storage. 



\subsubsection{System storage and guaranteed delivery of events}
\label{sec:guarnateed-delivery-of-events}
For a level-$N$ system, the stable storages servicing the individual level-$N$ units are also designated as \emph{system storages}. 
%Figure \ref{fig:system-storages} depicts a system comprising of 4 super-super-clusters and the replication schemes that exist in different parts of the system. 
For events issued by clients attached to nodes within these $u^N$ units,  these system storage nodes have the additional responsibility that they maintain events in stable storage till such time that they are sure that all the  other $u^N$ units within the system have received that event.  
When an event is issued within a super unit $u^N_i$, the destinations are computed as described in the event routing protocol. 
However, before the event is allowed to leave unit $u^N_i$, it must be stored onto the stable storage that provides nodes in $u^N_i$ with the minimum replication granularity of $r_N$. 
%Thus, in figure \ref{fig:system-storages}, for an event issued by a client attached to a node in \textbf{SSC-B}, that event must be stored to the system storage in \textbf{SSC-B.SC-4.h} before it can be routed to units \textbf{SSC-A,SSC-C} and \textbf{SSC-D}.

%\begin{figure}[ht]
%\centering
%\epsfig{file=images/SystemStorages.eps,width=427.00pt, height=386.00pt}
%\caption{\label{fig:system-storages} Systems storages and the guaranteed delivery of events}
%\end{figure}
 
The system storage node maintains the list of all known $u^N$ destinations within the system. This destination list is associated with every event that is stored by the system storage. 
Associated with these events is a \emph{sequence number}, which is different from the epoch number associated with the events that clients receive. 
Further, sequence numbers associated with events are used \emph{only} by the system storages to conjecture the events that they should have received from any other system storage within the system. These sequence numbers are \emph{not} used by the clients or the broker nodes within the system to detect missing events.  
Once the event is stored to such a system storage, it is ready to be sent across to the other $u^N$ destinations within the system.  Also, for an event that is issued by a client within  $u^N_i$, the event is stored to stable storage (to ensure routing to other $u^N$ units within the system) within $u^N_i$ and not at any other system storages at the other $u^N$ units within the system. 
When the events are being sent across gateway $g^N$ for dissemination to other $u^N$ units, every event has a sequence number associated with it and also the unit $u^N_i$ in which this event was issued. 
This is useful since the $r_N$ replicators (which serve as system storages) in other units can know which unit to send the acknowledgements (either positive or negative)  to. 
%Thus for an event $e$ issued by a client in \textbf{SSC-B} what we store is -- \texttt{<seqNumber, e, (SSC-A, SSC-C,SSC-D)>}.
%Every system storage also keeps track of the last sequence number that was received from a certain unit $u^N_i$.  Thus the system storage in \textbf{SSC-B} would keep track of the last received sequence numbers for events published by clients in \textbf{SSC-A,SSC-C} and \textbf{SSC-D}. Each system storage can now keep track of the events that it should receive from a certain unit $u^N_i$. Consider the case where the system store node at \textbf{SSC-A} has received events with sequence numbers $s^B_1,s^B_2, \cdots, s^B_{100}$ from unit \textbf{SSC-B}. When this system store  receives an event with sequence number $s^B_{103}$ from \textbf{SSC-B}, based on the last sequence number that it received, $s^B_{100}$, it knows that it has missed events with sequence numbers $s^B_{101}, s^B_{102}$. This system storage node then issues a NAK to retrieve those events. When the system storage at \textbf{SSC-B} receives this NAK($s^B_{101}, s^B_{102}$), it reissues those events to the requesting system store. The system storage at \textbf{SSC-A} does not assign an epoch (the system storage can assign an epoch because it is also a level-$N$ storage) or route the event with sequence number $s^B_{103}$ till such time that it receives events with $s^B_{101}, s^B_{102}$ from \textbf{SSC-B}. If an event with sequence number $s^B_{98}$ (assigned by \textbf{SSC-B}) is received, the system storage at \textbf{SSC-A} discards this event since it knows that it has already processed this event just as it has processed events with sequence numbers $s^B_{97}, s^B_{96}, \cdots s^B_0$. 

%Upon receipt of an event with an associated sequence number, other system storages issue an \emph{ACK(seqNum)} to facilitate garbage collection. The fact that sequence numbers assigned by any system storage increase monotonically allows us to sustain the loss of acknowledgement messages. This is because the receipt of an acknowledgement for an event $e$ (stored with sequence number n) \emph{ACK(n)} implies the receipt of events with sequence numbers $n, n-1, n-2, \cdots $ issued by the system store. The receipt of an \emph{ACK} from a certain unit (issued by the unit's system storage) results in that unit being removed from the destination lists  associated with events with sequence numbers $n, n-1, n-2, \cdots $. When the destinations associated with an event is reduced to zero the event is garbage collected. Thus, in our example, for events $e_1, e_2, \cdots, e_n$ issued by clients in \textbf{SSC-B} and stored to the system store at  \textbf{SSC-B.SC-4.h} with sequence numbers  $s^B_1,s^B_2, \cdots, s^B_n$ the destination list associated with every event stored by the system storage comprises of  \textbf{SSC-A,SSC-C} and \textbf{SSC-D}. The receipt of an \emph{ACK(n)} from \textbf{SSC-A} results in the removal of \textbf{SSC-A} from the destination lists associated with the stored events with sequence numbers $s^B_1,s^B_2, \cdots, s^B_n$. When the destination list associated with any of these stored events is reduced to zero that event is garbage collected. 



%\subsubsection{The upward propagation of events}
%When an event is issued by a client attached to a broker node, other clients interested in that event do not receive the event till such time that there is an epoch associated with the event.  This epoch is dependent on the replication granularity that exists at the corresponding broker nodes. The epoch that is associated with an event should be the epoch that is assigned by the servicing storage for the broker node in question. Events with epochs assigned by replicators $r_\ell$ are valid only within the unit $u^\ell_i$ that the node belongs to. When an event is issued by a client, the reissue behavior ensures that the event is stored onto a stable storage. If this stable storage is not the system storage (responsible for $r_N$), the stable storage node is responsible for storing this event and not scheduling it for garbage collection, till such time that it receives a notification from the system storage regarding the receipt of that event. Besides this, for an $N$-level system, the event is not allowed to leave the unit $u^N$ till such time that there is a sequence number (assigned by the system storage) associated with it. We use figure \ref{fig:replication-strategy-event-propagation} to explain the routing of events to persistent clients.  
%\begin{figure}[ht]
%\centering
%\epsfig{file=images/Replication-StableStorageIssues-propagationOfEvents.eps,width=330.00pt, height=315.00pt}
%\caption{\label{fig:replication-strategy-event-propagation} The propagation of events in the system}
%\end{figure}

%For an event $e$,  issued by a client attached to node  \textbf{SSC-B.SC-5.l.15}, the client reissue behavior ensures that the event is stored to the cluster storage at node \textbf{14}. Even if the reference count associated with this event is reduced to zero, the cluster storage cannot garbage collect this event till such time that the event is stored to the system storage at node \textbf{1}. It is now the responsibility of node \textbf{14}, to ensure the storage of  the event $e$ to the system storage and also to initiate corresponding reissues to ensure the same. This event is not allowed to leave \textbf{SSC-B}, though a gateway does exist at the super-cluster \textbf{SSC-B.SC-5}. Once the event is stored to the system storage at node \textbf{1}, it is allowed to leave the super-super-cluster.  
%This indication is stored in the event via the sequence number assigned by the system storage. An given event will be stored only at one system storage. The system storage at node \textbf{1} should also send a notification to node \textbf{14} indicating that the event was stored to the system storage. The event $e$ is then replicated at node \textbf{9} and \textbf{19}, if there are any clients attached to super-clusters \textbf{SC-4} and \textbf{SC-6} respectively. This event which is being disseminated in say \textbf{SC-6}, would have 2 epochs associated with it --- one assigned by the level-3 store at node \textbf{1} and the other assigned by the level-2 store at node \textbf{19}, besides the sequence number assigned by the system storage at node  \textbf{1}. 

\subsubsection{Stable storage failures}
When a stable storage node fails, the events that it stored would not be available to the system.
A new client trying to retrieve its events is prevented from doing so.
The stable storage also misses any garbage collect notifications that were intended for it. We require this stable storage to recover within a finite amount of time.  
\begin{requirement}
\label{requirement:stable-storages-failing}
A stable storage cannot remain failed forever, and must recover within a finite amount of time.
\end{requirement}


%\subsubsection{Handling events for a disconnected client}
%\label{sec:routing-events:disconnected-client}
%This problem pertains to one of the most important issues that needs to be addressed by our system. A client node has intermittent connection semantics, and is allowed to leave the system for prolonged durations and still expect to receive all the events that it \emph{missed} in the interim, along with real time events.Events are routed based on a clients persistent profile and the persistent profile is what would be stored at the last broker node that it was connected to. The broker node also has a persistent profile which is the sum of the profiles of all the client nodes that are attached to it and all the disconnected clients which were last attached to it. The persistent profile of the broker node is itself stored at the cluster gatekeeper. Consistency issues pertaining to out of order delivery of real time events and recovery events aside, our solution to this problem is to delegate this responsibility to the broker node that the client was attached to prior to a disconnect/leave.



\subsubsection{Routing events to a reconnected client}
\label{sec:routing-events-to-reconnected-clients}
When a client is not present in the system, the event is not acknowledged and thus can not be garbage collected by the replicator that this client was being serviced by. The events are thus available for the construction of recovery queues when the client connects back into the system. 
The recovering client in question could be both a roaming client or a client which has reconnected after a prolonged disconnect. 
Associated with every client is the epoch number associated with the last event that it received or the last profile change initiated by the client.  
The routing for the client is based on the node that the client was last attached to. 
It is this node that serves as a proxy for the client. 
If this node fails it is the cluster gateway, of the cluster that the node belonged to, which serves as a proxy for the client. 
As mentioned earlier, in our system a node/unit can fail and remain failed forever.

%One of the disadvantages of having a client keep track of the servicing stable storages is that when the client is operating in the disconnected mode, there could be other stable storages which are servicing the unit to which the client was last connected. However, the client is not aware of this new stable storage and could possibly lose events which it was supposed to receive.

Stable storages at  higher levels are aware of the finer grained replication schemes that exist within its unit. 
If a coarser grained stable storage is servicing the broker the client was last attached to,  the system would use the higher level stable storage to retrieve the client's interim events. Otherwise the system would delegate this retrieval process to the finer grained stable storage, which now services the client's lower level GES context.

%\begin{conjecture}
%\label{conjecture:locating-stable-storages}
%A client's logical address provides the system with the  stable storage that should be used for the construction of queues containing events that were missed by the client.
%\end{conjecture}

%It is possible that this stable storage is unavailable during a subsequent client reconnect and construction of event queues. From Requirement \ref{requirement:stable-storages-failing} it is clear that these storages would recover within some finite amount of time. During such a recovery the system should be able to reconstruct the event queues which it failed to create and route these event queues to the client.  This requires that --
%\begin{itemize}
%\item[(a)] The unit keeps track of all the requests for event queue construction that it failed to service.
%\item[(b)] Unserviced clients notify the unit about its location, every time it issues a roam.
%\end{itemize}

For a profile $\omega$ associated with a client, when a disconnected client joins the system it presents the node the it connects to in its present incarnation the following --
\begin{itemize}
\item[(a)] The logical address of the broker node that this client was attached to in its previous incarnation.
\item[(b)] The last epoch $\xi$ received from the replicator within the replication granularity $r_\ell$ of the sub system that it was formerly attached to. 
%	\subitem The replication granularity of the sub-system that the client was formerly attached to would have changed. The client however need not deal with this. The process of adding finer/coarse grained stable storages ensures that the epoch associated with the client is sufficient to complete a full recovery.  
\item[(c)] A list of the profile ID's associated with client's profile $\omega$.
\end{itemize}

Item (a) provides us with the stable storage that has stored events for the client. Item (b) provides us with the precise instant of time from which point on, event queues of events needs to be constructed and routed to the client's new location. Item (c) provides for the precise recovery of the disconnected client .  Details regarding the precise recovery mechanism can be found in \cite{ges, ges-framework}.

%\subsubsection{Locating the stable store }
%When the client reconnects back into the system, based on the logical address of the broker node that this client was last connected to, the stable storage responsible for assigning epochs to clients attached to that particular broker node is located. This works as follows, the request is first forwarded to the broker node that the client was last attached to. Based on the replication granularity $r_\ell$ currently available at the node, the recovery request is forwarded to level-$\ell$ servicing storage. The replication granularity of the sub-system that the client was formerly attached could have been greater than $r_\ell$. However the process of adding stable storages accounts for the fact that the epochs would be consistent for recovery. In the event that the broker node is down, the information could be retrieved from the cluster gateway for the cluster that the node is a part of. Since unlike a broker node without a stable store, a storage node is not allowed to remain failed forever -- the servicing storage will always be retrieved. 

%\subsubsection{The epochs used in the recovery process}
%Let $\xi_n$ be the last epoch contained in the event routed to a client in its last incarnation. As discussed in section \ref{sec:storage-format} an event $e$ is stored in the following format -- $< \xi^e, (d^e_0,d^e_1, \cdots, d^e_n),$ $ (p^e_0,p^e_1,$ $ \cdots, p^e_n),$ $e.type,$ $e.length,$ $e.serialize >$.  For any recovering client we first locate the client's servicing storage, based on the logical address of the broker node that this client was attached to in its previous incarnation. Among the events that have been stored at this stable storage, what we are interested in, are those events, which have an epoch greater than  $\xi_n$.  We then compute the second epoch $\xi_m$ associated with the recovery request. This is the epoch at the \emph{located} stable storage when the recovery request was received. This epoch indicates the point from which point on queues need not be constructed, since the real time events would be routed to the client by the sub-system that it is presently attached to. The set of events at the located stable storage, which form the preliminary set of events to be considered for recovery, are events with epochs greater than  $\xi_n$ and less than or equal to $\xi_m$. The number of events in this preliminary set would be less than or equal to $(m-n)$ since some of these intermediate events could have been garbage collected. 
 
%Within this set of events, the events that could potentially be routed to the client are those for which the unit that the broker node (the client was attached to in its previous incarnation) is a part of, is one of the destinations. This operation of computing potential recovery events based on the epochs and the destination list can be performed by a simple filter at the located stable storage. 

%\subsubsection{Profile ID's and the recovery events}
%The individual profile predicates $\delta\omega$ corresponding to the profile ID's are marked for removal from the profile graph, at the subsystem the client was attached to in its previous incarnation.  Using the profile-ID's we can compute the events that need to be received by the client; within the set of recovery events computed in the earlier section. This is very important since with the set computed in the previous section, in general, the number of events that should not be received at the recovering client far exceeds the number of events that should be received by the client.   The stable store then proceeds to propagate this removal of the profile ID's both to higher level gatekeepers just as in the profile propagation protocol and also to the lower level gatekeepers down to the broker node which last hosted this client. 

%When the client issues an event recovery process, the logical address of the client is changed to its present address. The recovery events each have a destination list which is internal to the event. This destination list comprises of a single entry -- the logical address of the broker node that the client is now attached to.   These recovery events are now managed by the stable storage servicing the broker node that the client is now attached to. This stable storage is responsible for issuing acknowledgements in response to the receipt of the recovery events. Upon receipt of acknowledgements from the new storage, the corresponding predicate count associated with the event at the old storage is decremented by one. If this count is reduced to zero, the destination is removed from the destination list. When this destination list is reduced to zero, the event is garbage collected at the old stable storage. Upon receiving every such recovery event, the epoch associated with the client's profile is advanced to the epoch contained in the latest recovery event that the client received. The epoch in this case would be assigned by the stable store that is presently servicing the client.  The profile predicates associated with the client's profile is propagated using the profile propagation protocol. 

% The client could once again roam while these events are being routed to its present logical address. In this case the broker node that the client was attached to prior to the roam,  is now responsible for ensuring that the client does not loose any events that it is interested in.  In case of client roam or storage failures during reconnection there is another epoch that is associated with the client. This pertains to the time from which point on events \emph{need not} be routed. Of course every recovery of a failed stable storage is a new epoch, and for clients which could not be serviced during the time the storage had failed, this is the epoch from which no events should be used in the construction of local queues.


%\subsubsection{Advantages of this scheme}
%This scheme ensures that any given event is received by all the persistent clients that had expressed an interest in it. The scheme withstands the failure of nodes/units, with all nodes within these units remaining failed forever. The only constraint that is imposed is that the stable storage not remain failed for ever and that it recover within  a finite amount of time. In the case of stable storage failures, the higher level stable storage would have stored events destined for the unit that was being serviced by the now failed stable storage. This higher level stable storage then release these stored events to the recovering lower level stable storage when it recovers after the failure. The scheme allows us to respond to node failures (associated link failures), gateway failures and network partitions. When partitions heal the units exchange data destined for each half of the partition. This scheme supports the roaming of clients and also accounts for the garbage collection of events stored onto stable storage in response to clients initiating roam and constructing local queues to receive missed events. 

%\subsection{Summary}
%In this section we presented our replication scheme, and outlined how different nodes within a given super-unit could be served by different stable storages. We laid down the restrictions imposed on the number/location of stable storages within the system. We outlined some of the common failure scenarios involved, and presented a scheme for the detection of partitions that these failure scenarios lead to. We then introduced the concept of epochs and discussed issues which demonstrate its usefulness. Combining the replication scheme and the concept of epochs, and adding the notion of  system storages we arrived at our scheme for guaranteed delivery. We then describe our scheme for handling messages for disconnected clients and also for clients reconnecting back into the system.  






\section{Results}
\label{results}
In this section we present results pertaining to the performance of the Grid Event Service (GES) protocols. We first proceed with outlining our experimental setups. We use two different topologies with different clustering coefficients. The factors that we measure include latencies in the delivery of events and variance in the latencies. We measure these factors under varying publish rates, event sizes, event disseminations and system connectivity. We intend to highlight the benefits of our routing protocols and how these protocols perform under the varying system conditions, which were listed earlier. 

\subsection{Experimental Setup}
The system comprises of 22 broker node processes organized into the topology shown in the Figure \ref{fig:testing-topology}. This set up is used so that the effects of queuing delays at higher publish rates, event sizes and matching rates are magnified. 

Each broker node process is hosted on 1 physical Sun SPARC Ultra-5 machine  (128 MB RAM, 333 MHz), with no SPARC Ultra-5 machine hosting two or more broker node processes. 
For the purpose of gathering performance numbers we have one publisher in the system and one \emph{measuring subscriber} (the client where we do our measurements). 
The publisher and the \emph{measuring} subscriber reside on the same SPARC Ultra-5 machine and are attached to nodes \textbf{22} and \textbf{10} respectively in the topology outlined in figure \ref{fig:testing-topology}.  In addition to this there are 100 subscribing client processes, with 5 client processes attached to every other broker node (nodes \textbf{22} and \textbf{10} do not have any other clients besides the publisher and measuring subscriber respectively) within the system. The 100 client node processes all reside on a SPARC Ultra-60 (512 MB RAM, 360 MHz) machine. 
The publisher is responsible for issuing events, while the subscribers are responsible for registering their interest in receiving events. The run-time environment for all the broker node and client processes is Solaris JVM (JDK 1.2.1, native threads, JIT). 

\begin{figure}[ht]
\centering
\epsfig{file=images/Testing-Topology.eps,width=347.00pt, height=346.00pt}
\caption{\label{fig:testing-topology} Testing Topology - (I) }
\end{figure}
 

\subsection{Factors to be measured}
Once the publisher starts issuing events the factor that we are most interested in is the \emph{latency} in the reception of events. This latency corresponds to the response times experienced at each of the clients. 
We measure the latencies at the client under varying conditions of \emph{publish rates}, \emph{event sizes} and \emph{matching rates}. 
Publish rate corresponds to the rate at which events are being issued by the publisher. 
Event size corresponds to the size of the individual events being published by the publisher.  
Matching rate is the percentage of events that are actually supposed to be
receieved at a client. In most publish subscribe systems, at any given time for a certain number of events being present in the system, any given client is generally interested in a very small subset of these events. Varying the matching rates allows us to simulate such a scenario, and perform measurements under conditions of varying selectivity.  
For a sample of events received at a client we calculate the \emph{mean latency} for the sample of received events and the \emph{variance} in the sample of these events.
%and the \emph{system throughput} measured in terms of the number of events received per second at the measuring subscriber.We also measure the highest and lowest event latencies within the sample of events that have been received. 
Another very important factor that needs to be measured is the change in latencies as the connectivity between the nodes in a broker network is increased. This increase in connectivity has the effect of reducing the number of broker hops that an event has to take prior to being received at a client. The effects of change in latencies with decreasing broker hops is discussed in section \ref{sec:pathlengths-and-latencies}. 

\subsubsection{Measuring the factors}
For events published by the publisher the number of tag-value pairs contained in every event is 6, with the matching being determined by varying the value contained in the fourth tag. 
The profile for all the clients in the system, thus have their first 3 $<$\emph{tag=value}$>$ pairs identical to the first 3 pairs contained in every published event. 
This scheme also ensures that for every event for which destinations are being computed there is some amount of processing being done. 
Clients attached to different broker nodes specify an interest in the type of events that they are interested in. This matching rate is controlled by the publisher, which publishes events with different footprints. 
Since we are aware of the footprints for the events published by the publisher, we can accordingly specify profiles, which will allow us to control the dissemination within the system. When we vary the matching rate we are varying the percentage of events published by the publisher that are actually being received by clients within the system. 
%Thus, when we say that the matching rate is set at 50\%, any given subscribing client within the system  will receive only 50\% of the events published by the publisher. 
%To vary the publish rates, we control the \emph{sleep time} associated with the publisher thread, and also the number of events that it publishes at a time, once the publisher thread \emph{wakes up}. This requires some preliminary tuning. Once the values for the \emph{sleep time} and the number of events that are published at a time have been fixed (for the publisher and the broker node that it is attached to), we proceed to compute the real publish rates for the sample of events that we send. This is the publish rate that we report in our results. 

For each matching rate we vary the size of the events from 30 to 500 bytes, and vary the publish rates at the publisher from 1 Event/Sec to around 1000 Events/second.  For each of these cases we measure the latencies in the reception of events. 
To compute latencies we have the publishing client and the \emph{measuring} subscriber residing on the same machine. 
Events issued by the publisher are \emph{timestamped} and when they are received at the subscribing client the difference between the \emph{present time} and the \emph{timestamp} contained  in the received event constitutes the latency in the dissemination of the event at the subscriber via the broker network. 
%In case the publisher and the subscriber are on two different machines, with   access to different underlying system clocks, we would need to synchronize the clocks and also account for the drift in clock rates prior to computing the latencies in event reception. 
Having the publisher and one of the subscribers on the same physical machine with access to the same underlying clock, obviates the need for clock synchronization and also accounts for clock drifts. 
%It should be noted that though the publisher and the \emph{measuring} subscriber are on the same machine, they are connected to two different broker nodes within the broker network, as depicted in figure \ref{fig:testing-topology}. In fact it takes 9 broker hops for an event issued by the publisher to be received at the measuring subscriber.  

\subsection{Discussion of Results}
In this section we discuss the latencies gathered for varying values of publish rates, event sizes and matching rates. We then proceed to include a small discussion on system throughputs at the clients. We also discuss the trends in the variance of the latencies, associated with the sample of  events received at a client. The results also discuss the latencies involved in the delivery of events to persistent clients in units with different replication schemes. The delays are in the range of 1-2 mSec for every broker hop. Implementing certain sections of the networking code in C/C++ and then employing JNI to provide access to these native routines could improve upon this delay significantly. The only disadvantage that would result is that we would need to compile programs separately for different platforms. However the brokers written in Java and Java/JNI could still continue to inter-operate with each other. Thus users can decide which parts of the broker network would run the JNI-optimized implementation and which would not. 
 
\subsubsection{Latencies for the routing of events to clients}
At high publish rates and increasing event sizes, the effects of queuing delays come into the picture. This queuing delay is a result of the events being added to the queue faster than they can be processed. 
In general, the mean latency associated with the delivery of events to a client is directly proportional to the size of the events and the rate at which these events were published. The latencies are the lowest for smaller events issued at low publish rates. The mean latency is further influenced by the matching rates for events issued by the publisher.  The results clearly demonstrate the effects of flooding/queuing that take place at high publish rates and high event sizes and high matching rates at a client.  
It is clear that as the matching rate reduces the latencies involved also reduce, this effect is more pronounced for cases involving events of a larger size at higher publish rates. 



Figure \ref{fig:22servers102clients-Latency100-50-10} 
%and  \ref{fig:22servers102clients-Latency100-33-3}, 
depicts the pattern of decreasing latencies with decreasing matching rates. Table \ref{table:latency-variance} outlines the reduction in latencies and the variance associated with the latencies corresponding to the sample of events received at a client. 
%The latencies vary from  391.85 \emph{mSecs} to 52.0 \emph{mSecs}, with the \emph{$<$publish rate, event size$>$} varying from $<$952 \emph{events/Sec} , 450 \emph{Bytes}$>$ for a matching rate of 100\% to $<$952 \emph{events/Sec}, 400 \emph{Bytes}$>$ for a matching rate of 10\%. 
This reduction in the latencies for decreasing matching rates, is a result of the routing algorithms that we have in place. 
These routing algorithms ensure that events are routed only to those parts of the system where there are clients, which are interested in the receipt of those events. 
%The routing algorithms are very selective about the links that are employed for event dissemination. 
%Thus, events are queued only at those broker nodes which -- 
%\begin{itemize}
%\item Have attached clients interested in those events 
%\item Are en route to broker nodes which are interested in these events. These broker nodes generally fall in the shortest path to reach the destination node.
%\end{itemize}
In the flooding approach, all events would still have been routed to all clients irrespective of the matching rates. 
Additional results for latencies at different matching rates, system throughput measurements and changes in variance in the latency samples can be found in \cite{ges, ges-results, ges-framework}.

\begin{figure}[htp]
\centering
\psfig{file=images/results/22servers102clientsMatchRates100-50-10.ps, width=6.0in, height=4.1in}
\caption{\label{fig:22servers102clients-Latency100-50-10} Latencies for match rates of 100\%, 50\% and 10\%}
\end{figure}

%\begin{figure}[htp]
%\centering
%\psfig{file=images/results/22servers102clientsMatchRates100-33-3.ps, width=6.0in, height=4.1in}
%\caption{\label{fig:22servers102clients-Latency100-33-3} Latencies for match rates of 100\%, 33\% and 3\%}
%\end{figure}

\begin{table}[htp]
\centering
\begin{tabular}{|p{0.5in}|p{1.35in}|p{1.35in}|p{0.8in}|p{0.8in}|} \hline
\textbf{Match} & \multicolumn{2}{c|}{Latency Range @ $<$ publish Rate, event Size $>$ } & \multicolumn{2}{c|}{Variance Range } \\ \cline{2-5}
\textbf{Rate}  & Latency $\mathcal{L}_a$ & Latency $\mathcal{L}_a$ & Variance  $\mathcal{V}_a$ @ $\mathcal{L}_a$ & Variance $\mathcal{V}_b$ @ $\mathcal{L}_b$ \\ \hline

100\% & 15.54 \emph{mSec} @  & 391.85 \emph{mSec} @ &  2.3684  & 69,713.93 \\

& $<$1\emph{evt/Sec}, 50\emph{B}$>$&$<$952\emph{evts/Sec}, 450\emph{B}$>$&$mSec^2$ & $mSec^2$\\ \hline

50\% & 13.02 \emph{mSec} @& 178.66 \emph{mSec} @& 56.8196  & 14,634 \\
    &  $<$20\emph{evts/Sec}, 50\emph{B}$>$ &  $<$952\emph{evts/Sec}, 350\emph{B}$>$ & $mSec^2$ & $mSec^2$ \\ \hline

33\% & 15.18 \emph{mSec} @ & 95.969 \emph{mSec} @ & 26.57 $mSec^2$ &  1,263 $mSec^2$ \\ 
     & $<$20\emph{evts/Sec}, 50\emph{B}$>$ & $<$952\emph{evts/Sec}, 425 \emph{B}$>$ &  & \\ \hline

25\% & 14.40 \emph{mSec} @ & 66.6 \emph{mSec} @&  0.24 $mSec^2$ & 587.04 \\
     & $<$20\emph{evts/Sec}, 50\emph{B}$>$ & $<$961\emph{evts/Sec}, 400\emph{B}$>$ & &  $mSec^2$\\ \hline



20\% & 15.35 \emph{mSec} @ & 62.35 \emph{mSec} & 12.02 $mSec^2$ & 312 $mSec^2$ \\
&  $<$20\emph{evts/Sec}, 50\emph{B}$>$ & $<$952\emph{evts/Sec}, 400\emph{B}$>$ & & \\ \hline


 
10\% & 14.40 \emph{mSec} @ & 52.0 \emph{mSec} @& 0.44 $mSec^2$& 103 $mSec^2$\\
  & $<$20 \emph{evts/Sec}, 50\emph{B}$>$  & $<$952\emph{evts/Sec}, 400\emph{B}$>$ & & \\ \hline


5 \% & 14.0 \emph{mSec} @ & 47.6 \emph{mSec} @ & 0.44 $mSec^2$ &  87.44 $mSec^2$ \\
&  $<$20\emph{evts/Sec}, 50\emph{B}$>$ & $<$961\emph{evts/Sec}, 425\emph{B}$>$ & & \\ \hline
\end{tabular}
\caption{\label{table:latency-variance} Latency and Variance Range Table}
\end{table}

%\subsection{System Throughput}
%We also depict the system throughputs at the client under conditions of varying event sizes and publish rates. We choose to depict the system throughputs at a matching rate of 100\%. At matching rates other than 100\% only the relevant events are being routed to the clients. The events received do not reveal the true throughputs that can be achieved at a client. Figure \ref{fig:22servers102clients-1} depicts the system throughputs achieved at a client under conditions of different publish rates and event sizes. The maximum throughput achieved was 480.76 \emph{events/Sec} at a publish rate of 492 \emph{events/Sec} with the sample of events being of size 75 bytes.  
%\begin{figure}[htp]
%\centering
%\psfig{file=images/results/systemThroughput.ps, width=6.0in, height=4.1in}
%\caption{\label{fig:22servers102clients-1} System Throughput}
%\end{figure}

%\subsection{Variance}
%Variance for the sample of received events at a client, demonstrate how queueing delays can add up to increase the mean latency. Variance also snapshots how this mean latency has high deviations from the highest and lowest latencies contained in the sample of latencies, associated with the events that are received at a client.  The variance in the sample of events varies from 69713 $mSec^2$ to 133.76 $mSec^2$ for $<$952 \emph{events/Sec} , 450 \emph{Bytes}$>$ at matching rates of 100\% to $<$877 \emph{events/Sec}, 450 \emph{Bytes}$>$ at matching rates of 5\%. Thus variance in the sample of events for higher event sizes at higher publish rates also reduces with decreasing matching rates for the published events. 

%\section{Performance Numbers}
%\begin{itemize} \item[(b)] Testing for persistent clients. This includes reception latencies under the above scenarios and also under differing constraints of replication granularities. A client would be at a path which could have multiple replications taking place prior to reception at client. There should also be a test for the number of events that are duplicate detected, under different replication granularity schemes. This tells us about the number of trips an event makes. \item[(c)] Internal events v/s external events. The internal/external refer to the events being issued within a unit $u^{N}$ in a N+1 system. \item[(d)] Compare the latencies involved with the \emph{event routing} scheme we have devised and the \emph{flooding} approach (under different matching rates and increasing publish rates)  \end{itemize}

\begin{figure}[htp]
\centering
\psfig{file=images/results/22servers102clientsPersistentClientsMatchRate50.ps, width=6.0in, height=4.1in}
\caption{\label{fig:22servers102clients-PersistentClients-Match50} Match Rates of 50\% - Persistent Clients (singular and double replication)}
\end{figure}
\subsubsection{Persistent Clients}
In figure \ref{fig:testing-topology} we have also outlined the replication scheme that exists in the system. When an event arrives at node \textbf{1}, the event is first stored to the level-3 stable store so that it has an epoch associated with it. The event is then forwarded for dissemination within the unit. Clients attached to any of the nodes in super-cluster \textbf{SC-6} have a replication granularity of $r_2$. When the events issued by the publisher in the test topology of figure \ref{fig:testing-topology} are  being disseminated, when clients attached to nodes in \textbf{SC-6} receive the event, that event would have been replicated twice (once at node \textbf{1} and once at node \textbf{8}). For testing purposes we set up another \emph{measuring} subscriber at node \textbf{7} in addition to the subscriber that we would set up at node \textbf{10}. When an event is received by the subscriber attached to node \textbf{7} the event would have been replicated only once, at node \textbf{1}. These \emph{measuring} subscribers allow us to measure the response times involved for singular and double replications experienced at clients attached to nodes \textbf{7} and \textbf{10} respectively. 
Every node (with the exception of nodes \textbf{7} and \textbf{10}) in the system has 5 persistent subscribing clients attached to it, for a total of 102 persistent subscribing clients. The publisher and the 2 \emph{measuring} subscribers are all hosted on the same machine for reasons discussed earlier. Figure \ref{fig:22servers102clients-PersistentClients-Match50} depicts the latencies in delivery of events at persistent clients, with singular and double replications for a matching rate of 50\%. 




\subsubsection{Pathlengths and Latencies}
\label{sec:pathlengths-and-latencies}
The topology in figure \ref{fig:testing-topology} allows us to magnify the latencies, which occur by having the queuing delays at individual broker hops add up. In that topology the number of broker hops taken by an event prior to delivery at the measuring subscriber is 9. We now proceed with testing for the topology outlined in figure \ref{fig:testing-topology-2}. The layout of the broker nodes is essentially identical to the earlier one, with the addition of links between nodes resulting in a strongly connected network. We have 5 subscribing clients at each of the broker nodes. The mapping of broker nodes and subscribing client nodes to the physical machines is also identical to the earlier topology. As can be seen the addition of super-cluster link between super-clusters \textbf{SC-5} and \textbf{SC-6}, and level-0 links between nodes \textbf{8} and \textbf{10} in cluster \textbf{SC-6.n} reduces the number of broker hops, for the shortest path from the publisher to the measuring subscriber at node \textbf{10},  from 9 to 4. 

\begin{figure}[ht]
\centering
\epsfig{file=images/Testing-Topology2.eps,width=312.00pt, height=353.00pt}
\caption{\label{fig:testing-topology-2} Testing Topology - Latencies versus broker hops}
\end{figure}

In this setting we are interested in the changes in latencies as the number of broker hops vary. We measure the latencies at three different locations, the measuring subscriber at node \textbf{10} has a broker hop of 4 while the measuring subscribers at nodes \textbf{1} and \textbf{22} have broker hops of 2 and 1 respectively for events published by the publisher at node \textbf{22}. In general, as the number of broker hops reduce the latencies also reduce. The patterns for changes in latency as the event size and publish rates increase is however similar to our earlier cases. We depict our results, gathered at the three measuring subscribers for a matching rate of 50\%. The pattern of decreasing latencies with a decrease in the number of broker hops is clear by looking at figure \ref{fig:22servers102clients-ServerHops-Match50}. Additional results at different matching rates can be found in \cite{ges-results}.
%through \ref{fig:22servers102clients-ServerHops-Match10}. 
\begin{figure}[htp]
\centering
\psfig{file=images/results/22servers102clientsServerHopsMatchRate50.ps, width=6.0in, height=4.1in}
\caption{\label{fig:22servers102clients-ServerHops-Match50} Broker Hops - Match Rate 50\%}
\end{figure}

%\begin{figure}[htp]
%\centering
%\psfig{file=images/results/22servers102clientsServerHopsMatchRate25.ps, width=6.0in, height=4.1in}
%\caption{\label{fig:22servers102clients-ServerHops-Match25} Broker Hops - Match Rate 25\%}
%\end{figure}

%\begin{figure}[htp]
%\centering
%\psfig{file=images/results/22servers102clientsServerHopsMatchRate10.ps, width=6.0in, height=4.1in}
%\caption{\label{fig:22servers102clients-ServerHops-Match10} Broker Hops - Match Rate 10\%}
%\end{figure}

\section{Conclusion}
\label{conclusion}
In this paper, we have presented the Grid Event Service (GES), a distributed event service designed to run on a very large network of broker nodes. 
The delivery guarantee needs to be met across client roams and also in the presence of broker failures. 
GES comprises of a suite of protocols, which are responsible for the organization of nodes, creation of abbreviated \emph{system views}, management of profiles and the hierarchical dissemination of content based on these profiles. The broker topology ensures that pathlengths would only increase logarithmically with geometric increases in the size of the broker network. The feature of having multiple links between two units/super-units ensures a greater degree of fault tolerance. Links could fail, and the routing to the affected units is performed using the alternate links. 
%The protocols in the GES protocol suite exchange information collected and processed by the other protocols. Thus when a new connection is added the information is used to update the connectivity graph, which is used to identify the relevant nodes for the propagation of  profiles to. This information contained in the profile graphs is then used for the hierarchical dissemination of content. All these protocols can run concurrently, adding a lot of flexibility to the overall system.  

The \emph{system views} at each of the broker nodes respond to changes in system inter-connections, aiding in the detection of partitions and the calculation of new routes to reach units within the system.  
The organization of connection information ensures that connection losses (or additions) are incorporated into the connectivity graph hosted at the broker nodes. 
%Certain sections of the routing cache are invalidated in response to this addition (or loss) of connections. This invalidation and subsequent calculation of best \emph{hops} to reach units (at different levels) ensure that the paths computed are consistent with the state of the network, and include only valid/active links.
%The event routing protocol uses the profile information available at the unit gatekeepers to compute the sub-units that the event should be routed to. To reach these destinations every node, at which this event is received, employs the \emph{best hops} to reach those destinations.  This \emph{best hop} is computed based on the cost of traversal and also the number of links connecting the different units. Thus, in our system, based on the organization of profiles and subsequent matching of events, the only units to which an event is routed to are those that have clients interested in that event.
 The protocols in GES ensure that the routing is intelligent and can handle sparse/dense interest in certain sections of the system. GES's ability to handle the complete spectrum of interests equally well, lends itself as a very scalable solution under conditions of varying publish rates, matching rates and message sizes. 

The paper outlined a scheme for the delivery of events in the presence of broker node failures. In our scheme a unit could fail and remain failed forever. The only requirement that we impose is that if a stable storage fails, it should recover within a finite amount of time. The replication strategy, that we adopted allows us to add stable storages and also to withstand stable storage failures. The replication strategy, epochs associated with received events and profile ID's associated with client profiles allowed us to account for a very precise recovery of events for clients with prolonged disconnects or those which have roamed the network. 

GES could be extended very easily to support dynamic topologies. Based on the concentration of clients at specific locations, bandwidth utilization can be optimized with the creation of dynamic brokers at some of the clients. This scheme fits very well with our failure model for system units, where they can remain failed forever. Detection schemes can be employed to detect slow nodes, which serve as performance bottlenecks. Forcing these affected nodes to fail then reconfigures the system. 
%GES immediately employs newly added units in the routing of events and responds to unit failures equally well. Similarly links can be added and removed in a similar fashion to optimize the routing for events. The routing decisions at each broker node are based on the current state of the system. GES thus provides the base infrastructures for dynamic topologies. 

%GES is intended to be a part of the Grid Collaborative Portal (GCP). The features of location transparency, intelligent routing, replication and persistent delivery of events lends itself very easily to aid in the development of the GCP. The application domains into which GES can be easily extended include collaborative systems, peer-to-peer (P2P) systems and messaging for hand-held devices such as the Grid Event Service Micro Edition (GESME). GESME, which would account for the conversion of GCP messages to be handled by the hand-held devices could add considerable value to GES. 

%The location transparency feature contained within GES where a client can roam the network in response to failure suspicions (correct or incorrect) or re-join the system after a prolonged disconnect, and attach itself to any of the nodes within the system and still receive all the events it was supposed to receive is a significant contribution. 
Clients can connect to local brokers instead of reconnecting all the way back to the remote broker that it was last connected to. This scheme optimizes bandwidth utilization. This optimization is very pronounced when there is a high concentration of clients accessing the remote broker. 
The failure model of the system, which allows a broker node or a unit/super-unit of broker nodes to fail and remain failed forever and still satisfy delivery guarantees is another significant contribution, which also allows the system to be easily extensible. 
This model ensures that clients need not wait for a broker to recover after this broker has failed. During failures clients do not experience a denial of service in this model. The service, as mentioned earlier, extends very naturally into dynamic topologies allowing for the dynamic instantiation and purging of brokers and connections. Changes in network fabric are incorporated by the routing algorithms, which ensure that the routing decisions made at a node are based on the current state of the system.  
The replication strategy presented in this paper, could be augmented to include mirror storages, which maintain information identical to that of the stable storages, and take over in the event of stable storage failures. This feature would add additional robustness and reduce the time required to recover from stable storage failures. 

The results in section \ref{results} demonstrated the efficiency of the routing algorithms and confirmed the advantages of our dissemination scheme, which intelligently routes messages. 
Industrial strength JMS solutions, which support the publish subscribe paradigm generally are optimized for a small network of brokers. The seamless integration of multiple broker nodes in our framework and the failure model that we impose on broker nodes provides for very easy maintenance of the broker network.

%The explosive developments in the area of pervasive computing have resulted in the need for building distributed network centric services. However these network services should also be easily extensible, scalable and have a reasonable failure model, which causes a denial of service only under the most extreme of failure scenarios. Another important feature is the ability to access another part of this distributed service for better response times or convenience. This paper makes significant contributions towards providing solutions to these issue. 





%\include{intro}
%\include{topology}
%\include{eventDelivery}
%\include{reliableDelivery}
%\include{results}

%\nocite{*}
\bibliography{GESReferences}
\bibliographystyle{plain}
\end{document}



