начало документации

остановка и запуск планировщика

Author: Vladimir Ershov

README.md

@@ -36,11 +36,13 @@ PG_MODULE_MAGIC;
 volatile sig_atomic_t got_sighup = false;
 volatile sig_atomic_t got_sigterm = false;
 
-
-static char *sched_databases = "";
-static char *sched_nodename = "master";
-static char *sched_transaction_state = "undefined";
-static int sched_max_workers = 2;
+/* Custom GUC variables */
+static char *scheduler_databases = "";
+static char *scheduler_nodename = "master";
+static char *scheduler_transaction_state = "undefined";
+static int  scheduler_max_workers = 2;
+static bool scheduler_service_enabled = false;
+/* Custom GUC done */
 
 extern void
 worker_spi_sighup(SIGNAL_ARGS)
@@ -127,6 +129,15 @@ TimestampTz _round_timestamp_to_minute(TimestampTz ts)
 #endif
 }
 
+bool is_scheduler_enabled(void)
+{
+	const char *opt;
+
+	opt = GetConfigOption("schedule.enabled", false, true);
+	if(memcmp(opt, "on", 2) == 0) return true;
+	return false;
+}
+
 
 /** END of SOME UTILS **/
 
@@ -242,9 +253,10 @@ char_array_t *readBasesToCheck(void)
 void parent_scheduler_main(Datum arg)
 {
 	int rc = 0, i;
-	char_array_t *names;
+	char_array_t *names = NULL;
 	schd_managers_poll_t *poll;
 	schd_manager_share_t *shared;
+	bool refresh = false;
 
 	init_worker_mem_ctx("Parent scheduler context");
 
@@ -270,17 +282,36 @@ void parent_scheduler_main(Datum arg)
 		{
 			got_sighup = false;
 			ProcessConfigFile(PGC_SIGHUP);
-			names = readBasesToCheck();
-			if(isBaseListChanged(names, poll))
+			refresh = false;
+			names = NULL;
+			if(is_scheduler_enabled() != poll->enabled)
+			{
+				if(poll->enabled)
+				{
+					poll->enabled = false;
+					stopAllManagers(poll);
+					set_supervisor_pgstatus(poll);
+				}
+				else
+				{
+					refresh = true;
+					poll->enabled = true;
+					names = readBasesToCheck();
+				}
+			}
+			else
+			{
+				names = readBasesToCheck();
+				if(isBaseListChanged(names, poll)) refresh = true;
+				else destroyCharArray(names);
+			}
+
+			if(refresh)
 			{
 				refreshManagers(names, poll);
 				set_supervisor_pgstatus(poll);
+				destroyCharArray(names);
 			}
-			destroyCharArray(names);
-		}
-		else if(got_sigterm)
-		{
-			/* TODO STOP ALL MANAGERS */
 		}
 		else 
 		{
@@ -341,7 +372,7 @@ pg_scheduler_startup(void)
 	worker.bgw_main_arg = 0;
 	strcpy(worker.bgw_name, "pgpro scheduler");
 
-	elog(LOG, "Register WORKER");
+	/* elog(LOG, "Register WORKER"); */
 
 
 	RegisterBackgroundWorker(&worker); 
@@ -354,9 +385,9 @@ void _PG_init(void)
 		"schedule.database",
 		"On which databases scheduler could be run",
 		NULL,
-		&sched_databases,
+		&scheduler_databases,
 		"",
-		PGC_SUSET,
+		PGC_SIGHUP,
 		0,
 		NULL,
 		NULL,
@@ -366,9 +397,9 @@ void _PG_init(void)
 		"schedule.nodename",
 		"The name of scheduler node",
 		NULL,
-		&sched_nodename,
+		&scheduler_nodename,
 		"master",
-		PGC_SUSET,
+		PGC_SIGHUP,
 		0,
 		NULL,
 		NULL,
@@ -378,7 +409,7 @@ void _PG_init(void)
 		"schedule.transaction_state",
 		"State of scheduler executor transaction",
 		"If not under scheduler executor process the variable has no mean and has a value = 'undefined', possible values: progress, success, failure",
-		&sched_transaction_state , 
+		&scheduler_transaction_state , 
 		"undefined",
 		PGC_INTERNAL,
 		0,
@@ -390,11 +421,23 @@ void _PG_init(void)
 		"schedule.max_workers",
 		"How much workers can serve scheduler on one database",
 		NULL,
-		&sched_max_workers,
+		&scheduler_max_workers,
 		2,
 		1,
 		100,
-		PGC_SUSET,
+		PGC_SIGHUP,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+	DefineCustomBoolVariable(
+		"schedule.enabled",
+		"Enable schedule service",
+		NULL,
+		&scheduler_service_enabled,
+		false,
+		PGC_SIGHUP,
 		0,
 		NULL,
 		NULL,

src/pgpro_scheduler.c

@@ -39,5 +39,6 @@ char *make_date_from_timestamp(TimestampTz ts);
 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);
+bool is_scheduler_enabled(void);
 
 #endif

src/pgpro_scheduler.h

@@ -14,6 +14,7 @@
 #include "storage/procarray.h"
 #include "pgpro_scheduler.h"
 #include "utils/resowner.h"
+#include "memutils.h"
 
 #include "postmaster/bgworker.h"
 
@@ -40,6 +41,14 @@ char *supervisor_state(schd_managers_poll_t *poll)
 	char *status;
 	int len;
 
+	if(!poll->enabled)
+	{
+		status = palloc(sizeof(char) * 9);
+		memcpy(status, "disabled", 8);
+		status[8] = 0;
+		return status;
+	}
+
 	len = dbnames ? strlen(dbnames): 0;
 	if(len == 0)
 	{
@@ -130,7 +139,8 @@ int stopAllManagers(schd_managers_poll_t *poll)
 		destroyManagerRecord(poll->workers[i]);
 	}
 	pfree(poll->workers);
-	pfree(poll);
+	poll->workers = NULL;
+	poll->n = 0;
 
 	return 1;
 }
@@ -145,10 +155,17 @@ void destroyManagerRecord(schd_manager_t *man)
 schd_managers_poll_t *initSchedulerManagerPool(char_array_t *names)
 {
 	int i;
-	schd_managers_poll_t *p = palloc(sizeof(schd_managers_poll_t));
+	schd_managers_poll_t *p = worker_alloc(sizeof(schd_managers_poll_t));
 
 	p->n = 0;
 	p->workers = NULL;
+	p->enabled = true;
+
+	if(!is_scheduler_enabled())
+	{
+		p->enabled = false;
+		return p;
+	}
 
 	if(names->n > 0)
 	{
@@ -335,8 +352,6 @@ int refreshManagers(char_array_t *names, schd_managers_poll_t *poll)
 	char_array_t *same = makeCharArray();
 	char_array_t *delete = makeCharArray();
 
-	elog(LOG, "NEW names: %d", names->n);
-
 	for(i=0; i < names->n; i++)
 	{
 		found = 0;

src/sched_manager_poll.c

@@ -32,6 +32,7 @@ typedef struct {
 typedef struct {
 	int n;
 	schd_manager_t **workers;
+	bool enabled;
 } schd_managers_poll_t;
 
 void changeChildBgwState(schd_manager_share_t *, schd_manager_status_t);

src/sched_manager_poll.h

@@ -145,7 +145,6 @@ void executor_worker_main(Datum arg)
 		{
 			/* success = false; */
 			status = SchdExecutorError;
-			SetConfigOption("schedule.transaction_state", "failure", PGC_INTERNAL, PGC_S_SESSION);
 			if(error)
 			{
 				push_executor_error(&EE, "error on %d: %s", i+1, error);
@@ -156,6 +155,7 @@ void executor_worker_main(Datum arg)
 				push_executor_error(&EE, "error on %d: code: %d", i+1, ret);
 			}
 			ABORT_SPI_SNAP();
+			SetConfigOption("schedule.transaction_state", "failure", PGC_INTERNAL, PGC_S_SESSION);
 			executor_onrollback(job, &EE);
 
 			break;

src/scheduler_executor.c

@@ -1223,6 +1223,23 @@ void clean_at_table(scheduler_manager_ctx_t *ctx)
 	STOP_SPI_SNAP();
 }
 
+bool check_parent_stop_signal(scheduler_manager_ctx_t *ctx)
+{
+	schd_manager_share_t *shared;
+
+	shared = dsm_segment_address(ctx->seg);
+	if(shared->setbyparent)
+	{
+		shared->setbyparent = false;
+		if(shared->status == SchdManagerStop)
+		{
+			elog(LOG, "Recieve stop signal from parent");
+			return true;
+		}
+	}
+	return false;
+}
+
 void set_slots_stat_report(scheduler_manager_ctx_t *ctx)
 {
 	char state[128];
@@ -1314,6 +1331,7 @@ void manager_worker_main(Datum arg)
 				if(rc & WL_LATCH_SET)
 				{
 					_pdebug("got latch from some bgworker");
+					if(check_parent_stop_signal(ctx)) break;
 					scheduler_check_slots(ctx);
 					set_slots_stat_report(ctx);
 					_pdebug("quit got latch");

src/scheduler_manager.c

@@ -94,5 +94,6 @@ int update_cron_texttime(scheduler_manager_ctx_t *ctx, int cron_id, TimestampTz
 int mark_job_broken(scheduler_manager_ctx_t *ctx, int cron_id, char *reason);
 void manager_fatal_error(scheduler_manager_ctx_t *ctx, int ecode, char *message, ...) pg_attribute_printf(3, 4);
 void set_slots_stat_report(scheduler_manager_ctx_t *ctx);
+bool check_parent_stop_signal(scheduler_manager_ctx_t *ctx);
 
 #endif