Merge branch 'PGPROEE9_6_scheduler' into PGPROEE9_6

Author: vbwagner

contrib/pgpro_scheduler/Makefile

@@ -20,3 +20,6 @@ else
 	include $(top_builddir)/src/Makefile.global
 	include $(top_srcdir)/contrib/contrib-global.mk
 endif
+
+#check: temp-install
+#	$(prove_check)

contrib/pgpro_scheduler/pgpro_scheduler--2.0.sql

@@ -5,9 +5,10 @@ SET search_path TO schedule;
 
 CREATE TYPE job_status_t AS ENUM ('working', 'done', 'error');
 CREATE TYPE job_at_status_t AS ENUM ('submitted', 'processing', 'done');
+CREATE SEQUENCE schedule.at_jobs_submitted_id_seq;
 
 CREATE TABLE at_jobs_submitted(
-   id SERIAL PRIMARY KEY,
+   id int PRIMARY KEY,
    node text,
    name text,
    comments text,
@@ -37,6 +38,8 @@ ALTER TABLE at_jobs_done ADD status boolean;
 ALTER TABLE at_jobs_done ADD reason text;
 ALTER TABLE at_jobs_done ADD done_time timestamp with time zone default now();
 
+ALTER TABLE at_jobs_submitted ALTER id SET default nextval('schedule.at_jobs_submitted_id_seq');
+
 
 CREATE TABLE cron(
    id SERIAL PRIMARY KEY,
@@ -281,7 +284,12 @@ DECLARE
   cron_id INTEGER;
 BEGIN
   cron_id := NEW.id; 
-  IF NOT NEW.active OR NEW.broken OR NEW.rule <> OLD.rule OR NEW.postpone <> OLD.postpone  THEN
+  IF NOT NEW.active OR NEW.broken OR
+  	coalesce(NEW.rule <> OLD.rule, true) OR
+	coalesce(NEW.postpone <> OLD.postpone, true)  OR
+	coalesce(NEW.start_date <> OLD.start_date, true) OR
+	coalesce(NEW.end_date <> OLD.end_date, true)
+  THEN
      DELETE FROM at WHERE cron = cron_id AND active = false;
   END IF;
   RETURN OLD;
@@ -1203,9 +1211,9 @@ BEGIN
 	END IF;
 
 	IF usename = '___all___' THEN
-		sql_cmd := 'SELECT * FROM log as l , cron as cron WHERE cron.id = l.cron';
+		sql_cmd := 'SELECT * FROM log as l LEFT OUTER JOIN cron ON cron.id = l.cron';
 	ELSE
-		sql_cmd := 'SELECT * FROM log as l , cron as cron WHERE cron.executor = ''' || usename || ''' AND cron.id = l.cron';
+		sql_cmd := 'SELECT * FROM log as l LEFT OUTER JOIN cron ON cron.executor = ''' || usename || ''' AND cron.id = l.cron';
 	END IF;
 
 	FOR ii IN EXECUTE sql_cmd LOOP
@@ -1338,7 +1346,7 @@ CREATE VIEW all_job_status AS
 		attempt, resubmit_limit, postpone as max_wait_interval,
 		max_run_time as max_duration, submit_time,
 		start_time, status as is_success, reason as error, done_time,
-		'processing'::job_at_status_t status
+		'done'::job_at_status_t status
 	FROM schedule.at_jobs_done 
 		UNION 
 	SELECT

contrib/pgpro_scheduler/src/char_array.c

@@ -25,6 +25,7 @@ char_array_t *makeCharArray(void)
 
 char_array_t *sortCharArray(char_array_t *a)
 {
+	if(a->n <= 1) return a;
 	qsort(a->data, a->n, sizeof(char *), __sort_char_string);
 
 	return a;

contrib/pgpro_scheduler/src/memutils.c

@@ -29,5 +29,6 @@ void *worker_alloc(Size size)
 
 void delete_worker_mem_ctx(void)
 {
+	MemoryContextSwitchTo(TopMemoryContext);
 	MemoryContextDelete(SchedulerWorkerContext);
 }

contrib/pgpro_scheduler/src/pgpro_scheduler.c

@@ -212,7 +212,7 @@ char_array_t *readBasesToCheck(void)
 	pgstat_report_activity(STATE_RUNNING, "read configuration");
 	result = makeCharArray();
 
-	value = GetConfigOption("schedule.database", 1, 0);
+	value = GetConfigOption("schedule.database", true, false);
 	if(!value || strlen(value) == 0)
 	{
 		return result;
@@ -254,6 +254,7 @@ char_array_t *readBasesToCheck(void)
 	pfree(clean_value);
 	if(names->n == 0)
 	{
+		destroyCharArray(names);
 		return result;
 	}
 
@@ -264,45 +265,39 @@ char_array_t *readBasesToCheck(void)
 		appendStringInfo(&sql, "'%s'", names->data[i]);
 		if(i + 1  != names->n) appendStringInfo(&sql, ",");
 	} 
+	destroyCharArray(names);
 	appendStringInfo(&sql, ")");
-	SetCurrentStatementStartTimestamp();
-	StartTransactionCommand();
-	SPI_connect();
-	PushActiveSnapshot(GetTransactionSnapshot());
+
+	START_SPI_SNAP();
 
 	ret = SPI_execute(sql.data, true, 0);
 	if (ret != SPI_OK_SELECT)
 	{
-		SPI_finish();
-		PopActiveSnapshot();
-		CommitTransactionCommand();
+		STOP_SPI_SNAP();
+		elog(ERROR, "cannot select from pg_database");
 	}
-	destroyCharArray(names);
 	processed  = SPI_processed;
 	if(processed == 0)
 	{
-		SPI_finish();
-		PopActiveSnapshot();
-		CommitTransactionCommand();
+		STOP_SPI_SNAP();
 		return result;
 	}
 	for(i=0; i < processed; i++)
 	{
 		clean_value = SPI_getvalue(SPI_tuptable->vals[i], SPI_tuptable->tupdesc, 1);
 		pushCharArray(result, clean_value);
 	}
-	SPI_finish();
-	PopActiveSnapshot();
-	CommitTransactionCommand();
+	STOP_SPI_SNAP();
 	sortCharArray(result);
+
 	return result;
 }
 
 void parent_scheduler_main(Datum arg)
 {
 	int rc = 0, i;
 	char_array_t *names = NULL;
-	schd_managers_poll_t *poll;
+	schd_managers_poll_t *pool;
 	schd_manager_share_t *shared;
 	bool refresh = false;
 
@@ -319,10 +314,10 @@ void parent_scheduler_main(Datum arg)
 
 	BackgroundWorkerInitializeConnection("postgres", NULL);
 	names = readBasesToCheck();
-	poll = initSchedulerManagerPool(names);
+	pool = initSchedulerManagerPool(names);
 	destroyCharArray(names);
 
-	set_supervisor_pgstatus(poll);
+	set_supervisor_pgstatus(pool);
 
 	while(!got_sigterm)
 	{
@@ -334,62 +329,62 @@ void parent_scheduler_main(Datum arg)
 			ProcessConfigFile(PGC_SIGHUP);
 			refresh = false;
 			names = NULL;
-			if(is_scheduler_enabled() != poll->enabled)
+			if(is_scheduler_enabled() != pool->enabled)
 			{
-				if(poll->enabled)
+				if(pool->enabled)
 				{
-					poll->enabled = false;
-					stopAllManagers(poll);
-					set_supervisor_pgstatus(poll);
+					pool->enabled = false;
+					stopAllManagers(pool);
+					set_supervisor_pgstatus(pool);
 				}
 				else
 				{
 					refresh = true;
-					poll->enabled = true;
+					pool->enabled = true;
 					names = readBasesToCheck();
 				}
 			}
-			else if(poll->enabled)
+			else if(pool->enabled)
 			{
 				names = readBasesToCheck();
-				if(isBaseListChanged(names, poll)) refresh = true;
+				if(isBaseListChanged(names, pool)) refresh = true;
 				else destroyCharArray(names);
 			}
 
 			if(refresh)
 			{
-				refreshManagers(names, poll);
-				set_supervisor_pgstatus(poll);
+				refreshManagers(names, pool);
+				set_supervisor_pgstatus(pool);
 				destroyCharArray(names);
 			}
 		}
 		else 
 		{
-			for(i=0; i < poll->n; i++)
+			for(i=0; i < pool->n; i++)
 			{
-				shared = dsm_segment_address(poll->workers[i]->shared);
+				shared = dsm_segment_address(pool->workers[i]->shared);
 
 				if(shared->setbychild)
 				{
-				/* elog(LOG, "got status change from: %s", poll->workers[i]->dbname); */
+				/* elog(LOG, "got status change from: %s", pool->workers[i]->dbname); */
 					shared->setbychild = false;
 					if(shared->status == SchdManagerConnected)
 					{
-						poll->workers[i]->connected = true;
+						pool->workers[i]->connected = true;
 					}
 					else if(shared->status == SchdManagerQuit)
 					{
-						removeManagerFromPoll(poll, poll->workers[i]->dbname, 1, true);
-						set_supervisor_pgstatus(poll);
+						removeManagerFromPoll(pool, pool->workers[i]->dbname, 1, true);
+						set_supervisor_pgstatus(pool);
 					}
 					else if(shared->status == SchdManagerDie)
 					{
-						removeManagerFromPoll(poll, poll->workers[i]->dbname, 1, false);
-						set_supervisor_pgstatus(poll);
+						removeManagerFromPoll(pool, pool->workers[i]->dbname, 1, false);
+						set_supervisor_pgstatus(pool);
 					}
 					else
 					{
-						elog(WARNING, "manager: %s set strange status: %d", poll->workers[i]->dbname, shared->status);
+						elog(WARNING, "manager: %s set strange status: %d", pool->workers[i]->dbname, shared->status);
 					}
 				}
 			}
@@ -399,7 +394,7 @@ void parent_scheduler_main(Datum arg)
 		CHECK_FOR_INTERRUPTS();
 		ResetLatch(MyLatch);
 	}
-	stopAllManagers(poll);
+	stopAllManagers(pool);
 	delete_worker_mem_ctx();
 
 	proc_exit(0);
@@ -413,7 +408,7 @@ pg_scheduler_startup(void)
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |
 		BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
-	worker.bgw_restart_time = BGW_NEVER_RESTART;
+	worker.bgw_restart_time = 10;
 	worker.bgw_main = NULL;
 	worker.bgw_notify_pid = 0;
 	worker.bgw_main_arg = Int32GetDatum(0);

contrib/pgpro_scheduler/src/sched_manager_poll.c

@@ -43,7 +43,7 @@ char *supervisor_state(schd_managers_poll_t *poll)
 
 	if(!poll->enabled)
 	{
-		status = palloc(sizeof(char) * 9);
+		status = worker_alloc(sizeof(char) * 9);
 		memcpy(status, "disabled", 8);
 		status[8] = 0;
 		return status;
@@ -52,13 +52,13 @@ char *supervisor_state(schd_managers_poll_t *poll)
 	len = dbnames ? strlen(dbnames): 0;
 	if(len == 0)
 	{
-		status = palloc(sizeof(char)*26);
+		status = worker_alloc(sizeof(char)*26);
 		memcpy(status, "waiting databases to set", 25);
 		status[25] = 0;
 	}
 	else
 	{
-		status = palloc(sizeof(char) * (len + 10));
+		status = worker_alloc(sizeof(char) * (len + 10));
 		memcpy(status, "work on: ", 9);
 		memcpy(status+9, dbnames, len);
 		status[len+9] = 0;
@@ -82,7 +82,7 @@ char *poll_dbnames(schd_managers_poll_t *poll)
 		if(i < (poll->n - 1))
 			appendStringInfo(&string, ", ");
 	}
-	out = palloc(sizeof(char) * (string.len + 1));
+	out = worker_alloc(sizeof(char) * (string.len + 1));
 	memcpy(out, string.data, string.len);
 	out[string.len] = 0;
 	pfree(string.data);
@@ -297,29 +297,18 @@ int addManagerToPoll(schd_managers_poll_t *poll, char *name, int sort)
 	schd_manager_t *man;
 	schd_manager_share_t *shm_data;
 	Size segsize;
-/*	shm_toc_estimator e;
-	shm_toc *toc; */
 	dsm_segment *seg;
 
-/*	shm_toc_initialize_estimator(&e);
-	shm_toc_estimate_chunk(&e, sizeof(schd_manager_share_t));
-	shm_toc_estimate_keys(&e, 1);
-	segsize = shm_toc_estimate(&e); */
 	segsize = (Size)sizeof(schd_manager_share_t);
 
 	CurrentResourceOwner = ResourceOwnerCreate(NULL, "pgpro_scheduler");
 	seg = dsm_create(segsize, 0);
 
-	man = palloc(sizeof(schd_manager_t));
-	man->dbname = palloc(sizeof(char *) * (strlen(name) + 1));
+	man = worker_alloc(sizeof(schd_manager_t));
+	man->dbname = worker_alloc(sizeof(char *) * (strlen(name) + 1));
 	man->connected = false;
 	memcpy(man->dbname, name, strlen(name) + 1);
 	man->shared = seg;
-/*	toc = shm_toc_create(PGPRO_SHM_TOC_MAGIC, dsm_segment_address(man->shared),
-	                         segsize);
-
-	shm_data = shm_toc_allocate(toc, sizeof(schd_manager_share_t));
-	shm_toc_insert(toc, 0, shm_data); */
 	shm_data = dsm_segment_address(man->shared);
 
 	shm_data->setbyparent = true;
@@ -331,7 +320,7 @@ int addManagerToPoll(schd_managers_poll_t *poll, char *name, int sort)
 		pos = poll->n++;
 		poll->workers = poll->workers ?
 			repalloc(poll->workers, sizeof(schd_manager_t *) * poll->n):
-			palloc(sizeof(schd_manager_t *));
+			worker_alloc(sizeof(schd_manager_t *));
 		poll->workers[pos] = man;
 		if(sort) _sortPollManagers(poll);