From 2bc447b370def11f29d4610638ed8d148fced3cc Mon Sep 17 00:00:00 2001 From: Peter Smith Date: Tue, 10 Jun 2025 17:38:11 +1000 Subject: [PATCH v20250610] VCI module - ENABLE_WOSROS_TRANS_DEFERRAL code --- contrib/vci/include/vci_mem.h | 4 + contrib/vci/include/vci_memory_entry.h | 4 + contrib/vci/include/vci_port.h | 35 ++ contrib/vci/include/vci_port_linux.h | 2 + contrib/vci/port/vci_linux.c | 600 +++++++++++++++++++++++++++++++++ contrib/vci/storage/vci_memory_entry.c | 90 +++++ contrib/vci/storage/vci_ros_daemon.c | 110 ++++++ contrib/vci/vci_read_guc.c | 13 + contrib/vci/vci_shmem.c | 5 + 9 files changed, 863 insertions(+) diff --git a/contrib/vci/include/vci_mem.h b/contrib/vci/include/vci_mem.h index 7f9d2cb..6e73178 100644 --- a/contrib/vci/include/vci_mem.h +++ b/contrib/vci/include/vci_mem.h @@ -19,6 +19,7 @@ #include "utils/palloc.h" #include "vci.h" +#include "vci_port.h" #include "vci_ros.h" #include "vci_memory_entry.h" @@ -64,6 +65,9 @@ typedef struct VciGucStruct bool enable_plainagg; bool enable_hashjoin; bool enable_nestloop; +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + bool enable_wosros_trans_deferral; +#endif /* GUC parameters for internal use */ bool enable_ros_control_daemon; diff --git a/contrib/vci/include/vci_memory_entry.h b/contrib/vci/include/vci_memory_entry.h index b0b9d4b..797bfd9 100644 --- a/contrib/vci/include/vci_memory_entry.h +++ b/contrib/vci/include/vci_memory_entry.h @@ -101,6 +101,10 @@ extern void vci_TouchMemoryEntry(vci_id_t *vciid, Oid tsid); extern bool vci_GetWosRosConvertingVCI(vci_wosros_conv_worker_arg_t *vci_info); extern void vci_freeMemoryEntry(vci_id_t *vciid); +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL +extern int vci_GetUnmonitoredDevicesHavingVCI(vci_dev_t *device_array); +#endif + extern void vci_ReconfigureUnmonitoredMemoryEntry(void); extern void vci_update_memoryentry_in_devloadinfo(void); extern void vci_MoveTranslatedVCI2Tail(void); diff --git a/contrib/vci/include/vci_port.h b/contrib/vci/include/vci_port.h index 7cec2c7..c9f228e 100644 --- a/contrib/vci/include/vci_port.h +++ b/contrib/vci/include/vci_port.h @@ -111,6 +111,12 @@ typedef struct */ unsigned int devstat_num; +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + + /* Status of devices */ + vci_devstat_t devstat[VCI_DISKSTAT_RECORD_NUM]; +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + bool contain_vci; vci_memory_entry_list_t *memory_entry_queue; @@ -123,6 +129,35 @@ typedef struct } vci_devload_t; +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + +/* + * Set of device names + */ +typedef struct +{ + char devname[VCI_PATH_MAX]; + dlist_node link; +} vci_dev_t; + +/* + * Stores device info for updating mount point info + */ +typedef struct +{ + vci_dev_t *dev_set; /* Set of device name */ + dlist_head device_set; /* Set of devices to be monitored */ +} vci_update_info_t; + +extern bool vci_device_is_in_array(vci_dev_t *device_array, int size, const char *devname); +extern bool vci_prepare_updating_device_info(vci_devload_t *devload_array, int num_devload_info, int max_monitoring_devnum, + vci_dev_t *unmonitored_device_array, int unmonitored_device_num, vci_update_info_t *update_info); +extern void vci_close_updating_device_info(vci_update_info_t *update_info); +extern void vci_set_devload_info(vci_devload_t *devload_array, dlist_head *free_memory_entry_queue_list, + int *num_devload_info, int max_array_size, dlist_head *monitoring_dev_list); + +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + extern void vci_init_devloadinfo(vci_devload_t *devinfo, const char *devname); extern void vci_init_tablespace2dev(void); extern void vci_fini_tablespace2dev(void); diff --git a/contrib/vci/include/vci_port_linux.h b/contrib/vci/include/vci_port_linux.h index 2513217..905bbd8 100644 --- a/contrib/vci/include/vci_port_linux.h +++ b/contrib/vci/include/vci_port_linux.h @@ -45,4 +45,6 @@ typedef struct unsigned int rq_ticks; /* Field 11 */ } vci_devstat_t; +#define ENABLE_WOSROS_TRANS_DEFERRAL + #endif /* VCI_PORT_LINUX_H */ diff --git a/contrib/vci/port/vci_linux.c b/contrib/vci/port/vci_linux.c index 81ab5da..86f9368 100644 --- a/contrib/vci/port/vci_linux.c +++ b/contrib/vci/port/vci_linux.c @@ -61,6 +61,17 @@ static LWLock *mntpoint2dev_lock; */ static dlist_head mntpoint2dev; +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL +static int vci_get_mountpoint_num(void); +static bool dev_is_in_set(dlist_head *set, const char *devname); +static void add_relation_to_mountpoint2device_mapping(const char *mountpoint, const char *devname); +static bool mountpoint_set_mounted(const char *mntpoint, const char *devname); +static void reset_mounted_flag(void); +static void release_nonmounted_mntpoint2dev(void); +static int read_mounts(vci_dev_t *dev_array, int max_monitored_device_num); +static bool add_dev_to_set(dlist_head *set, vci_dev_t *dev); +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + static FILE *fopen_with_retry(const char *path, const char *mode); static char *fgets_with_retry(char *s, int size, FILE *stream); @@ -268,3 +279,592 @@ vci_IsLowIOLoad(vci_devload_t *devinfo) return ret; } +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + +/* + * Returns number of mount points in /proc/mounts + */ +static int +vci_get_mountpoint_num(void) +{ + int num_mountpoints = 0; + char line[LINESIZE]; + FILE *fp; + + if ((fp = fopen_with_retry(VCI_MOUNTS_FILE, "r")) == NULL) + return 0; + + while (fgets_with_retry(line, LINESIZE, fp) != NULL) + { + char devname[PATH_MAX], + mntpoint[PATH_MAX]; + int num; + + num = sscanf(line, "%s %s %*s %*s %*u %*u", devname, mntpoint); + if (num != 2) + ereport(PANIC, (errcode_for_file_access(), + errmsg_internal("%s is invalid", VCI_MOUNTS_FILE))); + + num_mountpoints++; + } + + fclose(fp); + + return num_mountpoints; +} + +/** + * check whether dev is in set or not. + * + * @param[in] set set of pairs of a mount point and a device name. + * @param[in] dev device name + * + * @return true if dev is contained in set, false otherwise + */ +static bool +dev_is_in_set(dlist_head *set, const char *devname) +{ + bool found = false; + dlist_iter iter; + + dlist_foreach(iter, set) + { + vci_dev_t *item; + + item = dlist_container(vci_dev_t, link, iter.cur); + if (strcmp(item->devname, devname) == 0) + { + found = true; + break; + } + } + + return found; +} + +/* + * Updates the device infor from the /proc/diskstats + * + * @param[in,out] devload_array an array of vci_devload_info_t, which represents a set of devices. + * @param[in,out] free_memory_entry_queue_list a list of memory entry queues, each element of which are used for a vci_devload_t type value. + * @param[in,out] num_devload_info the number of elements of devload_array + * @param[in] max_array_size the size of devload_array + * @param[in] monitoring_dev_list a set of names of devices to be monitored. + */ +void +vci_set_devload_info(vci_devload_t *devload_array, dlist_head *free_memory_entry_queue_list, + int *num_devload_info, int max_array_size, dlist_head *monitoring_dev_list) +{ + FILE *fp; + char line[LINESIZE]; + + if ((fp = fopen_with_retry(VCI_DISKSTAT_FILE, "r")) == NULL) + ereport(PANIC, (errcode_for_file_access(), + errmsg_internal("cannot read %s", VCI_DISKSTAT_FILE))); + + while (fgets_with_retry(line, LINESIZE, fp) != NULL) + { + char devname[PATH_MAX]; + char path[PATH_MAX]; + unsigned long rd_ios, + wr_ios; + int num; + + num = sscanf(line, "%*d %*d %s %lu %*u %*u %*u %lu", devname, &rd_ios, &wr_ios); + + if (num != 3) + ereport(PANIC, (errcode_for_file_access(), + errmsg_internal("%s is invalid", VCI_DISKSTAT_FILE))); + + sprintf(path, "/sys/block/%s", devname); /* allow virtual device */ + + if (dev_is_in_set(monitoring_dev_list, devname)) + { + vci_devload_t *prev_stat; + + if ((prev_stat = vci_find_devloadinfo(devload_array, *num_devload_info, devname)) == NULL) + { + if (access(path, F_OK) == 0 /* device exists */ && + (rd_ios != 0 || wr_ios != 0) /* active device */ ) + { + vci_devload_t *tmp; + + assert(*num_devload_info < max_array_size); + + tmp = &(devload_array[*num_devload_info]); + vci_init_devloadinfo(tmp, devname); + tmp->await = VCI_INITIAL_AWAIT; + tmp->alive = true; + + Assert(!dlist_is_empty(free_memory_entry_queue_list)); + tmp->memory_entry_queue = + dlist_container(vci_memory_entry_list_t, link, + dlist_pop_head_node(free_memory_entry_queue_list)); + dlist_init(&(tmp->memory_entry_queue->head)); + + *num_devload_info += 1; + } + else + { + if (access(path, F_OK) == 0 && (rd_ios == 0 && wr_ios == 0)) + elog(DEBUG1, "%s is not active: rd_ios = 0 & wr_ios = 0", devname); + } + } + else + { + elog(DEBUG1, "alive %s", devname); + prev_stat->alive = true; + } + } + } + + fclose(fp); +} + + +/* + * Add a mountpoint-devname pair to mntpoint2dev + * + * @param[in] mountpoint mountpoint name + * @param[in] devname device name + */ +static void +add_relation_to_mountpoint2device_mapping(const char *mountpoint, const char *devname) +{ + mountpoint_dev_pair_t *pair; + bool reach_last = false; + + pair = (mountpoint_dev_pair_t *) palloc(sizeof(mountpoint_dev_pair_t)); + + strcpy(pair->mountpoint, mountpoint); + pair->len = strlen(pair->mountpoint); + strcpy(pair->devname, devname); + pair->mounted = true; + + LWLockAcquire(mntpoint2dev_lock, LW_EXCLUSIVE); + + if (dlist_is_empty(&mntpoint2dev)) + dlist_push_head(&mntpoint2dev, &(pair->link)); + else + { + dlist_node *n; + mountpoint_dev_pair_t *item; + + n = dlist_head_node(&mntpoint2dev); + item = dlist_container(mountpoint_dev_pair_t, link, n); + while (!reach_last && strcmp(item->mountpoint, pair->mountpoint) > 0) + { + if (dlist_has_next(&mntpoint2dev, n)) + { + n = dlist_next_node(&mntpoint2dev, n); + item = dlist_container(mountpoint_dev_pair_t, link, n); + } + else + reach_last = true; + } + + if (reach_last) + dlist_push_tail(&mntpoint2dev, &(pair->link)); + else + dlist_insert_before(n, &(pair->link)); + } + + LWLockRelease(mntpoint2dev_lock); +} + +/** + * set flag 'mounted' if the mount point specified by 'mntpoint' exists in + * the current mountpoint -> device mapping. devname can be changed to + * VCI_PSEUDO_UNMONITORED_DEVICE. Therefore, it is reset to the original device name. + * @param[in] mntpoint mountpoint + * @param[in] devname device name + * @return true if the current set contains mntpoint, false otherwise + */ +static bool +mountpoint_set_mounted(const char *mntpoint, const char *devname) +{ + bool ret = false; + dlist_iter iter; + + LWLockAcquire(mntpoint2dev_lock, LW_EXCLUSIVE); + + dlist_foreach(iter, &mntpoint2dev) + { + mountpoint_dev_pair_t *pair; + + pair = dlist_container(mountpoint_dev_pair_t, link, iter.cur); + if (strcmp(mntpoint, pair->mountpoint) == 0) + { + pair->mounted = true; + strcpy(pair->devname, devname); + ret = true; + break; + } + } + + LWLockRelease(mntpoint2dev_lock); + + return ret; +} + +/** + * set flag 'mounted' as false in the current path->dev mapping before reading /proc/mounts + */ +static void +reset_mounted_flag(void) +{ + dlist_iter iter; + + LWLockAcquire(mntpoint2dev_lock, LW_EXCLUSIVE); + dlist_foreach(iter, &mntpoint2dev) + { + mountpoint_dev_pair_t *pair; + + pair = dlist_container(mountpoint_dev_pair_t, link, iter.cur); + pair->mounted = false; + } + LWLockRelease(mntpoint2dev_lock); +} + +/* + * Remove mount info which is currently not mounted + */ +static void +release_nonmounted_mntpoint2dev(void) +{ + dlist_mutable_iter miter; + + LWLockAcquire(mntpoint2dev_lock, LW_EXCLUSIVE); + dlist_foreach_modify(miter, &mntpoint2dev) + { + mountpoint_dev_pair_t *pair; + + pair = dlist_container(mountpoint_dev_pair_t, link, miter.cur); + if (!pair->mounted) + { + dlist_delete(miter.cur); + pfree(pair); + } + } + LWLockRelease(mntpoint2dev_lock); +} + +/* + * Read /proc/mounts to create mountpoint -> device mapping + * @param[in,out] dev_array array of devices the devices in /proc/mounts is stored + * @param[in] max_monitored_device_num + * + * @return number of attached devices + */ +static int +read_mounts(vci_dev_t *dev_array, int max_monitored_device_num) +{ + int mounted_device_num; + FILE *fp; + char buffer[PATH_MAX]; + + reset_mounted_flag(); + + if ((fp = fopen_with_retry(VCI_MOUNTS_FILE, "r")) == NULL) + ereport(PANIC, (errcode_for_file_access(), + errmsg_internal("could not open file \"%s\": %m", VCI_MOUNTS_FILE))); + + mounted_device_num = 0; + while (fgets_with_retry(buffer, sizeof(buffer), fp) != NULL) + { + int len; + char mountpoint[PATH_MAX]; + char devfile[PATH_MAX]; /* form like /dev/sda */ + char *devname; /* pointer in devpath to the head of device name, like sda */ + char path[PATH_MAX]; + int num; + + num = sscanf(buffer, "%s %s %*s %*s %*u %*u", devfile, mountpoint); + if (num != 2) + ereport(PANIC, (errcode_for_file_access(), + errmsg_internal("%s is invalid", VCI_MOUNTS_FILE))); + + if (strncmp(devfile, "/dev/", strlen("/dev/")) == 0) + { + /* dev is a partition name. Translate that to a device name. */ + + char resolved_path[PATH_MAX]; + + if (realpath(devfile, resolved_path) != NULL) + { + len = strlen(resolved_path); + len--; + while (len > 0 && isdigit(resolved_path[len])) + { + resolved_path[len] = '\0'; + len--; + } + while (len > 0 && resolved_path[len] != '/') + { + len--; + } + len++; + + devname = &resolved_path[len]; + + if (strcmp(RAMFS, devname) != 0) + { + sprintf(path, "/sys/block/%s", devname); + if (access(path, F_OK) == 0) + { + if (!vci_device_is_in_array(dev_array, mounted_device_num, devname)) + { + strcpy(dev_array[mounted_device_num].devname, devname); + mounted_device_num++; + } + + if (!mountpoint_set_mounted(mountpoint, devname)) + { + if (mounted_device_num < max_monitored_device_num) + { + add_relation_to_mountpoint2device_mapping(mountpoint, devname); + } + else + { + mounted_device_num = -1; + goto SIZE_EXCESS; + } + } + } + } + } + } + } + + fclose(fp); + + release_nonmounted_mntpoint2dev(); + +#if 1 + LWLockAcquire(mntpoint2dev_lock, LW_SHARED); + { + dlist_iter iter; + + dlist_foreach(iter, &mntpoint2dev) + { + mountpoint_dev_pair_t *pair; + + pair = dlist_container(mountpoint_dev_pair_t, link, iter.cur); + elog(DEBUG1, "mapping: mountpoint [%s] -> device [%s]", pair->mountpoint, pair->devname); + } + } + LWLockRelease(mntpoint2dev_lock); +#endif + +SIZE_EXCESS: + return mounted_device_num; + +} + +/** + * set := set + {dev} + * + * @param[in] set set of vci_dev_t + * @param[in] dev an element added to set + * + * @return the number of elements in set + */ +static bool +add_dev_to_set(dlist_head *set, vci_dev_t *dev) +{ + bool ret; + + if (!dev_is_in_set(set, dev->devname)) + { + ret = true; + dlist_push_head(set, &(dev->link)); + } + else + ret = false; + + return ret; +} + +/** + * check whether devname is in dev_array or not. + * @param[in] dev_array array of device names + * @param[in] size size of dev_array + * @param[in] devname device name + * @return true is devame is in dev_array, false otherwise + */ +bool +vci_device_is_in_array(vci_dev_t *device_array, int size, const char *devname) +{ + int j; + bool found = false; + + for (j = 0; j < size; j++) + { + if (strcmp(devname, device_array[j].devname) == 0) + { + found = true; + break; + } + } + + return found; +} + +/* + * Determine devices to be monitored by reading /proc/mounts. + * + * @param[in] devload_array array of vci_devload_array_t + * @param[in] num_devload_info the number of elements set in devload_array + * @param[in] max_monitoring_devnum size devload_array + * @param[in] unmonitored_device_array array of unmonitored devices which have VCIs. + * @param[in] unmonitored_device_num the size of ts_array + * @param[out] update_info set of devices which should be monitored. + * + * @return: true if all devices can be monitored, false otherwise + */ +bool +vci_prepare_updating_device_info(vci_devload_t *devload_array, int num_devload_info, int max_monitoring_devnum, + vci_dev_t *unmonitored_device_array, int unmonitored_device_num, vci_update_info_t *update_info) +{ + bool ret = true; + int mntpoint_num; + int mounted_device_num; + + int i; + + dlist_head *device_set = &(update_info->device_set); /* list of vci_dev_t */ + int monitored_device_num; + + /* read /proc/mounts and set mountpoint informaiton */ + mounted_device_num = -1; + while (mounted_device_num < 0) + { + mntpoint_num = vci_get_mountpoint_num() + 1; /* +1 for the root of + * path tree (path="") */ + + update_info->dev_set = (vci_dev_t *) palloc(sizeof(vci_dev_t) * mntpoint_num); + + mounted_device_num = read_mounts(update_info->dev_set, mntpoint_num); + } + +#if 1 + elog(DEBUG1, "devices returned by read_mounts"); + for (i = 0; i < mounted_device_num; i++) + elog(DEBUG1, "device: [%s]", update_info->dev_set[i].devname); +#endif + + /* + * += 1 for storing a path "" and a device name + * VCI_PSEUDO_UNMONITORED_DEVICE. The vci_mntpoint_element_t type value + * becomes the root of the mountpoint path tree. + */ + mntpoint_num += 1; + + dlist_init(device_set); + + /* + * Set alive flag to false if the device is the target but VCI is not on + * it + */ + monitored_device_num = 0; + for (i = 0; i < num_devload_info; i++) + { + if (!devload_array[i].contain_vci || + strcmp(devload_array[i].devname, VCI_PSEUDO_UNMONITORED_DEVICE) != 0) + devload_array[i].alive = false; + else + monitored_device_num++; + } + + for (i = 0; i < mounted_device_num; i++) + { + char *devname = update_info->dev_set[i].devname; + + if (monitored_device_num < max_monitoring_devnum) + { + if (vci_device_is_in_array(unmonitored_device_array, unmonitored_device_num, devname) && + add_dev_to_set(device_set, &(update_info->dev_set[i]))) + monitored_device_num++; + } + else + break; + } + + LWLockAcquire(mntpoint2dev_lock, LW_EXCLUSIVE); + { + dlist_iter iter; + + dlist_foreach(iter, &mntpoint2dev) + { + mountpoint_dev_pair_t *pair; + + pair = dlist_container(mountpoint_dev_pair_t, link, iter.cur); + + if (monitored_device_num < max_monitoring_devnum) + { + vci_dev_t *dev = NULL; + + for (i = 0; i < mntpoint_num; i++) + { + if (strcmp(update_info->dev_set[i].devname, pair->devname) == 0) + { + dev = &(update_info->dev_set[i]); + break; + } + } + Assert(dev != NULL); + + if (add_dev_to_set(device_set, dev)) + monitored_device_num++; + } + else + { + if (!dev_is_in_set(device_set, pair->devname)) + { + strcpy(pair->devname, VCI_PSEUDO_UNMONITORED_DEVICE); + + elog(DEBUG1, "%s %s cannot be monitored because of size limitation", + pair->mountpoint, pair->devname); + + ret = false; + } + } + + } + + } + LWLockRelease(mntpoint2dev_lock); + +#if 1 + { + dlist_iter iter; + + elog(DEBUG1, "devices returned by vci_prepare_updating_device_info"); + dlist_foreach(iter, device_set) + { + vci_dev_t *item; + + item = dlist_container(vci_dev_t, link, iter.cur); + + elog(DEBUG1, "dev: [%s]", item->devname); + } + + } +#endif + + return ret; + +} + +/* + * Releases memory used to update device info + * + * @param[in] update_info stores device information in /proc/mounts + */ +void +vci_close_updating_device_info(vci_update_info_t *update_info) +{ + pfree(update_info->dev_set); +} + +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ diff --git a/contrib/vci/storage/vci_memory_entry.c b/contrib/vci/storage/vci_memory_entry.c index b0b0192..e20f51d 100644 --- a/contrib/vci/storage/vci_memory_entry.c +++ b/contrib/vci/storage/vci_memory_entry.c @@ -736,6 +736,42 @@ vci_GetWosRosConvertingVCI(vci_wosros_conv_worker_arg_t *vciinfo) elog(LOG, "wos->ros conversion on device (%s): vci oid=%u dbid=%u", dl->devname, vciinfo->oid, vciinfo->dbid); +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + if (VciGuc.enable_wosros_trans_deferral && + entry->tsid == InvalidOid) + { + Oid default_tsid; + + Assert(vciinfo->dbid != InvalidOid); + + /* start transaction */ + SetCurrentStatementStartTimestamp(); + StartTransactionCommand(); + PushActiveSnapshot(GetTransactionSnapshot()); + + default_tsid = getDefaultTablespaceOid(vciinfo->dbid); + + /* close transaction */ + PopActiveSnapshot(); + CommitTransactionCommand(); + + if (default_tsid == InvalidOid) + { + /* the database was dropped. */ + elog(DEBUG2, + "the database %d was dropped", vciinfo->dbid); + found = false; + } + else if (default_tsid != entry->real_tsid) + { + + entry->real_tsid = default_tsid; + + dlist_delete(ret); + registerMemoryEntry2Device(entry); + } + } +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ } return found; @@ -757,6 +793,60 @@ vci_TouchMemoryEntry(vci_id_t *vciid, Oid tsid) vci_ReleaseMemoryEntry(entry); } +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL +/** + * @param[out] device_array array of names of devices whose IO-load is not monitored. + * + * @return the number of unmonitored devices + */ +int +vci_GetUnmonitoredDevicesHavingVCI(vci_dev_t *device_array) +{ + int device_num; + vci_devload_t *unmonitored_dev; + dlist_iter iter; + dlist_head *unmonitored_memory_entry_queue; + + LWLockAcquire(VciShmemAddr->memory_entries->lock, LW_EXCLUSIVE); + + unmonitored_dev = + vci_find_devloadinfo(VciShmemAddr->devload_array, + VciShmemAddr->num_devload_info, VCI_PSEUDO_UNMONITORED_DEVICE); + Assert(unmonitored_dev != NULL); + + unmonitored_memory_entry_queue = &(unmonitored_dev->memory_entry_queue->head); + + device_num = 0; + dlist_foreach(iter, (unmonitored_memory_entry_queue)) + { + const char *devname; + vci_memory_entry_t *item = dlist_container(vci_memory_entry_t, link, iter.cur); + char *tablespace_path; + bool free_flag = true; + + if ((tablespace_path = getTablespacePath(item->real_tsid)) == NULL) + { + tablespace_path = DataDir; + free_flag = false; + } + + devname = vci_convert_tablespace_to_devname(tablespace_path); + if (!vci_device_is_in_array(device_array, device_num, devname)) + { + strcpy(device_array[device_num].devname, devname); + device_num++; + } + + if (free_flag) + pfree(tablespace_path); + + } + LWLockRelease(VciShmemAddr->memory_entries->lock); + + return device_num; +} +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + void vci_ReconfigureUnmonitoredMemoryEntry(void) { diff --git a/contrib/vci/storage/vci_ros_daemon.c b/contrib/vci/storage/vci_ros_daemon.c index d23a623..6f7e95b 100644 --- a/contrib/vci/storage/vci_ros_daemon.c +++ b/contrib/vci/storage/vci_ros_daemon.c @@ -208,6 +208,10 @@ void vci_ROS_control_daemon_main(Datum main_arg) { bool excess_devnum = false; +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + vci_update_info_t update_info; + vci_dev_t *unmonitored_device_array; +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ pg_bindtextdomain(TEXTDOMAIN); @@ -242,6 +246,24 @@ vci_ROS_control_daemon_main(Datum main_arg) /* initialize tablespace->dev map */ vci_init_tablespace2dev(); +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + unmonitored_device_array = palloc(sizeof(vci_dev_t) * VciShmemAddr->memory_entries->capacity_hash_entries); + Assert(unmonitored_device_array != NULL); + + if (VciGuc.enable_wosros_trans_deferral) + { + LWLockAcquire(VciShmemAddr->io_load_lock, LW_EXCLUSIVE); + vci_prepare_updating_device_info(VciShmemAddr->devload_array, VciShmemAddr->num_devload_info, VciShmemAddr->max_devices, + unmonitored_device_array, 0, &update_info); + vci_set_devload_info(VciShmemAddr->devload_array, + &VciShmemAddr->free_memory_entry_queue_list, + &(VciShmemAddr->num_devload_info), + VciShmemAddr->max_devices, &(update_info.device_set)); + vci_close_updating_device_info(&update_info); + LWLockRelease(VciShmemAddr->io_load_lock); + } +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + /* Main loop */ while (!gotSigterm) { @@ -274,6 +296,90 @@ vci_ROS_control_daemon_main(Datum main_arg) /* Check VCI' database is exists */ vci_RemoveMemoryEntryOnDroppedDatabase(); +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + if (VciGuc.enable_wosros_trans_deferral) + { + bool all_dev_monitored; + vci_devload_t *unmonitored_dev; + dlist_head *memory_entry_on_unmonitored_dev; + int unmonitored_device_num; + + unmonitored_device_num = vci_GetUnmonitoredDevicesHavingVCI(unmonitored_device_array); + + all_dev_monitored = + vci_prepare_updating_device_info(VciShmemAddr->devload_array, VciShmemAddr->num_devload_info, VciShmemAddr->max_devices, + unmonitored_device_array, unmonitored_device_num, &update_info); + + if (!excess_devnum && !all_dev_monitored) + { + ereport(WARNING, + (errcode(ERRCODE_INTERNAL_ERROR), + errmsg("The number of devices exceeds the upper bound"))); + excess_devnum = true; + } + excess_devnum = !all_dev_monitored; + + vci_set_devload_info(VciShmemAddr->devload_array, + &VciShmemAddr->free_memory_entry_queue_list, + &(VciShmemAddr->num_devload_info), + VciShmemAddr->max_devices, &(update_info.device_set)); + vci_close_updating_device_info(&update_info); + + unmonitored_dev = vci_find_devloadinfo(VciShmemAddr->devload_array, + VciShmemAddr->num_devload_info, + VCI_PSEUDO_UNMONITORED_DEVICE); + Assert(unmonitored_dev != NULL); + memory_entry_on_unmonitored_dev = &(unmonitored_dev->memory_entry_queue->head); + + /* remove the entries of unmarked devices */ + for (i = 0; i < VciShmemAddr->num_devload_info; i++) + { + vci_devload_t *item = &(VciShmemAddr->devload_array[i]); + + if (!item->alive) + { + + dlist_head *memory_entry_queue = &(item->memory_entry_queue->head); + + elog(DEBUG1, "ros daemon: stop monitoring %s", item->devname); + while (!dlist_is_empty(memory_entry_queue)) + { + dlist_node *tmp; /* vci_memory_entry_t */ + vci_memory_entry_t *entry; + + tmp = dlist_pop_head_node(memory_entry_queue); + entry = dlist_container(vci_memory_entry_t, link, tmp); + dlist_push_tail(memory_entry_on_unmonitored_dev, &(entry->link)); + } + + VciShmemAddr->num_devload_info--; + memmove(&(VciShmemAddr->devload_array[i]), + &(VciShmemAddr->devload_array[VciShmemAddr->num_devload_info]), + sizeof(vci_devload_t)); + item = &(VciShmemAddr->devload_array[VciShmemAddr->num_devload_info]); + vci_init_devloadinfo(item, VCI_FREE_DEVLOAD_INFO_NAME); + + Assert(dlist_is_empty(&(item->memory_entry_queue->head))); + dlist_push_head(&(VciShmemAddr->free_memory_entry_queue_list), &(item->memory_entry_queue->link)); + item->await = VCI_FREE_DEVLOAD_AWAIT; + } + + vci_ReconfigureUnmonitoredMemoryEntry(); + } + +#if 1 + elog(DEBUG1, "ros daemon: monitoring devices"); + for (i = 0; i < VciShmemAddr->num_devload_info; i++) + { + elog(DEBUG1, "ros daemon: %s", VciShmemAddr->devload_array[i].devname); + } +#endif + + qsort(VciShmemAddr->devload_array, VciShmemAddr->num_devload_info, + sizeof(vci_devload_t), compare_devload); + } +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + vci_update_memoryentry_in_devloadinfo(); if (gotSigterm) @@ -382,6 +488,10 @@ reload_configuration: done: +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + pfree(unmonitored_device_array); +#endif /* ENABLE_WOSROS_TRANS_DEFERRAL */ + vci_fini_tablespace2dev(); /* diff --git a/contrib/vci/vci_read_guc.c b/contrib/vci/vci_read_guc.c index 33b4ab1..9eb2626 100644 --- a/contrib/vci/vci_read_guc.c +++ b/contrib/vci/vci_read_guc.c @@ -202,6 +202,19 @@ static struct config_bool VciConfigureNamesBool[] = NULL, NULL, NULL, }, +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + { + { + "vci.enable_wosros_trans_deferral", + PGC_POSTMASTER, DEVELOPER_OPTIONS, + "Enables deferral of WOS->ROS transforamtion", + NULL, + }, + &VciGuc.enable_wosros_trans_deferral, + false, + NULL, NULL, NULL, + }, +#endif }; static struct config_int VciConfigureNamesInt[] = diff --git a/contrib/vci/vci_shmem.c b/contrib/vci/vci_shmem.c index 37bc921..089660a 100644 --- a/contrib/vci/vci_shmem.c +++ b/contrib/vci/vci_shmem.c @@ -92,7 +92,12 @@ vci_shmem_startup_routine(void) Size size; #endif +#ifdef ENABLE_WOSROS_TRANS_DEFERRAL + if (!VciGuc.enable_wosros_trans_deferral) + VciGuc.max_devices = 0; +#else VciGuc.max_devices = 0; +#endif if (IsPostmasterEnvironment) { -- 1.8.3.1