external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix some bugs

pull/7254/head
EmelSimsek 2023-10-12 13:59:31 +03:00 committed by Emel Şimşek
parent 87a5fcff41
commit 0ba47298a7
2 changed files with 63 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/backend/distributed/shared_library_init.c
View File

@ -481,6 +481,7 @@ _PG_init(void)
#endif
InitializeMaintenanceDaemon();
InitializeMaintenanceDaemonForAdminDB();
/* initialize coordinated transaction management */
InitializeTransactionManagement();

141
src/backend/distributed/utils/maintenanced.c
View File

@ -134,54 +134,48 @@ static void WarnMaintenanceDaemonNotStarted(void);
void
InitializeMaintenanceDaemon(void)
{
elog(LOG, "InitializeMaintenanceDaemon");
prev_shmem_startup_hook = shmem_startup_hook;
shmem_startup_hook = MaintenanceDaemonShmemInit;
}
/*
* InitializeMaintenanceDaemonForAdminDB is called in _PG_Init
* at which stage we are not in a transaction or have databaseOid
*/
void
InitializeMaintenanceDaemonForAdminDB(void)
{
elog(LOG, "InitializeMaintenanceDaemonForAdmin");
BackgroundWorker worker;
BackgroundWorkerHandle *handle = NULL;
memset(&worker, 0, sizeof(worker));
memset(&worker, 0, sizeof(worker));
SafeSnprintf(worker.bgw_name, sizeof(worker.bgw_name),
"Citus Maintenance Daemon: %u/%u",
0, 0);
SafeSnprintf(worker.bgw_name, sizeof(worker.bgw_name),
"Citus Maintenance Daemon: %u/%u",
0, 0);
/* request ability to connect to target database */
worker.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
/* request ability to connect to target database */
worker.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
/*
* No point in getting started before able to run query, but we do
* want to get started on Hot-Standby.
*/
worker.bgw_start_time = BgWorkerStart_ConsistentState;
/*
* No point in getting started before able to run query, but we do
* want to get started on Hot-Standby.
*/
worker.bgw_start_time = BgWorkerStart_ConsistentState;
/* Restart after a bit after errors, but don't bog the system. */
worker.bgw_restart_time = 5;
strcpy_s(worker.bgw_library_name,
sizeof(worker.bgw_library_name), "citus");
strcpy_s(worker.bgw_function_name, sizeof(worker.bgw_library_name),
"CitusMaintenanceDaemonMain");
/* Restart after a bit after errors, but don't bog the system. */
worker.bgw_restart_time = 5;
strcpy_s(worker.bgw_library_name,
sizeof(worker.bgw_library_name), "citus");
strcpy_s(worker.bgw_function_name, sizeof(worker.bgw_library_name),
"CitusMaintenanceDaemonMain");
worker.bgw_main_arg = (Datum)0;
worker.bgw_notify_pid = MyProcPid;
if (!RegisterDynamicBackgroundWorker(&worker, &handle)) {
elog(LOG, "RegisterDynamicBackgroundWorker failed for admin");
}
pid_t pid;
WaitForBackgroundWorkerStartup(handle, &pid);
pfree(handle);
worker.bgw_main_arg = (Datum) 0;
RegisterBackgroundWorker(&worker);
}
/*
* InitializeMaintenanceDaemonBackend, called at backend start and
* configuration changes, is responsible for starting a per-database
@ -190,7 +184,6 @@ InitializeMaintenanceDaemonForAdminDB(void)
void
InitializeMaintenanceDaemonBackend(void)
{
elog(LOG, "InitializeMaintenanceDaemonBackend");
Oid extensionOwner = CitusExtensionOwner();
bool found;
@ -322,7 +315,6 @@ WarnMaintenanceDaemonNotStarted(void)
void
CitusMaintenanceDaemonMain(Datum main_arg)
{
elog(LOG, "CitusMaintenanceDaemonMain");
Oid databaseOid = DatumGetObjectId(main_arg);
TimestampTz nextStatsCollectionTime USED_WITH_LIBCURL_ONLY =
TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 60 * 1000);
@ -340,70 +332,66 @@ CitusMaintenanceDaemonMain(Datum main_arg)
MaintenanceDaemonDBData *myDbData = NULL;
if (databaseOid == 0)
if (databaseOid == 0)
{
/* TODO : Get the admin database name from GUC contro_db*/
char* databaseName = "postgres";
char *databaseName = "postgres";
BackgroundWorkerInitializeConnection(databaseName, NULL, 0);
// Now we have a valid MyDatabaseId.
// Insert the daemon instance to the hash table.
bool found;
/* Now we have a valid MyDatabaseId. */
/* Insert the daemon instance to the hash table. */
bool found;
LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
myDbData = (MaintenanceDaemonDBData *) hash_search(
MaintenanceDaemonDBHash,
&MyDatabaseId,
HASH_ENTER_NULL,
&found);
MaintenanceDaemonDBHash,
&MyDatabaseId,
HASH_ENTER_NULL,
&found);
if (!myDbData)
{
LWLockRelease(&MaintenanceDaemonControl->lock);
return;
}
if (!myDbData)
{
LWLockRelease(&MaintenanceDaemonControl->lock);
return;
}
if (!found)
{
/* ensure the values in MaintenanceDaemonDBData are zero */
memset(((char *) myDbData) + sizeof(Oid), 0,
sizeof(MaintenanceDaemonDBData) - sizeof(Oid));
myDbData->userOid = 0;
myDbData->workerPid = 0;
myDbData->triggerNodeMetadataSync = false;
}
if (!found)
{
/* ensure the values in MaintenanceDaemonDBData are zero */
memset(((char *) myDbData) + sizeof(Oid), 0,
sizeof(MaintenanceDaemonDBData) - sizeof(Oid));
}
before_shmem_exit(MaintenanceDaemonShmemExit, ObjectIdGetDatum(MyDatabaseId));
databaseOid = MyDatabaseId;
myDbData->userOid = GetSessionUserId();
}
else
{
LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
myDbData = (MaintenanceDaemonDBData *)
hash_search(MaintenanceDaemonDBHash, &databaseOid,
HASH_FIND, NULL);
if (!myDbData || myDbData->workerPid != 0)
{
/*
* When the database crashes, background workers are restarted, but
* the state in shared memory is lost. In that case, we exit and
* wait for a session to call InitializeMaintenanceDaemonBackend
* to properly add it to the hash.
*/
myDbData = (MaintenanceDaemonDBData *)
hash_search(MaintenanceDaemonDBHash, &databaseOid,
HASH_FIND, NULL);
if (!myDbData || myDbData->workerPid != 0)
{
/*
* When the database crashes, background workers are restarted, but
* the state in shared memory is lost. In that case, we exit and
* wait for a session to call InitializeMaintenanceDaemonBackend
* to properly add it to the hash.
*/
LWLockRelease(&MaintenanceDaemonControl->lock);
proc_exit(0);
LWLockRelease(&MaintenanceDaemonControl->lock);
proc_exit(0);
}
before_shmem_exit(MaintenanceDaemonShmemExit, main_arg);
BackgroundWorkerInitializeConnectionByOid(databaseOid, myDbData->userOid, 0);
}
/*
* We do metadata sync in a separate background worker. We need its
* handle to be able to check its status.
@ -945,7 +933,6 @@ MaintenanceDaemonShmemSize(void)
void
MaintenanceDaemonShmemInit(void)
{
elog(LOG, "MaintenanceDaemonShmemInit");
bool alreadyInitialized = false;
HASHCTL hashInfo;
@ -997,7 +984,6 @@ MaintenanceDaemonShmemInit(void)
static void
MaintenanceDaemonShmemExit(int code, Datum arg)
{
elog(LOG, "MaintenanceDaemonShmemExit");
Oid databaseOid = DatumGetObjectId(arg);
LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
@ -1108,7 +1094,6 @@ LockCitusExtension(void)
void
StopMaintenanceDaemon(Oid databaseId)
{
elog(LOG, "StopMaintenanceDaemon");
bool found = false;
pid_t workerPid = 0;
@ -1140,7 +1125,6 @@ StopMaintenanceDaemon(Oid databaseId)
void
TriggerNodeMetadataSync(Oid databaseId)
{
elog(LOG, "TriggerNodeMetadataSync");
bool found = false;
LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
@ -1168,7 +1152,6 @@ TriggerNodeMetadataSync(Oid databaseId)
static bool
MetadataSyncTriggeredCheckAndReset(MaintenanceDaemonDBData *dbData)
{
elog(LOG, "MetadataSyncTriggeredCheckAndReset");
LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
bool metadataSyncTriggered = dbData->triggerNodeMetadataSync;