[prev in list] [next in list] [prev in thread] [next in thread] 

List:       openais
Subject:    [Openais] New evt recovery patch
From:       Mark Haverkamp <markh () osdl ! org>
Date:       2005-03-28 17:35:04
Message-ID: 1112031304.5560.8.camel () markh1 ! pdx ! osdl ! net
[Download RAW message or body]

After thinking about the retained event recovery with merged partitions,
I realized that I wasn't distributing newly received events to processes
with associated channels already open.  This patch checks the newly
received retained events and selects them to be delivered to those
processes with the associated channels open and have the proper
subscription filters.

Mark.


-- 
Mark Haverkamp <markh@osdl.org>

["ais_evt_recovery_2.patch" (ais_evt_recovery_2.patch)]

===== exec/evt.c 1.25 vs edited =====
--- 1.25/exec/evt.c	2005-03-22 15:22:34 -08:00
+++ edited/exec/evt.c	2005-03-24 10:30:36 -08:00
@@ -2886,6 +2886,45 @@
 	return 0;
 }
 
+static int
+try_deliver_event(struct event_data *evt, 
+		struct event_svr_channel_instance *eci)
+{
+	struct list_head *l, *l1;
+	struct event_svr_channel_open *eco;
+	struct event_svr_channel_subscr *ecs;
+	int delivered_event = 0;
+	/*
+	 * Check open channels
+	 */
+	for (l = eci->esc_open_chans.next; l != &eci->esc_open_chans; l = l->next) {
+		eco = list_entry(l, struct event_svr_channel_open, eco_entry);
+		/*
+		 * See if enabled to receive
+		 */
+		if (!(eco->eco_flags & SA_EVT_CHANNEL_SUBSCRIBER)) {
+				continue;
+		}
+
+		/*
+		 * Check subscriptions
+		 */
+		for (l1 = eco->eco_subscr.next; l1 != &eco->eco_subscr; l1 = l1->next) {
+			ecs = list_entry(l1, struct event_svr_channel_subscr, ecs_entry);
+			/*
+			 * Apply filter rules and deliver if patterns
+			 * match filters.
+			 * Only deliver one event per open channel
+			 */
+			if (event_match(evt, ecs) == SA_AIS_OK) {
+				deliver_event(evt, eco, ecs);
+				delivered_event++;
+				break;
+			}
+		}
+	}
+	return delivered_event;
+}
 
 /*
  * Receive the network event message and distribute it to local subscribers
@@ -2902,10 +2941,7 @@
 	 */
 	struct lib_event_data *evtpkt = msg;
 	struct event_svr_channel_instance *eci;
-	struct event_svr_channel_open *eco;
-	struct event_svr_channel_subscr *ecs;
 	struct event_data *evt;
-	struct list_head *l, *l1;
 	SaClmClusterNodeT *cn;
 
 	log_printf(LOG_LEVEL_DEBUG, "Remote event data received from %s\n",
@@ -2980,34 +3016,7 @@
 		retain_event(evt);
 	}
 
-	/*
-	 * Check open channels
-	 */
-	for (l = eci->esc_open_chans.next; l != &eci->esc_open_chans; l = l->next) {
-		eco = list_entry(l, struct event_svr_channel_open, eco_entry);
-		/*
-		 * See if enabled to receive
-		 */
-		if (!(eco->eco_flags & SA_EVT_CHANNEL_SUBSCRIBER)) {
-				continue;
-		}
-
-		/*
-		 * Check subscriptions
-		 */
-		for (l1 = eco->eco_subscr.next; l1 != &eco->eco_subscr; l1 = l1->next) {
-			ecs = list_entry(l1, struct event_svr_channel_subscr, ecs_entry);
-			/*
-			 * Apply filter rules and deliver if patterns
-			 * match filters.
-			 * Only deliver one event per open channel
-			 */
-			if (event_match(evt, ecs) == SA_AIS_OK) {
-				deliver_event(evt, eco, ecs);
-				break;
-			}
-		}
-	}
+	try_deliver_event(evt, eci);
 	free_event_data(evt);
 
 
@@ -3034,13 +3043,17 @@
 		int endian_conversion_required)
 {
 	/*
-	 * - retain events that have a retention time
+	 * - calculate remaining retention time
 	 * - Find assocated channel
+	 * - Scan list of subscribers
+	 * - Apply filters
+	 * - Deliver events that pass the filter test
 	 */
 	struct lib_event_data *evtpkt = msg;
 	struct event_svr_channel_instance *eci;
 	struct event_data *evt;
 	struct member_node_data *md;
+	int num_delivered;
 	SaTimeT now;
 
 	now = clust_time_now();
@@ -3128,6 +3141,9 @@
 		}
 			
 		retain_event(evt);
+		num_delivered = try_deliver_event(evt, eci);
+		log_printf(RECOVERY_EVENT_DEBUG, "Delivered to %d subscribers\n",
+				num_delivered);
 		free_event_data(evt);
 	}
 


_______________________________________________
Openais mailing list
Openais@lists.osdl.org
http://lists.osdl.org/mailman/listinfo/openais


[prev in list] [next in list] [prev in thread] [next in thread]