public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v52 1/7] sequential scan for dshash 2+ messages / 2 participants [nested] [flat]
* [PATCH v52 1/7] sequential scan for dshash @ 2020-03-13 07:58 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Kyotaro Horiguchi @ 2020-03-13 07:58 UTC (permalink / raw) Dshash did not allow scan the all entries sequentially. This adds the functionality. The interface is similar but a bit different both from that of dynahash and simple dshash search functions. One of the most significant differences is the sequential scan interface of dshash always needs a call to dshash_seq_term when scan ends. Another is locking. Dshash holds partition lock when returning an entry, dshash_seq_next() also holds lock when returning an entry but callers shouldn't release it, since the lock is essential to continue a scan. The seqscan interface allows entry deletion while a scan. The in-scan deletion should be performed by dshash_delete_current(). --- src/backend/lib/dshash.c | 160 ++++++++++++++++++++++++++++++- src/include/lib/dshash.h | 22 +++++ src/tools/pgindent/typedefs.list | 1 + 3 files changed, 182 insertions(+), 1 deletion(-) diff --git a/src/backend/lib/dshash.c b/src/backend/lib/dshash.c index e0c763be32..520bfa0979 100644 --- a/src/backend/lib/dshash.c +++ b/src/backend/lib/dshash.c @@ -127,6 +127,10 @@ struct dshash_table #define NUM_SPLITS(size_log2) \ (size_log2 - DSHASH_NUM_PARTITIONS_LOG2) +/* How many buckets are there in a given size? */ +#define NUM_BUCKETS(size_log2) \ + (((size_t) 1) << (size_log2)) + /* How many buckets are there in each partition at a given size? */ #define BUCKETS_PER_PARTITION(size_log2) \ (((size_t) 1) << NUM_SPLITS(size_log2)) @@ -153,6 +157,10 @@ struct dshash_table #define BUCKET_INDEX_FOR_PARTITION(partition, size_log2) \ ((partition) << NUM_SPLITS(size_log2)) +/* Choose partition based on bucket index. */ +#define PARTITION_FOR_BUCKET_INDEX(bucket_idx, size_log2) \ + ((bucket_idx) >> NUM_SPLITS(size_log2)) + /* The head of the active bucket for a given hash value (lvalue). */ #define BUCKET_FOR_HASH(hash_table, hash) \ (hash_table->buckets[ \ @@ -324,7 +332,7 @@ dshash_destroy(dshash_table *hash_table) ensure_valid_bucket_pointers(hash_table); /* Free all the entries. */ - size = ((size_t) 1) << hash_table->size_log2; + size = NUM_BUCKETS(hash_table->size_log2); for (i = 0; i < size; ++i) { dsa_pointer item_pointer = hash_table->buckets[i]; @@ -592,6 +600,156 @@ dshash_memhash(const void *v, size_t size, void *arg) return tag_hash(v, size); } +/* + * dshash_seq_init/_next/_term + * Sequentially scan through dshash table and return all the + * elements one by one, return NULL when no more. + * + * dshash_seq_term should always be called when a scan finished. + * The caller may delete returned elements midst of a scan by using + * dshash_delete_current(). exclusive must be true to delete elements. + */ +void +dshash_seq_init(dshash_seq_status *status, dshash_table *hash_table, + bool exclusive) +{ + status->hash_table = hash_table; + status->curbucket = 0; + status->nbuckets = 0; + status->curitem = NULL; + status->pnextitem = InvalidDsaPointer; + status->curpartition = -1; + status->exclusive = exclusive; +} + +/* + * Returns the next element. + * + * Returned elements are locked and the caller must not explicitly release + * it. It is released at the next call to dshash_next(). + */ +void * +dshash_seq_next(dshash_seq_status *status) +{ + dsa_pointer next_item_pointer; + + if (status->curitem == NULL) + { + int partition; + + Assert(status->curbucket == 0); + Assert(!status->hash_table->find_locked); + + /* first shot. grab the first item. */ + partition = + PARTITION_FOR_BUCKET_INDEX(status->curbucket, + status->hash_table->size_log2); + LWLockAcquire(PARTITION_LOCK(status->hash_table, partition), + status->exclusive ? LW_EXCLUSIVE : LW_SHARED); + status->curpartition = partition; + + /* resize doesn't happen from now until seq scan ends */ + status->nbuckets = + NUM_BUCKETS(status->hash_table->control->size_log2); + ensure_valid_bucket_pointers(status->hash_table); + + next_item_pointer = status->hash_table->buckets[status->curbucket]; + } + else + next_item_pointer = status->pnextitem; + + Assert(LWLockHeldByMeInMode(PARTITION_LOCK(status->hash_table, + status->curpartition), + status->exclusive ? LW_EXCLUSIVE : LW_SHARED)); + + /* Move to the next bucket if we finished the current bucket */ + while (!DsaPointerIsValid(next_item_pointer)) + { + int next_partition; + + if (++status->curbucket >= status->nbuckets) + { + /* all buckets have been scanned. finish. */ + return NULL; + } + + /* Check if move to the next partition */ + next_partition = + PARTITION_FOR_BUCKET_INDEX(status->curbucket, + status->hash_table->size_log2); + + if (status->curpartition != next_partition) + { + /* + * Move to the next partition. Lock the next partition then release + * the current, not in the reverse order to avoid concurrent + * resizing. Avoid dead lock by taking lock in the same order + * with resize(). + */ + LWLockAcquire(PARTITION_LOCK(status->hash_table, + next_partition), + status->exclusive ? LW_EXCLUSIVE : LW_SHARED); + LWLockRelease(PARTITION_LOCK(status->hash_table, + status->curpartition)); + status->curpartition = next_partition; + } + + next_item_pointer = status->hash_table->buckets[status->curbucket]; + } + + status->curitem = + dsa_get_address(status->hash_table->area, next_item_pointer); + status->hash_table->find_locked = true; + status->hash_table->find_exclusively_locked = status->exclusive; + + /* + * The caller may delete the item. Store the next item in case of deletion. + */ + status->pnextitem = status->curitem->next; + + return ENTRY_FROM_ITEM(status->curitem); +} + +/* + * Terminates the seqscan and release all locks. + * + * Should be always called when finishing or exiting a seqscan. + */ +void +dshash_seq_term(dshash_seq_status *status) +{ + status->hash_table->find_locked = false; + status->hash_table->find_exclusively_locked = false; + + if (status->curpartition >= 0) + LWLockRelease(PARTITION_LOCK(status->hash_table, status->curpartition)); +} + +/* Remove the current entry while a seq scan. */ +void +dshash_delete_current(dshash_seq_status *status) +{ + dshash_table *hash_table = status->hash_table; + dshash_table_item *item = status->curitem; + size_t partition = PARTITION_FOR_HASH(item->hash); + + Assert(status->exclusive); + Assert(hash_table->control->magic == DSHASH_MAGIC); + Assert(hash_table->find_locked); + Assert(hash_table->find_exclusively_locked); + Assert(LWLockHeldByMeInMode(PARTITION_LOCK(hash_table, partition), + LW_EXCLUSIVE)); + + delete_item(hash_table, item); +} + +/* Get the current entry while a seq scan. */ +void * +dshash_get_current(dshash_seq_status *status) +{ + return ENTRY_FROM_ITEM(status->curitem); +} + /* * Print debugging information about the internal state of the hash table to * stderr. The caller must hold no partition locks. diff --git a/src/include/lib/dshash.h b/src/include/lib/dshash.h index c069ec9de7..a6ea377173 100644 --- a/src/include/lib/dshash.h +++ b/src/include/lib/dshash.h @@ -59,6 +59,21 @@ typedef struct dshash_parameters struct dshash_table_item; typedef struct dshash_table_item dshash_table_item; +/* + * Sequential scan state. The detail is exposed to let users know the storage + * size but it should be considered as an opaque type by callers. + */ +typedef struct dshash_seq_status +{ + dshash_table *hash_table; /* dshash table working on */ + int curbucket; /* bucket number we are at */ + int nbuckets; /* total number of buckets in the dshash */ + dshash_table_item *curitem; /* item we are currently at */ + dsa_pointer pnextitem; /* dsa-pointer to the next item */ + int curpartition; /* partition number we are at */ + bool exclusive; /* locking mode */ +} dshash_seq_status; + /* Creating, sharing and destroying from hash tables. */ extern dshash_table *dshash_create(dsa_area *area, const dshash_parameters *params, @@ -80,6 +95,13 @@ extern bool dshash_delete_key(dshash_table *hash_table, const void *key); extern void dshash_delete_entry(dshash_table *hash_table, void *entry); extern void dshash_release_lock(dshash_table *hash_table, void *entry); +/* seq scan support */ +extern void dshash_seq_init(dshash_seq_status *status, dshash_table *hash_table, + bool exclusive); +extern void *dshash_seq_next(dshash_seq_status *status); +extern void dshash_seq_term(dshash_seq_status *status); +extern void dshash_delete_current(dshash_seq_status *status); +extern void *dshash_get_current(dshash_seq_status *status); /* Convenience hash and compare functions wrapping memcmp and tag_hash. */ extern int dshash_memcmp(const void *a, const void *b, size_t size, void *arg); extern dshash_hash dshash_memhash(const void *v, size_t size, void *arg); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index e4d2debb3c..21fd98af66 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2954,6 +2954,7 @@ dshash_hash dshash_hash_function dshash_parameters dshash_partition +dshash_seq_status dshash_table dshash_table_control dshash_table_handle -- 2.27.0 ----Next_Part(Wed_Mar_10_17_51_37_2021_192)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v52-0002-Add-conditional-lock-feature-to-dshash.patch" ^ permalink raw reply [nested|flat] 2+ messages in thread
* Re: Refactor calculations to use instr_time @ 2023-02-17 10:53 Nazir Bilal Yavuz <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Nazir Bilal Yavuz @ 2023-02-17 10:53 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: pgsql-hackers Hi, On 2/16/23 19:13, Andres Freund wrote: >> +#define WALSTAT_ACC(fld, var_to_add) \ >> + (stats_shmem->stats.fld += var_to_add.fld) >> +#define WALLSTAT_ACC_INSTR_TIME_TYPE(fld) \ >> + (stats_shmem->stats.fld += INSTR_TIME_GET_MICROSEC(PendingWalStats.fld)) >> + WALSTAT_ACC(wal_records, diff); >> + WALSTAT_ACC(wal_fpi, diff); >> + WALSTAT_ACC(wal_bytes, diff); >> + WALSTAT_ACC(wal_buffers_full, PendingWalStats); >> + WALSTAT_ACC(wal_write, PendingWalStats); >> + WALSTAT_ACC(wal_sync, PendingWalStats); >> + WALLSTAT_ACC_INSTR_TIME_TYPE(wal_write_time); >> + WALLSTAT_ACC_INSTR_TIME_TYPE(wal_sync_time); >> #undef WALSTAT_ACC >> - >> LWLockRelease(&stats_shmem->lock); > WALSTAT_ACC is undefined, but WALLSTAT_ACC_INSTR_TIME_TYPE isn't. > > I'd not remove the newline before LWLockRelease(). > > >> /* >> diff --git a/src/include/pgstat.h b/src/include/pgstat.h >> index db9675884f3..295c5eabf38 100644 >> --- a/src/include/pgstat.h >> +++ b/src/include/pgstat.h >> @@ -445,6 +445,21 @@ typedef struct PgStat_WalStats >> TimestampTz stat_reset_timestamp; >> } PgStat_WalStats; >> >> +/* Created for accumulating wal_write_time and wal_sync_time as a >> instr_time > Minor code-formatting point: In postgres we don't put code in the same line as > a multi-line comment starting with the /*. So either > > /* single line comment */ > or > /* > * multi line > * comment > */ Thanks for the review. I updated the patch. Regards, Nazir Bilal Yavuz Microsoft Attachments: [text/x-diff] v2-0001-Refactor-instr_time-calculations.patch (5.2K, ../../[email protected]/3-v2-0001-Refactor-instr_time-calculations.patch) download | inline diff: From e3723aca8b79b07190834b0cfd1440dbbf706862 Mon Sep 17 00:00:00 2001 From: Nazir Bilal Yavuz <[email protected]> Date: Thu, 2 Feb 2023 15:06:48 +0300 Subject: [PATCH v2] Refactor instr_time calculations --- src/backend/access/transam/xlog.c | 6 ++---- src/backend/storage/file/buffile.c | 6 ++---- src/backend/utils/activity/pgstat_wal.c | 27 +++++++++++++------------ src/include/pgstat.h | 18 ++++++++++++++++- 4 files changed, 35 insertions(+), 22 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index f9f0f6db8d1..3c35dc1ca23 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -2206,8 +2206,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible) instr_time duration; INSTR_TIME_SET_CURRENT(duration); - INSTR_TIME_SUBTRACT(duration, start); - PendingWalStats.wal_write_time += INSTR_TIME_GET_MICROSEC(duration); + INSTR_TIME_ACCUM_DIFF(PendingWalStats.wal_write_time, duration, start); } PendingWalStats.wal_write++; @@ -8201,8 +8200,7 @@ issue_xlog_fsync(int fd, XLogSegNo segno, TimeLineID tli) instr_time duration; INSTR_TIME_SET_CURRENT(duration); - INSTR_TIME_SUBTRACT(duration, start); - PendingWalStats.wal_sync_time += INSTR_TIME_GET_MICROSEC(duration); + INSTR_TIME_ACCUM_DIFF(PendingWalStats.wal_sync_time, duration, start); } PendingWalStats.wal_sync++; diff --git a/src/backend/storage/file/buffile.c b/src/backend/storage/file/buffile.c index 0a51624df3b..e55f86b675e 100644 --- a/src/backend/storage/file/buffile.c +++ b/src/backend/storage/file/buffile.c @@ -469,8 +469,7 @@ BufFileLoadBuffer(BufFile *file) if (track_io_timing) { INSTR_TIME_SET_CURRENT(io_time); - INSTR_TIME_SUBTRACT(io_time, io_start); - INSTR_TIME_ADD(pgBufferUsage.temp_blk_read_time, io_time); + INSTR_TIME_ACCUM_DIFF(pgBufferUsage.temp_blk_read_time, io_time, io_start); } /* we choose not to advance curOffset here */ @@ -544,8 +543,7 @@ BufFileDumpBuffer(BufFile *file) if (track_io_timing) { INSTR_TIME_SET_CURRENT(io_time); - INSTR_TIME_SUBTRACT(io_time, io_start); - INSTR_TIME_ADD(pgBufferUsage.temp_blk_write_time, io_time); + INSTR_TIME_ACCUM_DIFF(pgBufferUsage.temp_blk_write_time, io_time, io_start); } file->curOffset += bytestowrite; diff --git a/src/backend/utils/activity/pgstat_wal.c b/src/backend/utils/activity/pgstat_wal.c index e8598b2f4e0..ab2869a0e24 100644 --- a/src/backend/utils/activity/pgstat_wal.c +++ b/src/backend/utils/activity/pgstat_wal.c @@ -21,7 +21,7 @@ #include "executor/instrument.h" -PgStat_WalStats PendingWalStats = {0}; +PgStat_PendingWalUsage PendingWalStats = {0}; /* * WAL usage counters saved from pgWALUsage at the previous call to @@ -89,24 +89,25 @@ pgstat_flush_wal(bool nowait) * previous counters from the current ones. */ WalUsageAccumDiff(&diff, &pgWalUsage, &prevWalUsage); - PendingWalStats.wal_records = diff.wal_records; - PendingWalStats.wal_fpi = diff.wal_fpi; - PendingWalStats.wal_bytes = diff.wal_bytes; if (!nowait) LWLockAcquire(&stats_shmem->lock, LW_EXCLUSIVE); else if (!LWLockConditionalAcquire(&stats_shmem->lock, LW_EXCLUSIVE)) return true; -#define WALSTAT_ACC(fld) stats_shmem->stats.fld += PendingWalStats.fld - WALSTAT_ACC(wal_records); - WALSTAT_ACC(wal_fpi); - WALSTAT_ACC(wal_bytes); - WALSTAT_ACC(wal_buffers_full); - WALSTAT_ACC(wal_write); - WALSTAT_ACC(wal_sync); - WALSTAT_ACC(wal_write_time); - WALSTAT_ACC(wal_sync_time); +#define WALSTAT_ACC(fld, var_to_add) \ + (stats_shmem->stats.fld += var_to_add.fld) +#define WALLSTAT_ACC_INSTR_TIME_TYPE(fld) \ + (stats_shmem->stats.fld += INSTR_TIME_GET_MICROSEC(PendingWalStats.fld)) + WALSTAT_ACC(wal_records, diff); + WALSTAT_ACC(wal_fpi, diff); + WALSTAT_ACC(wal_bytes, diff); + WALSTAT_ACC(wal_buffers_full, PendingWalStats); + WALSTAT_ACC(wal_write, PendingWalStats); + WALSTAT_ACC(wal_sync, PendingWalStats); + WALLSTAT_ACC_INSTR_TIME_TYPE(wal_write_time); + WALLSTAT_ACC_INSTR_TIME_TYPE(wal_sync_time); +#undef WALLSTAT_ACC_INSTR_TIME_TYPE #undef WALSTAT_ACC LWLockRelease(&stats_shmem->lock); diff --git a/src/include/pgstat.h b/src/include/pgstat.h index db9675884f3..cc103df7d2e 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -445,6 +445,22 @@ typedef struct PgStat_WalStats TimestampTz stat_reset_timestamp; } PgStat_WalStats; +/* + * Created for accumulating wal_write_time and wal_sync_time as a instr_time + * but instr_time can't be used as a type where it ends up on-disk + * because its units may change. PgStat_WalStats type is used for + * in-memory/on-disk data. So, PgStat_PendingWalUsage is created for + * accumulating intervals as a instr_time. + */ +typedef struct PgStat_PendingWalUsage +{ + PgStat_Counter wal_buffers_full; + PgStat_Counter wal_write; + PgStat_Counter wal_sync; + instr_time wal_write_time; + instr_time wal_sync_time; +} PgStat_PendingWalUsage; + /* * Functions in pgstat.c @@ -758,7 +774,7 @@ extern PGDLLIMPORT SessionEndType pgStatSessionEndCause; */ /* updated directly by backends and background processes */ -extern PGDLLIMPORT PgStat_WalStats PendingWalStats; +extern PGDLLIMPORT PgStat_PendingWalUsage PendingWalStats; #endif /* PGSTAT_H */ -- 2.39.1 ^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2023-02-17 10:53 UTC | newest] Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-03-13 07:58 [PATCH v52 1/7] sequential scan for dshash Kyotaro Horiguchi <[email protected]> 2023-02-17 10:53 Re: Refactor calculations to use instr_time Nazir Bilal Yavuz <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox