working copy before tests

Author: Vladimir Ershov

src/pgpro_scheduler.c

@@ -49,7 +49,7 @@
 PG_MODULE_MAGIC;
 #endif
 
-int _makeTheOnlyNodesToSelect(sheduler_nodes_data_t *dst, char *nodename);
+int _makeTheOnlyNodesToSelect(scheduler_nodes_data_t *dst, char *nodename);
 
 static const char *show_scheduler_nodename(void);
 static const char *show_scheduler_database(void);
@@ -507,11 +507,11 @@ char_array_t *readBasesToCheck(void)
 	return result;
 }
 
-int _makeTheOnlyNodesToSelect(sheduler_nodes_data_t *dst, char *nodename)
+int _makeTheOnlyNodesToSelect(scheduler_nodes_data_t *dst, char *nodename)
 {
 	dst->n = 1;
 	dst->argtypes = palloc(sizeof(Oid) * 1);
-	dst->values = palloc(sizeof(Datum) * 1)
+	dst->values = palloc(sizeof(Datum) * 1);
 	dst->cond = my_copy_string(" = $1 ");
 
 	dst->values[0] = CStringGetTextDatum(nodename);
@@ -520,7 +520,7 @@ int _makeTheOnlyNodesToSelect(sheduler_nodes_data_t *dst, char *nodename)
 	return dst->n;
 }
 
-int getNodesToSelect(sheduler_nodes_data_t *dst, char *nodename)
+int getNodesToSelect(scheduler_nodes_data_t *dst, char *nodename)
 {
 #ifdef _WITH_MTM
 	StringInfo cond;
@@ -535,21 +535,21 @@ int getNodesToSelect(sheduler_nodes_data_t *dst, char *nodename)
 	}
 	dst->n = MTM_nodes->need_help + 1;
 	dst->argtypes = palloc(sizeof(Oid) * dst->n);
-	dst->values = palloc(sizeof(Datum) * dst->n)
+	dst->values = palloc(sizeof(Datum) * dst->n);
 	dst->values[0] = CStringGetTextDatum(nodename);
 	dst->argtypes[0] = TEXTOID;
 
-	cond = initStringInfo();
-	appendStringInfo(cond, "($1");
+	cond = makeStringInfo();
+	appendStringInfo(cond, "in ($1");
 	for(i=0; i < MTM_nodes->nodes_num; i++)
 	{
 		if(MTM_nodes->nodes[i].need_help)
 		{
 			n++;
 			dst->values[n] =
-				CStringGetTextDatum(MTM_nodes->nodes[i].nodename);
+				CStringGetTextDatum(MTM_nodes->nodes[i].name);
 			dst->argtypes[n] = TEXTOID;
-			appendStringInfo(cond, ",$%d", n+1)
+			appendStringInfo(cond, ",$%d", n+1);
 		}
 	}
 	appendStringInfo(cond, ")");
@@ -563,7 +563,7 @@ int getNodesToSelect(sheduler_nodes_data_t *dst, char *nodename)
 #endif
 }
 
-void destroyNodesToSelect(sheduler_nodes_data_t *nodes)
+void destroyNodesToSelect(scheduler_nodes_data_t *nodes)
 {
 	pfree(nodes->values);
 	pfree(nodes->argtypes);

src/pgpro_scheduler.h

@@ -34,7 +34,7 @@ typedef struct {
 	Datum *values;
 	Oid *argtypes;
 	char *cond;
-} sheduler_nodes_data_t;
+} scheduler_nodes_data_t;
 
 extern void worker_spi_sighup(SIGNAL_ARGS);
 extern void worker_spi_sigterm(SIGNAL_ARGS);
@@ -63,7 +63,7 @@ char *get_scheduler_nodename(MemoryContext mem);
 char_array_t *_split_string_to_char_array(char *str, bool doclean);
 Oid get_scheduler_schema_oid(void);
 
-int getNodesToSelect(sheduler_nodes_data_t *nodes, char *nodename);
-void destroyNodesToSelect(sheduler_nodes_data_t *nodes);
+int getNodesToSelect(scheduler_nodes_data_t *nodes, char *nodename);
+void destroyNodesToSelect(scheduler_nodes_data_t *nodes);
 
 #endif

src/scheduler_executor.c

@@ -35,6 +35,7 @@
 #include "pgpro_scheduler.h"
 #include "scheduler_executor.h"
 #include "scheduler_job.h"
+#include "scheduler_mtm.h"
 #include "scheduler_spi_utils.h"
 #include "memutils.h"
 #include "utils/elog.h"
@@ -711,6 +712,10 @@ void at_executor_worker_main(Datum arg)
 	BackgroundWorkerInitializeConnection(shared->database, NULL);
 	SetConfigOption("application_name", "pgp-s at executor", PGC_USERSET, PGC_S_SESSION);
 	pgstat_report_activity(STATE_RUNNING, "initialize");
+#ifdef _WITH_MTM
+	/* need to connect to mtm nodes status struct */
+	init_scheduler_mtm_nodes();
+#endif
 
 	START_SPI_SNAP();
 	reloid = RangeVarGetRelid(makeRangeVarFromNameList(

src/scheduler_job.c

@@ -7,6 +7,7 @@
 #include "scheduler_job.h"
 #include "lib/stringinfo.h"
 #include "scheduler_spi_utils.h"
+#include "scheduler_mtm.h"
 #include "utils/timestamp.h"
 #include "utils/builtins.h"
 #include "memutils.h"
@@ -170,19 +171,35 @@ job_t *get_at_job_for_process(MemoryContext mem, char *nodename, char **error)
 	bool nulls[17];
 	int i;
 	char *oldpath;
-
-	/* const char *get_job_sql = "select * from at_jobs_submitted s where ((not exists ( select * from at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from at_jobs_process p where p.id = any(s.depends_on)) AND s.depends_on is NOT NULL and s.at IS NULL) OR ( s.at IS NOT NULL AND  at <= 'now' and (last_start_available is NULL OR last_start_available > 'now'))) and node = $1 and not canceled order by at,  submit_time limit 1 FOR UPDATE SKIP LOCKED";  */
-	const char *get_job_sql = "select * from at_jobs_submitted s where ((not exists ( select * from at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from at_jobs_process p where p.id = any(s.depends_on)) AND s.depends_on is NOT NULL and s.at IS NULL AND not exists ( select * from at_jobs_done d where d.id = any(s.depends_on) and d.status=false)) OR ( s.at IS NOT NULL AND  at <= 'now' and (last_start_available is NULL OR last_start_available > 'now'))) and node = $1 and not canceled order by at,  submit_time limit 1 FOR UPDATE SKIP LOCKED ";
+	scheduler_nodes_data_t nodes;
+	StringInfo get_job_sql;
 	const char *insert_sql = "insert into at_jobs_process values($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)";
 	spi_response_t *r;
 	spi_response_t *r2;
 
+	get_job_sql = makeStringInfo();
+	getNodesToSelect(&nodes, nodename);
+
+	appendStringInfo(get_job_sql, "select * from at_jobs_submitted s where \
+		((not exists ( select * from at_jobs_submitted s2 where \
+		s2.id = any(s.depends_on)) AND not exists \
+		( select * from at_jobs_process p where p.id = any(s.depends_on)) AND \
+		s.depends_on is NOT NULL and s.at IS NULL AND not exists \
+		( select * from at_jobs_done d where d.id = any(s.depends_on) and \
+		d.status=false)) OR ( s.at IS NOT NULL AND  at <= 'now' and \
+		(last_start_available is NULL OR last_start_available > 'now'))) and \
+		node %s and not canceled order by at,  submit_time \
+		limit 1 FOR UPDATE SKIP LOCKED ", nodes.cond);
+
 	oldpath = set_schema(NULL, true); 
 	*error = NULL;
-	values[0] = CStringGetTextDatum(nodename);
 
-	r = execute_spi_sql_with_args(mem, get_job_sql, 1, 
-										argtypes, values, NULL);
+	r = execute_spi_sql_with_args(mem, get_job_sql->data, nodes.n, 
+										nodes.argtypes, nodes.values, NULL);
+	pfree(get_job_sql->data);
+	pfree(get_job_sql);
+	destroyNodesToSelect(&nodes);
+
 	if(r->retval != SPI_OK_SELECT)
 	{
 		set_schema(oldpath, false);
@@ -204,6 +221,7 @@ job_t *get_at_job_for_process(MemoryContext mem, char *nodename, char **error)
 
 	job->cron_id = scheduler_atjob_id_OID == INT8OID ? 
 		get_int64_from_spi(r, 0, 1, 0): get_int_from_spi(r, 0, 1, 0);
+	job->node = get_text_from_spi(mem, r, 0, 2);
 	job->start_at = get_timestamp_from_spi(r, 0, 5, 0);
 	job->dosql[0] = get_text_from_spi(mem, r, 0, 6);
 	job->sql_params = get_textarray_from_spi(mem, r, 0, 7, &job->sql_params_n);
@@ -348,18 +366,18 @@ job_t *_cron_get_jobs_to_do(MemoryContext mem, char *nodename, int *n, int *is_e
 {
 	job_t *jobs = NULL;
 	int ret, got, i;
-	stringInfo get_job_sql;
-	sheduler_nodes_data_t nodes;
+	StringInfo get_job_sql;
+	scheduler_nodes_data_t nodes;
 
 	get_job_sql = makeStringInfo();
-	appendStringInfo(&get_job_sql,
+	appendStringInfo(get_job_sql,
 		"select at.start_at, at.last_start_available, at.cron, max_run_time, \
 		 cron.max_instances, cron.executor, cron.next_time_statement, \
 		 cron.node from at at, cron cron where start_at <= 'now' and \
 		 not at.active and (last_start_available is NULL OR \
 		 last_start_available > 'now') and at.cron = cron.id AND cron.node ");
 	getNodesToSelect(&nodes, nodename);
-	appendStringInfo(&get_job_sql, " %s order by at.start_at limit %d", nodes.cond, limit);
+	appendStringInfo(get_job_sql, " %s order by at.start_at limit %d", nodes.cond, limit);
 
 
 	*is_error = *n = 0;
@@ -390,6 +408,8 @@ job_t *_cron_get_jobs_to_do(MemoryContext mem, char *nodename, int *n, int *is_e
 	{
 		*is_error = 1;
 	}
+	pfree(get_job_sql->data);
+	pfree(get_job_sql);
 	destroyNodesToSelect(&nodes);
 	STOP_SPI_SNAP();
 	return jobs;
@@ -400,12 +420,16 @@ job_t *get_expired_at_jobs(char *nodename, int *n, int *is_error)
 	StringInfoData sql;
 	job_t *jobs = NULL;
 	int ret, got, i;
+	scheduler_nodes_data_t nodes;
 
 	*n = *is_error = 0;
+	getNodesToSelect(&nodes, nodename);
 	initStringInfo(&sql);
-	appendStringInfo(&sql, "select at, last_start_available, id from ONLY at_jobs_submitted where last_start_available < 'now' and node = '%s'", nodename);
-	ret = SPI_execute(sql.data, true, 0);
+	appendStringInfo(&sql, "select at, last_start_available, id, node from ONLY at_jobs_submitted where last_start_available < 'now' and node %s", nodes.cond);
+
+	ret = SPI_execute_with_args(sql.data, nodes.n, nodes.argtypes, nodes.values, NULL, true, 0);
 	pfree(sql.data);
+	destroyNodesToSelect(&nodes);
 
 	if(ret == SPI_OK_SELECT)
 	{
@@ -421,7 +445,7 @@ job_t *get_expired_at_jobs(char *nodename, int *n, int *is_error)
 				jobs[i].last_start_avail = get_timestamp_from_spi(NULL, i, 2, 0);
 				jobs[i].cron_id = scheduler_atjob_id_OID == INT8OID ?
 					get_int64_from_spi(NULL, i, 3, 0): get_int_from_spi(NULL, i, 3, 0);
-				jobs[i].node = my_copy_string(nodename);
+				jobs[i].node = get_text_from_spi(CurrentMemoryContext, NULL, i, 4);
 			}
 		}
 	}
@@ -437,12 +461,17 @@ job_t *get_expired_cron_jobs(char *nodename, int *n, int *is_error)
 	StringInfoData sql;
 	job_t *jobs = NULL;
 	int ret, got, i;
+	scheduler_nodes_data_t nodes;
 
 	*n = *is_error = 0;
 	initStringInfo(&sql);
-	appendStringInfo(&sql, "select start_at, last_start_available, cron, started, active from at where last_start_available < 'now' and not active and node = '%s'", nodename);
-	ret = SPI_execute(sql.data, true, 0);
+	appendStringInfo(&sql, "select start_at, last_start_available, cron, started, active, node from at where last_start_available < 'now' and not active and node ");
+	getNodesToSelect(&nodes, nodename);
+	appendStringInfoString(&sql, nodes.cond);
+
+	ret = SPI_execute_with_args(sql.data, nodes.n, nodes.argtypes, nodes.values, NULL, true, 0);
 	pfree(sql.data);
+	destroyNodesToSelect(&nodes);
 
 	if(ret == SPI_OK_SELECT)
 	{
@@ -457,7 +486,7 @@ job_t *get_expired_cron_jobs(char *nodename, int *n, int *is_error)
 				jobs[i].start_at = get_timestamp_from_spi(NULL, i, 1, 0);
 				jobs[i].last_start_avail = get_timestamp_from_spi(NULL, i, 2, 0);
 				jobs[i].cron_id = get_int_from_spi(NULL, i, 3, 0);
-				jobs[i].node = my_copy_string(nodename);
+				jobs[i].node = get_text_from_spi(CurrentMemoryContext,NULL, i, 6);
 			}
 		}
 	}