LuoGe
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

!1428 发布订阅增强：解决一个异常的subscription会导致正常的subscription也无法启动worker的问题 

Merge pull request !1428 from pengjiong/pubsub_enhance_launcher
This commit is contained in:
opengauss-bot 2022-01-05 01:48:31 +00:00 committed by Gitee
parent 345c633c0b f2af76de00
commit 1e7740b754
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
1 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/gausskernel/storage/replication/logical/launcher.cpp
View File

@ -483,6 +483,7 @@ void ApplyLauncherMain()
sigjmp_buf localSigjmpBuf;
TimestampTz last_start_time = 0;
char username[NAMEDATALEN];
int nextLaunchSub = 0;
/* we are a postmaster subprocess now */
IsUnderPostmaster = true;
@ -650,6 +651,7 @@ void ApplyLauncherMain()
sublist = get_subscription_list();
/* Start the missing workers for enabled subscriptions. */
List *pendingSubList = NIL;
foreach (lc, sublist) {
Subscription *sub = (Subscription *)lfirst(lc);
LogicalRepWorker *w;
@ -662,15 +664,27 @@ void ApplyLauncherMain()
w = logicalrep_worker_find(sub->oid);
LWLockRelease(LogicalRepWorkerLock);
/* Add to pending list if the subscription has no work attached */
if (w == NULL) {
logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, sub->owner);
last_start_time = now;
wait_time = wal_retrieve_retry_interval;
/* Limit to one worker per mainloop cycle. */
break;
pendingSubList = lappend(pendingSubList, sub);
}
}
/*
* Try to launch the subscription worker one by one, to avoid launch an abnormal
* subscription again and again and normal subscription become starve.
*/
if (pendingSubList != NIL) {
nextLaunchSub = nextLaunchSub % list_length(pendingSubList);
Subscription *readyToLaunchSub = (Subscription*)list_nth(pendingSubList, nextLaunchSub);
nextLaunchSub++;
logicalrep_worker_launch(readyToLaunchSub->dbid, readyToLaunchSub->oid,
readyToLaunchSub->name, readyToLaunchSub->owner);
last_start_time = now;
wait_time = wal_retrieve_retry_interval;
}
/* Switch back to original memory context. */
MemoryContextSwitchTo(oldctx);
/* Clean the temporary memory. */