shceduler for 10

Author: Vladimir Ershov

src/pgpro_scheduler.c

@@ -392,7 +392,12 @@ void parent_scheduler_main(Datum arg)
 			}
 		}
 		rc = WaitLatch(MyLatch,
+#if PG_VERSION_NUM < 100000
 			WL_LATCH_SET | WL_POSTMASTER_DEATH, 0);
+#else
+			WL_LATCH_SET | WL_POSTMASTER_DEATH, 0, 
+			PG_WAIT_EXTENSION);
+#endif
 		CHECK_FOR_INTERRUPTS();
 		ResetLatch(MyLatch);
 	}
@@ -411,7 +416,9 @@ pg_scheduler_startup(void)
 		BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_restart_time = 10;
+#if PG_VERSION_NUM < 100000
 	worker.bgw_main = NULL;
+#endif
 	worker.bgw_notify_pid = 0;
 	worker.bgw_main_arg = Int32GetDatum(0);
 	worker.bgw_extra[0] = 0;

src/sched_manager_poll.c

@@ -211,7 +211,9 @@ pid_t registerManagerWorker(schd_manager_t *man)
 		BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_restart_time = BGW_NEVER_RESTART;
+#if PG_VERSION_NUM < 100000
 	worker.bgw_main = NULL;
+#endif
 	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(man->shared));
 	sprintf(worker.bgw_library_name, "pgpro_scheduler");
 	sprintf(worker.bgw_function_name, "manager_worker_main");

src/scheduler_executor.c

@@ -27,6 +27,7 @@
 #include "utils/builtins.h"
 #include "utils/timestamp.h"
 #include "utils/memutils.h"
+#include "utils/regproc.h"
 #include "utils/guc.h"
 
 #include "pgpro_scheduler.h"
@@ -160,7 +161,12 @@ void executor_worker_main(Datum arg)
 		}
 
 		pgstat_report_activity(STATE_IDLE, "waiting for a job");
+#if PG_VERSION_NUM < 100000
 		rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH, 0L);
+#else
+		rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH, 0L,
+			PG_WAIT_EXTENSION);
+#endif
 		ResetLatch(MyLatch);
 		if(rc && rc & WL_POSTMASTER_DEATH) break;
 	}
@@ -237,8 +243,6 @@ int do_one_job(schd_executor_share_t *shared, schd_executor_status_t *status)
 
 	pgstat_report_activity(STATE_RUNNING, "process job");
 	CHECK_FOR_INTERRUPTS();
-	/* rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 0);
-	ResetLatch(MyLatch); */
 	SetConfigOption("schedule.transaction_state", "running", PGC_INTERNAL, PGC_S_SESSION);
 
 	if(job->same_transaction)
@@ -737,7 +741,12 @@ void at_executor_worker_main(Datum arg)
 		{
 			pgstat_report_activity(STATE_IDLE, "waiting for a job");
 			rc = WaitLatch(MyLatch,
+#if PG_VERSION_NUM < 100000
 				WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_TIMEOUT, 1000L);
+#else
+				WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_TIMEOUT, 1000L,
+				PG_WAIT_EXTENSION);
+#endif
 			ResetLatch(MyLatch);
 			if(rc && rc & WL_POSTMASTER_DEATH) break;
 			lets_sleep = false;

src/scheduler_manager.c

@@ -24,6 +24,7 @@
 #include "utils/snapmgr.h"
 #include "utils/builtins.h"
 #include "utils/timestamp.h"
+#include "utils/regproc.h"
 #include <sys/time.h>
 #include "utils/lsyscache.h"
 #include "catalog/namespace.h"
@@ -304,7 +305,12 @@ int refresh_scheduler_manager_context(scheduler_manager_ctx_t *ctx)
 			if(got_sighup) return 0;   /* need to refresh it again */
 		}
 		rc = WaitLatch(MyLatch,
+#if PG_VERSION_NUM < 100000
 				WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 500L);
+#else
+				WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 500L, 
+				PG_WAIT_EXTENSION);
+#endif
 		ResetLatch(MyLatch);
 	}
 
@@ -883,7 +889,9 @@ int launch_executor_worker(scheduler_manager_ctx_t *ctx, scheduler_manager_slot_
 					BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_restart_time = BGW_NEVER_RESTART;
+#if PG_VERSION_NUM < 100000
 	worker.bgw_main = NULL;
+#endif
 	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
 	sprintf(worker.bgw_library_name, "pgpro_scheduler");
 	sprintf(worker.bgw_function_name, "executor_worker_main");
@@ -1705,7 +1713,9 @@ int start_at_worker(scheduler_manager_ctx_t *ctx, int pos)
 					BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_restart_time = BGW_NEVER_RESTART;
+#if PG_VERSION_NUM < 100000
 	worker.bgw_main = NULL;
+#endif
 	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
 	sprintf(worker.bgw_library_name, "pgpro_scheduler");
 	sprintf(worker.bgw_function_name, "at_executor_worker_main");
@@ -1882,7 +1892,12 @@ void manager_worker_main(Datum arg)
 
 		delete_worker_mem_ctx(old);
 		rc = WaitLatch(MyLatch,
+#if PG_VERSION_NUM < 100000
 			WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 1500L);
+#else
+			WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 1500L, 
+			PG_WAIT_EXTENSION);
+#endif
 		ResetLatch(MyLatch);
 	}
 	scheduler_manager_stop(ctx);

test/perl/runtest.pl

@@ -42,6 +42,29 @@
     exit(-1);
 }
 
+my $data = $dbh->selectall_arrayref('show shared_preload_libraries', { Slice => {} });
+if($data)
+{
+	my $v = $data->[0]->{shared_preload_libraries};
+	unless($v =~ /pgpro_scheduler/)
+	{
+		my $cdata  =  $dbh->selectall_arrayref('show config_file', { Slice => {} });
+		if($cdata)
+		{
+			print STDERR "Put 'pgpro_scheduler' into 'shared_preload_libraries' $cdata->[0]->{config_file}.\n";
+			exit 1;
+		}
+		print STDERR "Put 'pgpro_scheduler' into 'shared_preload_libraries'.\n";
+		exit 1;
+		
+	}
+}
+else
+{
+	print STDERR "Cannot get information about 'shared_preload_libraries' check config.\n";
+	exit 2;
+}
+
 my @sqls = ( 
 	"ALTER SYSTEM SET schedule.enabled=off",
 	"SELECT pg_reload_conf()",
@@ -86,7 +109,7 @@
 );
 my $harness = TAP::Harness->new( \%args );
 my @tests = glob( 't/*.t' );
-### @tests = ('t/jobMaxRunTime.t');
+## @tests = ('t/jobNextTime.t'); - PGPROEE10 тут валится
 $harness->runtests(@tests);
 
 

test/perl/t/jobWithBlock.t

@@ -45,13 +45,13 @@ $query = "VACUUM FULL ANALYZE;";
 my $sth = $dbh->prepare($query);
 $sth->execute();
 
-sleep 60;
+sleep 120;
 $query = "SELECT count(*) FROM test_results";
 $sth = $dbh->prepare($query);
 ok($sth->execute()) or (print $DBI::errstr and $dbh->disconnect() and BAIL_OUT);
 
 my $result = $sth->fetchrow_array() and $sth->finish();
-ok ($result > 1) or print "Count <= 1\n";
+ok ($result > 1) or print "Count $result <= 1\n";
 
 my $query = "DELETE FROM test_results;";
 $dbh->do($query);