Christopher Browne <cbbrowne at ca.afilias.info> writes:
> Suggestion:
>
> - With 2 variables, I can store the number of event rows pulled for
>   this and the last iteration, last_events, and current_events
>
> - We set the query to do "limit (2*sync_group_maxsize)"
>
> - If (last_events + current_events) = 4*sync_group_maxsize, then we
>   sleep for a configurable period, *and* drop back into LISTENing mode.
>
> That allows all of this logic to take place at the end of
> remoteListen_receive_events().
>
> It means that the slon never completely ceases to add events to the
> queue, but:
>
>  a) The events aren't enormously big, so we should be running out of
>     memory due to other things way before running out due to event
>     bloat;
>
>  b) If we decelerate it sufficiently, that should be helpful enough.
>
> I'll see about a patch for this.

Actually, it's a bit simpler than I thought; I already had "ntuples"
to work with, so it's only requiring one extra variable.

Index: remote_listen.c
===================================================================
RCS file: /home/cvsd/slony1/slony1-engine/src/slon/remote_listen.c,v
retrieving revision 1.43
diff -c -u -r1.43 remote_listen.c
--- remote_listen.c	23 Apr 2008 22:29:12 -0000	1.43
+++ remote_listen.c	30 Apr 2008 22:02:47 -0000
@@ -65,6 +65,8 @@
 extern char *lag_interval;
 int remote_listen_timeout;
 
+static int last_event_sel = 0;
+
 /* ----------
  * slon_remoteListenThread
  *
@@ -697,7 +699,11 @@
 	{
 		slon_appendquery(&query, ")");
 	}
-	slon_appendquery(&query, " order by e.ev_origin, e.ev_seqno");
+	/* Limit the result set size to:
+            sync_group_maxsize * 2, if it's set
+			100, if sync_group_maxsize isn't set */
+	slon_appendquery(&query, " order by e.ev_origin, e.ev_seqno limit %d",
+					 (sync_group_maxsize>0)? sync_group_maxsize * 2 : 100);
 
 	rtcfg_unlock();
 
@@ -785,24 +791,25 @@
 	}
 
 	if (ntuples > 0) {
-		poll_sleep = 0;
-		poll_state = SLON_POLLSTATE_POLL;
+			if ((ntuples + last_event_sel) >= (4 * sync_group_maxsize)) {
+					poll_state = SLON_POLLSTATE_LISTEN;
+					slon_log(SLON_INFO, "remoteListenThread_%d: events queued in two iterations (%d) > sync_group_maxsize*4 (%d)\n",
+							 node->no_id, (ntuples + last_event_sel), (4 * sync_group_maxsize));
+					slon_log(SLON_INFO, "remoteListenThread_%d: sleep 10s, return to LISTEN mode\n",
+							 node->no_id, (ntuples + last_event_sel), (4 * sync_group_maxsize));
+					sched_msleep(node, 10000);
+			} else {
+					poll_sleep = 0;
+					poll_state = SLON_POLLSTATE_POLL;
+			}
 	} else {
-		poll_sleep = poll_sleep * 2 + sync_interval;
-		if (poll_sleep > sync_interval_timeout) {
-			poll_sleep = sync_interval_timeout;
-			poll_state = SLON_POLLSTATE_LISTEN;
-		}
+			poll_sleep = poll_sleep * 2 + sync_interval;
+			if (poll_sleep > sync_interval_timeout) {
+					poll_sleep = sync_interval_timeout;
+					poll_state = SLON_POLLSTATE_LISTEN;
+			}
 	}
 	PQclear(res);
-
+	last_event_sel = ntuples;
 	return 0;
 }
-
-/*
- * Local Variables:
- *	tab-width: 4
- *	c-indent-level: 4
- *	c-basic-offset: 4
- * End:
- */
-- 
(reverse (concatenate 'string "ofni.sailifa.ac" "@" "enworbbc"))
<http://dba2.int.libertyrms.com/>
Christopher Browne
(416) 673-4124 (land)