Online upgrade implementation

Author: knizhnik

src/backend/port/sysv_shmem.c

@@ -243,7 +243,7 @@ InternalIpcMemoryCreate(IpcMemoryKey memKey, Size size)
 	}
 
 	/* Register on-exit routine to delete the new segment */
-	//on_shmem_exit(IpcMemoryDelete, Int32GetDatum(shmid));
+	on_shmem_exit(IpcMemoryDelete, Int32GetDatum(shmid));
 
 	/* OK, should be able to attach to the segment */
 	memAddress = shmat(shmid, requestedAddress, PG_SHMAT_FLAGS);
@@ -291,9 +291,12 @@ IpcMemoryDetach(int status, Datum shmaddr)
 static void
 IpcMemoryDelete(int status, Datum shmId)
 {
-	if (shmctl(DatumGetInt32(shmId), IPC_RMID, NULL) < 0)
-		elog(LOG, "shmctl(%d, %d, 0) failed: %m",
-			 DatumGetInt32(shmId), IPC_RMID);
+	if (!IsOnlineUpgrade)
+	{
+		if (shmctl(DatumGetInt32(shmId), IPC_RMID, NULL) < 0)
+			elog(LOG, "shmctl(%d, %d, 0) failed: %m",
+				 DatumGetInt32(shmId), IPC_RMID);
+	}
 }
 
 /*

src/backend/postmaster/bgworker.c

@@ -152,6 +152,24 @@ BackgroundWorkerShmemSize(void)
 	return size;
 }
 
+void
+BackgroundWorkerShmemAttach(void)
+{
+	slist_iter	siter;
+	int			slotno = 0;
+
+	slist_foreach(siter, &BackgroundWorkerList)
+	{
+		RegisteredBgWorker *rw;
+
+		rw = slist_container(RegisteredBgWorker, rw_lnode, siter.cur);
+		Assert(slotno < max_worker_processes);
+		rw->rw_shmem_slot = slotno;
+		rw->rw_worker.bgw_notify_pid = 0;	/* might be reinit after crash */
+		++slotno;
+	}
+}
+
 /*
  * Initialize shared memory.
  */
@@ -377,7 +395,7 @@ BackgroundWorkerStateChange(void)
 		rw->rw_worker.bgw_notify_pid = slot->worker.bgw_notify_pid;
 		if (!PostmasterMarkPIDForWorkerNotify(rw->rw_worker.bgw_notify_pid))
 		{
-			elog(DEBUG1, "worker notification PID %lu is not valid",
+			elog(LOG, "worker notification PID %lu is not valid",
 				 (long) rw->rw_worker.bgw_notify_pid);
 			rw->rw_worker.bgw_notify_pid = 0;
 		}
@@ -480,6 +498,7 @@ ReportBackgroundWorkerExit(slist_mutable_iter *cur)
 		rw->rw_worker.bgw_restart_time == BGW_NEVER_RESTART)
 		ForgetBackgroundWorker(cur);
 
+	elog(LOG, "Notify PID=%d", notify_pid);
 	if (notify_pid != 0)
 		kill(notify_pid, SIGUSR1);
 }
@@ -572,6 +591,18 @@ BackgroundWorkerEntry(int slotno)
 	memcpy(&myEntry, &slot->worker, sizeof myEntry);
 	return &myEntry;
 }
+
+void*
+GetBackgroundWorkerEntries(void)
+{
+	return BackgroundWorkerData;
+}
+
+void
+SetBackgroundWorkerEntries(void* entries)
+{
+	BackgroundWorkerData = entries;
+}
 #endif
 
 /*
@@ -1147,6 +1178,7 @@ WaitForBackgroundWorkerShutdown(BackgroundWorkerHandle *handle)
 		CHECK_FOR_INTERRUPTS();
 
 		status = GetBackgroundWorkerPid(handle, &pid);
+		elog(LOG, "GetBackgroundWorkerPid(%d) = %d", pid, status);
 		if (status == BGWH_STOPPED)
 			break;
 
@@ -1184,6 +1216,7 @@ TerminateBackgroundWorker(BackgroundWorkerHandle *handle)
 
 	/* Set terminate flag in shared memory, unless slot has been reused. */
 	LWLockAcquire(BackgroundWorkerLock, LW_EXCLUSIVE);
+	elog(LOG, "Terminate worker slot %d", handle->slot);
 	if (handle->generation == slot->generation)
 	{
 		slot->terminate = true;

src/backend/postmaster/postmaster.c

@@ -511,6 +511,7 @@ typedef struct
 	bool		IsBinaryUpgrade;
 	int			max_safe_fds;
 	int			MaxBackends;
+	void*       BackgroundWorkers;
 #ifdef WIN32
 	HANDLE		PostmasterHandle;
 	HANDLE		initial_signal_pipe;
@@ -563,10 +564,16 @@ HANDLE		PostmasterHandle;
 static void
 UpgradePostgres(void)
 {
+#ifdef EXEC_BACKEND
 	char* PostmasterArgs[] = {"/home/knizhnik/postgresql/dist/bin/postgres", "-U", "-D", "." ,NULL};
 	BackendParameters param;
 	FILE	   *fp;
 
+	IsOnlineUpgrade = true;
+	TerminateChildren(SIGTERM);
+	if (CheckpointerPID != 0)
+		signal_child(CheckpointerPID, SIGUSR2);
+
 	if (!save_backend_variables(&param, NULL))
 		return;				/* log made by save_backend_variables */
 
@@ -600,8 +607,27 @@ UpgradePostgres(void)
 
 	elog(LOG, "Upgrade postgres");
 	execv(PostmasterArgs[0], PostmasterArgs);
+#else
+	elog(LOG, "Online upgrade is possible only postgres configured with EXEC_BACKEND");
+#endif
+}
+
+
+static void
+RestoreBackendList(void)
+{
+#ifdef EXEC_BACKEND
+	int i;
+	for (i = MaxLivePostmasterChildren() - 1; i >= 0; i--)
+	{
+		Backend* bp = (Backend *) &ShmemBackendArray[i];
+		if (bp->bkend_type == BACKEND_TYPE_NORMAL)
+			dlist_push_head(&BackendList, &bp->elem);
+	}
+#endif
 }
 
+
 /*
  * Postmaster main entry point
  */
@@ -900,8 +926,10 @@ PostmasterMain(int argc, char *argv[])
 	}
 
 	if (IsOnlineUpgrade)
+	{
 		read_backend_variables("postmaster.params", NULL);
-
+		RegisterUnlinkLockFileCallback();
+	}
 
 	/*
 	 * Locate the proper configuration files and data directory, and read
@@ -1038,20 +1066,23 @@ PostmasterMain(int argc, char *argv[])
 	 */
 	process_shared_preload_libraries();
 
-	/*
-	 * Now that loadable modules have had their chance to register background
-	 * workers, calculate MaxBackends.
-	 */
-	if (!IsOnlineUpgrade)
-		InitializeMaxBackends();
-
 	/*
 	 * Set up shared memory and semaphores.
 	 */
 	if (IsOnlineUpgrade)
+	{
 		PGSharedMemoryReAttach();
+		RestoreBackendList();
+	}
 	else
+	{
+		/*
+		 * Now that loadable modules have had their chance to register background
+		 * workers, calculate MaxBackends.
+		 */
+		InitializeMaxBackends();
 		reset_shared();
+	}
 
 	/*
 	 * Estimate number of openable files.  This must happen after setting up
@@ -1435,10 +1466,18 @@ PostmasterMain(int argc, char *argv[])
 	/*
 	 * We're ready to rock and roll...
 	 */
-	StartupPID = StartupDataBase();
-	Assert(StartupPID != 0);
-	StartupStatus = STARTUP_RUNNING;
-	pmState = PM_STARTUP;
+	if (!IsOnlineUpgrade)
+	{
+		StartupPID = StartupDataBase();
+		Assert(StartupPID != 0);
+		StartupStatus = STARTUP_RUNNING;
+		pmState = PM_STARTUP;
+	}
+	else
+	{
+		BackgroundWorkerShmemAttach();
+		pmState = PM_RUN;
+	}
 
 	/* Some workers may be scheduled to start now */
 	maybe_start_bgworkers();
@@ -1696,6 +1735,7 @@ ServerLoop(void)
 	last_lockfile_recheck_time = last_touch_time = time(NULL);
 
 	nSockets = initMasks(&readmask);
+	IsOnlineUpgrade = false;
 
 	for (;;)
 	{
@@ -4103,7 +4143,10 @@ SignalSomeChildren(int signal, int target)
 static void
 TerminateChildren(int signal)
 {
-	SignalChildren(signal);
+	if (IsOnlineUpgrade)
+		SignalSomeChildren(signal, BACKEND_TYPE_WALSND | BACKEND_TYPE_WORKER);
+	else
+		SignalChildren(signal);
 	if (StartupPID != 0)
 	{
 		signal_child(StartupPID, signal);
@@ -6180,6 +6223,7 @@ save_backend_variables(BackendParameters *param, Port *port,
 	param->PgStartTime = PgStartTime;
 	param->PgReloadTime = PgReloadTime;
 	param->first_syslogger_file_time = first_syslogger_file_time;
+	param->BackgroundWorkers = GetBackgroundWorkerEntries();
 
 	param->redirection_done = redirection_done;
 	param->IsBinaryUpgrade = IsBinaryUpgrade;
@@ -6417,6 +6461,7 @@ restore_backend_variables(BackendParameters *param, Port *port)
 	PgStartTime = param->PgStartTime;
 	PgReloadTime = param->PgReloadTime;
 	first_syslogger_file_time = param->first_syslogger_file_time;
+	SetBackgroundWorkerEntries(param->BackgroundWorkers);
 
 	redirection_done = param->redirection_done;
 	IsBinaryUpgrade = param->IsBinaryUpgrade;

src/backend/replication/logical/launcher.c

@@ -978,6 +978,9 @@ ApplyLauncherMain(Datum main_arg)
 
 	before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
 
+	while (LogicalRepCtx->launcher_pid != 0)
+		pg_usleep(1000000);
+
 	Assert(LogicalRepCtx->launcher_pid == 0);
 	LogicalRepCtx->launcher_pid = MyProcPid;
 

src/backend/storage/ipc/shmem.c

@@ -101,6 +101,7 @@ InitShmemAccess(void *seghdr)
 	ShmemSegHdr = shmhdr;
 	ShmemBase = (void *) shmhdr;
 	ShmemEnd = (char *) ShmemBase + shmhdr->totalsize;
+
 }
 
 /*

src/backend/storage/lmgr/proc.c

@@ -542,11 +542,20 @@ InitAuxiliaryProcess(void)
 	/*
 	 * Find a free auxproc ... *big* trouble if there isn't one ...
 	 */
-	for (proctype = 0; proctype < NUM_AUXILIARY_PROCS; proctype++)
+	while (true)
 	{
-		auxproc = &AuxiliaryProcs[proctype];
-		if (auxproc->pid == 0)
-			break;
+		for (proctype = 0; proctype < NUM_AUXILIARY_PROCS; proctype++)
+		{
+			auxproc = &AuxiliaryProcs[proctype];
+			if (auxproc->pid == 0)
+				break;
+		}
+		if (IsOnlineUpgrade && proctype >= NUM_AUXILIARY_PROCS)
+		{
+			pg_usleep(1000000);
+			continue;
+		}
+		break;
 	}
 	if (proctype >= NUM_AUXILIARY_PROCS)
 	{

src/backend/utils/init/miscinit.c

@@ -860,6 +860,20 @@ UnlinkLockFiles(int status, Datum arg)
 			(errmsg("database system is shut down")));
 }
 
+
+static void
+UnlinkPostmasterLock(int status, Datum arg)
+{
+	unlink("postmaster.pid");
+}
+
+void
+RegisterUnlinkLockFileCallback(void)
+{
+	on_proc_exit(UnlinkPostmasterLock, 0);
+}
+
+
 /*
  * Create a lockfile.
  *

src/include/miscadmin.h

@@ -451,5 +451,6 @@ extern bool has_rolreplication(Oid roleid);
 /* in access/transam/xlog.c */
 extern bool BackupInProgress(void);
 extern void CancelBackup(void);
+extern void RegisterUnlinkLockFileCallback(void);
 
 #endif							/* MISCADMIN_H */

src/include/postmaster/bgworker.h

@@ -157,5 +157,6 @@ extern void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, ui
 /* Block/unblock signals in a background worker process */
 extern void BackgroundWorkerBlockSignals(void);
 extern void BackgroundWorkerUnblockSignals(void);
+extern void BackgroundWorkerShmemAttach(void);
 
 #endif							/* BGWORKER_H */

src/include/postmaster/bgworker_internals.h

@@ -58,6 +58,8 @@ extern void StartBackgroundWorker(void) pg_attribute_noreturn();
 
 #ifdef EXEC_BACKEND
 extern BackgroundWorker *BackgroundWorkerEntry(int slotno);
+extern void SetBackgroundWorkerEntries(void* entries);
+extern void* GetBackgroundWorkerEntries(void);
 #endif
 
 #endif							/* BGWORKER_INTERNALS_H */