attachment:slon.diff von PostgreSQL

   1 Index: slon.c
   2 ===================================================================
   3 RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/slon/slon.c,v
   4 retrieving revision 1.28
   5 diff -r1.28 slon.c
   6 66a67,69
   7 >         FILE *pidfile=0;
   8 > 	char *pidfilename=0;
   9 > 	pid_t pid=0;
  10 67a71
  11 > 	int			logpid=0;
  12 69a74
  13 >         int			bgrun = 0;
  14 76c81
  15 < 	while ((c = getopt(argc, argv, "d:s:t:g:c:h")) != EOF)
  16 ---
  17 > 	while ((c = getopt(argc, argv, "d:s:t:g:c:h:p:b")) != EOF)
  18 132a138,143
  19 >                         case 'p':       logpid=1;
  20 > 					pidfilename=(char*)malloc(strlen(optarg)+1);
  21 > 					strcpy(pidfilename,optarg);
  22 > 					break;
  23 >                         case 'b':       bgrun=1;
  24 > 					break;
  25 159,200c170,171
  26 < 	if (sync_interval_timeout != 0 && sync_interval_timeout <= sync_interval)
  27 < 		sync_interval_timeout = sync_interval * 2;
  28 < 
  29 < 	/*
  30 < 	 * Remember the cluster name and build the properly quoted 
  31 < 	 * namespace identifier
  32 < 	 */
  33 < 	slon_pid = getpid();
  34 < 	rtcfg_cluster_name	= (char *)argv[optind];
  35 < 	rtcfg_namespace		= malloc(strlen(argv[optind]) * 2 + 4);
  36 < 	cp2 = rtcfg_namespace;
  37 < 	*cp2++ = '"';
  38 < 	*cp2++ = '_';
  39 < 	for (cp1 = (char *)argv[optind]; *cp1; cp1++)
  40 < 	{
  41 < 		if (*cp1 == '"')
  42 < 			*cp2++ = '"';
  43 < 		*cp2++ = *cp1;
  44 < 	}
  45 < 	*cp2++ = '"';
  46 < 	*cp2 = '\0';
  47 < 
  48 < 	/*
  49 < 	 * Remember the connection information for the local node.
  50 < 	 */
  51 < 	rtcfg_conninfo = (char *)argv[++optind];
  52 < 
  53 < 	/*
  54 < 	 * Connect to the local database for reading the initial configuration
  55 < 	 */
  56 < 	startup_conn = PQconnectdb(rtcfg_conninfo);
  57 < 	if (startup_conn == NULL)
  58 < 	{
  59 < 		slon_log(SLON_FATAL, "main: PQconnectdb() failed\n");
  60 < 		slon_exit(-1);
  61 < 	}
  62 < 	if (PQstatus(startup_conn) != CONNECTION_OK)
  63 < 	{
  64 < 		slon_log(SLON_FATAL, "main: Cannot connect to local database - %s",
  65 < 				PQerrorMessage(startup_conn));
  66 < 		PQfinish(startup_conn);
  67 < 		slon_exit(-1);
  68 ---
  69 >         if ( bgrun == 1 ){
  70 >         	pid=fork();
  71 201a173,184
  72 > 	if ( pid == 0 || bgrun == 0){
  73 > 		if (logpid==1){
  74 > 			pidfile=fopen(pidfilename,"w");
  75 > 			if(pidfile){
  76 > 				fprintf(pidfile,"%d",getpid());
  77 > 				fclose(pidfile);
  78 > 			} else {
  79 > 				fprintf(stderr,"couldn't open %s pidfile\n",pidfile);
  80 > 			}
  81 > 		}
  82 > 		if (sync_interval_timeout != 0 && sync_interval_timeout <= sync_interval)
  83 > 			sync_interval_timeout = sync_interval * 2;
  84 203,212c186,203
  85 < 	/*
  86 < 	 * Get our local node ID
  87 < 	 */
  88 < 	rtcfg_nodeid = db_getLocalNodeId(startup_conn);
  89 < 	if (rtcfg_nodeid < 0)
  90 < 	{
  91 < 		slon_log(SLON_FATAL, "main: Node is not initialized properly\n");
  92 < 		slon_exit(-1);
  93 < 	}
  94 < 	slon_log(SLON_CONFIG, "main: local node id = %d\n", rtcfg_nodeid);
  95 ---
  96 > 		/*
  97 > 		 * Remember the cluster name and build the properly quoted 
  98 > 		 * namespace identifier
  99 > 		 */
 100 > 		slon_pid = getpid();
 101 > 		rtcfg_cluster_name	= (char *)argv[optind];
 102 > 		rtcfg_namespace		= malloc(strlen(argv[optind]) * 2 + 4);
 103 > 		cp2 = rtcfg_namespace;
 104 > 		*cp2++ = '"';
 105 > 		*cp2++ = '_';
 106 > 		for (cp1 = (char *)argv[optind]; *cp1; cp1++)
 107 > 		{
 108 > 			if (*cp1 == '"')
 109 > 				*cp2++ = '"';
 110 > 			*cp2++ = *cp1;
 111 > 		}
 112 > 		*cp2++ = '"';
 113 > 		*cp2 = '\0';
 114 214,218c205,225
 115 < 	/*
 116 < 	 * Start the event scheduling system
 117 < 	 */
 118 < 	if (sched_start_mainloop() < 0)
 119 < 		slon_exit(-1);
 120 ---
 121 > 		/*
 122 > 		 * Remember the connection information for the local node.
 123 > 		 */
 124 > 		rtcfg_conninfo = (char *)argv[++optind];
 125 > 
 126 > 		/*
 127 > 		 * Connect to the local database for reading the initial configuration
 128 > 		 */
 129 > 		startup_conn = PQconnectdb(rtcfg_conninfo);
 130 > 		if (startup_conn == NULL)
 131 > 		{
 132 > 			slon_log(SLON_FATAL, "main: PQconnectdb() failed\n");
 133 > 			slon_exit(-1);
 134 > 		}
 135 > 		if (PQstatus(startup_conn) != CONNECTION_OK)
 136 > 		{
 137 > 			slon_log(SLON_FATAL, "main: Cannot connect to local database - %s",
 138 > 					PQerrorMessage(startup_conn));
 139 > 			PQfinish(startup_conn);
 140 > 			slon_exit(-1);
 141 > 		}
 142 220c227,236
 143 < 	slon_log(SLON_CONFIG, "main: loading current cluster configuration\n");
 144 ---
 145 > 		/*
 146 > 		 * Get our local node ID
 147 > 		 */
 148 > 		rtcfg_nodeid = db_getLocalNodeId(startup_conn);
 149 > 		if (rtcfg_nodeid < 0)
 150 > 		{
 151 > 			slon_log(SLON_FATAL, "main: Node is not initialized properly\n");
 152 > 			slon_exit(-1);
 153 > 		}
 154 > 		slon_log(SLON_CONFIG, "main: local node id = %d\n", rtcfg_nodeid);
 155 222,231c238,258
 156 < 	/*
 157 < 	 * Begin a transaction
 158 < 	 */
 159 < 	res = PQexec(startup_conn, 
 160 < 			"start transaction; "
 161 < 			"set transaction isolation level serializable;");
 162 < 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
 163 < 	{
 164 < 		slon_log(SLON_FATAL, "Cannot start transaction - %s",
 165 < 				PQresultErrorMessage(res));
 166 ---
 167 > 		/*
 168 > 		 * Start the event scheduling system
 169 > 		 */
 170 > 		if (sched_start_mainloop() < 0)
 171 > 			slon_exit(-1);
 172 > 
 173 > 		slon_log(SLON_CONFIG, "main: loading current cluster configuration\n");
 174 > 
 175 > 		/*
 176 > 		 * Begin a transaction
 177 > 		 */
 178 > 		res = PQexec(startup_conn, 
 179 > 				"start transaction; "
 180 > 				"set transaction isolation level serializable;");
 181 > 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
 182 > 		{
 183 > 			slon_log(SLON_FATAL, "Cannot start transaction - %s",
 184 > 					PQresultErrorMessage(res));
 185 > 			PQclear(res);
 186 > 			slon_exit(-1);
 187 > 		}
 188 233,235d259
 189 < 		slon_exit(-1);
 190 < 	}
 191 < 	PQclear(res);
 192 237,271c261,280
 193 < 	/*
 194 < 	 * Read configuration table sl_node
 195 < 	 */
 196 < 	dstring_init(&query);
 197 < 	slon_mkquery(&query, 
 198 < 			"select no_id, no_active, no_comment, "
 199 < 			"    (select coalesce(max(con_seqno),0) from %s.sl_confirm "
 200 < 			"        where con_origin = no_id and con_received = %d) "
 201 < 			"        as last_event "
 202 < 			"from %s.sl_node "
 203 < 			"order by no_id; ",
 204 < 			rtcfg_namespace, rtcfg_nodeid, rtcfg_namespace);
 205 < 	res = PQexec(startup_conn, dstring_data(&query));
 206 < 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 207 < 	{
 208 < 		slon_log(SLON_FATAL, "main: Cannot get node list - %s",
 209 < 				PQresultErrorMessage(res));
 210 < 		PQclear(res);
 211 < 		dstring_free(&query);
 212 < 		slon_exit(-1);
 213 < 	}
 214 < 	for (i = 0, n = PQntuples(res); i < n; i++)
 215 < 	{
 216 < 		int		no_id		= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 217 < 		int		no_active	= (*PQgetvalue(res, i, 1) == 't') ? 1 : 0;
 218 < 		char   *no_comment	= PQgetvalue(res, i, 2);
 219 < 		int64	last_event;
 220 < 
 221 < 		if (no_id == rtcfg_nodeid)
 222 < 		{
 223 < 			/*
 224 < 			 * Complete our own local node entry
 225 < 			 */
 226 < 			rtcfg_nodeactive  = no_active;
 227 < 			rtcfg_nodecomment = strdup(no_comment);
 228 ---
 229 > 		/*
 230 > 		 * Read configuration table sl_node
 231 > 		 */
 232 > 		dstring_init(&query);
 233 > 		slon_mkquery(&query, 
 234 > 				"select no_id, no_active, no_comment, "
 235 > 				"    (select coalesce(max(con_seqno),0) from %s.sl_confirm "
 236 > 				"        where con_origin = no_id and con_received = %d) "
 237 > 				"        as last_event "
 238 > 				"from %s.sl_node "
 239 > 				"order by no_id; ",
 240 > 				rtcfg_namespace, rtcfg_nodeid, rtcfg_namespace);
 241 > 		res = PQexec(startup_conn, dstring_data(&query));
 242 > 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 243 > 		{
 244 > 			slon_log(SLON_FATAL, "main: Cannot get node list - %s",
 245 > 					PQresultErrorMessage(res));
 246 > 			PQclear(res);
 247 > 			dstring_free(&query);
 248 > 			slon_exit(-1);
 249 273c282
 250 < 		else
 251 ---
 252 > 		for (i = 0, n = PQntuples(res); i < n; i++)
 253 275,287c284,312
 254 < 			/*
 255 < 			 * Add a remote node
 256 < 			 */
 257 < 			slon_scanint64(PQgetvalue(res, i, 3), &last_event);
 258 < 			rtcfg_storeNode(no_id, no_comment);
 259 < 			rtcfg_setNodeLastEvent(no_id, last_event);
 260 < 
 261 < 			/*
 262 < 			 * If it is active, remember for activation just before
 263 < 			 * we start processing events.
 264 < 			 */
 265 < 			if (no_active)
 266 < 				rtcfg_needActivate(no_id);
 267 ---
 268 > 			int		no_id		= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 269 > 			int		no_active	= (*PQgetvalue(res, i, 1) == 't') ? 1 : 0;
 270 > 			char   *no_comment	= PQgetvalue(res, i, 2);
 271 > 			int64	last_event;
 272 > 
 273 > 			if (no_id == rtcfg_nodeid)
 274 > 			{
 275 > 				/*
 276 > 				 * Complete our own local node entry
 277 > 				 */
 278 > 				rtcfg_nodeactive  = no_active;
 279 > 				rtcfg_nodecomment = strdup(no_comment);
 280 > 			}
 281 > 			else
 282 > 			{
 283 > 				/*
 284 > 				 * Add a remote node
 285 > 				 */
 286 > 				slon_scanint64(PQgetvalue(res, i, 3), &last_event);
 287 > 				rtcfg_storeNode(no_id, no_comment);
 288 > 				rtcfg_setNodeLastEvent(no_id, last_event);
 289 > 
 290 > 				/*
 291 > 				 * If it is active, remember for activation just before
 292 > 				 * we start processing events.
 293 > 				 */
 294 > 				if (no_active)
 295 > 					rtcfg_needActivate(no_id);
 296 > 			}
 297 289,303d313
 298 < 	}
 299 < 	PQclear(res);
 300 < 
 301 < 	/*
 302 < 	 * Read configuration table sl_path - the interesting pieces
 303 < 	 */
 304 < 	slon_mkquery(&query, 
 305 < 			"select pa_server, pa_conninfo, pa_connretry "
 306 < 			"from %s.sl_path where pa_client = %d",
 307 < 			rtcfg_namespace, rtcfg_nodeid);
 308 < 	res = PQexec(startup_conn, dstring_data(&query));
 309 < 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 310 < 	{
 311 < 		slon_log(SLON_FATAL, "main: Cannot get path config - %s",
 312 < 				PQresultErrorMessage(res));
 313 305,312d314
 314 < 		dstring_free(&query);
 315 < 		slon_exit(-1);
 316 < 	}
 317 < 	for (i = 0, n = PQntuples(res); i < n; i++)
 318 < 	{
 319 < 		int		pa_server		= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 320 < 		char   *pa_conninfo		= PQgetvalue(res, i, 1);
 321 < 		int		pa_connretry	= (int) strtol(PQgetvalue(res, i, 2), NULL, 10);
 322 314,316c316,336
 323 < 		rtcfg_storePath(pa_server, pa_conninfo, pa_connretry);
 324 < 	}
 325 < 	PQclear(res);
 326 ---
 327 > 		/*
 328 > 		 * Read configuration table sl_path - the interesting pieces
 329 > 		 */
 330 > 		slon_mkquery(&query, 
 331 > 				"select pa_server, pa_conninfo, pa_connretry "
 332 > 				"from %s.sl_path where pa_client = %d",
 333 > 				rtcfg_namespace, rtcfg_nodeid);
 334 > 		res = PQexec(startup_conn, dstring_data(&query));
 335 > 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 336 > 		{
 337 > 			slon_log(SLON_FATAL, "main: Cannot get path config - %s",
 338 > 					PQresultErrorMessage(res));
 339 > 			PQclear(res);
 340 > 			dstring_free(&query);
 341 > 			slon_exit(-1);
 342 > 		}
 343 > 		for (i = 0, n = PQntuples(res); i < n; i++)
 344 > 		{
 345 > 			int		pa_server		= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 346 > 			char   *pa_conninfo		= PQgetvalue(res, i, 1);
 347 > 			int		pa_connretry	= (int) strtol(PQgetvalue(res, i, 2), NULL, 10);
 348 318,329c338,339
 349 < 	/*
 350 < 	 * Read configuration table sl_listen - the interesting pieces
 351 < 	 */
 352 < 	slon_mkquery(&query, 
 353 < 			"select li_origin, li_provider "
 354 < 			"from %s.sl_listen where li_receiver = %d",
 355 < 			rtcfg_namespace, rtcfg_nodeid);
 356 < 	res = PQexec(startup_conn, dstring_data(&query));
 357 < 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 358 < 	{
 359 < 		slon_log(SLON_FATAL, "main: Cannot get listen config - %s",
 360 < 				PQresultErrorMessage(res));
 361 ---
 362 > 			rtcfg_storePath(pa_server, pa_conninfo, pa_connretry);
 363 > 		}
 364 331,337d340
 365 < 		dstring_free(&query);
 366 < 		slon_exit(-1);
 367 < 	}
 368 < 	for (i = 0, n = PQntuples(res); i < n; i++)
 369 < 	{
 370 < 		int		li_origin	= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 371 < 		int		li_provider	= (int) strtol(PQgetvalue(res, i, 1), NULL, 10);
 372 339,341c342,361
 373 < 		rtcfg_storeListen(li_origin, li_provider);
 374 < 	}
 375 < 	PQclear(res);
 376 ---
 377 > 		/*
 378 > 		 * Read configuration table sl_listen - the interesting pieces
 379 > 		 */
 380 > 		slon_mkquery(&query, 
 381 > 				"select li_origin, li_provider "
 382 > 				"from %s.sl_listen where li_receiver = %d",
 383 > 				rtcfg_namespace, rtcfg_nodeid);
 384 > 		res = PQexec(startup_conn, dstring_data(&query));
 385 > 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 386 > 		{
 387 > 			slon_log(SLON_FATAL, "main: Cannot get listen config - %s",
 388 > 					PQresultErrorMessage(res));
 389 > 			PQclear(res);
 390 > 			dstring_free(&query);
 391 > 			slon_exit(-1);
 392 > 		}
 393 > 		for (i = 0, n = PQntuples(res); i < n; i++)
 394 > 		{
 395 > 			int		li_origin	= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 396 > 			int		li_provider	= (int) strtol(PQgetvalue(res, i, 1), NULL, 10);
 397 343,354c363,364
 398 < 	/*
 399 < 	 * Read configuration table sl_set
 400 < 	 */
 401 < 	slon_mkquery(&query, 
 402 < 			"select set_id, set_origin, set_comment "
 403 < 			"from %s.sl_set",
 404 < 			rtcfg_namespace);
 405 < 	res = PQexec(startup_conn, dstring_data(&query));
 406 < 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 407 < 	{
 408 < 		slon_log(SLON_FATAL, "main: Cannot get set config - %s",
 409 < 				PQresultErrorMessage(res));
 410 ---
 411 > 			rtcfg_storeListen(li_origin, li_provider);
 412 > 		}
 413 356,363d365
 414 < 		dstring_free(&query);
 415 < 		slon_exit(-1);
 416 < 	}
 417 < 	for (i = 0, n = PQntuples(res); i < n; i++)
 418 < 	{
 419 < 		int		set_id		= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 420 < 		int		set_origin	= (int) strtol(PQgetvalue(res, i, 1), NULL, 10);
 421 < 		char   *set_comment = PQgetvalue(res, i, 2);
 422 365,367c367,387
 423 < 		rtcfg_storeSet(set_id, set_origin, set_comment);
 424 < 	}
 425 < 	PQclear(res);
 426 ---
 427 > 		/*
 428 > 		 * Read configuration table sl_set
 429 > 		 */
 430 > 		slon_mkquery(&query, 
 431 > 				"select set_id, set_origin, set_comment "
 432 > 				"from %s.sl_set",
 433 > 				rtcfg_namespace);
 434 > 		res = PQexec(startup_conn, dstring_data(&query));
 435 > 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 436 > 		{
 437 > 			slon_log(SLON_FATAL, "main: Cannot get set config - %s",
 438 > 					PQresultErrorMessage(res));
 439 > 			PQclear(res);
 440 > 			dstring_free(&query);
 441 > 			slon_exit(-1);
 442 > 		}
 443 > 		for (i = 0, n = PQntuples(res); i < n; i++)
 444 > 		{
 445 > 			int		set_id		= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 446 > 			int		set_origin	= (int) strtol(PQgetvalue(res, i, 1), NULL, 10);
 447 > 			char   *set_comment = PQgetvalue(res, i, 2);
 448 369,381c389,390
 449 < 	/*
 450 < 	 * Read configuration table sl_subscribe - our subscriptions only
 451 < 	 */
 452 < 	slon_mkquery(&query, 
 453 < 			"select sub_set, sub_provider, sub_forward, sub_active "
 454 < 			"from %s.sl_subscribe "
 455 < 			"where sub_receiver = %d",
 456 < 			rtcfg_namespace, rtcfg_nodeid);
 457 < 	res = PQexec(startup_conn, dstring_data(&query));
 458 < 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 459 < 	{
 460 < 		slon_log(SLON_FATAL, "main: Cannot get subscription config - %s",
 461 < 				PQresultErrorMessage(res));
 462 ---
 463 > 			rtcfg_storeSet(set_id, set_origin, set_comment);
 464 > 		}
 465 383,397d391
 466 < 		dstring_free(&query);
 467 < 		slon_exit(-1);
 468 < 	}
 469 < 	for (i = 0, n = PQntuples(res); i < n; i++)
 470 < 	{
 471 < 		int		sub_set			= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 472 < 		int		sub_provider	= (int) strtol(PQgetvalue(res, i, 1), NULL, 10);
 473 < 		char   *sub_forward		= PQgetvalue(res, i, 2);
 474 < 		char   *sub_active		= PQgetvalue(res, i, 3);
 475 < 
 476 < 		rtcfg_storeSubscribe(sub_set, sub_provider, sub_forward);
 477 < 		if (*sub_active == 't')
 478 < 			rtcfg_enableSubscription(sub_set, sub_provider, sub_forward);
 479 < 	}
 480 < 	PQclear(res);
 481 399,410c393,420
 482 < 	/*
 483 < 	 * Remember the last known local event sequence
 484 < 	 */
 485 < 	slon_mkquery(&query,
 486 < 			"select coalesce(max(ev_seqno), -1) from %s.sl_event "
 487 < 			"where ev_origin = '%d'",
 488 < 			rtcfg_namespace, rtcfg_nodeid);
 489 < 	res = PQexec(startup_conn, dstring_data(&query));
 490 < 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 491 < 	{
 492 < 		slon_log(SLON_FATAL, "main: Cannot get last local eventid - %s",
 493 < 				PQresultErrorMessage(res));
 494 ---
 495 > 		/*
 496 > 		 * Read configuration table sl_subscribe - our subscriptions only
 497 > 		 */
 498 > 		slon_mkquery(&query, 
 499 > 				"select sub_set, sub_provider, sub_forward, sub_active "
 500 > 				"from %s.sl_subscribe "
 501 > 				"where sub_receiver = %d",
 502 > 				rtcfg_namespace, rtcfg_nodeid);
 503 > 		res = PQexec(startup_conn, dstring_data(&query));
 504 > 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 505 > 		{
 506 > 			slon_log(SLON_FATAL, "main: Cannot get subscription config - %s",
 507 > 					PQresultErrorMessage(res));
 508 > 			PQclear(res);
 509 > 			dstring_free(&query);
 510 > 			slon_exit(-1);
 511 > 		}
 512 > 		for (i = 0, n = PQntuples(res); i < n; i++)
 513 > 		{
 514 > 			int		sub_set			= (int) strtol(PQgetvalue(res, i, 0), NULL, 10);
 515 > 			int		sub_provider	= (int) strtol(PQgetvalue(res, i, 1), NULL, 10);
 516 > 			char   *sub_forward		= PQgetvalue(res, i, 2);
 517 > 			char   *sub_active		= PQgetvalue(res, i, 3);
 518 > 
 519 > 			rtcfg_storeSubscribe(sub_set, sub_provider, sub_forward);
 520 > 			if (*sub_active == 't')
 521 > 				rtcfg_enableSubscription(sub_set, sub_provider, sub_forward);
 522 > 		}
 523 412,418c422,439
 524 < 		dstring_free(&query);
 525 < 		slon_exit(-1);
 526 < 	}
 527 < 	if (PQntuples(res) == 0)
 528 < 		strcpy(rtcfg_lastevent, "-1");
 529 < 	else
 530 < 		if (PQgetisnull(res, 0, 0))
 531 ---
 532 > 
 533 > 		/*
 534 > 		 * Remember the last known local event sequence
 535 > 		 */
 536 > 		slon_mkquery(&query,
 537 > 				"select coalesce(max(ev_seqno), -1) from %s.sl_event "
 538 > 				"where ev_origin = '%d'",
 539 > 				rtcfg_namespace, rtcfg_nodeid);
 540 > 		res = PQexec(startup_conn, dstring_data(&query));
 541 > 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 542 > 		{
 543 > 			slon_log(SLON_FATAL, "main: Cannot get last local eventid - %s",
 544 > 					PQresultErrorMessage(res));
 545 > 			PQclear(res);
 546 > 			dstring_free(&query);
 547 > 			slon_exit(-1);
 548 > 		}
 549 > 		if (PQntuples(res) == 0)
 550 421,435c442,462
 551 < 			strcpy(rtcfg_lastevent, PQgetvalue(res, 0, 0));
 552 < 	PQclear(res);
 553 < 	dstring_free(&query);
 554 < 	slon_log(SLON_DEBUG2, 
 555 < 			"main: last local event sequence = %s\n", 
 556 < 			rtcfg_lastevent);
 557 < 
 558 < 	/*
 559 < 	 * Rollback the transaction we used to get the config snapshot
 560 < 	 */
 561 < 	res = PQexec(startup_conn, "rollback transaction;");
 562 < 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
 563 < 	{
 564 < 		slon_log(SLON_FATAL, "main: Cannot rollback transaction - %s",
 565 < 				PQresultErrorMessage(res));
 566 ---
 567 > 			if (PQgetisnull(res, 0, 0))
 568 > 				strcpy(rtcfg_lastevent, "-1");
 569 > 			else
 570 > 				strcpy(rtcfg_lastevent, PQgetvalue(res, 0, 0));
 571 > 		PQclear(res);
 572 > 		dstring_free(&query);
 573 > 		slon_log(SLON_DEBUG2, 
 574 > 				"main: last local event sequence = %s\n", 
 575 > 				rtcfg_lastevent);
 576 > 
 577 > 		/*
 578 > 		 * Rollback the transaction we used to get the config snapshot
 579 > 		 */
 580 > 		res = PQexec(startup_conn, "rollback transaction;");
 581 > 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
 582 > 		{
 583 > 			slon_log(SLON_FATAL, "main: Cannot rollback transaction - %s",
 584 > 					PQresultErrorMessage(res));
 585 > 			PQclear(res);
 586 > 			slon_exit(-1);
 587 > 		}
 588 437,439d463
 589 < 		slon_exit(-1);
 590 < 	}
 591 < 	PQclear(res);
 592 441,444c465,468
 593 < 	/*
 594 < 	 * Done with the startup, don't need the local connection any more.
 595 < 	 */
 596 < 	PQfinish(startup_conn);
 597 ---
 598 > 		/*
 599 > 		 * Done with the startup, don't need the local connection any more.
 600 > 		 */
 601 > 		PQfinish(startup_conn);
 602 446c470
 603 < 	slon_log(SLON_CONFIG, "main: configuration complete - starting threads\n");
 604 ---
 605 > 		slon_log(SLON_CONFIG, "main: configuration complete - starting threads\n");
 606 448,463c472,487
 607 < 	/*
 608 < 	 * Create the local event thread that is monitoring
 609 < 	 * the local node for administrative events to adjust the
 610 < 	 * configuration at runtime.
 611 < 	 * We wait here until the local listen thread has checked that
 612 < 	 * there is no other slon daemon running.
 613 < 	 */
 614 < 	pthread_mutex_lock(&slon_wait_listen_lock);
 615 < 	if (pthread_create(&local_event_thread, NULL, localListenThread_main, NULL) < 0)
 616 < 	{
 617 < 		slon_log(SLON_FATAL, "main: cannot create localListenThread - %s\n",
 618 < 				strerror(errno));
 619 < 		slon_abort();
 620 < 	}
 621 < 	pthread_cond_wait(&slon_wait_listen_cond, &slon_wait_listen_lock);
 622 < 	pthread_mutex_unlock(&slon_wait_listen_lock);
 623 ---
 624 > 		/*
 625 > 		 * Create the local event thread that is monitoring
 626 > 		 * the local node for administrative events to adjust the
 627 > 		 * configuration at runtime.
 628 > 		 * We wait here until the local listen thread has checked that
 629 > 		 * there is no other slon daemon running.
 630 > 		 */
 631 > 		pthread_mutex_lock(&slon_wait_listen_lock);
 632 > 		if (pthread_create(&local_event_thread, NULL, localListenThread_main, NULL) < 0)
 633 > 		{
 634 > 			slon_log(SLON_FATAL, "main: cannot create localListenThread - %s\n",
 635 > 					strerror(errno));
 636 > 			slon_abort();
 637 > 		}
 638 > 		pthread_cond_wait(&slon_wait_listen_cond, &slon_wait_listen_lock);
 639 > 		pthread_mutex_unlock(&slon_wait_listen_lock);
 640 465,468c489,503
 641 < 	/*
 642 < 	 * Enable all nodes that are active
 643 < 	 */
 644 < 	rtcfg_doActivate();
 645 ---
 646 > 		/*
 647 > 		 * Enable all nodes that are active
 648 > 		 */
 649 > 		rtcfg_doActivate();
 650 > 
 651 > 		/*
 652 > 		 * Create the local cleanup thread that will remove old
 653 > 		 * events and log data.
 654 > 		 */
 655 > 		if (pthread_create(&local_cleanup_thread, NULL, cleanupThread_main, NULL) < 0)
 656 > 		{
 657 > 			slon_log(SLON_FATAL, "main: cannot create cleanupThread - %s\n",
 658 > 					strerror(errno));
 659 > 			slon_abort();
 660 > 		}
 661 470,479c505,514
 662 < 	/*
 663 < 	 * Create the local cleanup thread that will remove old
 664 < 	 * events and log data.
 665 < 	 */
 666 < 	if (pthread_create(&local_cleanup_thread, NULL, cleanupThread_main, NULL) < 0)
 667 < 	{
 668 < 		slon_log(SLON_FATAL, "main: cannot create cleanupThread - %s\n",
 669 < 				strerror(errno));
 670 < 		slon_abort();
 671 < 	}
 672 ---
 673 > 		/*
 674 > 		 * Create the local sync thread that will generate SYNC
 675 > 		 * events if we had local database updates.
 676 > 		 */
 677 > 		if (pthread_create(&local_sync_thread, NULL, syncThread_main, NULL) < 0)
 678 > 		{
 679 > 			slon_log(SLON_FATAL, "main: cannot create syncThread - %s\n",
 680 > 					strerror(errno));
 681 > 			slon_abort();
 682 > 		}
 683 481,490c516,525
 684 < 	/*
 685 < 	 * Create the local sync thread that will generate SYNC
 686 < 	 * events if we had local database updates.
 687 < 	 */
 688 < 	if (pthread_create(&local_sync_thread, NULL, syncThread_main, NULL) < 0)
 689 < 	{
 690 < 		slon_log(SLON_FATAL, "main: cannot create syncThread - %s\n",
 691 < 				strerror(errno));
 692 < 		slon_abort();
 693 < 	}
 694 ---
 695 > 		/*
 696 > 		 * Wait until the scheduler has shut down all remote connections
 697 > 		 */
 698 > 		slon_log(SLON_DEBUG1, "main: running scheduler mainloop\n");
 699 > 		if (sched_wait_mainloop() < 0)
 700 > 		{
 701 > 			slon_log(SLON_FATAL, "main: scheduler returned with error\n");
 702 > 			slon_abort();
 703 > 		}
 704 > 		slon_log(SLON_DEBUG1, "main: scheduler mainloop returned\n");
 705 492,501c527,533
 706 < 	/*
 707 < 	 * Wait until the scheduler has shut down all remote connections
 708 < 	 */
 709 < 	slon_log(SLON_DEBUG1, "main: running scheduler mainloop\n");
 710 < 	if (sched_wait_mainloop() < 0)
 711 < 	{
 712 < 		slon_log(SLON_FATAL, "main: scheduler returned with error\n");
 713 < 		slon_abort();
 714 < 	}
 715 < 	slon_log(SLON_DEBUG1, "main: scheduler mainloop returned\n");
 716 ---
 717 > 		/*
 718 > 		 * Wait for all remote threads to finish
 719 > 		 */
 720 > 		main_thread = pthread_self();
 721 > 		main_argv = argv;
 722 > 		signal(SIGALRM, sigalrmhandler);
 723 > 		alarm(20);
 724 503,509c535
 725 < 	/*
 726 < 	 * Wait for all remote threads to finish
 727 < 	 */
 728 < 	main_thread = pthread_self();
 729 < 	main_argv = argv;
 730 < 	signal(SIGALRM, sigalrmhandler);
 731 < 	alarm(20);
 732 ---
 733 > 		rtcfg_joinAllRemoteThreads();
 734 511c537
 735 < 	rtcfg_joinAllRemoteThreads();
 736 ---
 737 > 		alarm(0);
 738 513c539,552
 739 < 	alarm(0);
 740 ---
 741 > 		/*
 742 > 		 * Wait for the local threads to finish
 743 > 		 */
 744 > 		if (pthread_join(local_event_thread, NULL) < 0)
 745 > 			slon_log(SLON_ERROR, "main: cannot join localListenThread - %s\n",
 746 > 					strerror(errno));
 747 > 
 748 > 		if (pthread_join(local_cleanup_thread, NULL) < 0)
 749 > 			slon_log(SLON_ERROR, "main: cannot join cleanupThread - %s\n",
 750 > 					strerror(errno));
 751 > 
 752 > 		if (pthread_join(local_sync_thread, NULL) < 0)
 753 > 			slon_log(SLON_ERROR, "main: cannot join syncThread - %s\n",
 754 > 					strerror(errno));
 755 515,528c554,561
 756 < 	/*
 757 < 	 * Wait for the local threads to finish
 758 < 	 */
 759 < 	if (pthread_join(local_event_thread, NULL) < 0)
 760 < 		slon_log(SLON_ERROR, "main: cannot join localListenThread - %s\n",
 761 < 				strerror(errno));
 762 < 
 763 < 	if (pthread_join(local_cleanup_thread, NULL) < 0)
 764 < 		slon_log(SLON_ERROR, "main: cannot join cleanupThread - %s\n",
 765 < 				strerror(errno));
 766 < 
 767 < 	if (pthread_join(local_sync_thread, NULL) < 0)
 768 < 		slon_log(SLON_ERROR, "main: cannot join syncThread - %s\n",
 769 < 				strerror(errno));
 770 ---
 771 > 		if (slon_restart_request)
 772 > 		{
 773 > 			slon_log(SLON_DEBUG1, "main: restart requested\n");
 774 > 			execvp(argv[0], argv);
 775 > 			slon_log(SLON_FATAL,
 776 > 					"main: cannot restart via execvp(): %s\n", strerror(errno));
 777 > 			exit(-1);
 778 > 		}
 779 530,536c563,569
 780 < 	if (slon_restart_request)
 781 < 	{
 782 < 		slon_log(SLON_DEBUG1, "main: restart requested\n");
 783 < 		execvp(argv[0], argv);
 784 < 		slon_log(SLON_FATAL,
 785 < 				"main: cannot restart via execvp(): %s\n", strerror(errno));
 786 < 		exit(-1);
 787 ---
 788 > 		/*
 789 > 		 * That's it.
 790 > 		 */
 791 > 		slon_log(SLON_DEBUG1, "main: done\n");
 792 > 		unlink(pidfilename);
 793 > 		free(pidfilename);
 794 > 		
 795 538,542d570
 796 < 
 797 < 	/*
 798 < 	 * That's it.
 799 < 	 */
 800 < 	slon_log(SLON_DEBUG1, "main: done\n");