Re: Proposal : For Auto-Prewarm.

On 2017-03-13 18:45:00 +0530, Mithun Cy wrote:
> I have implemented a similar logic now. The prewarm bgworker will
> launch a sub-worker per database in the dump file. And, each
> sub-worker will load its database block info. The sub-workers will be
> launched only after previous one is finished. All of this will only
> start if the database has reached a consistent state.

Hm. For replay performance it'd possibly be good to start earlier,
before reaching consistency. Is there an issue starting earlier?

> diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
> new file mode 100644
> index 0000000..f4b34ca
> --- /dev/null
> +++ b/contrib/pg_prewarm/autoprewarm.c
> @@ -0,0 +1,1137 @@
> +/*-------------------------------------------------------------------------
> + *
> + * autoprewarm.c
> + *
> + * -- Automatically prewarm the shared buffer pool when server restarts.

Don't think we ususally use -- here.

> + * Copyright (c) 2013-2017, PostgreSQL Global Development Group

Hm, that's a bit of a weird date range.

> + * IDENTIFICATION
> + * contrib/pg_prewarm.c/autoprewarm.c
> + *-------------------------------------------------------------------------
> + */

The pg_prewarm.c in there looks like some search & replace gone awry.

> +#include "postgres.h"
> +#include <unistd.h>
> +
> +/* These are always necessary for a bgworker. */
> +#include "miscadmin.h"
> +#include "postmaster/bgworker.h"
> +#include "storage/ipc.h"
> +#include "storage/latch.h"
> +#include "storage/lwlock.h"
> +#include "storage/proc.h"
> +#include "storage/shmem.h"
> +
> +/* These are necessary for prewarm utilities. */
> +#include "pgstat.h"
> +#include "storage/buf_internals.h"
> +#include "storage/smgr.h"
> +#include "utils/memutils.h"
> +#include "utils/resowner.h"
> +#include "utils/guc.h"
> +#include "catalog/pg_class.h"
> +#include "catalog/pg_type.h"
> +#include "executor/spi.h"
> +#include "access/xact.h"
> +#include "utils/rel.h"
> +#include "port/atomics.h"

I'd rather just sort these alphabetically.

I think this should rather be in the initial header.

> +/*
> + * autoprewarm :
> + *
> + * What is it?
> + * ===========
> + * A bgworker which automatically records information about blocks which were
> + * present in buffer pool before server shutdown and then prewarm the buffer
> + * pool upon server restart with those blocks.
> + *
> + * How does it work?
> + * =================
> + * When the shared library "pg_prewarm" is preloaded, a
> + * bgworker "autoprewarm" is launched immediately after the server has reached
> + * consistent state. The bgworker will start loading blocks recorded in the
> + * format BlockInfoRecord
> + * <<DatabaseId,TableSpaceId,RelationId,Forknum,BlockNum>> in
> + * $PGDATA/AUTOPREWARM_FILE, until there is a free buffer left in the buffer
> + * pool. This way we do not replace any new blocks which were loaded either by
> + * the recovery process or the querying clients.

s/until there is a/until there is no/?

> +/*
> + * ============================================================================
> + * =========================== SIGNAL HANDLERS ===========================
> + * ============================================================================
> + */

Hm...

> +static void sigtermHandler(SIGNAL_ARGS);
> +static void sighupHandler(SIGNAL_ARGS);

I don't think that's a casing we commonly use. We mostly use CamelCase
or underscore_case.

> +/*
> + * Signal handler for SIGUSR1.
> + */
> +static void
> +sigusr1Handler(SIGNAL_ARGS)
> +{
> + int save_errno = errno;
> +
> + if (MyProc)
> + SetLatch(&MyProc->procLatch);
> +
> + errno = save_errno;
> +}

Hm, what's this one for?

> +/*
> + * Shared state information about the running autoprewarm bgworker.
> + */
> +typedef struct AutoPrewarmSharedState
> +{
> + pg_atomic_uint32 current_task; /* current tasks performed by
> + * autoprewarm workers. */
> +} AutoPrewarmSharedState;

Hm. Why do we need atomics here? I thought there's no concurrency?

> +/*
> + * sort_cmp_func - compare function used for qsort().
> + */
> +static int
> +sort_cmp_func(const void *p, const void *q)
> +{

rename to blockinfo_cmp?

> +static AutoPrewarmTask
> +get_autoprewarm_task(AutoPrewarmTask todo_task)
> +{
> + bool found;
> +
> + state = NULL;
> +
> + LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
> + state = ShmemInitStruct("autoprewarm",
> + sizeof(AutoPrewarmSharedState),
> + &found);
> + if (!found)
> + pg_atomic_write_u32(&(state->current_task), todo_task);

Superflous parens (repeated a lot).

> + LWLockRelease(AddinShmemInitLock);
> +
> + /* If found check if we can go ahead. */
> + if (found)
> + {
> + if (pg_atomic_read_u32(&(state->current_task)) ==
> + TASK_PREWARM_BUFFERPOOL)

You repeat the read in every branch - why don't you store it in a
variable instead?

That aside, the use of an atomic doesn't seem to actually gain us
anything here. If we need control over concurrency it seems a lot
better to instead use a lwlock or spinlock. There's no contention here,
using lock-free stuff just increases complexity without a corresponding
benefit.

> + {
> + if (todo_task == TASK_PREWARM_BUFFERPOOL)
> + {
> + /*
> + * we were prewarming and we are back to do same, time to
> + * abort prewarming and move to dumping.
> + */

I'm not sure what "back to do same" should mean here - changing to a
different type of task surely is not the same.

> + pg_atomic_write_u32(&(state->current_task),
> + TASK_DUMP_BUFFERPOOL_INFO);
> + return TASK_DUMP_BUFFERPOOL_INFO;
> + }
> + else
> + return TASK_END; /* rest all cannot proceed further. */

What does that comment mean?

> + }
> + else if (pg_atomic_read_u32(&(state->current_task)) ==
> + TASK_DUMP_IMMEDIATE_ONCE)
> + {
> + uint32 current_state = TASK_DUMP_IMMEDIATE_ONCE;
> +
> + /* We cannot do a TASK_PREWARM_BUFFERPOOL but rest can go ahead */
> + if (todo_task == TASK_DUMP_IMMEDIATE_ONCE)
> + return TASK_DUMP_IMMEDIATE_ONCE;
> +
> + if (todo_task == TASK_PREWARM_BUFFERPOOL)
> + todo_task = TASK_DUMP_BUFFERPOOL_INFO; /* skip to do dump only */
> +
> + /*
> + * first guy who can atomically set the current_task get the
> + * opportunity to proceed further
> + */
> + if (pg_atomic_compare_exchange_u32(&(state->current_task),
> + &current_state,
> + TASK_DUMP_BUFFERPOOL_INFO))
> + {
> + /* Wow! We won the race proceed with the task. */
> + return TASK_DUMP_BUFFERPOOL_INFO;
> + }
> + else
> + return TASK_END;

Note that it's not generally guaranteed that any
pg_atomic_compare_exchange_u32 actually wins, it could temporarily fail
for all.

> +/*
> + * getnextblockinfo -- given a BlkType get its next BlockInfoRecord from the
> + * dump file.
> + */
> +static BlkType
> +getnextblockinfo(FILE *file, BlockInfoRecord *currblkinfo, BlkType reqblock,
> + BlockInfoRecord *newblkinfo)
> +{
> + BlkType nextblk;
> +
> + while (true)
> + {
> + /* get next block. */
> + if (5 != fscanf(file, "%u,%u,%u,%u,%u\n", &(newblkinfo->database),
> + &(newblkinfo->spcNode), &(newblkinfo->filenode),
> + (uint32 *) &(newblkinfo->forknum),
> + &(newblkinfo->blocknum)))
> + return BLKTYPE_END; /* No more valid entry hence stop processing. */

Hm. Is it actually helpful to store the file as text? That's commonly
going to increase the size of the file quite considerably, no?

> +/*
> + * GetRelOid -- given a filenode get its relation oid.
> + */
> +static Oid
> +get_reloid(Oid filenode)
> +{

Function and comment don't agree on naming.

But what is this actually used for? I thought Robert, in
http://archives.postgresql.org/message-id/CA%2BTgmoa%3DUqCL2mR%2B9WTq05tB3Up-z4Sv2wkzkDxDwBP7Mj_2_w%40mail.gmail.com
suggested storing the filenode in the dump, and then to use
RelidByRelfilenode to get the corresponding relation?

It seems a lot better to use relfilenodes, because otherwise table
rewrites will lead to reloading wrong things.

> + int ret;
> + Oid relationid;
> + bool isnull;
> + Datum value[1] = {ObjectIdGetDatum(filenode)};
> + StringInfoData buf;
> + Oid ptype[1] = {OIDOID};
> +
> + initStringInfo(&buf);
> + appendStringInfo(&buf,
> + "select oid from pg_class where pg_relation_filenode(oid) = $1");
> +
> + ret = SPI_execute_with_args(buf.data, 1, (Oid *) &ptype, (Datum *) &value,
> + NULL, true, 1);
> +
> + if (ret != SPI_OK_SELECT)
> + ereport(FATAL, (errmsg("SPI_execute failed: error code %d", ret)));
> +
> + if (SPI_processed < 1)
> + return InvalidOid;
> +
> + relationid = DatumGetObjectId(SPI_getbinval(SPI_tuptable->vals[0],
> + SPI_tuptable->tupdesc,
> + 1, &isnull));
> + if (isnull)
> + return InvalidOid;
> +
> + return relationid;
> +}

Doing this via SPI doesn't strike me as a good idea - that's really
quite expensive. Why not call the underlying function directly?

> +/*
> + * load_one_database -- start of prewarm sub-worker, this will try to load
> + * blocks of one database starting from block info position passed by main
> + * prewarm worker.
> + */
> +void
> +load_one_database(Datum main_arg)
> +{

> + /* check if file exists and open file in read mode. */
> + snprintf(dump_file_path, sizeof(dump_file_path), "%s", AUTOPREWARM_FILE);
> + file = fopen(dump_file_path, PG_BINARY_R);
> + if (!file)
> + return; /* No file to load. */

Shouldn't this be an error case? In which case is it ok for the file to
be gone after we launched the worker?

> + /*
> + * It should be a block info belonging to a new database. Or else dump
> + * file is corrupted better to end the loading of bocks now.
> + */
> + if (loadblocktype != BLKTYPE_NEW_DATABASE)
> + goto end_load; /* should we raise a voice here? */

Yes, this should raise an error.

> + case BLKTYPE_NEW_RELATION:
> +
> + /*
> + * release lock on previous relation.
> + */
> + if (rel)
> + {
> + relation_close(rel, AccessShareLock);
> + rel = NULL;
> + }
> +
> + loadblocktype = BLKTYPE_NEW_RELATION;
> +
> + /*
> + * lock new relation.
> + */
> + reloid = get_reloid(toload_block.filenode);
> +
> + if (!OidIsValid(reloid))
> + break;
> +
> + rel = try_relation_open(reloid, AccessShareLock);
> + if (!rel)
> + break;
> + RelationOpenSmgr(rel);

Now I'm confused. Your get_reloid used pg_relation_filenode() to map
from relation oid to filenode - and then you're using it to lock the
relation? Something's wrong.

> + case BLKTYPE_NEW_FORK:
> +
> + /*
> + * check if fork exists and if block is within the range
> + */
> + loadblocktype = BLKTYPE_NEW_FORK;
> + if ( /* toload_block.forknum > InvalidForkNumber &&
> + * toload_block.forknum <= MAX_FORKNUM && */
> + !smgrexists(rel->rd_smgr, toload_block.forknum))
> + break;

Huh? What's with that commented out section of code?

> + case BLKTYPE_NEW_BLOCK:
> +
> + /* check if blocknum is valid and with in fork file size. */
> + if (toload_block.blocknum >= nblocks)
> + {
> + /* move to next forknum. */
> + loadblocktype = BLKTYPE_NEW_FORK;
> + break;
> + }

Hm. Why does the size of the underlying file allow us to skip to the
next fork? Don't we have to read all the pending dump records?

> + buf = ReadBufferExtended(rel, toload_block.forknum,
> + toload_block.blocknum, RBM_NORMAL,
> + NULL);
> + if (BufferIsValid(buf))
> + {
> + ReleaseBuffer(buf);
> + }
> +
> + loadblocktype = BLKTYPE_NEW_BLOCK;
> + break;

Hm. RBM_NORMAL will error out in a bunch of cases, is that ok?

> + if (have_dbconnection)
> + {
> + SPI_finish();
> + PopActiveSnapshot();
> + CommitTransactionCommand();
> + }
> + return;
> +}

Are we really ok keeping open a transaction through all of this? That
could potentially be quite long, no? How about doing that on a per-file
basis, or even moving to session locks alltogether?

> +/* This sub-module is for periodically dumping buffer pool's block info into
> + * a dump file AUTOPREWARM_FILE.
> + * Each entry of block info looks like this:
> + * <DatabaseId,TableSpaceId,RelationId,Forknum,BlockNum> and we shall call it
> + * as BlockInfoRecord.
> + *
> + * Contents of AUTOPREWARM_FILE has been formated such a way that
> + * blockInfoRecord of each database can be given to different prewarm workers.
> + *
> + * format of AUTOPREWAM_FILE
> + * =======================================
> + * [offset position of database map table]
> + * [sorted BlockInfoRecords..............]
> + * [database map table]
> + * =======================================

This doesn't mention storing things as ascii, instead of binary...

> + * The [database map table] is sequence of offset in file which will point to
> + * first BlockInfoRecords of each database in the dump. The prewarm worker
> + * will read this offset one by one in sequence and ask its subworker to seek
> + * to this position and then start loading the BlockInfoRecords one by one
> + * until it see a BlockInfoRecords of a different database than it is actually
> + * connected to.
> + * NOTE : We store off_t inside file so the dump file will not be portable to
> + * be used across systems where sizeof off_t is different from each other.
> + */

Why are we using off_t? Shouldn't this just be BlockNumber?

> +static uint32
> +dump_now(void)
> +{
> + static char dump_file_path[MAXPGPATH],

> +
> + for (num_blocks = 0, i = 0; i < NBuffers; i++)
> + {
> + uint32 buf_state;
> +
> + bufHdr = GetBufferDescriptor(i);
> +
> + /* lock each buffer header before inspecting. */
> + buf_state = LockBufHdr(bufHdr);
> +
> + if (buf_state & BM_TAG_VALID)
> + {
> + block_info_array[num_blocks].database = bufHdr->tag.rnode.dbNode;
> + block_info_array[num_blocks].spcNode = bufHdr->tag.rnode.spcNode;
> + block_info_array[num_blocks].filenode = bufHdr->tag.rnode.relNode;
> + block_info_array[num_blocks].forknum = bufHdr->tag.forkNum;
> + block_info_array[num_blocks].blocknum = bufHdr->tag.blockNum;
> + ++num_blocks;
> + }
> +
> + UnlockBufHdr(bufHdr, buf_state);

> + }
> +
> + /* sorting now only to avoid sorting while loading. */

"sorting while loading"? You mean random accesses?

> + pg_qsort(block_info_array, num_blocks, sizeof(BlockInfoRecord),
> + sort_cmp_func);

> + snprintf(transient_dump_file_path, sizeof(dump_file_path),
> + "%s.%d", AUTOPREWARM_FILE, MyProcPid);
> + file = fopen(transient_dump_file_path, "w");
> + if (file == NULL)
> + ereport(ERROR,
> + (errcode_for_file_access(),
> + errmsg("autoprewarm: could not open \"%s\": %m",
> + dump_file_path)));

What if that file already exists? You're not truncating it. Also, what
if we error out in the middle of this? We'll leak an fd. I think this
needs to use OpenTransientFile etc.

> + snprintf(dump_file_path, sizeof(dump_file_path),
> + "%s", AUTOPREWARM_FILE);
> + ret = fprintf(file, "%020jd\n", (intmax_t) 0);
> + if (ret < 0)
> + {
> + fclose(file);
> + ereport(ERROR,
> + (errcode_for_file_access(),
> + errmsg("autoprewarm: error writing to \"%s\" : %m",
> + dump_file_path)));
> + }
> +
> + database_map_table[num_db++] = ftello(file);
> +
> + for (i = 0; i < num_blocks; i++)
> + {
> + if (i > 0 && block_info_array[i].database != prev_database)
> + {
> + if (num_db == database_map_table_size)
> + {
> + database_map_table_size *= 2; /* double and repalloc. */
> + database_map_table =
> + (off_t *) repalloc(database_map_table,
> + sizeof(off_t) * database_map_table_size);
> + }
> + fflush(file);
> + database_map_table[num_db++] = ftello(file);
> + }
> +
> + ret = fprintf(file, "%u,%u,%u,%u,%u\n",
> + block_info_array[i].database,
> + block_info_array[i].spcNode,
> + block_info_array[i].filenode,
> + (uint32) block_info_array[i].forknum,
> + block_info_array[i].blocknum);
> + if (ret < 0)
> + {
> + fclose(file);
> + ereport(ERROR,
> + (errcode_for_file_access(),
> + errmsg("autoprewarm: error writing to \"%s\" : %m",
> + dump_file_path)));
> + }
> +
> + prev_database = block_info_array[i].database;
> + }

I think we should check for interrupts somewhere in that (and the
preceding) loop.

> +/*
> + * dump_block_info_periodically - at regular intervals, which is defined by GUC
> + * dump_interval, dump the info of blocks which are present in buffer pool.
> + */
> +void
> +dump_block_info_periodically()
> +{

Suggest adding void to the parameter list.

> + pg_time_t last_dump_time = (pg_time_t) time(NULL);
> +
> + while (!got_sigterm)
> + {
> + int rc;
> + pg_time_t now;
> + int elapsed_secs = 0,
> + timeout = AT_PWARM_DEFAULT_DUMP_INTERVAL;
> +
> + if (dump_interval > AT_PWARM_DUMP_AT_SHUTDOWN_ONLY)
> + {
> + now = (pg_time_t) time(NULL);
> + elapsed_secs = now - last_dump_time;
> +
> + if (elapsed_secs > dump_interval)
> + {
> + dump_now();
> + if (got_sigterm)
> + return; /* got shutdown signal just after a dump. And,
> + * I think better to return now. */
> + last_dump_time = (pg_time_t) time(NULL);
> + elapsed_secs = 0;
> + }
> +
> + timeout = dump_interval - elapsed_secs;
> + }

I suggest using GetCurrenttimstamp() and TimestampDifferenceExceeds()
instead.

> + /* Has been set not to dump. Nothing more to do. */
> + if (dump_interval == AT_PWARM_OFF)
> + return;
> +
> + ResetLatch(&MyProc->procLatch);
> + rc = WaitLatch(&MyProc->procLatch,
> + WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
> + timeout * 1000, PG_WAIT_EXTENSION);
> +
> + if (rc & WL_POSTMASTER_DEATH)
> + proc_exit(1);
> +
> + /*
> + * In case of a SIGHUP, just reload the configuration.
> + */
> + if (got_sighup)
> + {
> + got_sighup = false;
> + ProcessConfigFile(PGC_SIGHUP);
> + }
> + }
> +
> + /* One last block meta info dump while postmaster shutdown. */
> + if (dump_interval != AT_PWARM_OFF)
> + dump_now();

Uh, afaics we'll also do this if somebody SIGTERMed the process
interactively?

> +/* Extension's entry point. */
> +void
> +_PG_init(void)
> +{
> + BackgroundWorker autoprewarm;
> +
> + /* Define custom GUC variables. */
> + DefineCustomIntVariable("pg_prewarm.dump_interval",
> + "Sets the maximum time between two buffer pool dumps",
> + "If set to Zero, timer based dumping is disabled."
> + " If set to -1, stops the running autoprewarm.",
> + &dump_interval,
> + AT_PWARM_DEFAULT_DUMP_INTERVAL,
> + AT_PWARM_OFF, INT_MAX / 1000,
> + PGC_SIGHUP,
> + GUC_UNIT_S,
> + NULL,
> + NULL,
> + NULL);
> +
> + /* if not run as a preloaded library, nothing more to do here! */
> + if (!process_shared_preload_libraries_in_progress)
> + return;
> +
> + DefineCustomStringVariable("pg_prewarm.default_database",
> + "default database to connect if dump has not recorded same.",
> + NULL,
> + &default_database,
> + "postgres",
> + PGC_POSTMASTER,
> + 0,
> + NULL,
> + NULL,
> + NULL);

I don't think it's a good idea to make guc registration depending on
process_shared_preload_libraries_in_progress.

You should also use EmitWarningsOnPlaceholders() somewhere here.

I also wonder whether we don't need to use prefetch to actually make
this fast enough.

I think it's pretty clear that this needs a bit more work and thus won't
be ready for v10. Moved to the next CF.

- Andres