util functions but some mem alloc problems near ALTER SYSTEM

Author: Vladimir Ershov

README.md

@@ -28,6 +28,13 @@ Install extension as follows:
 	$ sudo make USE_PGXS=1 install
 	$ psql <DBNAME> -c "CREATE EXTENSION pgpro_scheduler"
 
+Install extension to use it with multimaster(mmts) extension:
+
+	$ cd pgpro_scheduler
+	$ make USE_PGXS=1 WITH_MTM=<path to mmts extension>
+	$ sudo make USE_PGXS=1 install
+	$ psql <DBNAME> -c "CREATE EXTENSION pgpro_scheduler"
+
 ## Configuration
 
 The extension defines a number of PostgreSQL variables (GUC). This variables 

pgpro_scheduler--2.2.sql

@@ -1450,6 +1450,26 @@ CREATE FUNCTION nodename()
   AS 'MODULE_PATHNAME', 'nodename'
   LANGUAGE C IMMUTABLE;
 
+CREATE FUNCTION enable()
+  RETURNS boolean
+  AS 'MODULE_PATHNAME', 'scheduler_enable'
+  LANGUAGE C IMMUTABLE;
+
+CREATE FUNCTION disable()
+  RETURNS boolean
+  AS 'MODULE_PATHNAME', 'scheduler_disable'
+  LANGUAGE C IMMUTABLE;
+
+CREATE FUNCTION start()
+  RETURNS boolean
+  AS 'MODULE_PATHNAME', 'scheduler_start'
+  LANGUAGE C IMMUTABLE;
+
+CREATE FUNCTION stop()
+  RETURNS boolean
+  AS 'MODULE_PATHNAME', 'scheduler_stop'
+  LANGUAGE C IMMUTABLE;
+
 --------------
 -- TRIGGERS --
 --------------

src/char_array.c

@@ -43,6 +43,57 @@ char_array_t *pushCharArray(char_array_t *a, const char *str)
 	return a;
 }
 
+char *joinCharArray(char_array_t *a, char *clue)
+{
+	int i;
+	char *ret;
+	char *ptr;
+	int realn = 0;
+	size_t size = 0;
+	int clue_len;
+
+	if(a->n >  0)
+	{
+		for(i=0; i < a->n; i++)
+		{
+			if(a->data[i])
+			{
+				realn++;
+				size += strlen(a->data[i]);
+			}
+		}
+	}
+	if(realn == 0)
+	{
+		ret = (char *)palloc(sizeof(char) * 1);
+		if(!ret) elog(ERROR, "chararray: cannot alloc memory (join)");
+		ret[0] = 0;
+		return ret;
+	}
+	clue_len = strlen(clue);
+	size += clue_len * realn + 1;
+	ret = (char *)palloc(sizeof(char) * realn);
+	if(!ret) elog(ERROR, "chararray: cannot alloc memory (join)");
+	
+	ptr = ret;
+	for(i=0; i < a->n; i++)
+	{
+		if(a->data[i])
+		{
+			memcpy(ptr, a->data[i], strlen(a->data[i]));
+			ptr += strlen(a->data[i]);
+			if(--realn > 0)
+			{
+				memcpy(ptr, clue, clue_len);
+				ptr += clue_len;
+			}
+		}
+	}
+	ret[size-1] = 0;
+
+	return ret;
+}
+
 void destroyCharArray(char_array_t *a)
 {
 	int i;

src/char_array.h

@@ -10,6 +10,7 @@ int __sort_char_string(const void *a, const void *b);
 char_array_t *makeCharArray(void);
 char_array_t *pushCharArray(char_array_t *a, const char *str);
 char_array_t *sortCharArray(char_array_t *a);
+char *joinCharArray(char_array_t *a, char *clue);
 void destroyCharArray(char_array_t *data);
 
 

src/memutils.c

@@ -3,6 +3,7 @@
 #include "memutils.h"
 
 MemoryContext SchedulerWorkerContext = NULL;
+bool worker_context_faked = false;
 
 MemoryContext init_mem_ctx(const char *name)
 {
@@ -17,6 +18,15 @@ bool is_worker_context_initialized(void)
 	return SchedulerWorkerContext == NULL ? false: true;
 }
 
+MemoryContext fake_worker_mem_ctx(MemoryContext ctx)
+{
+	AssertState(SchedulerWorkerContext == NULL);
+	SchedulerWorkerContext = ctx;
+	worker_context_faked = true;
+
+	return SchedulerWorkerContext;
+}
+
 MemoryContext init_worker_mem_ctx(const char *name)
 {
 	AssertState(SchedulerWorkerContext == NULL);
@@ -54,20 +64,23 @@ void *worker_alloc(Size size)
 
 void drop_worker_context(void)
 {
-	if(SchedulerWorkerContext)
+	if(SchedulerWorkerContext && !worker_context_faked)
 	{
 		MemoryContextDelete(SchedulerWorkerContext);
-		SchedulerWorkerContext = NULL;
 	}
+	SchedulerWorkerContext = NULL;
+	worker_context_faked = false;
+	
 }
 
 void delete_worker_mem_ctx(MemoryContext old)
 {
 	if(!old) old = TopMemoryContext;
 
 	MemoryContextSwitchTo(old);
-	MemoryContextDelete(SchedulerWorkerContext);
+	if(!worker_context_faked) MemoryContextDelete(SchedulerWorkerContext);
 	SchedulerWorkerContext = NULL;
+	worker_context_faked = false;
 }
 
 char *_mcopy_string(MemoryContext ctx, char *str)

src/memutils.h

@@ -8,6 +8,7 @@ extern MemoryContext SchedulerWorkerContext;
 
 MemoryContext init_worker_mem_ctx(const char *name);
 MemoryContext init_mem_ctx(const char *name);
+MemoryContext fake_worker_mem_ctx(MemoryContext ctx);
 MemoryContext switch_to_worker_context(void);
 void *worker_alloc(Size size);
 void delete_worker_mem_ctx(MemoryContext toswitch);

src/pgpro_scheduler.c

@@ -1,3 +1,5 @@
+#define ENABLE_LIST_COMPAT	1
+
 #include "postgres.h"
 
 #include "miscadmin.h"
@@ -33,6 +35,7 @@
 #include "access/sysattr.h"
 #include "access/htup_details.h"
 #include "utils/fmgroids.h"
+#include "nodes/pg_list.h"
 
 
 #include "char_array.h"
@@ -479,17 +482,14 @@ char_array_t *readBasesToCheck(void)
 	destroyCharArray(names);
 	appendStringInfo(&sql, ")");
 
-	elog(LOG, "SCHEDULER: select from PG_DATABASES %s", sql.data);
 	START_SPI_SNAP();
-	elog(LOG, "SCHEDULER: %s", sql.data);
 
 	ret = SPI_execute(sql.data, true, 0);
 	if (ret != SPI_OK_SELECT)
 	{
 		STOP_SPI_SNAP();
 		elog(ERROR, "cannot select from pg_database");
 	}
-	elog(LOG, "SCHEDULER: select from PG_DATABASES done");
 	processed  = SPI_processed;
 	if(processed == 0)
 	{
@@ -571,6 +571,41 @@ void destroyNodesToSelect(scheduler_nodes_data_t *nodes)
 	nodes->n = -1;
 }
 
+/* Alter system to set some valiable */
+
+void alter_system(char *name, char *value)
+{
+	AlterSystemStmt *stmt;
+	A_Const *vv;
+
+	stmt = makeNode(AlterSystemStmt);
+
+	stmt->setstmt = makeNode(VariableSetStmt);
+	stmt->setstmt->kind = VAR_SET_VALUE;
+	stmt->setstmt->is_local = false;
+	stmt->setstmt->name = name;
+
+	vv = makeNode(A_Const);
+	vv->val.type = T_String;
+	vv->val.val.str = value;
+	vv->location = -1;
+	stmt->setstmt->args = makeList1(vv);
+
+	AlterSystemSetConfigFile(stmt);
+}
+
+/* Sends SIGHUP to postmaster to reload config files */
+
+bool reload_config(void)
+{
+	if(kill(PostmasterPid, SIGHUP))
+	{
+		ereport(WARNING, (errmsg("failed to send signal to postmaster: %m")));
+		return false;
+	}
+	return true;
+}
+
 void parent_scheduler_main(Datum arg)
 {
 	int rc = 0, i;
@@ -589,19 +624,14 @@ void parent_scheduler_main(Datum arg)
 	pqsignal(SIGHUP, worker_spi_sighup);
 	pqsignal(SIGTERM, worker_spi_sigterm);
 	BackgroundWorkerUnblockSignals();
-	elog(LOG, "SCHEDULER: going to make connection");
 
 	BackgroundWorkerInitializeConnection("postgres", NULL);
-	elog(LOG, "SCHEDULER: connected");
 	names = readBasesToCheck();
-	elog(LOG, "SCHEDULER: init pool");
 	pool = initSchedulerManagerPool(names);
-	elog(LOG, "SCHEDULER: init pool initialized");
 
 	destroyCharArray(names);
 
 	set_supervisor_pgstatus(pool);
-	elog(LOG, "SCHEDULER: start main cycle");
 
 	while(!got_sigterm)
 	{
@@ -907,4 +937,127 @@ cron_string_to_json_text(PG_FUNCTION_ARGS)
 	PG_RETURN_NULL();
 }
 
+PG_FUNCTION_INFO_V1(scheduler_enable);
+Datum
+scheduler_enable(PG_FUNCTION_ARGS)
+{
+	if(!superuser())
+	{
+		elog(ERROR, "Only superuser allowed to enable pgpro_scheduler");
+	}
+	alter_system("schedule.enabled", "on");
+
+	PG_RETURN_BOOL(reload_config());
+}
+
+PG_FUNCTION_INFO_V1(scheduler_disable);
+Datum
+scheduler_disable(PG_FUNCTION_ARGS)
+{
+	if(!superuser())
+	{
+		elog(ERROR, "Only superuser allowed to disable pgpro_scheduler");
+	}
+	alter_system("schedule.enabled", "off");
+
+	PG_RETURN_BOOL(reload_config());
+}
+
+PG_FUNCTION_INFO_V1(scheduler_start);
+Datum
+scheduler_start(PG_FUNCTION_ARGS)
+{
+	char *dbname;
+	char *current;
+	char *new;
+	char_array_t *in_use;
+	int i;
 
+	if(!superuser())
+	{
+		elog(ERROR, "Only superuser allowed to start pgpro_scheduler");
+	}
+	init_worker_mem_ctx("scheduler start");
+
+	dbname = get_database_name(MyDatabaseId);
+	current = (char *)GetConfigOption("schedule.database", false, true);
+	in_use = _split_string_to_char_array(current, true);
+	if(in_use == NULL || in_use->n == 0)
+	{
+		alter_system("schedule.database", dbname);
+	}
+	else
+	{
+		for(i=0; i < in_use->n; i++)
+		{
+			if(strcmp(dbname, in_use->data[i]) == 0)
+			{
+				elog(WARNING, "Database %s already has pgpro_scheduler running", dbname);
+				drop_worker_context();
+				PG_RETURN_BOOL(false);
+
+			}
+		}
+	}
+	if(!in_use) in_use = makeCharArray();
+	pushCharArray(in_use, dbname);
+	new = joinCharArray(in_use, ", ");
+
+	elog(NOTICE, "set: [%d] %s", (int)strlen(new), new);
+	alter_system("schedule.database", new);
+
+	drop_worker_context();
+	PG_RETURN_BOOL(reload_config());
+}
+
+PG_FUNCTION_INFO_V1(scheduler_stop);
+Datum
+scheduler_stop(PG_FUNCTION_ARGS)
+{
+	char *dbname;
+	char *current;
+	char *new;
+	char_array_t *in_use;
+	int i;
+	bool found = false;
+
+	if(!superuser())
+	{
+		elog(ERROR, "Only superuser allowed to stop pgpro_scheduler");
+	}
+	init_worker_mem_ctx("scheduler stop");
+
+	dbname = get_database_name(MyDatabaseId);
+	current = (char *)GetConfigOption("schedule.database", false, true);
+	in_use = _split_string_to_char_array(current, true);
+	if(in_use == NULL || in_use->n == 0)
+	{
+		elog(WARNING, "Database %s is not running pgpro_scheduler", dbname);
+		drop_worker_context();
+		PG_RETURN_BOOL(false);
+	}
+	else
+	{
+		for(i=0; i < in_use->n; i++)
+		{
+			if(strcmp(dbname, in_use->data[i]) == 0)
+			{
+				pfree(in_use->data[i]);
+				in_use->data[i] = NULL;
+				found = true;
+			}
+		}
+	}
+	if(found)
+	{
+		new = joinCharArray(in_use, ", ");
+		elog(NOTICE, "set: [%d] %s", (int)strlen(new), new);
+		alter_system("schedule.database", new);
+		drop_worker_context();
+		PG_RETURN_BOOL(reload_config());
+	}
+
+	elog(WARNING, "Database %s is not running pgpro_scheduler", dbname);
+	drop_worker_context();
+	PG_RETURN_BOOL(false);
+}