parallel.h

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * parallel.h
 *    Infrastructure for launching parallel workers
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * src/include/access/parallel.h
 *
 *-------------------------------------------------------------------------
 */
 
#ifndef PARALLEL_H
#define PARALLEL_H
 
#include "access/xlogdefs.h"
#include "lib/ilist.h"
#include "postmaster/bgworker.h"
#include "storage/shm_mq.h"
#include "storage/shm_toc.h"
 
typedef void (*parallel_worker_main_type) (dsm_segment *seg, shm_toc *toc);
 
typedef struct ParallelWorkerInfo
{
    BackgroundWorkerHandle *bgwhandle;
    shm_mq_handle *error_mqh;
    int32       pid;
} ParallelWorkerInfo;
 
typedef struct ParallelContext
{
    dlist_node  node;
    SubTransactionId subid;
    int         nworkers;       /* Maximum number of workers to launch */
    int         nworkers_to_launch; /* Actual number of workers to launch */
    int         nworkers_launched;
    char       *library_name;
    char       *function_name;
    ErrorContextCallback *error_context_stack;
    shm_toc_estimator estimator;
    dsm_segment *seg;
    void       *private_memory;
    shm_toc    *toc;
    ParallelWorkerInfo *worker;
    int         nknown_attached_workers;
    bool       *known_attached_workers;
} ParallelContext;
 
typedef struct ParallelWorkerContext
{
    dsm_segment *seg;
    shm_toc    *toc;
} ParallelWorkerContext;
 
extern PGDLLIMPORT volatile sig_atomic_t ParallelMessagePending;
extern PGDLLIMPORT int ParallelWorkerNumber;
extern PGDLLIMPORT bool InitializingParallelWorker;
 
#define     IsParallelWorker()      (ParallelWorkerNumber >= 0)
 
extern ParallelContext *CreateParallelContext(const char *library_name,
                                              const char *function_name, int nworkers);
extern void InitializeParallelDSM(ParallelContext *pcxt);
extern void ReinitializeParallelDSM(ParallelContext *pcxt);
extern void ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch);
extern void LaunchParallelWorkers(ParallelContext *pcxt);
extern void WaitForParallelWorkersToAttach(ParallelContext *pcxt);
extern void WaitForParallelWorkersToFinish(ParallelContext *pcxt);
extern void DestroyParallelContext(ParallelContext *pcxt);
extern bool ParallelContextActive(void);
 
extern void HandleParallelMessageInterrupt(void);
extern void HandleParallelMessages(void);
extern void AtEOXact_Parallel(bool isCommit);
extern void AtEOSubXact_Parallel(bool isCommit, SubTransactionId mySubId);
extern void ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end);
 
extern void ParallelWorkerMain(Datum main_arg);
 
#endif                          /* PARALLEL_H */