avoid with query

Author: Vladimir Ershov

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,

src/pgpro_scheduler.c

@@ -276,6 +276,7 @@ char_array_t *readBasesToCheck(void)
 		SPI_finish();
 		PopActiveSnapshot();
 		CommitTransactionCommand();
+		elog(ERROR, "cannot select from pg_database");
 	}
 	destroyCharArray(names);
 	processed  = SPI_processed;

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);
@@ -310,8 +310,8 @@ int addManagerToPoll(schd_managers_poll_t *poll, char *name, int sort)
 	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;
@@ -331,7 +331,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);
 

src/scheduler_executor.c

@@ -696,23 +696,24 @@ void at_executor_worker_main(Datum arg)
 int process_one_job(schd_executor_share_state_t *shared, schd_executor_status_t *status)
 {
 	char *error = NULL;
+	char *set_error = NULL;
 	job_t *job;
-	int ret;
+	int ret, set_ret;
 	char buff[512];
 
 	*status = shared->status = SchdExecutorWork;
 
 	pgstat_report_activity(STATE_RUNNING, "initialize at job");
 	START_SPI_SNAP();
 
-	job = get_next_at_job_with_lock(shared->nodename, &error);
-
+	/* job = get_next_at_job_with_lock(shared->nodename, &error); */
+	job = get_at_job_for_process(shared->nodename, &error);
 	if(!job)
 	{
 		if(error)
 		{
 			shared->status = SchdExecutorIdling;
-			elog(LOG, "AT EXECUTOR: ERROR: %s", error);
+			elog(LOG, "AT EXECUTOR ERROR: get job: %s", error);
 			pfree(error);
 			ABORT_SPI_SNAP();
 			return -1;
@@ -722,30 +723,35 @@ int process_one_job(schd_executor_share_state_t *shared, schd_executor_status_t
 		return 0;
 	}
 	current_job_id = job->cron_id;
-	if(!move_at_job_process(job->cron_id))
+/*	if(!move_at_job_process(job->cron_id))
 	{
 		elog(LOG, "AT EXECUTOR: error move to process");
 		ABORT_SPI_SNAP();
 		return -1;
-	}
+	} */
 	STOP_SPI_SNAP(); /* Commit changes */
-
-	START_SPI_SNAP();
 	pgstat_report_activity(STATE_RUNNING, "job initialized");
+	START_SPI_SNAP();
 
 	ResetAllOptions();
 	if(set_session_authorization_by_name(job->executor, &error) == InvalidOid)
 	{
 		if(error)
 		{
-			set_at_job_done(job, error, 0);
+			set_ret = set_at_job_done(job, error, 0, &set_error);
 			pfree(error);
 		}
 		else
 		{
-			set_at_job_done(job, "Unknown set session auth error", 0);
+			set_ret = set_at_job_done(job, "Unknown set session auth error", 0, &set_error);
 		}
 		shared->status = SchdExecutorIdling;
+		if(set_ret < 0)
+		{
+			elog(LOG, "AT-EXECUTOR ERROR: move after auth: %s", set_error);
+			ABORT_SPI_SNAP();
+			return -1;
+		}
 		STOP_SPI_SNAP();
 		return 1;
 	}
@@ -778,26 +784,40 @@ int process_one_job(schd_executor_share_state_t *shared, schd_executor_status_t
 	{
 		if(error)
 		{
-			set_at_job_done(job, error, resubmit_current_job);
+			set_ret = set_at_job_done(job, error, resubmit_current_job, &set_error);
 			pfree(error);
 		}
 		else
 		{
 			sprintf(buff, "error in command: code: %d", ret);
-			set_at_job_done(job, buff, resubmit_current_job);
+			set_ret = set_at_job_done(job, buff, resubmit_current_job, &set_error);
 		}
 	}
 	else
 	{
-		set_at_job_done(job, NULL, resubmit_current_job);
+		set_ret = set_at_job_done(job, NULL, resubmit_current_job, &set_error);
 	}
-	STOP_SPI_SNAP();
 
 	resubmit_current_job = 0;
 	current_job_id = -1;
 	pgstat_report_activity(STATE_RUNNING, "finish job processing");
+	if(set_ret > 0)
+	{
+		STOP_SPI_SNAP();
+		return 1;
+	}
+	if(set_error)
+	{
+		elog(LOG, "AT_EXECUTOR ERROR: set log: %s", set_error);
+		pfree(set_error);
+	}
+	else
+	{
+		elog(LOG, "AT_EXECUTOR ERROR: set log: unknown error");
+	}
+	ABORT_SPI_SNAP();
 
-	return 1;
+	return -1;
 }
 
 Oid set_session_authorization_by_name(char *rolename, char **error)