Re: Background worker with Listen

Why not use libpq in worker? i.e. your worker works just like a pure PG client.

In my project, I uses worker in this way and it works well. I do not
use any backend API to access the database.

2016-04-21 15:51 GMT+08:00 Ihnat Peter | TSS Group a.s. <Ihnat(at)tssgroup(dot)sk>:
> I am trying to create background worker which listens to notifications and
> do some work after receiving one.
>
> I got 2 problems:
>
> - Worker is receiving notifications from every channel not only the
> registered channel (in my case “foo”)
>
> - Notifications are not logged in the server log – I cannot store
> the payloads for further work
>
> Any help is welcomed.
>
>
>
> Here is the code:
>
> PG_MODULE_MAGIC;
>
>
>
> void _PG_init(void);
>
> void _PG_fini(void);
>
>
>
> static volatile sig_atomic_t got_sigterm = false;
>
> static volatile sig_atomic_t got_sigusr1 = false;
>
> static char *notify_database = NULL;
>
> static emit_log_hook_type prev_log_hook = NULL;
>
>
>
> static void
>
> bgw_sigterm(SIGNAL_ARGS)
>
> {
>
> int save_errno = errno;
>
> got_sigterm = true;
>
> if (MyProc)
>
> SetLatch(&MyProc->procLatch);
>
> errno = save_errno;
>
> }
>
>
>
> static void
>
> bgw_sigusr1(SIGNAL_ARGS)
>
> {
>
> int save_errno = errno;
>
> got_sigusr1 = true;
>
> if (MyProc)
>
> SetLatch(&MyProc->procLatch);
>
> errno = save_errno;
>
> }
>
>
>
> static void
>
> notify_main(Datum main_arg)
>
> {
>
> pqsignal(SIGTERM, bgw_sigterm);
>
> pqsignal(SIGUSR1, bgw_sigusr1);
>
>
>
> BackgroundWorkerUnblockSignals();
>
> BackgroundWorkerInitializeConnection(notify_database, NULL);
>
> EnableNotifyInterrupt();
>
>
>
> pgstat_report_activity(STATE_RUNNING, "background_worker");
>
> StartTransactionCommand();
>
> Async_Listen("foo");
>
> CommitTransactionCommand();
>
> pgstat_report_activity(STATE_IDLE, NULL);
>
>
>
> while (!got_sigterm)
>
> {
>
> int rc;
>
>
>
> rc = WaitLatch(&MyProc->procLatch,
> WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 10000L);
>
> ResetLatch(&MyProc->procLatch);
>
>
>
> if (rc & WL_POSTMASTER_DEATH)
>
> proc_exit(1);
>
>
>
> if (got_sigusr1)
>
> {
>
> got_sigusr1 = false;
>
> elog(INFO, "
> background_worker: notification received");
>
> // DO SOME WORK WITH STORED
> NOTIFICATIONS
>
> }
>
>
>
> }
>
>
>
> elog(LOG, "background_worker: finished");
>
> proc_exit(0);
>
> }
>
>
>
> static void
>
> store_notification(ErrorData *edata)
>
> {
>
> // HERE STORE THE NOTIFICATION FROM SERVER LOG
>
>
>
> if (prev_log_hook)
>
> (*prev_log_hook) (edata);
>
> }
>
>
>
> void
>
> _PG_init(void)
>
> {
>
> BackgroundWorker worker;
>
> DefineCustomStringVariable("postgres", NULL, NULL,
> &notify_database,
>
> "postgres",
>
> PGC_POSTMASTER, 0, NULL, NULL,
> NULL);
>
>
>
> MemSet(&worker, 0, sizeof(BackgroundWorker));
>
> snprintf(worker.bgw_name, BGW_MAXLEN, "background_worker");
>
> worker.bgw_flags = BGWORKER_SHMEM_ACCESS |
> BGWORKER_BACKEND_DATABASE_CONNECTION;
>
> worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
>
> worker.bgw_main = notify_main;
>
> worker.bgw_restart_time = 10;
>
> worker.bgw_main_arg = (Datum) 0;
>
> worker.bgw_notify_pid = 0;
>
> RegisterBackgroundWorker(&worker);
>
>
>
> prev_log_hook = emit_log_hook;
>
> emit_log_hook = store_notification;
>
> }
>
>
>
> void
>
> _PG_fini(void)
>
> {
>
> emit_log_hook = prev_log_hook;
>
> }