starts and wait so far

Author: Vladimir Ershov

pgpro_scheduler--2.0.sql

@@ -3,9 +3,8 @@
 CREATE SCHEMA IF NOT EXISTS schedule;
 
 CREATE TYPE schedule.job_status AS ENUM ('working', 'done', 'error');
-CREATE TYPE schedule.at_job_status AS ENUM ('submitted', 'working', 'done', 'error');
 
-CREATE TABLE schedule.at_jobs(
+CREATE TABLE schedule.at_jobs_submitted(
    id SERIAL PRIMARY KEY,
    node text,
    name text,
@@ -16,16 +15,27 @@ CREATE TABLE schedule.at_jobs(
    onrollback_statement text,
    executor text,
    owner text,
+   last_start_available timestamp with time zone,
    postpone interval,
    max_run_time	interval,
    max_instances integer default 1,
-   status at_job_status default 'submited',
-   submit_time timestamp with time zone default now(),
-   started timestamp with time zone,
-   finished timestamp with time zone,
-   reason text
+   submit_time timestamp with time zone default now()
 );
-CREATE INDEX at_jobs_status_node_at_idx on schedule.at (status, node,  at);
+CREATE INDEX at_jobs_submitted_node_at_idx on schedule.at_jobs_submitted (node,  at);
+
+CREATE TABLE schedule.at_jobs_process(
+	start_time timestamp with time zone default now()
+) INHERITS (schedule.at_jobs_submitted);
+
+CREATE INDEX at_jobs_process_node_at_idx on schedule.at_jobs_process (node,  at);
+
+CREATE TABLE schedule.at_jobs_done(
+	status boolean,
+	reason text,
+	done_time timestamp with time zone default now()
+) INHERITS (schedule.at_jobs_process);
+
+CREATE INDEX at_jobs_done_node_at_idx on schedule.at_jobs_done (node,  at);
 
 CREATE TABLE schedule.cron(
    id SERIAL PRIMARY KEY,

src/pgpro_scheduler.c

@@ -46,6 +46,7 @@ char *scheduler_databases = NULL;
 char *scheduler_nodename = NULL;
 char *scheduler_transaction_state = NULL;
 int  scheduler_max_workers = 2;
+int  scheduler_at_max_workers = 2;
 bool scheduler_service_enabled = false;
 char *scheduler_schema = NULL;
 /* Custom GUC done */
@@ -109,17 +110,18 @@ int get_integer_from_string(char *s, int start, int len)
 	return atoi(buff);
 }
 
-char *make_date_from_timestamp(TimestampTz ts)
+char *make_date_from_timestamp(TimestampTz ts, bool hires)
 {
 	struct pg_tm dt;
-	char *str = worker_alloc(sizeof(char) * 17);
+	char *str = worker_alloc(sizeof(char) * 19);
 	int tz;
 	fsec_t fsec;
 	const char *tzn;
 
 	timestamp2tm(ts, &tz, &dt, &fsec, &tzn, NULL ); 
-	sprintf(str, "%04d-%02d-%02d %02d:%02d", dt.tm_year , dt.tm_mon,
-			dt.tm_mday, dt.tm_hour, dt.tm_min);
+	sprintf(str, "%04d-%02d-%02d %02d:%02d:%02d", dt.tm_year , dt.tm_mon,
+			dt.tm_mday, dt.tm_hour, dt.tm_min, dt.tm_sec);
+	if(!hires) str[16] = 0;
 	return str;
 }
 
@@ -457,7 +459,7 @@ void _PG_init(void)
 	);
 	DefineCustomIntVariable(
 		"schedule.max_workers",
-		"How much workers can serve scheduler on one database",
+		"How much workers can serve scheduled jobs on one database",
 		NULL,
 		&scheduler_max_workers,
 		2,
@@ -469,6 +471,20 @@ void _PG_init(void)
 		NULL,
 		NULL
 	);
+	DefineCustomIntVariable(
+		"schedule.at_max_workers",
+		"How much workers can serve at jobs on one database",
+		NULL,
+		&scheduler_at_max_workers,
+		2,
+		1,
+		100,
+		PGC_SUSET,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
 	DefineCustomBoolVariable(
 		"schedule.enabled",
 		"Enable schedule service",

src/pgpro_scheduler.h

@@ -35,7 +35,7 @@ pid_t registerManagerWorker(schd_manager_t *man);
 
 void reload_db_role_config(char *dbname);
 TimestampTz timestamp_add_seconds(TimestampTz to, int add);
-char *make_date_from_timestamp(TimestampTz ts);
+char *make_date_from_timestamp(TimestampTz ts, bool hires);
 int get_integer_from_string(char *s, int start, int len);
 TimestampTz get_timestamp_from_string(char *str);
 TimestampTz _round_timestamp_to_minute(TimestampTz ts);

src/scheduler_executor.c

@@ -389,72 +389,30 @@ void set_pg_var(bool result, executor_error_t *ee)
 
 job_t *initializeExecutorJob(schd_executor_share_t *data)
 {
-	const char *sql = "select at.last_start_available, cron.same_transaction, cron.do_sql, cron.executor, cron.postpone, cron.max_run_time as time_limit, cron.max_instances, cron.onrollback_statement , cron.next_time_statement from schedule.at at, schedule.cron cron where start_at = $1 and  at.active and at.cron = cron.id AND cron.node = $2 AND cron.id = $3";
-	Oid argtypes[3] = { TIMESTAMPTZOID, TEXTOID, INT4OID};
-	Datum args[3];
 	job_t *J;
-	int ret;
 	char *error = NULL;
-	char *ts;
-
-	args[0] = TimestampTzGetDatum(data->start_at);
-	args[1] = PointerGetDatum(cstring_to_text(data->nodename));
-	args[2] = Int32GetDatum(data->cron_id);
 
-	START_SPI_SNAP();
-	ret = execute_spi_sql_with_args(sql, 3, argtypes, args, NULL, &error);
+	J = data->type == CronJob ?
+		get_cron_job(data->cron_id, data->start_at, data->nodename, &error):
+		get_at_job(data->cron_id, data->nodename, &error);
 
 	if(error)
 	{
-		snprintf(data->message,
-			PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
-			"cannot retrive job: %s", error);
+		snprintf(data->message, PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
+			"%s", error);
 		elog(LOG, "EXECUTOR: %s", data->message);
 		pfree(error);
-		PopActiveSnapshot();
-		AbortCurrentTransaction();
-		SPI_finish();
 		return NULL;
 	}
-
-	if(ret == SPI_OK_SELECT)
+	if(!J)
 	{
-		if(SPI_processed == 0)
-		{
-			STOP_SPI_SNAP();
-			ts = make_date_from_timestamp(data->start_at);
-			snprintf(data->message,
-				PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
-				"cannot find job: %d @ %s [%s]",
-				data->cron_id, ts, data->nodename);
-			elog(LOG, "EXECUTOR: %s", data->message);
-			pfree(ts);
-			return NULL;
-		}
-		J = worker_alloc(sizeof(job_t));
-
-		J->cron_id = data->cron_id;
-		J->start_at = data->start_at;
-		J->node = _copy_string(data->nodename);
-		J->same_transaction = get_boolean_from_spi(0, 2, false);
-		J->dosql = get_textarray_from_spi(0, 3, &J->dosql_n);
-		J->executor = get_text_from_spi(0, 4);
-		J->onrollback = get_text_from_spi(0, 8);
-		J->next_time_statement = get_text_from_spi(0, 9);
-
-		STOP_SPI_SNAP();
-
-		return J;
+		snprintf(data->message, PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
+			"unknown error get job");
+		elog(LOG, "EXECUTOR: %s", data->message);
+		return NULL;
 	}
-	snprintf(data->message,
-		PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
-		"error while retrive job information: %d", ret);
-	elog(LOG, "EXECUTOR: %s", data->message);
-
-	PopActiveSnapshot();
-	AbortCurrentTransaction();
-	SPI_finish();
-	return NULL;
+
+	return J;
 }
 
 int push_executor_error(executor_error_t *e, char *fmt, ...)

src/scheduler_executor.h

@@ -20,6 +20,7 @@ typedef struct {
 	char user[NAMEDATALEN];
 
 	int cron_id;
+	task_type_t type;
 	TimestampTz start_at;
 
 	schd_executor_status_t status;