Fix for bug 776. Don't require retained events to expire

before deleting the channel on unlink.



git-svn-id: http://svn.fedorahosted.org/svn/corosync/trunk@780 fd59a12c-fef9-0310-b244-a6a79926bd2f

exec/evt.c

@@ -950,6 +950,24 @@ static void delete_channel(struct event_svr_channel_instance *eci)
 	}
 }
 
+/*
+ * Free up an event structure if it isn't being used anymore.
+ */
+static void
+free_event_data(struct event_data *edp)
+{
+	if (--edp->ed_ref_count) {
+		return;
+	}
+	log_printf(LOG_LEVEL_DEBUG, "Freeing event ID: 0x%llx\n", 
+			edp->ed_event.led_event_id);
+	if (edp->ed_delivered) {
+		free(edp->ed_delivered);
+	}
+
+	free(edp);
+}
+
 /*
  * Mark a channel for deletion.
  */
@@ -957,7 +975,7 @@ static void unlink_channel(struct event_svr_channel_instance *eci,
 		uint64_t unlink_id)
 {
 	struct event_data *edp;
-	struct list_head *l;
+	struct list_head *l, *nxt;
 
 	log_printf(CHAN_UNLINK_DEBUG, "Unlink request: %s, id 0x%llx\n",
 			eci->esc_channel_name.value, unlink_id);
@@ -977,14 +995,25 @@ static void unlink_channel(struct event_svr_channel_instance *eci,
 	list_add(&eci->esc_entry, &esc_unlinked_head);
 
 	/*
-	 * Scan the retained event list and tag any associated with this channel
-	 * with the unlink ID so that they get routed properly.
+	 * Scan the retained event list and remove any retained events.
+	 * Since no new opens can occur there won't be any need of sending
+	 * retained events on the channel.
 	 */
-	for (l = retained_list.next; l != &retained_list; l = l->next) {
+	for (l = retained_list.next; l != &retained_list; l = nxt) {
+		nxt = l->next;
 		edp = list_entry(l, struct event_data, ed_retained);
 		if ((edp->ed_my_chan == eci) && 
 				(edp->ed_event.led_chan_unlink_id == EVT_CHAN_ACTIVE)) {
-			edp->ed_event.led_chan_unlink_id = unlink_id;
+			poll_timer_delete(aisexec_poll_handle, edp->ed_timer_handle);
+			edp->ed_event.led_retention_time = 0;
+			list_del(&edp->ed_retained);
+			list_init(&edp->ed_retained);
+			edp->ed_my_chan->esc_retained_count--;
+
+			log_printf(CHAN_UNLINK_DEBUG, 
+				"Unlink: Delete retained event id 0x%llx\n",
+			edp->ed_event.led_event_id);
+			free_event_data(edp);
 		}
 	}
 
@@ -1303,24 +1332,6 @@ static SaErrorT get_event_id(uint64_t *event_id)
 
 
 
-/*
- * Free up an event structure if it isn't being used anymore.
- */
-static void
-free_event_data(struct event_data *edp)
-{
-	if (--edp->ed_ref_count) {
-		return;
-	}
-	log_printf(LOG_LEVEL_DEBUG, "Freeing event ID: 0x%llx\n", 
-			edp->ed_event.led_event_id);
-	if (edp->ed_delivered) {
-		free(edp->ed_delivered);
-	}
-
-	free(edp);
-}
-
 /*
  * Timer handler to delete expired events.
  *
@@ -3050,9 +3061,11 @@ static int evt_remote_evt(void *msg, struct in_addr source_addr,
 	evtpkt->led_in_addr = source_addr;
 	evtpkt->led_receive_time = clust_time_now();
 
-	log_printf(CHAN_UNLINK_DEBUG, 
+	if (evtpkt->led_chan_unlink_id != EVT_CHAN_ACTIVE) {
+		log_printf(CHAN_UNLINK_DEBUG, 
 				"evt_remote_evt(0): chan %s, id 0x%llx\n",
 					evtpkt->led_chan_name.value, evtpkt->led_chan_unlink_id);
+	}
 	eci = find_channel(&evtpkt->led_chan_name, evtpkt->led_chan_unlink_id);
 	/*
 	 * We may have had some events that were already queued when an

test/Makefile

@@ -54,12 +54,12 @@ TEST_SRC =  testclm.c testamf.c testamf1.c testamf2.c testamf3.c \
 		testckpt.c ckptstress.c testparse.c ckptbench.c  \
 		ckptbenchth.c testevt.c testevs.c evsbench.c \
 		subscription.c publish.c evtbench.c \
-		sa_error.c
+		sa_error.c unlink.c
 
 all: testclm testamf testamf1 testamf2 testamf3 testamf4 testamf5 \
 	testamf6 testamfth testckpt ckptstress testparse ckptbench \
 	ckptbenchth ckpt-rd ckpt-wr testevt testevs \
-	evsbench subscription publish evtbench
+	evsbench subscription publish evtbench unlink
 
 testparse: testparse.o $(LIBRARIES)
 	$(CC) $(LDFLAGS) -o testparse testparse.o ../exec/parse.o ../exec/print.o ../exec/mempool.o ../exec/util.o
@@ -124,6 +124,9 @@ subscription: subscription.o sa_error.o $(LIBRARIES)
 publish: publish.o sa_error.o $(LIBRARIES)
 	$(CC) $(LDFLAGS) -o publish publish.o sa_error.o $(LIBS)
 
+unlink: unlink.o sa_error.o $(LIBRARIES)
+	$(CC) $(LDFLAGS) -o unlink unlink.o sa_error.o $(LIBS)
+
 evtbench: evtbench.o sa_error.o $(LIBRARIES)
 	$(CC) $(LDFLAGS) -o evtbench evtbench.o sa_error.o $(LIBS)
 
@@ -140,7 +143,7 @@ clean:
 	rm -f *.o testclm testamf testamf1 testamf2 testamf3 testamf4 \
 	testamf5 testamf6 testamfth testckpt ckptstress testparse testtimer \
 	ckptbench ckptbenchth testevt testevs ckpt-wr ckpt-rd \
-	evsbench subscription publish evtbench
+	evsbench subscription publish evtbench unlink
 %.o: %.c
 	$(CC) $(CFLAGS) $(EXTRA_CFLAGS) -c -o $@ $<
 
@@ -149,28 +152,27 @@ depend:
 # DO NOT DELETE
 
 testclm.o: ../include/ais_types.h ../include/saClm.h
-testamf.o: ../include/ais_types.h ../include/ais_amf.h ../include/ipc_evs.h
-testamf.o: ../include/evs.h ../include/saClm.h ../include/ipc_gen.h
+testamf.o: ../include/ais_types.h ../include/ais_amf.h ../include/saClm.h
 testamf1.o: ../include/ais_types.h ../include/ais_amf.h
 testamf2.o: ../include/ais_types.h ../include/ais_amf.h
 testamf3.o: ../include/ais_types.h ../include/ais_amf.h
 testamf4.o: ../include/ais_types.h ../include/ais_amf.h
 testamf5.o: ../include/ais_types.h ../include/ais_amf.h
 testamf6.o: ../include/ais_types.h ../include/ais_amf.h
-testamfth.o: ../include/ais_types.h ../include/ais_amf.h ../include/ipc_evs.h
-testamfth.o: ../include/evs.h ../include/saClm.h ../include/ipc_gen.h
-testckpt.o: ../include/ais_types.h ../include/saCkpt.h
+testamfth.o: ../include/ais_types.h ../include/ais_amf.h ../include/saClm.h
+testckpt.o: ../include/ais_types.h ../include/saCkpt.h sa_error.h
 ckptstress.o: ../include/ais_types.h ../include/saCkpt.h
 testparse.o: ../include/ais_types.h ../exec/parse.h ../include/list.h
 testparse.o: ../exec/aispoll.h ../exec/totempg.h ../exec/totemsrp.h
-testparse.o: ../exec/print.h ../include/saClm.h ../include/ais_types.h
+testparse.o: ../exec/totem.h ../exec/parse.h ../exec/print.h
+testparse.o: ../include/saClm.h ../include/ais_types.h
 ckptbench.o: ../include/ais_types.h ../include/saCkpt.h
 ckptbenchth.o: ../include/ais_types.h ../include/saCkpt.h
 testevt.o: ../include/ais_types.h ../include/saEvt.h
 testevs.o: ../include/evs.h
-testevsth.o: ../include/evs.h
 evsbench.o: ../include/ais_types.h ../include/evs.h
 subscription.o: ../include/ais_types.h ../include/saEvt.h
 publish.o: ../include/ais_types.h ../include/saEvt.h
 evtbench.o: ../include/ais_types.h ../include/saEvt.h
 sa_error.o: ../include/ais_types.h
+unlink.o: ../include/ais_types.h ../include/saEvt.h

test/unlink.c

@@ -0,0 +1,95 @@
+/*
+ * Test program for event service
+ */
+
+#include <stdio.h>
+#include <string.h>
+#include <sys/poll.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <getopt.h>
+#include <sys/time.h>
+#include "ais_types.h"
+#include "saEvt.h"
+
+#define TRY_WAIT 2
+
+extern int get_sa_error(SaAisErrorT, char *, int);
+char result_buf[256];
+int result_buf_len = sizeof(result_buf);
+
+SaVersionT version = { 'B', 0x01, 0x01 };
+
+SaEvtCallbacksT callbacks = {
+	0,
+	0
+};
+
+
+char channel[256] = "EVENT_TEST_CHANNEL";
+
+	
+int
+do_unlink()
+{
+	SaEvtHandleT handle;
+	SaNameT channel_name;
+
+	SaAisErrorT result;
+	 
+	do {
+		result = saEvtInitialize (&handle, &callbacks, &version);
+	} while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
+	if (result != SA_AIS_OK) {
+		get_sa_error(result, result_buf, result_buf_len);
+		printf("Event Initialize result: %s\n", result_buf);
+		return(result);
+	}
+
+	strcpy(channel_name.value, channel);
+	channel_name.length = strlen(channel);
+	do {
+	result = saEvtChannelUnlink(handle, &channel_name);
+	} while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
+	if (result != SA_AIS_OK) {
+		get_sa_error(result, result_buf, result_buf_len);
+		printf("ERROR: channel unlink result: %s\n", result_buf);
+	}
+
+	do {
+		result = saEvtFinalize(handle);
+	} while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
+	if (result != SA_AIS_OK) {
+		get_sa_error(result, result_buf, result_buf_len);
+		printf("ERROR: Event Finalize result: %s\n", result_buf);
+	}
+	return 0;
+}
+
+int main (int argc, char **argv)
+{
+	static const char opts[] = "c:";
+
+	int option;
+
+	while (1) {
+		option = getopt(argc, argv, opts);
+		if (option == -1) 
+			break;
+
+		switch (option) {
+
+		case 'c':
+			strcpy(channel, optarg);
+			break;
+		default:
+			printf("invalid arg: \"%s\"\n", optarg);
+			return 1;
+		}
+	}
+	do_unlink();
+
+	return 0;
+}