fix sleepy

Author: Vladimir Ershov

src/scheduler_executor.c

@@ -671,12 +671,12 @@ void at_executor_worker_main(Datum arg)
 
 		if(lets_sleep)
 		{
-			elog(LOG, "sleeping");
 			pgstat_report_activity(STATE_IDLE, "waiting for a job");
 			rc = WaitLatch(MyLatch,
 				WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_TIMEOUT, 1000L);
 			ResetLatch(MyLatch);
 			if(rc && rc & WL_POSTMASTER_DEATH) break;
+			lets_sleep = false;
 		}
 	}
 
@@ -711,15 +711,17 @@ int process_one_job(schd_executor_share_t *shared, schd_executor_status_t *statu
 
 	if(!job)
 	{
-		STOP_SPI_SNAP();
 		if(error)
 		{
 			shared->status = SchdExecutorIdling;
 			snprintf(shared->message, PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
 				"Cannot get job: %s", error);
+			elog(LOG, "AT EXECUTOR: ERROR: %s", error);
 			pfree(error);
+			ABORT_SPI_SNAP();
 			return -1;
 		}
+		STOP_SPI_SNAP();
 		shared->status = SchdExecutorIdling;
 		return 0;
 	}
@@ -784,7 +786,6 @@ int process_one_job(schd_executor_share_t *shared, schd_executor_status_t *statu
 			sprintf(buff, "error in command: code: %d", ret);
 			set_at_job_done(job, buff, resubmit_current_job);
 		}
-		
 	}
 	else
 	{

src/scheduler_manager.c

@@ -681,33 +681,19 @@ int set_job_on_free_slot(scheduler_manager_ctx_t *ctx, job_t *job)
 	schd_executor_share_t *sdata;
 	PGPROC *worker;
 	bool started = false;
-	struct timeval tv; 
-	double begin, elapsed;
 
 	p = job->type == CronJob ? &(ctx->cron) : &(ctx->at);
 	if(p->free == 0)
 	{
 		return -1;
 	}
 
-	SetCurrentStatementStartTimestamp();
-
-	gettimeofday(&tv, NULL);
-	begin = ((double)tv.tv_sec)*1000 + ((double)tv.tv_usec)/1000;
+	START_SPI_SNAP();
 
-	StartTransactionCommand();
-	PushActiveSnapshot(GetTransactionSnapshot());
-	SPI_connect();
 	ret = job->type == CronJob ?
 		set_cron_job_started(job): set_at_job_started(job);
-	SPI_finish();
-	PopActiveSnapshot();
-
-	gettimeofday(&tv, NULL);
-	elapsed = ((double)tv.tv_sec)*1000 + ((double)tv.tv_usec)/1000 - begin;
-	elog(LOG, "move job %f", elapsed);
-	CommitTransactionCommand();
 
+	START_SPI_SNAP();
 
 	if(ret)
 	{
@@ -878,8 +864,6 @@ int scheduler_start_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
 	char *ts;
 	scheduler_manager_pool_t *p;
 	TimestampTz *check_time;
-	struct timeval tv;
-	double begin, elapsed;
 
 	if(type == CronJob)
 	{
@@ -900,16 +884,9 @@ int scheduler_start_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
 		return 1;
 	}
 
-	gettimeofday(&tv, NULL);
-	begin = ((double)tv.tv_sec)*1000 + ((double)tv.tv_usec)/1000;
 
 	jobs = get_jobs_to_do(ctx->nodename, type, &njobs, &is_error, p->free);
 
-	gettimeofday(&tv, NULL);
-	elapsed = ((double)tv.tv_sec)*1000 + ((double)tv.tv_usec)/1000 - begin;
-	elog(LOG, "get todo %d = %f", type, elapsed);
-
-
 	nwaiting = njobs;
 	if(is_error)
 	{
@@ -929,19 +906,16 @@ int scheduler_start_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
 		N = p->free;
 		if(N > nwaiting) N = nwaiting; 
 
-		/* START_SPI_SNAP(); */
 
 		for(i = start_i; i < N + start_i; i++)
 		{
 			ni = type == CronJob ?
 				how_many_instances_on_work(ctx, &(jobs[i])): 100000;
 			if(type == CronJob && ni >= jobs[i].max_instances)
 			{
-				/* START_SPI_SNAP(); */
 				set_job_error(&jobs[i], "max instances limit reached");
 				move_job_to_log(&jobs[i], false, false);
 				destroy_job(&jobs[i], 0);
-				/* STOP_SPI_SNAP(); */
 				jobs[i].cron_id = -1;
 			}
 			else
@@ -964,21 +938,13 @@ int scheduler_start_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
 						elog(ERROR, "Cannot set job to free slot type=%d, id=%d", 
 									jobs[i].type, jobs[i].cron_id);
 					}
-		/*			START_SPI_SNAP(); */
 					move_job_to_log(&jobs[i], false, false);
 					destroy_job(&jobs[i], 0);
 					jobs[i].cron_id = -1;
-					/* STOP_SPI_SNAP(); */
 				}
-	gettimeofday(&tv, NULL);
-	elapsed = ((double)tv.tv_sec)*1000 + ((double)tv.tv_usec)/1000 - begin;
-	elog(LOG, "get todo %d set one job = %f", type, elapsed);
 			}
 		}
 
-		/* STOP_SPI_SNAP(); */
-
-
 		if(N < nwaiting)
 		{
 			start_i += N;
@@ -1597,7 +1563,7 @@ int start_at_worker(scheduler_manager_ctx_t *ctx, int pos)
 	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");
-	snprintf(worker.bgw_name, BGW_MAXLEN, "scheduler executor %s", shm_data->database);
+	snprintf(worker.bgw_name, BGW_MAXLEN, "scheduler at-executor %s", shm_data->database);
 	worker.bgw_notify_pid = MyProcPid;
 
 	if(!RegisterDynamicBackgroundWorker(&worker, &(item->handler)))

src/scheduler_spi_utils.c

@@ -246,6 +246,7 @@ Datum select_onedatumvalue_sql(const char *sql, bool *is_null)
 	int ret;
 	Datum datum = 0;
 
+	SetCurrentStatementStartTimestamp();
 	ret = SPI_execute(sql, true, 0);
 	if(ret == SPI_OK_SELECT)
 	{
@@ -266,6 +267,7 @@ int select_count_with_args(const char *sql, int n, Oid *argtypes, Datum *values,
 	Datum datum;
 	bool is_null;
 
+	SetCurrentStatementStartTimestamp();
 	ret = SPI_execute_with_args(sql, n, argtypes, values, nulls, true, 0);
 	if(ret == SPI_OK_SELECT)
 	{
@@ -303,6 +305,7 @@ int execute_spi_sql_with_args(const char *sql, int n, Oid *argtypes, Datum *valu
 
 	PG_TRY();
 	{
+		SetCurrentStatementStartTimestamp();
 		ret = SPI_execute_with_args(sql, n, argtypes, values, nulls, false, 0);
 	}
 	PG_CATCH();
@@ -389,6 +392,7 @@ int execute_spi_params_prepared(const char *sql, int nparams, char **params, cha
 		plan = SPI_prepare(sql, nparams, paramtypes);
 		if(plan)
 		{
+			SetCurrentStatementStartTimestamp();
 			ret = SPI_execute_plan(plan, values, NULL, false, 0);
 		}
 	}