step2-job_cache.diff

*** src/backend/storage/ipc/shmqueue.c	2fbe744d1fc4b2a45f1eaa1704276a23edd3f542
--- src/backend/storage/ipc/shmqueue.c	de5c363e57d6086b863eb3b580f235e4dec92fd9
*************** SHMQueueInit(SHM_QUEUE *queue)
*** 44,57 ****
   * SHMQueueIsDetached -- TRUE if element is not currently
   *		in a queue.
   */
- #ifdef NOT_USED
  bool
  SHMQueueIsDetached(SHM_QUEUE *queue)
  {
  	Assert(ShmemAddrIsValid(queue));
  	return (queue->prev == NULL);
  }
- #endif
  
  /*
   * SHMQueueElemInit -- clear an element's links
--- 44,55 ----
============================================================
*** src/backend/tcop/postgres.c	62792c84c23e8b39d78a49f4113cdea2828c0cb7
--- src/backend/tcop/postgres.c	778b93a7067312d73d4c68561d8fa23476c0a292
*************** ProcessInterrupts(void)
*** 2936,2947 ****
  		}
  		if (IsAutoVacuumWorkerProcess())
  		{
  			ImmediateInterruptOK = false;		/* not idle anymore */
  			DisableNotifyInterrupt();
  			DisableCatchupInterrupt();
  			ereport(ERROR,
  					(errcode(ERRCODE_QUERY_CANCELED),
! 					 errmsg("canceling autovacuum task")));
  		}
  		if (RecoveryConflictPending)
  		{
--- 2936,2948 ----
  		}
  		if (IsAutoVacuumWorkerProcess())
  		{
+ 			/* FIXME: make sure this works fine for all background jobs */
  			ImmediateInterruptOK = false;		/* not idle anymore */
  			DisableNotifyInterrupt();
  			DisableCatchupInterrupt();
  			ereport(ERROR,
  					(errcode(ERRCODE_QUERY_CANCELED),
! 					 errmsg("bg worker [%d/%d]: canceling autovacuum task", MyProcPid, MyBackendId)));
  		}
  		if (RecoveryConflictPending)
  		{
============================================================
*** src/backend/utils/init/globals.c	0c8c71495665f7b1077f5c3f3e5a02c98c92be3a
--- src/backend/utils/init/globals.c	626dcd5939ff8882d155015f0aecae78e2945728
*************** volatile uint32 CritSectionCount = 0;
*** 31,36 ****
--- 31,37 ----
  volatile bool ImmediateInterruptOK = false;
  volatile uint32 InterruptHoldoffCount = 0;
  volatile uint32 CritSectionCount = 0;
+ volatile bool BgWorkerCleanupInProgress = false;
  
  int			MyProcPid;
  pg_time_t	MyStartTime;
============================================================
*** src/include/storage/shmem.h	068b4aff617ae86bce155915e1a8137cb16337c0
--- src/include/storage/shmem.h	8a0aa9faddf9bdc9bd1cdf7032ca89f9ce52bcc2
*************** extern void SHMQueueInit(SHM_QUEUE *queu
*** 119,124 ****
--- 119,125 ----
   * prototypes for functions in shmqueue.c
   */
  extern void SHMQueueInit(SHM_QUEUE *queue);
+ extern bool SHMQueueIsDetached(SHM_QUEUE *queue);
  extern void SHMQueueElemInit(SHM_QUEUE *queue);
  extern void SHMQueueDelete(SHM_QUEUE *queue);
  extern void SHMQueueInsertBefore(SHM_QUEUE *queue, SHM_QUEUE *elem);
============================================================
*** src/include/miscadmin.h	726d181c386b7dbd5202e3b19819876f7dadc0f7
--- src/include/miscadmin.h	a1e372be7cde09b2aeecb5d6ba5c59003fd0e831
*************** extern PGDLLIMPORT volatile uint32 CritS
*** 75,80 ****
--- 75,82 ----
  extern PGDLLIMPORT volatile uint32 InterruptHoldoffCount;
  extern PGDLLIMPORT volatile uint32 CritSectionCount;
  
+ extern volatile bool BgWorkerCleanupInProgress;
+ 
  /* in tcop/postgres.c */
  extern void ProcessInterrupts(void);
  
============================================================
*** src/include/storage/lwlock.h	3a8cf5b9c3534985e33cac96819ca385878887d2
--- src/include/storage/lwlock.h	2aea3a2fcddca3a9f9eea62f65d08930d221beb5
*************** typedef enum LWLockId
*** 66,71 ****
--- 66,72 ----
  	AddinShmemInitLock,
  	AutovacuumLock,
  	AutovacuumScheduleLock,
+ 	CoordinatorDatabasesLock,
  	SyncScanLock,
  	RelationMappingLock,
  	AsyncCtlLock,
============================================================
*** src/backend/storage/ipc/procarray.c	62107b3a36554510ce82208eaa98cec1b40f3e14
--- src/backend/storage/ipc/procarray.c	903e6def62dfd4119752e20cb16711b66fdf6996
*************** BackendPidGetProc(int pid)
*** 1652,1657 ****
--- 1652,1692 ----
  }
  
  /*
+  * BackendIdGetProc -- get a backend's PGPROC given its backendId
+  *
+  * Returns NULL if not found.  Note that it is up to the caller to be
+  * sure that the question remains meaningful for long enough for the
+  * answer to be used ...
+  */
+ PGPROC *
+ BackendIdGetProc(BackendId id)
+ {
+ 	PGPROC	   *result = NULL;
+ 	ProcArrayStruct *arrayP = procArray;
+ 	int			index;
+ 
+ 	if (id == InvalidBackendId)
+ 		return NULL;
+ 
+ 	LWLockAcquire(ProcArrayLock, LW_SHARED);
+ 
+ 	for (index = 0; index < arrayP->numProcs; index++)
+ 	{
+ 		PGPROC	   *proc = arrayP->procs[index];
+ 
+ 		if (proc->backendId == id)
+ 		{
+ 			result = proc;
+ 			break;
+ 		}
+ 	}
+ 
+ 	LWLockRelease(ProcArrayLock);
+ 
+ 	return result;
+ }
+ 
+ /*
   * BackendXidGetPid -- get a backend's pid given its XID
   *
   * Returns 0 if not found or it's a prepared transaction.  Note that
============================================================
*** src/include/storage/procarray.h	dd1cc5e6251c6a4d901434e91504b134e9e2b1dc
--- src/include/storage/procarray.h	2b020c22a35cecbbf0da861c0699d3546dfa9bac
*************** extern PGPROC *BackendPidGetProc(int pid
*** 52,57 ****
--- 52,58 ----
  extern bool HaveTransactionsInCommit(TransactionId *xids, int nxids);
  
  extern PGPROC *BackendPidGetProc(int pid);
+ extern PGPROC *BackendIdGetProc(BackendId id);
  extern int	BackendXidGetPid(TransactionId xid);
  extern bool IsBackendPid(int pid);
  
============================================================
*** src/backend/postmaster/autovacuum.c	3b9cbe9ba0d576b32d490f21969b073a7ec55b98
--- src/backend/postmaster/autovacuum.c	50ad2e93982867e91a47e8ca7af5f4be8b975d8f
***************
*** 86,94 ****
--- 86,97 ----
  #include "storage/ipc.h"
  #include "storage/pmsignal.h"
  #include "storage/proc.h"
+ #include "storage/procarray.h"
  #include "storage/procsignal.h"
+ #include "storage/shmem.h"
  #include "storage/sinvaladt.h"
  #include "tcop/tcopprot.h"
+ #include "utils/builtins.h"
  #include "utils/fmgroids.h"
  #include "utils/lsyscache.h"
  #include "utils/memutils.h"
*************** static int	default_freeze_table_age;
*** 139,147 ****
  static int	default_freeze_min_age;
  static int	default_freeze_table_age;
  
! /* Memory context for long-lived data */
! static MemoryContext AutovacMemCxt;
  
  /* struct to keep track of databases in launcher */
  typedef struct avl_dbase
  {
--- 142,155 ----
  static int	default_freeze_min_age;
  static int	default_freeze_table_age;
  
! /* Memory contexts for long-lived data */
! static MemoryContext AutovacLauncherMemCxt;
! static MemoryContext AutovacWorkerMemCxt;
  
+ /* job handling routines */
+ static void bgworker_job_initialize(worker_state new_state);
+ static void bgworker_job_completed(void);
+ 
  /* struct to keep track of databases in launcher */
  typedef struct avl_dbase
  {
*************** typedef struct avw_dbase
*** 153,164 ****
  /* struct to keep track of databases in worker */
  typedef struct avw_dbase
  {
! 	Oid			adw_datid;
  	char	   *adw_name;
  	TransactionId adw_frozenxid;
  	PgStat_StatDBEntry *adw_entry;
  } avw_dbase;
  
  /* struct to keep track of tables to vacuum and/or analyze, in 1st pass */
  typedef struct av_relation
  {
--- 161,177 ----
  /* struct to keep track of databases in worker */
  typedef struct avw_dbase
  {
! 	Oid			adw_datid;		/* hash key -- must be first */
  	char	   *adw_name;
  	TransactionId adw_frozenxid;
  	PgStat_StatDBEntry *adw_entry;
  } avw_dbase;
  
+ typedef struct cached_job {
+ 	Dlelem cj_links;
+ 	IMessage *cj_msg;
+ } cached_job;
+ 
  /* struct to keep track of tables to vacuum and/or analyze, in 1st pass */
  typedef struct av_relation
  {
*************** typedef struct autovac_table
*** 186,222 ****
  } autovac_table;
  
  /*-------------
-  * This struct holds information about a single worker's whereabouts.  We keep
-  * an array of these in shared memory, sized according to
-  * autovacuum_max_workers.
-  *
-  * wi_links			entry into free list or running list
-  * wi_dboid			OID of the database this worker is supposed to work on
-  * wi_tableoid		OID of the table currently being vacuumed
-  * wi_backend_id	id of the running worker backend, NULL if not started
-  * wi_launchtime	Time at which this worker was launched
-  * wi_cost_*		Vacuum cost-based delay parameters current in this worker
-  *
-  * All fields are protected by AutovacuumLock, except for wi_tableoid which is
-  * protected by AutovacuumScheduleLock (which is read-only for everyone except
-  * that worker itself).
-  *-------------
-  */
- typedef struct WorkerInfoData
- {
- 	SHM_QUEUE	wi_links;
- 	Oid			wi_dboid;
- 	Oid			wi_tableoid;
- 	BackendId	wi_backend_id;
- 	TimestampTz wi_launchtime;
- 	int			wi_cost_delay;
- 	int			wi_cost_limit;
- 	int			wi_cost_limit_base;
- } WorkerInfoData;
- 
- typedef struct WorkerInfoData *WorkerInfo;
- 
- /*-------------
   * The main autovacuum shmem struct.  On shared memory we store this main
   * struct and the array of WorkerInfo structs.	This struct keeps:
   *
--- 199,204 ----
*************** static AutoVacuumShmemStruct *AutoVacuum
*** 240,253 ****
  
  static AutoVacuumShmemStruct *AutoVacuumShmem;
  
  /* the database list in the launcher, and the context that contains it */
  static Dllist *DatabaseList = NULL;
  static MemoryContext DatabaseListCxt = NULL;
  
  /* Pointer to my own WorkerInfo, valid on each worker */
  static WorkerInfo MyWorkerInfo = NULL;
  
- 
  #ifdef EXEC_BACKEND
  static pid_t avlauncher_forkexec(void);
  static pid_t avworker_forkexec(void);
--- 222,241 ----
  
  static AutoVacuumShmemStruct *AutoVacuumShmem;
  
+ /*
+  * Table of databases with at least one connected worker, resides in shared
+  * memory, protected by CoordinatorDatabasesLock
+  */
+ static HTAB *co_databases = NULL;
+ 
  /* the database list in the launcher, and the context that contains it */
  static Dllist *DatabaseList = NULL;
  static MemoryContext DatabaseListCxt = NULL;
  
  /* Pointer to my own WorkerInfo, valid on each worker */
  static WorkerInfo MyWorkerInfo = NULL;
+ static WorkerInfo terminatable_worker = NULL;
  
  #ifdef EXEC_BACKEND
  static pid_t avlauncher_forkexec(void);
  static pid_t avworker_forkexec(void);
*************** NON_EXEC_STATIC void AutoVacLauncherMain
*** 256,273 ****
  static void handle_imessage(IMessage *msg);
  NON_EXEC_STATIC void AutoVacLauncherMain(int argc, char *argv[]);
  
! static void CoordinatorMaybeTriggerAutoVacuum(void);
  
! static void do_start_worker(avw_dbase *avdb);
! static Oid	do_start_autovacuum_worker(void);
  static void launcher_determine_sleep(bool can_launch, bool recursing,
  									 struct timeval *nap);
! static void launch_worker(TimestampTz now);
  static List *get_database_list(void);
  static void rebuild_database_list(Oid newdb);
  static int	db_comparator(const void *a, const void *b);
  static void autovac_balance_cost(void);
  
  static void do_autovacuum(void);
  static void FreeWorkerInfo(int code, Datum arg);
  
--- 244,278 ----
  static void handle_imessage(IMessage *msg);
  NON_EXEC_STATIC void AutoVacLauncherMain(int argc, char *argv[]);
  
! static void init_co_database(co_database *codb);
! static co_database *get_co_database(Oid dboid);
! static void populate_co_databases(void);
  
! static bool can_deliver_cached_job(co_database *codb, IMessage *msg,
!                                    BackendId *target);
! static WorkerInfo get_idle_worker(co_database *codb);
! static void cache_job(IMessage *msg, co_database *codb);
! static void forward_job(IMessage *msg, co_database *codb, BackendId backend_id);
! static void dispatch_job(IMessage *msg, co_database *codb);
! static void process_cached_jobs(co_database *codb);
! 
! static bool CoordinatorCanLaunchWorker(TimestampTz current_time);
! static void manage_workers(bool can_launch);
! static void autovacuum_maybe_trigger_job(TimestampTz current_time,
! 										 bool can_launch);
! 
! static void do_start_worker(Oid dboid);
! static Oid autovacuum_select_database(void);
  static void launcher_determine_sleep(bool can_launch, bool recursing,
  									 struct timeval *nap);
! static void autovacuum_update_timing(Oid dbid, TimestampTz now);
  static List *get_database_list(void);
  static void rebuild_database_list(Oid newdb);
  static int	db_comparator(const void *a, const void *b);
  static void autovac_balance_cost(void);
  
+ static void add_as_idle_worker(Oid dbid, bool inc_connected_count);
+ 
  static void do_autovacuum(void);
  static void FreeWorkerInfo(int code, Datum arg);
  
*************** static void autovac_refresh_stats(void);
*** 293,298 ****
--- 298,316 ----
  
  
  
+ char *
+ decode_worker_state(worker_state state)
+ {
+ 	switch (state)
+ 	{
+ 		case WS_IDLE: return "WS_IDLE";
+ 		case WS_AUTOVACUUM: return "WS_AUTOVACUUM";
+ 
+ 		default: return "UNKNOWN STATE";
+ 	}
+ }
+ 
+ 
  /********************************************************************
   *					  AUTOVACUUM LAUNCHER CODE
   ********************************************************************/
*************** StartAutoVacLauncher(void)
*** 369,375 ****
--- 387,586 ----
  	return 0;
  }
  
+ void
+ init_co_database(co_database *codb)
+ {
+ 	Assert(ShmemAddrIsValid(codb));
+ 	SHMQueueInit(&codb->codb_idle_workers);
+ 	codb->codb_num_idle_workers = 0;
+ 
+ 	/*
+ 	 * While only the coordinator may fiddle with this list, as its entries
+ 	 * reside in that process' memory, it's safe to set the counters to 0
+ 	 * and initialize the list headers with NULL values using DLInitList().
+ 	 */
+ 	codb->codb_num_cached_jobs = 0;
+ 	DLInitList(&codb->codb_cached_jobs);
+ 
+ 	codb->codb_num_connected_workers = 0;
+ }
+ 
+ static void
+ cache_job(IMessage *msg, co_database *codb)
+ {
+ 	cached_job *job;
+ 
+ #ifdef COORDINATOR_DEBUG
+ 	elog(DEBUG5, "Coordinator: caching job of type %s for database %d",
+ 		 decode_imessage_type(msg->type), codb->codb_dboid);
+ #endif
+ 
+ 	job = palloc(sizeof(cached_job));
+ 	DLInitElem(&job->cj_links, job);
+ 	job->cj_msg = msg;
+ 	DLAddTail(&codb->codb_cached_jobs, &job->cj_links);
+ 	codb->codb_num_cached_jobs++;
+ }
+ 
  /*
+  * get_idle_worker
+  *
+  * Returns the first idle worker for a given database, removing it from its
+  * list of idle workers. The caller is expected to make sure that there is
+  * at least one idle worker and it must hold the CoordinatorDatabasesLock.
+  */
+ static WorkerInfo
+ get_idle_worker(co_database *codb)
+ {
+ 	WorkerInfo worker;
+ 
+ 	/* remove a worker from the list of idle workers */
+ 	worker = (WorkerInfo) SHMQueueNext(&codb->codb_idle_workers,
+ 									   &codb->codb_idle_workers,
+ 									   offsetof(WorkerInfoData, wi_links));
+ 	Assert(worker);
+ 	SHMQueueDelete(&worker->wi_links);
+ 	Assert(worker->wi_backend_id != InvalidBackendId);
+ 
+ 	/* maintain per-database counter */
+ 	codb->codb_num_idle_workers--;
+ 
+ 	return worker;
+ }
+ 
+ /*
+  * forward_job
+  *
+  * Takes an imessage and forwards it to the first idle backend for the given
+  * database as its next job to process. The caller must hold the
+  * CoordinatorDatabasesLock.
+  */
+ static void
+ forward_job(IMessage *msg, co_database *codb, BackendId backend_id)
+ {
+ 	/* various actions before job delivery depending on the message type */
+ 	switch (msg->type)
+ 	{
+ 		case IMSGT_TERM_WORKER:
+ 			break;
+ 
+ 		case IMSGT_PERFORM_VACUUM:
+ #ifdef COORDINATOR_DEBUG
+ 			elog(DEBUG1, "Coordinator: delivering msg %s of size %d for "
+ 				 "database %d to backend %d",
+ 				 decode_imessage_type(msg->type), msg->size, codb->codb_dboid,
+ 				 backend_id);
+ #endif
+ 			autovacuum_update_timing(codb->codb_dboid, GetCurrentTimestamp());
+ 			break;
+ 
+ 		default:
+ 			/* no-op */
+ 			break;
+ 	}
+ 
+ 	IMessageActivate(msg, backend_id);
+ }
+ 
+ /*
+  * dispatch_job
+  *
+  * Depending on the status of idle workers, this either forwards a job to an
+  * idle worker directly or caches it for later processing. The caller must
+  * hold the CoordinatorDatabasesLock.
+  */
+ void
+ dispatch_job(IMessage *msg, co_database *codb)
+ {
+     WorkerInfo worker;
+ 
+ 	if (codb->codb_num_idle_workers > 0)
+     {
+         worker = get_idle_worker(codb); 
+ 		forward_job(msg, codb, worker->wi_backend_id);
+     }
+ 	else
+ 		cache_job(msg, codb);
+ }
+ 
+ /*
+  * process_cached_jobs
+  *
+  * Dispatches cached jobs to idle background workers, as long as there are
+  * of both. Before delivering a job, an additional check is performed with
+  * can_deliver_cached_job(), which also chooses the background worker to run
+  * the job on.
+  */
+ static void
+ process_cached_jobs(co_database *codb)
+ {
+ 	BackendId	target;
+ 	cached_job *job;
+ 
+ #ifdef COORDINATOR_DEBUG
+ 	elog(DEBUG5, "Coordinator: cached jobs: %d, idle workers: %d",
+ 		 codb->codb_num_cached_jobs, codb->codb_num_idle_workers);
+ #endif
+ 
+ 	job = (cached_job*) DLGetHead(&codb->codb_cached_jobs);
+ 	while ((codb->codb_num_cached_jobs > 0) &&
+ 		   (codb->codb_num_idle_workers > 0) &&
+ 		   (job != NULL))
+ 	{
+ 		target = InvalidBackendId;
+ 		if (can_deliver_cached_job(codb, job->cj_msg, &target))
+ 		{
+ 			/* remove the job from the cache */
+ 			DLRemove(&job->cj_links);
+ 			codb->codb_num_cached_jobs--;
+ 			
+ 			/* forward the job to some idle worker and cleanup */
+ 			if (target == InvalidBackendId)
+ 				target = get_idle_worker(codb)->wi_backend_id;
+ 
+ 			forward_job(job->cj_msg, codb, target);
+ 			pfree(job);
+ 			
+ 			job = (cached_job*) DLGetHead(&codb->codb_cached_jobs);
+ 		}
+ 		else
+ 			job = (cached_job*) DLGetSucc(&job->cj_links);
+ 	}
+ }
+ 
+ /*
+  * populate_co_databases
+  *
+  * Called at startup of the coordinator to scan pg_database. Schedules an
+  * initial VACUUM job on the template database to populate pg_stat.
+  */
+ static void
+ populate_co_databases()
+ {
+ 	List     *dblist;
+ 	ListCell *cell;
+ 	IMessage *msg;
+ 
+ 	dblist = get_database_list();
+ 	LWLockAcquire(CoordinatorDatabasesLock, LW_EXCLUSIVE);
+ 	foreach(cell, dblist)
+ 	{
+ 		avw_dbase  *avdb = lfirst(cell);
+ 		co_database *codb = get_co_database(avdb->adw_datid);
+ 		if (codb->codb_dboid == TemplateDbOid)
+ 		{
+ 			/*
+ 			 * Create a cached job as an imessage to ourselves, but without
+ 			 * activating it. It can get forwarded to a backend later on.
+ 			 */
+ 			msg = IMessageCreate(IMSGT_PERFORM_VACUUM, 0);
+ 			cache_job(msg, codb);
+ 		}
+ 	}
+ 	LWLockRelease(CoordinatorDatabasesLock);
+ }
+ 
+ /*
   * Main loop for the autovacuum launcher process.
   */
  NON_EXEC_STATIC void
*************** AutoVacLauncherMain(int argc, char *argv
*** 377,382 ****
--- 588,594 ----
  {
  	sigjmp_buf	local_sigjmp_buf;
  	IMessage   *msg = NULL;
+ 	bool        can_launch;
  
  	/* we are a postmaster subprocess now */
  	IsUnderPostmaster = true;
*************** AutoVacLauncherMain(int argc, char *argv
*** 450,461 ****
  	 * that we can reset the context during error recovery and thereby avoid
  	 * possible memory leaks.
  	 */
! 	AutovacMemCxt = AllocSetContextCreate(TopMemoryContext,
! 										  "Autovacuum Launcher",
! 										  ALLOCSET_DEFAULT_MINSIZE,
! 										  ALLOCSET_DEFAULT_INITSIZE,
! 										  ALLOCSET_DEFAULT_MAXSIZE);
! 	MemoryContextSwitchTo(AutovacMemCxt);
  
  	/*
  	 * If an exception is encountered, processing resumes here.
--- 662,673 ----
  	 * that we can reset the context during error recovery and thereby avoid
  	 * possible memory leaks.
  	 */
! 	AutovacLauncherMemCxt = AllocSetContextCreate(TopMemoryContext,
! 												  "Autovacuum Launcher",
! 												  ALLOCSET_DEFAULT_MINSIZE,
! 												  ALLOCSET_DEFAULT_INITSIZE,
! 												  ALLOCSET_DEFAULT_MAXSIZE);
! 	MemoryContextSwitchTo(AutovacLauncherMemCxt);
  
  	/*
  	 * If an exception is encountered, processing resumes here.
*************** AutoVacLauncherMain(int argc, char *argv
*** 479,495 ****
  		EmitErrorReport();
  
  		/* Abort the current transaction in order to recover */
! 		AbortCurrentTransaction();
  
  		/*
  		 * Now return to normal top-level context and clear ErrorContext for
  		 * next time.
  		 */
! 		MemoryContextSwitchTo(AutovacMemCxt);
  		FlushErrorState();
  
  		/* Flush any leaked data in the top-level context */
! 		MemoryContextResetAndDeleteChildren(AutovacMemCxt);
  
  		/* don't leave dangling pointers to freed memory */
  		DatabaseListCxt = NULL;
--- 691,707 ----
  		EmitErrorReport();
  
  		/* Abort the current transaction in order to recover */
! 		AbortOutOfAnyTransaction();
  
  		/*
  		 * Now return to normal top-level context and clear ErrorContext for
  		 * next time.
  		 */
! 		MemoryContextSwitchTo(AutovacLauncherMemCxt);
  		FlushErrorState();
  
  		/* Flush any leaked data in the top-level context */
! 		MemoryContextResetAndDeleteChildren(AutovacLauncherMemCxt);
  
  		/* don't leave dangling pointers to freed memory */
  		DatabaseListCxt = NULL;
*************** AutoVacLauncherMain(int argc, char *argv
*** 520,525 ****
--- 732,742 ----
  	AutoVacuumShmem->av_launcherid = MyBackendId;
  
  	/*
+ 	 * Initial population of the database list from pg_database
+ 	 */
+ 	populate_co_databases();
+ 
+ 	/*
  	 * Create the initial database list.  The invariant we want this list to
  	 * keep is that it's ordered by decreasing next_time.  As soon as an entry
  	 * is updated to a higher time, it will be moved to the front (which is
*************** AutoVacLauncherMain(int argc, char *argv
*** 530,536 ****
  
  	for (;;)
  	{
! 		struct timeval nap;
  
  		/*
  		 * Emergency bailout if postmaster has died.  This is to avoid the
--- 747,754 ----
  
  	for (;;)
  	{
! 		TimestampTz		current_time;
! 		struct timeval	nap;
  
  		/*
  		 * Emergency bailout if postmaster has died.  This is to avoid the
*************** AutoVacLauncherMain(int argc, char *argv
*** 539,546 ****
  		if (!PostmasterIsAlive(true))
  			proc_exit(1);
  
! 		launcher_determine_sleep((AutoVacuumShmem->av_freeWorkers != NULL),
! 								 false, &nap);
  
  		/* Allow sinval catchup interrupts while sleeping */
  		EnableCatchupInterrupt();
--- 757,764 ----
  		if (!PostmasterIsAlive(true))
  			proc_exit(1);
  
! 		can_launch = (AutoVacuumShmem->av_freeWorkers != NULL);
! 		launcher_determine_sleep(can_launch, false, &nap);
  
  		/* Allow sinval catchup interrupts while sleeping */
  		EnableCatchupInterrupt();
*************** AutoVacLauncherMain(int argc, char *argv
*** 647,658 ****
  			msg = IMessageCheck();
  		}
  
  		/*
  		 * Periodically check and trigger autovacuum workers, if autovacuum
  		 * is enabled.
  		 */
  		if (autovacuum_enabled)
! 			CoordinatorMaybeTriggerAutoVacuum();
  	}
  
  	/* Normal exit from the autovac launcher is here */
--- 865,881 ----
  			msg = IMessageCheck();
  		}
  
+ 		current_time  = GetCurrentTimestamp();
+ 		can_launch = CoordinatorCanLaunchWorker(current_time);
+ 
  		/*
  		 * Periodically check and trigger autovacuum workers, if autovacuum
  		 * is enabled.
  		 */
  		if (autovacuum_enabled)
! 			autovacuum_maybe_trigger_job(current_time, can_launch);
! 
! 		manage_workers(can_launch);
  	}
  
  	/* Normal exit from the autovac launcher is here */
*************** handle_imessage(IMessage *msg)
*** 666,680 ****
  void
  handle_imessage(IMessage *msg)
  {
- 	IMessageType	msg_type;
  	BackendId		msg_sender;
  
! 	msg_type = msg->type;
  	msg_sender = msg->sender;
  
  #ifdef COORDINATOR_DEBUG
! 	elog(DEBUG3, "Coordinator: received %s from pid %d",
! 		 decode_imessage_type(msg->type), msg->sender);
  #endif
  
  	switch (msg->type)
--- 889,927 ----
  void
  handle_imessage(IMessage *msg)
  {
  	BackendId		msg_sender;
+ 	PGPROC         *proc;
+ 	TransactionId   local_xid = InvalidTransactionId;
+ 	co_database    *codb = NULL;
+ 	Oid             dboid = InvalidOid;
  
! 	/*
! 	 * Get the PGPROC entry of the sender and the related database info, if
! 	 * any.
! 	 */
  	msg_sender = msg->sender;
  
+ 	LWLockAcquire(ProcArrayLock, LW_SHARED);
+ 
+ 	proc = BackendIdGetProc(msg_sender);
+ 	if (proc)
+ 	{
+ 		local_xid = proc->xid;
+ 		dboid = proc->databaseId;
+ 	}
+ 
+ 	LWLockRelease(ProcArrayLock);
+ 
  #ifdef COORDINATOR_DEBUG
! 	if (proc)
! 		elog(DEBUG3, "Coordinator: received %s of size %d from backend %d\n"
! 			 "\t(connected to db %d, local xid %d)",
! 			 decode_imessage_type(msg->type), msg->size, msg_sender,
! 			 dboid, local_xid);
! 	else
! 		elog(DEBUG3, "Coordinator: received %s of size %d from backend %d\n"
! 			 "\t(for which no PGPROC could be found)",
! 			 decode_imessage_type(msg->type), msg->size, msg_sender);
  #endif
  
  	switch (msg->type)
*************** handle_imessage(IMessage *msg)
*** 687,701 ****
  			/* consume the message */
  			IMessageRemove(msg);
  
  			/*
! 			 * default to immediately command the new background
! 			 * worker to perform an autovacuum step.
  			 */
! 			if (IMSGT_REGISTER_WORKER)
! 			{
! 				msg = IMessageCreate(IMSGT_PERFORM_VACUUM, 0);
! 				IMessageActivate(msg, msg_sender);
! 			}
  
  			/*
  			 * Rebalance cost limits, as the worker has already reported its
--- 934,952 ----
  			/* consume the message */
  			IMessageRemove(msg);
  
+ 			LWLockAcquire(CoordinatorDatabasesLock, LW_SHARED);
+ 			codb = get_co_database(dboid);
+ 			process_cached_jobs(codb);
+ 			LWLockRelease(CoordinatorDatabasesLock);
+ 
  			/*
! 			 * We trigger a DatabaseList rebuild if it is still empty and
! 			 * after a job is done. This mainly covers the initialization
! 			 * phase after the first background worker is done with vacuuming
! 			 * template1 (and thus having populated pgstat).
  			 */
! 			if (DLGetHead(DatabaseList) == NULL)
! 				rebuild_database_list(InvalidOid);
  
  			/*
  			 * Rebalance cost limits, as the worker has already reported its
*************** handle_imessage(IMessage *msg)
*** 714,735 ****
  			IMessageRemove(msg);
  
  			/* trigger an autovacuum worker */
! 			do_start_autovacuum_worker();
! 
  			break;
  
  		default:
! 			elog(WARNING, "Unknown message type: %c, ignored!", msg->type);
  			IMessageRemove(msg);
  	}
  }
  
  static void
! CoordinatorMaybeTriggerAutoVacuum()
  {
! 	TimestampTz current_time = 0;
  	bool		can_launch;
- 	Dlelem	   *elem;
  
  	/* FIXME: indentation */
  	{
--- 965,1170 ----
  			IMessageRemove(msg);
  
  			/* trigger an autovacuum worker */
! 			dboid = autovacuum_select_database();
! 			if (dboid != InvalidOid)
! 			{
! 				LWLockAcquire(CoordinatorDatabasesLock, LW_SHARED);
! 				codb = get_co_database(dboid);
! 				msg = IMessageCreate(IMSGT_PERFORM_VACUUM, 0);
! 				dispatch_job(msg, codb);
! 				LWLockRelease(CoordinatorDatabasesLock);
! 			}
  			break;
  
  		default:
! 			elog(WARNING, "Coordinator: unknown message type: %c, ignored!",
! 				 msg->type);
  			IMessageRemove(msg);
  	}
  }
  
+ 
+ 
+ /*
+  * get_co_database
+  *
+  * Gets or creates the database info for replication in shared memory.
+  * Expects the caller to have the CoordinatorDatabasesLock.
+  */
+ co_database *
+ get_co_database(Oid dboid)
+ {
+     co_database *codb;
+     bool found;
+ 
+     codb = hash_search(co_databases, &dboid, HASH_ENTER, &found);
+     if (!found)
+         init_co_database(codb);
+ 
+     return codb;
+ }
+ 
+ static bool
+ can_deliver_cached_job(co_database *codb, IMessage *msg, BackendId *target)
+ {
+ #ifdef COORDINATOR_DEBUG
+ 	elog(DEBUG5, "Coordinator: checking deliverability of job type %s",
+ 		 decode_imessage_type(msg->type));
+ #endif
+ 
+ 	switch (msg->type)
+ 	{
+ 		case IMSGT_TERM_WORKER:
+ 		case IMSGT_PERFORM_VACUUM:
+ 			return true;
+ 
+ 		default:
+ 			elog(WARNING, "Coordinator: missing deliverability check for "
+ 				 "message type %s", decode_imessage_type(msg->type));
+ 			return false;
+ 	}
+ }
+ 
+ /*
+  * manage_workers
+  *
+  * Starts background workers for databases which have at least one cached
+  * job or which have less than min_background_workers connected. Within the
+  * same loop, the autovacuum_max_workers is checked and terminates a worker
+  * accordingly.
+  * 
+  * Note that at max one worker can be requested to start or stop per
+  * invocation.
+  */
  static void
! manage_workers(bool can_launch)
  {
! 	HASH_SEQ_STATUS			hash_status;
! 	co_database	           *codb;
! 	Oid                     launch_dboid = InvalidOid;
! 	float                   max_score = 0.0,
! 		                    score;
! 	bool                    worker_slots_available;
! 	int                     idle_workers_required;
! 	int                     job_workers_required;
! 
! 	LWLockAcquire(AutovacuumLock, LW_SHARED);
! 	worker_slots_available = (AutoVacuumShmem->av_freeWorkers != NULL);
! 	LWLockRelease(AutovacuumLock);
! 
! 	/*
! 	 * Terminate an unneeded worker that has been fetched from the list of
! 	 * idle workers in the last invocation. We defer sending the signal one
! 	 * invocation to make sure the coordinator had time to handle all
! 	 * pending messages from that worker. As idle workers don't ever send
! 	 * messages, we can safely assume there is no pending message from that
! 	 * worker by now.
! 	 */
! 	if (terminatable_worker != NULL)
! 	{
! 		IMessage *msg;
! 
! #ifdef COORDINATOR_DEBUG
! 		PGPROC *proc = BackendIdGetProc(terminatable_worker->wi_backend_id);
! 		if (proc)
! 			elog(DEBUG3, "Coordinator: terminating worker [%d/%d].",
! 				 proc->pid, terminatable_worker->wi_backend_id);
! 		else
! 			elog(WARNING, "Coordinator: terminating worker (no PGPROC, backend %d).",
! 				 terminatable_worker->wi_backend_id);
! #endif
! 
! 		msg = IMessageCreate(IMSGT_TERM_WORKER, 0);
! 		IMessageActivate(msg, terminatable_worker->wi_backend_id);
! 
! 		terminatable_worker = NULL;
! 	}
! 
! #ifdef COORDINATOR_DEBUG
! 	elog(DEBUG3, "Coordinator: manage_workers: can_launch: %s, slots_available: %s",
! 		 (can_launch ? "true" : "false"), (worker_slots_available ? "true" : "false"));
! #endif
! 
! 	/*
! 	 * Check the list of databases and fire the first pending request
! 	 * we find.
! 	 */
! 	idle_workers_required = 0;
! 	job_workers_required = 0;
! 	LWLockAcquire(CoordinatorDatabasesLock, LW_SHARED);
! 	hash_seq_init(&hash_status, co_databases);
! 	while ((codb = (co_database*) hash_seq_search(&hash_status)))
! 	{
! 		score = ((float) codb->codb_num_cached_jobs /
! 				 (float) (codb->codb_num_connected_workers + 1)) * 100.0;
! 
! 		if (codb->codb_num_idle_workers < min_spare_background_workers)
! 			score += (min_spare_background_workers -
! 					  codb->codb_num_idle_workers) * 10.0;
! 
! #ifdef COORDINATOR_DEBUG
! 		elog(DEBUG3, "Coordinator:     db %d, idle/conn: %d/%d, jobs: %d, score: %0.1f",
! 			 codb->codb_dboid, codb->codb_num_idle_workers,
! 			 codb->codb_num_connected_workers, codb->codb_num_cached_jobs,
! 			 score);
! #endif
! 
! 		if (codb->codb_num_cached_jobs &&
! 			(codb->codb_num_connected_workers == 0))
! 			job_workers_required++;
! 
! 		if (codb->codb_num_idle_workers < min_spare_background_workers)
! 			idle_workers_required += (min_spare_background_workers -
! 									  codb->codb_num_idle_workers);
! 
! 		/*
! 		 * FIXME: "misconfiguration" allows "starvation" in case the global
! 		 *        maximum is reached all with idle workers, but other dbs
! 		 *        w/o a single worker still have jobs.
! 		 */
! 		if (can_launch && ((codb->codb_num_cached_jobs > 0) ||
! 						   (codb->codb_num_idle_workers <
! 							min_spare_background_workers)))
! 		{
! 			if (can_launch && (score > max_score))
! 			{
! 				launch_dboid = codb->codb_dboid;
! 				max_score = score;
! 			}
! 		}
! 
! 		/*
! 		 * If we are above limit, we fetch an idle worker from the list
! 		 * and mark it as terminatable. Actual termination happens in
! 		 * the following invocation, see above.
! 		 */
! 		if ((terminatable_worker == NULL) &&
! 			(codb->codb_num_idle_workers > max_spare_background_workers))
! 			terminatable_worker = get_idle_worker(codb);
! 	}
! 	LWLockRelease(CoordinatorDatabasesLock);
! 
! 	if (!worker_slots_available && idle_workers_required > 0)
! 	{
! 		elog(WARNING, "Coordinator: no more background workers available, but requiring %d more, according to min_spare_background_workers.",
! 			 idle_workers_required);
! 	}
! 
! 	if (!worker_slots_available && job_workers_required > 0)
! 	{
! 		elog(WARNING, "Coordinator: no background workers avalibale, but %d databases have background jobs pending.",
! 			 job_workers_required);
! 	}
! 
! 	/* request a worker for the first database found, which needs one */
! 	if (OidIsValid(launch_dboid))
! 		do_start_worker(launch_dboid);
! }
! 
! static bool
! CoordinatorCanLaunchWorker(TimestampTz current_time)
! {
  	bool		can_launch;
  
  	/* FIXME: indentation */
  	{
*************** CoordinatorMaybeTriggerAutoVacuum()
*** 741,747 ****
  		 * other worker failed while starting up.
  		 */
  
- 		current_time = GetCurrentTimestamp();
  		LWLockAcquire(AutovacuumLock, LW_SHARED);
  
  		can_launch = (AutoVacuumShmem->av_freeWorkers != NULL);
--- 1176,1181 ----
*************** CoordinatorMaybeTriggerAutoVacuum()
*** 800,810 ****
  				can_launch = false;
  		}
  		LWLockRelease(AutovacuumLock);	/* either shared or exclusive */
  
! 		/* if we can't do anything, just return. */
! 		if (!can_launch)
! 			return;
  
  		/* We're OK to start a new worker */
  
  		elem = DLGetTail(DatabaseList);
--- 1234,1255 ----
  				can_launch = false;
  		}
  		LWLockRelease(AutovacuumLock);	/* either shared or exclusive */
+ 	}
  
! 	return can_launch;
! }
  
+ static void
+ autovacuum_maybe_trigger_job(TimestampTz current_time, bool can_launch)
+ {
+ 	Oid         dboid = InvalidOid;
+ 	Dlelem	   *elem;
+ 	IMessage   *msg;
+ 	co_database *codb;
+ 
+ 	/* FIXME: indentation */
+ 	{
+ 
  		/* We're OK to start a new worker */
  
  		elem = DLGetTail(DatabaseList);
*************** CoordinatorMaybeTriggerAutoVacuum()
*** 818,836 ****
  			 */
  			if (TimestampDifferenceExceeds(avdb->adl_next_worker,
  										   current_time, 0))
! 				launch_worker(current_time);
  		}
  		else
  		{
  			/*
! 			 * Special case when the list is empty: start a worker right away.
! 			 * This covers the initial case, when no database is in pgstats
! 			 * (thus the list is empty).  Note that the constraints in
! 			 * launcher_determine_sleep keep us from starting workers too
! 			 * quickly (at most once every autovacuum_naptime when the list is
! 			 * empty).
  			 */
! 			launch_worker(current_time);
  		}
  	}
  }
--- 1263,1301 ----
  			 */
  			if (TimestampDifferenceExceeds(avdb->adl_next_worker,
  										   current_time, 0))
! 			{
! 				dboid = autovacuum_select_database();
! 				if (OidIsValid(dboid))
! 				{
! 					LWLockAcquire(CoordinatorDatabasesLock, LW_EXCLUSIVE);
! 
! 					codb = get_co_database(dboid);
! 
! 					/*
! 					 * Only dispatch a job, if it can be processed immediately
! 					 * so we don't end up having lots of autovacuum requests
! 					 * in the job cache.
! 					 */
! 					if (can_launch || codb->codb_num_idle_workers > 0)
! 					{
! 						msg = IMessageCreate(IMSGT_PERFORM_VACUUM, 0);
! 						dispatch_job(msg, codb);
! 					}
! 
! 					LWLockRelease(CoordinatorDatabasesLock);
! 				}
! 			}
  		}
  		else
  		{
  			/*
! 			 * If the list is still empty, this is a no-op. Instead we simply
! 			 * wait until the initial vacuum on the template database is
! 			 * done. That will populate pg_stat.
  			 */
! #ifdef COORDINATOR_DEBUG
! 			elog(DEBUG5, "Coordinator: DatabaseList is still empty.");
! #endif
  		}
  	}
  }
*************** rebuild_database_list(Oid newdb)
*** 933,939 ****
  	/* use fresh stats */
  	autovac_refresh_stats();
  
! 	newcxt = AllocSetContextCreate(AutovacMemCxt,
  								   "AV dblist",
  								   ALLOCSET_DEFAULT_MINSIZE,
  								   ALLOCSET_DEFAULT_INITSIZE,
--- 1398,1404 ----
  	/* use fresh stats */
  	autovac_refresh_stats();
  
! 	newcxt = AllocSetContextCreate(AutovacLauncherMemCxt,
  								   "AV dblist",
  								   ALLOCSET_DEFAULT_MINSIZE,
  								   ALLOCSET_DEFAULT_INITSIZE,
*************** void
*** 1129,1141 ****
   * start a worker.
   */
  void
! do_start_worker(avw_dbase *avdb)
  {
  	WorkerInfo	worker;
  
  #ifdef COORDINATOR_DEBUG
! 	elog(DEBUG5, "Coordinator: requesting worker for database %s",
! 		 avdb->adw_name);
  #endif
  
  	LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
--- 1594,1607 ----
   * start a worker.
   */
  void
! do_start_worker(Oid dboid)
  {
  	WorkerInfo	worker;
  
+ 	Assert(OidIsValid(dboid));
+ 
  #ifdef COORDINATOR_DEBUG
! 	elog(DEBUG3, "Coordinator: requesting worker for database %d.", dboid);
  #endif
  
  	LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
*************** do_start_worker(avw_dbase *avdb)
*** 1151,1157 ****
  
  	AutoVacuumShmem->av_freeWorkers = (WorkerInfo) worker->wi_links.next;
  
! 	worker->wi_dboid = avdb->adw_datid;
  	worker->wi_backend_id = InvalidBackendId;
  	worker->wi_launchtime = GetCurrentTimestamp();
  
--- 1617,1623 ----
  
  	AutoVacuumShmem->av_freeWorkers = (WorkerInfo) worker->wi_links.next;
  
! 	worker->wi_dboid = dboid;
  	worker->wi_backend_id = InvalidBackendId;
  	worker->wi_launchtime = GetCurrentTimestamp();
  
*************** do_start_worker(avw_dbase *avdb)
*** 1163,1188 ****
  }
  
  /*
!  * do_start_autovacuum_worker
   *
!  * It determines what database to work on, sets up shared memory stuff and
!  * signals postmaster to start the worker.	It fails gracefully if invoked when
!  * autovacuum_workers are already active.
   *
!  * Return value is the OID of the database that the worker is going to process,
!  * or InvalidOid if no worker was actually started.
   */
  static Oid
! do_start_autovacuum_worker(void)
  {
! 	List	   *dblist;
! 	ListCell   *cell;
  	TransactionId xidForceLimit;
  	bool		for_xid_wrap;
  	avw_dbase  *avdb;
  	TimestampTz current_time;
  	bool		skipit = false;
! 	Oid			retval = InvalidOid;
  	MemoryContext tmpcxt,
  				oldcxt;
  
--- 1629,1653 ----
  }
  
  /*
!  * autovacuum_select_database
   *
!  * It determines what database to work on and sets up shared memory stuff. It
!  * fails gracefully if invoked when autovacuum_workers are already active.
   *
!  * Returns a pointer to the coordinator info struct of the database that the
!  * next worker should process, or NULL if no database needs vacuuming.
   */
  static Oid
! autovacuum_select_database(void)
  {
! 	List		*dblist;
! 	ListCell    *cell;
  	TransactionId xidForceLimit;
  	bool		for_xid_wrap;
  	avw_dbase  *avdb;
  	TimestampTz current_time;
  	bool		skipit = false;
! 	Oid         retval = InvalidOid;
  	MemoryContext tmpcxt,
  				oldcxt;
  
*************** do_start_autovacuum_worker(void)
*** 1315,1324 ****
  			avdb = tmp;
  	}
  
- 	/* Found a database -- process it */
  	if (avdb != NULL)
  	{
! 		do_start_worker(avdb);
  		retval = avdb->adw_datid;
  	}
  	else if (skipit)
--- 1780,1788 ----
  			avdb = tmp;
  	}
  
  	if (avdb != NULL)
  	{
! 		/* We've found a database that needs vacuuming, return its id */
  		retval = avdb->adw_datid;
  	}
  	else if (skipit)
*************** do_start_autovacuum_worker(void)
*** 1337,1361 ****
  }
  
  /*
!  * launch_worker
   *
!  * Wrapper for starting a worker from the launcher.  Besides actually starting
!  * it, update the database list to reflect the next time that another one will
!  * need to be started on the selected database.  The actual database choice is
!  * left to do_start_autovacuum_worker.
   *
   * This routine is also expected to insert an entry into the database list if
   * the selected database was previously absent from the list.
   */
  static void
! launch_worker(TimestampTz now)
  {
! 	Oid			dbid;
! 	Dlelem	   *elem;
  
! 	dbid = do_start_autovacuum_worker();
! 	if (OidIsValid(dbid))
  	{
  		/*
  		 * Walk the database list and update the corresponding entry.  If the
  		 * database is not on the list, we'll recreate the list.
--- 1801,1824 ----
  }
  
  /*
!  * autovacuum_update_timing
   *
!  * After having started an autovacuum job, the coordinator needs to update
!  * the database list to reflect the next time that another one will need to
!  * be started on the selected database.
   *
   * This routine is also expected to insert an entry into the database list if
   * the selected database was previously absent from the list.
   */
+ 
  static void
! autovacuum_update_timing(Oid dbid, TimestampTz now)
  {
! 	Dlelem		   *elem;
  
! 	/* FIXME: indentation */
  	{
+ 
  		/*
  		 * Walk the database list and update the corresponding entry.  If the
  		 * database is not on the list, we'll recreate the list.
*************** StartAutoVacWorker(void)
*** 1492,1497 ****
--- 1955,2132 ----
  }
  
  /*
+  * add_as_idle_worker
+  *
+  * Marks the current worker as idle by adding it to the database's list of
+  * idle worker backends. The caller is expected to hold the AutovacuumLock.
+  */
+ static void
+ add_as_idle_worker(Oid dbid, bool inc_connected_count)
+ {
+ 	co_database *codb;
+ 
+ 	Assert(SHMQueueIsDetached(&MyWorkerInfo->wi_links));
+ 
+ 	/* Lookup the corresponding database, or create an entry for it */
+ 	LWLockAcquire(CoordinatorDatabasesLock, LW_EXCLUSIVE);
+ 	codb = get_co_database(dbid);
+ 
+ 	if (inc_connected_count)
+ 		codb->codb_num_connected_workers++;
+ 
+ 	/* add as an idle worker */
+ 	SHMQueueInsertBefore(&codb->codb_idle_workers, &MyWorkerInfo->wi_links);
+ 	codb->codb_num_idle_workers++;
+ 
+ 	LWLockRelease(CoordinatorDatabasesLock);
+ }
+ 
+ /*
+  * bgworker_job_initialize
+  *
+  * Initializes the memory contexts for a background job.
+  */
+ void
+ bgworker_job_initialize(worker_state new_state)
+ {
+ 	/*
+ 	 * Note that the coordinator is responsible for dequeuing the worker from
+ 	 * the list of idle backends, but is shall *NOT* assign a worker state,
+ 	 * we do that from the worker exclusively.
+ 	 */
+ 	Assert(SHMQueueIsDetached(&MyWorkerInfo->wi_links));
+ 	Assert(MyWorkerInfo->wi_state == WS_IDLE);
+ 
+ 	MyWorkerInfo->wi_state = new_state;
+ 	switch (new_state)
+ 	{
+ 		case WS_IDLE:
+ 			Assert(false);    /* use bgworker_job_completed instead */
+ 			break;
+ 		case WS_AUTOVACUUM:
+ 			set_ps_display("bg worker: autovacuum", false);
+ 			break;
+ 		default:
+ 			set_ps_display("bg worker: unknown", false);
+ 	}
+ 
+ 	/*
+ 	 * StartTransactionCommand and CommitTransactionCommand will
+ 	 * automatically switch to other contexts.  None the less we need this
+ 	 * one for other book-keeping of the various background jobs across
+ 	 * transactions, for example to keep the list of relations to vacuum.
+ 	 */
+ 	Assert(AutovacWorkerMemCxt == NULL);
+ 	AutovacWorkerMemCxt = AllocSetContextCreate(TopMemoryContext,
+ 										   "Background Worker",
+ 										   ALLOCSET_DEFAULT_MINSIZE,
+ 										   ALLOCSET_DEFAULT_INITSIZE,
+ 										   ALLOCSET_DEFAULT_MAXSIZE);
+ 
+ 	MessageContext = AllocSetContextCreate(TopMemoryContext,
+ 										   "MessageContext",
+ 										   ALLOCSET_DEFAULT_MINSIZE,
+ 										   ALLOCSET_DEFAULT_INITSIZE,
+ 										   ALLOCSET_DEFAULT_MAXSIZE);
+ 
+ 	MemoryContextSwitchTo(AutovacWorkerMemCxt);
+ }
+ 
+ /*
+  * bgworker_job_completed
+  *
+  * Cleans up the memory contexts used for the worker's current job and
+  * informs the coordinator.
+  */
+ void
+ bgworker_job_completed(void)
+ {
+ 	/* Notify the coordinator of the job completion. */
+ #ifdef COORDINATOR_DEBUG
+ 	ereport(DEBUG3,
+ 			(errmsg("bg worker [%d]: job completed.", MyProcPid)));
+ #endif
+ 
+ 	/* reset the worker state */
+ 	bgworker_reset();
+ }
+ 
+ void
+ bgworker_reset(void)
+ {
+ 	BackendId AutovacuumLauncherId;
+ 	IMessage *msg;
+ 
+ 	elog(DEBUG5, "bg worker [%d/%d]: resetting",
+ 		 MyProcPid, MyBackendId);
+ 
+ 	Assert(MyWorkerInfo->wi_state != WS_IDLE);
+ 	Assert(!TransactionIdIsValid(GetTopTransactionIdIfAny()));
+ 
+ 	/* reset the worker state */
+ 	MyWorkerInfo->wi_state = WS_IDLE;
+ 	set_ps_display("bg worker: idle", false);
+ 
+ 	/* clean up memory contexts */
+ 	Assert(AutovacWorkerMemCxt);
+ 	MemoryContextSwitchTo(TopMemoryContext);
+ 	MemoryContextDelete(AutovacWorkerMemCxt);
+ 	AutovacWorkerMemCxt = NULL;
+ 	MemoryContextDelete(MessageContext);
+ 	MessageContext = NULL;
+ 
+ 	/* Reset the the process-local cleanup handler state. */
+ 	BgWorkerCleanupInProgress = false;
+ 
+ 	/* propagate as idle worker, inform the coordinator */
+ 	LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+ 	add_as_idle_worker(MyDatabaseId, false);
+ 	LWLockRelease(AutovacuumLock);
+ 
+ 	AutovacuumLauncherId = GetAutovacuumLauncherId();
+ 	if (AutovacuumLauncherId != InvalidBackendId)
+ 	{
+ 		msg = IMessageCreate(IMSGT_READY, 0);
+ 		IMessageActivate(msg, AutovacuumLauncherId);
+ 	}
+ 	else
+ 		elog(WARNING, "bg worker [%d/%d]: no coordinator?!?",
+ 			 MyProcPid, MyBackendId);
+ }
+ 
+ void
+ bgworker_job_failed(int errcode)
+ {
+ 	TransactionId xid;
+ 
+ 	xid = GetTopTransactionIdIfAny();
+ 
+ #ifdef COORDINATOR_DEBUG
+ 	ereport(DEBUG3,
+ 			(errmsg("bg worker [%d/%d]: job failed (xid: %d)",
+ 					MyProcPid, MyBackendId, xid)));
+ #endif
+ 
+ 	/*
+ 	 * Abort any transaction that might still be running and tell the
+ 	 * coordinator that we are ready to process the next background job.
+ 	 */
+ 	AbortOutOfAnyTransaction();
+ 
+ 	/*
+ 	 * Flush the error state.
+ 	 */
+ 	FlushErrorState();
+ 
+ 	/*
+ 	 * Make sure pgstat also considers our stat data as gone.
+ 	 */
+ 	pgstat_clear_snapshot();
+ 
+ 	Assert(!IMessageCheck());
+ }
+ 
+ /*
   * AutoVacWorkerMain
   */
  NON_EXEC_STATIC void
*************** AutoVacWorkerMain(int argc, char *argv[]
*** 1632,1641 ****
  	/* FIXME: indentation */
  	{
  
- 		/* insert into the running list */
- 		SHMQueueInsertBefore(&AutoVacuumShmem->av_runningWorkers,
- 							 &MyWorkerInfo->wi_links);
- 
  		/*
  		 * remove from the "starting" pointer, so that the launcher can start
  		 * a new worker if required
--- 2267,2272 ----
*************** AutoVacWorkerMain(int argc, char *argv[]
*** 1665,1677 ****
  		 */
  		InitPostgres(NULL, dbid, NULL, dbname);
  		SetProcessingMode(NormalProcessing);
! 		set_ps_display(dbname, false);
! 		ereport(DEBUG1,
! 				(errmsg("autovacuum: processing database \"%s\"", dbname)));
  	}
  
  	MyWorkerInfo->wi_backend_id = MyBackendId;
  
  	/* register with the coordinator */
  	if (coordinator_id != InvalidBackendId)
  	{
--- 2296,2323 ----
  		 */
  		InitPostgres(NULL, dbid, NULL, dbname);
  		SetProcessingMode(NormalProcessing);
! 		set_ps_display("bg worker: idle", false);
  	}
  
+ 	AutovacWorkerMemCxt = NULL;
+ 
  	MyWorkerInfo->wi_backend_id = MyBackendId;
+ 	MyWorkerInfo->wi_state = WS_IDLE;
  
+ #ifdef COORDINATOR_DEBUG
+ 	elog(DEBUG3, "bg worker [%d/%d]: connected to database %d",
+ 		 MyProcPid, MyBackendId, dbid);
+ #endif
+ 
+ 	/*
+ 	 * Add as an idle worker and notify the coordinator only *after* having
+ 	 * set MyProc->databaseId in InitPostgres, so the coordinator can
+ 	 * determine which database we are connected to.
+ 	 */
+ 	LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+ 	add_as_idle_worker(dbid, true);
+ 	LWLockRelease(AutovacuumLock);
+ 
  	/* register with the coordinator */
  	if (coordinator_id != InvalidBackendId)
  	{
*************** AutoVacWorkerMain(int argc, char *argv[]
*** 1684,1695 ****
--- 2330,2354 ----
  
  	while (!terminate_worker)
  	{
+ 		PG_TRY();
+ 		{
+ 		/* FIXME: indentation */
+ 
  		CHECK_FOR_INTERRUPTS();
  
  		ImmediateInterruptOK = true;
  		pg_usleep(1000000L);
  		ImmediateInterruptOK = false;
  
+ 		/*
+ 		 * FIXME: check different ways of terminating a background worker
+ 		 *        via ProcDiePending. How about postmaster initiated
+ 		 *        restarts?
+ 		 */
+ 		if (ProcDiePending)
+ 			elog(FATAL, "bg worker [%d/%d]: Terminated via ProcDie",
+ 				 MyProcPid, MyBackendId);
+ 
  		while ((msg = IMessageCheck()) && !terminate_worker)
  		{
  #ifdef COORDINATOR_DEBUG
*************** AutoVacWorkerMain(int argc, char *argv[]
*** 1702,1762 ****
  							MyDatabaseId)));
  #endif
  
- 			/*
- 			 * StartTransactionCommand and CommitTransactionCommand will
- 			 * automatically switch to other contexts.  None the less we need
- 			 * this one for other book-keeping of the various background
- 			 * jobs across transactions, for example to keep the list of
- 			 * relations to vacuum.
- 			 */
- 			AutovacMemCxt = AllocSetContextCreate(TopMemoryContext,
- 												  "Background Worker",
- 												  ALLOCSET_DEFAULT_MINSIZE,
- 												  ALLOCSET_DEFAULT_INITSIZE,
- 												  ALLOCSET_DEFAULT_MAXSIZE);
- 			MemoryContextSwitchTo(AutovacMemCxt);
- 
  			switch (msg->type)
  			{
  				case IMSGT_PERFORM_VACUUM:
  					/* immediately remove the message to free shared memory */
  					IMessageRemove(msg);
  
  					/* do an appropriate amount of work */
  					do_autovacuum();
  
- 
  					/*
! 					 * send an IMSGT_READY to inform the coordinator that we
! 					 * finished vacuuming.
  					 */
! 					msg = IMessageCreate(IMSGT_READY, 0);
! 					IMessageActivate(msg, GetAutovacuumLauncherId());
  
! 					/*
! 					 * for now, we always terminate the worker after an 
! 					 * autovacuum job.
! 					 */
! 					terminate_worker = true;
  					break;
  
  				default:
  					ereport(WARNING,
  							(errmsg("bg worker [%d]: invalid message type "
  									"'%c' ignored",
  									MyProcPid, msg->type)));
  
! 					/* keep shared memory clean */
! 					IMessageRemove(msg);
! 			}
  
! 			MemoryContextSwitchTo(TopMemoryContext);
! 			MemoryContextDelete(AutovacMemCxt);
  
! 			CHECK_FOR_INTERRUPTS();
  		}
  	}
  
  	/* All done, go away */
  	ereport(DEBUG1, (errmsg("bg worker [%d/%d]: terminating",
  							MyProcPid, MyBackendId)));
--- 2361,2458 ----
  							MyDatabaseId)));
  #endif
  
  			switch (msg->type)
  			{
+ 				case IMSGT_TERM_WORKER:
+ 					IMessageRemove(msg);
+ 					terminate_worker = true;
+ 					break;
+ 
  				case IMSGT_PERFORM_VACUUM:
  					/* immediately remove the message to free shared memory */
  					IMessageRemove(msg);
  
+ 					bgworker_job_initialize(WS_AUTOVACUUM);
+ 
+ 					/*
+ 					 * Add ourselves to the list of runningWorkers
+ 					 */
+ 					LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
+ 					SHMQueueInsertBefore(&AutoVacuumShmem->av_runningWorkers,
+ 										 &MyWorkerInfo->wi_links);
+ 					LWLockRelease(AutovacuumLock);
+ 
  					/* do an appropriate amount of work */
  					do_autovacuum();
  
  					/*
! 					 * Remove ourselves from the list of runningWorkers and
! 					 * mark as available background worker.
  					 */
! 					LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
! 					SHMQueueDelete(&MyWorkerInfo->wi_links);
! 					LWLockRelease(AutovacuumLock);
  
! 					bgworker_job_completed();
  					break;
  
  				default:
+ 					/* keep shared memory clean */
+ 					IMessageRemove(msg);
+ 
  					ereport(WARNING,
  							(errmsg("bg worker [%d]: invalid message type "
  									"'%c' ignored",
  									MyProcPid, msg->type)));
+ 			}
  
! 			CHECK_FOR_INTERRUPTS();
! 		}
  
! 		}
! 		PG_CATCH();
! 		{
! 			ErrorData *errdata;
! 			MemoryContext ecxt;
  
! 			ecxt = MemoryContextSwitchTo(AutovacWorkerMemCxt);
! 			errdata = CopyErrorData();
! 
! 			elog(WARNING, "bg worker [%d/%d]: caught error '%s' in %s:%d, state %s",
! 				 MyProcPid, MyBackendId, errdata->message,
! 				 errdata->filename, errdata->lineno,
! 				 decode_worker_state(MyWorkerInfo->wi_state));
! 
! 			/*
! 			 * Inform the coordinator about the failure.
! 			 */
! 			bgworker_job_failed(errdata->sqlerrcode);
! 
! 			if (errdata->sqlerrcode == ERRCODE_QUERY_CANCELED)
! 			{
! #ifdef DEBUG_CSET_APPL
! 				elog(DEBUG3, "bg worker [%d/%d]: cancelled active job.",
! 					 MyProcPid, MyBackendId);
! #endif
! 
! 				bgworker_reset();
! 			}
! 			else
! 			{
! 				elog(WARNING, "bg worker [%s:%d]: unexpected error %d: '%s'!\n"
! 					 "    triggered from %s:%d (in %s)\n",
! 					 __FILE__, __LINE__, errdata->sqlerrcode,
! 					 errdata->message, errdata->filename, errdata->lineno,
! 					 errdata->funcname);
! 				/* re-throw the error, so the backend quits */
! 				MemoryContextSwitchTo(ecxt);
! 				PG_RE_THROW();
! 			}
  		}
+ 		PG_END_TRY();
  	}
  
+ 
  	/* All done, go away */
  	ereport(DEBUG1, (errmsg("bg worker [%d/%d]: terminating",
  							MyProcPid, MyBackendId)));
*************** FreeWorkerInfo(int code, Datum arg)
*** 1769,1779 ****
  static void
  FreeWorkerInfo(int code, Datum arg)
  {
  	if (MyWorkerInfo != NULL)
  	{
  		LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
  
! 		SHMQueueDelete(&MyWorkerInfo->wi_links);
  		MyWorkerInfo->wi_links.next = (SHM_QUEUE *) AutoVacuumShmem->av_freeWorkers;
  		MyWorkerInfo->wi_dboid = InvalidOid;
  		MyWorkerInfo->wi_tableoid = InvalidOid;
--- 2465,2478 ----
  static void
  FreeWorkerInfo(int code, Datum arg)
  {
+ 	co_database *codb;
  	if (MyWorkerInfo != NULL)
  	{
  		LWLockAcquire(AutovacuumLock, LW_EXCLUSIVE);
  
! 		if (!SHMQueueIsDetached(&MyWorkerInfo->wi_links))
! 			SHMQueueDelete(&MyWorkerInfo->wi_links);
! 
  		MyWorkerInfo->wi_links.next = (SHM_QUEUE *) AutoVacuumShmem->av_freeWorkers;
  		MyWorkerInfo->wi_dboid = InvalidOid;
  		MyWorkerInfo->wi_tableoid = InvalidOid;
*************** FreeWorkerInfo(int code, Datum arg)
*** 1786,1791 ****
--- 2485,2497 ----
  		/* not mine anymore */
  		MyWorkerInfo = NULL;
  
+ 		/* decrease the conn count */
+ 		LWLockAcquire(CoordinatorDatabasesLock, LW_EXCLUSIVE);
+ 		codb = hash_search(co_databases, &MyDatabaseId, HASH_FIND, NULL);
+ 		Assert(codb);
+ 		codb->codb_num_connected_workers--;
+ 		LWLockRelease(CoordinatorDatabasesLock);
+ 
  		LWLockRelease(AutovacuumLock);
  	}
  }
*************** get_database_list(void)
*** 1910,1917 ****
  	StartTransactionCommand();
  	(void) GetTransactionSnapshot();
  
! 	/* Allocate our results in AutovacMemCxt, not transaction context */
! 	MemoryContextSwitchTo(AutovacMemCxt);
  
  	rel = heap_open(DatabaseRelationId, AccessShareLock);
  	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
--- 2616,2623 ----
  	StartTransactionCommand();
  	(void) GetTransactionSnapshot();
  
! 	/* Allocate our results in AutovacLauncherMemCxt, not transaction context */
! 	MemoryContextSwitchTo(AutovacLauncherMemCxt);
  
  	rel = heap_open(DatabaseRelationId, AccessShareLock);
  	scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
*************** get_database_list(void)
*** 1919,1925 ****
  	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
  	{
  		Form_pg_database pgdatabase = (Form_pg_database) GETSTRUCT(tup);
! 		avw_dbase  *avdb;
  
  		avdb = (avw_dbase *) palloc(sizeof(avw_dbase));
  
--- 2625,2631 ----
  	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
  	{
  		Form_pg_database pgdatabase = (Form_pg_database) GETSTRUCT(tup);
! 		avw_dbase *avdb;
  
  		avdb = (avw_dbase *) palloc(sizeof(avw_dbase));
  
*************** do_autovacuum(void)
*** 2004,2011 ****
  
  	ReleaseSysCache(tuple);
  
  	/* StartTransactionCommand changed elsewhere */
! 	MemoryContextSwitchTo(AutovacMemCxt);
  
  	/* The database hash where pgstat keeps shared relations */
  	shared = pgstat_fetch_stat_dbentry(InvalidOid);
--- 2710,2721 ----
  
  	ReleaseSysCache(tuple);
  
+ 	ereport(DEBUG1,
+ 			(errmsg("autovacuum: processing database \"%s\"",
+ 					NameStr(dbForm->datname))));
+ 
  	/* StartTransactionCommand changed elsewhere */
! 	MemoryContextSwitchTo(AutovacWorkerMemCxt);
  
  	/* The database hash where pgstat keeps shared relations */
  	shared = pgstat_fetch_stat_dbentry(InvalidOid);
*************** do_autovacuum(void)
*** 2219,2225 ****
  	 * create a memory context to act as fake PortalContext, so that the
  	 * contexts created in the vacuum code are cleaned up for each table.
  	 */
! 	PortalContext = AllocSetContextCreate(AutovacMemCxt,
  										  "Autovacuum Portal",
  										  ALLOCSET_DEFAULT_INITSIZE,
  										  ALLOCSET_DEFAULT_MINSIZE,
--- 2929,2935 ----
  	 * create a memory context to act as fake PortalContext, so that the
  	 * contexts created in the vacuum code are cleaned up for each table.
  	 */
! 	PortalContext = AllocSetContextCreate(AutovacWorkerMemCxt,
  										  "Autovacuum Portal",
  										  ALLOCSET_DEFAULT_INITSIZE,
  										  ALLOCSET_DEFAULT_MINSIZE,
*************** do_autovacuum(void)
*** 2291,2297 ****
  		 * that somebody just finished vacuuming this table.  The window to
  		 * the race condition is not closed but it is very small.
  		 */
! 		MemoryContextSwitchTo(AutovacMemCxt);
  		tab = table_recheck_autovac(relid, table_toast_map, pg_class_desc);
  		if (tab == NULL)
  		{
--- 3001,3007 ----
  		 * that somebody just finished vacuuming this table.  The window to
  		 * the race condition is not closed but it is very small.
  		 */
! 		MemoryContextSwitchTo(AutovacWorkerMemCxt);
  		tab = table_recheck_autovac(relid, table_toast_map, pg_class_desc);
  		if (tab == NULL)
  		{
*************** AutoVacuumShmemSize(void)
*** 2874,2885 ****
  	Size		size;
  
  	/*
! 	 * Need the fixed struct and the array of WorkerInfoData.
  	 */
  	size = sizeof(AutoVacuumShmemStruct);
  	size = MAXALIGN(size);
  	size = add_size(size, mul_size(autovacuum_max_workers,
  								   sizeof(WorkerInfoData)));
  	return size;
  }
  
--- 3584,3600 ----
  	Size		size;
  
  	/*
! 	 * Need the fixed struct and the array of WorkerInfoData, plus per
! 	 * database entries in a hash. As we only track databases which have at
! 	 * least one worker attached, we won't ever need more than
! 	 * autovacuum_max_workers entries.
  	 */
  	size = sizeof(AutoVacuumShmemStruct);
  	size = MAXALIGN(size);
  	size = add_size(size, mul_size(autovacuum_max_workers,
  								   sizeof(WorkerInfoData)));
+ 	size = add_size(size, hash_estimate_size(autovacuum_max_workers,
+ 											 sizeof(co_database)));
  	return size;
  }
  
*************** AutoVacuumShmemInit(void)
*** 2890,2895 ****
--- 3605,3611 ----
  void
  AutoVacuumShmemInit(void)
  {
+ 	HASHCTL     hctl;
  	bool		found;
  
  	AutoVacuumShmem = (AutoVacuumShmemStruct *)
*************** AutoVacuumShmemInit(void)
*** 2921,2926 ****
--- 3637,3651 ----
  	}
  	else
  		Assert(found);
+ 
+ 	hctl.keysize = sizeof(Oid);
+ 	hctl.entrysize = sizeof(co_database);
+ 	hctl.hash = oid_hash;
+ 	co_databases = ShmemInitHash("Coordinator Database Info",
+ 								 autovacuum_max_workers,
+ 								 autovacuum_max_workers,
+ 								 &hctl,
+ 								 HASH_ELEM | HASH_FUNCTION);
  }
  
  /*
============================================================
*** src/include/postmaster/autovacuum.h	9d1ae53662236fcc871f84a9b1036619931bffe6
--- src/include/postmaster/autovacuum.h	0937b36c82e931677a8fed1cdd638f0138155086
***************
*** 14,22 ****
--- 14,91 ----
  #ifndef AUTOVACUUM_H
  #define AUTOVACUUM_H
  
+ #include "postgres.h"
+ #include "pgstat.h"
+ #include "lib/dllist.h"
  #include "storage/imsg.h"
  #include "storage/lock.h"
  
+ /*
+  * Valid backend states for background workers.
+  */
+ typedef enum
+ {
+ 	WS_IDLE = 'I',
+ 
+ 	WS_AUTOVACUUM = 'V',
+ 
+ } worker_state;
+ 
+ #define IsIdleWorker(wi)			(IsAutoVacuumWorkerProcess() && (wi->wi_state == WS_IDLE))
+ #define IsAutoVacuumWorker(wi)      (IsAutoVacuumWorkerProcess() && (wi->wi_state == WS_AUTOVACUUM))
+ 
+ 
+ /*-------------
+  * This struct holds information about a single worker's whereabouts.  We keep
+  * an array of these in shared memory, sized according to
+  * autovacuum_max_workers.
+  *
+  * wi_links			entry into free list or running list
+  * wi_dboid			OID of the database this worker is supposed to work on
+  * wi_backend_id	id of the running worker backend, NULL if not started
+  * wi_launchtime	Time at which this worker was launched
+  *
+  * wi_tableoid		OID of the table currently being vacuumed
+  * wi_cost_*		Vacuum cost-based delay parameters current in this worker
+  *
+  * All fields are protected by AutovacuumLock, except for wi_tableoid which is
+  * protected by AutovacuumScheduleLock (which is read-only for everyone except
+  * that worker itself).
+  *-------------
+  */
+ typedef struct WorkerInfoData
+ {
+ 	SHM_QUEUE	wi_links;
+ 	Oid			wi_dboid;
+ 	BackendId   wi_backend_id;
+ 	TimestampTz wi_launchtime;
+ 	worker_state wi_state;
+ 
+ 	/* autovacuum specific fields */
+ 	Oid			wi_tableoid;
+ 	int			wi_cost_delay;
+ 	int			wi_cost_limit;
+ 	int			wi_cost_limit_base;
+ } WorkerInfoData;
+ 
+ typedef struct WorkerInfoData *WorkerInfo;
+ 
+ /* struct to keep track of databases in the coordinator */
+ typedef struct co_database
+ {
+ 	Oid					codb_dboid;
+ 
+ 	/* for internal use by the coordinator */
+ 	int                 codb_num_cached_jobs;
+ 	Dllist              codb_cached_jobs;
+ 
+ 	/* tracking of idle workers */
+ 	int				    codb_num_idle_workers;
+ 	SHM_QUEUE           codb_idle_workers;
+ 
+ 	int                 codb_num_connected_workers;
+ } co_database;
+ 
  /* GUC variables */
  extern bool autovacuum_enabled;
  extern int	autovacuum_max_workers;
*************** extern int	Log_autovacuum_min_duration;
*** 33,38 ****
--- 102,109 ----
  
  extern int	Log_autovacuum_min_duration;
  
+ extern char *decode_worker_state(worker_state state);
+ 
  /* Status inquiry functions */
  extern bool IsAutoVacuumLauncherProcess(void);
  extern bool IsAutoVacuumWorkerProcess(void);
*************** extern void AutoVacuumShmemInit(void);
*** 57,60 ****
--- 128,135 ----
  extern Size AutoVacuumShmemSize(void);
  extern void AutoVacuumShmemInit(void);
  
+ /* bgworker job management functions */
+ extern void bgworker_job_failed(int errcode);
+ extern void bgworker_reset(void);
+ 
  #endif   /* AUTOVACUUM_H */
============================================================
*** src/backend/storage/ipc/imsg.c	81f58ac6067b9e4ebfaaf1e8ada1df68a20e8493
--- src/backend/storage/ipc/imsg.c	c9dbb1c2c90a6b1a5019b4edb42c682f92dac469
*************** decode_imessage_type(const IMessageType 
*** 82,87 ****
--- 82,89 ----
  
  		case IMSGT_REGISTER_WORKER:
  			return "IMSGT_REGISTER_WORKER";
+ 		case IMSGT_TERM_WORKER:
+ 			return "IMSGT_TERM_WORKER";
  		case IMSGT_READY:
  			return "IMSGT_READY";
  
============================================================
*** src/include/storage/imsg.h	33f7d6772ba8d599a4eed0b882ccfa474a5cf258
--- src/include/storage/imsg.h	036c47ec0fee900cdff2ab7a5264d5396eed89dd
*************** typedef enum
*** 33,38 ****
--- 33,39 ----
  	 * registration and job management.
  	 */
  	IMSGT_REGISTER_WORKER = 'W',
+ 	IMSGT_TERM_WORKER = 'M',
  	IMSGT_READY = 'r',
  
  	/* inform the coordinator about a database that needs vacuuming to