granicus.if.org Git - postgresql/blob - src/backend/commands/async.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * async.c--
   4  *        Asynchronous notification
   5  *
   6  * Copyright (c) 1994, Regents of the University of California
   7  *
   8  *
   9  * IDENTIFICATION
  10  *        $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.40 1998/09/01 04:27:42 momjian Exp $
  11  *
  12  *-------------------------------------------------------------------------
  13  */
  14 /* New Async Notification Model:
  15  * 1. Multiple backends on same machine.  Multiple backends listening on
  16  *        one relation.
  17  *
  18  * 2. One of the backend does a 'notify <relname>'.  For all backends that
  19  *        are listening to this relation (all notifications take place at the
  20  *        end of commit),
  21  *        2.a  If the process is the same as the backend process that issued
  22  *                 notification (we are notifying something that we are listening),
  23  *                 signal the corresponding frontend over the comm channel.
  24  *        2.b  For all other listening processes, we send kill(SIGUSR2) to wake up
  25  *                 the listening backend.
  26  * 3. Upon receiving a kill(SIGUSR2) signal from another backend process
  27  *        notifying that one of the relation that we are listening is being
  28  *        notified, we can be in either of two following states:
  29  *        3.a  We are sleeping, wake up and signal our frontend.
  30  *        3.b  We are in middle of another transaction, wait until the end of
  31  *                 of the current transaction and signal our frontend.
  32  * 4. Each frontend receives this notification and processes accordingly.
  33  *
  34  * -- jw, 12/28/93
  35  *
  36  */
  37
  38 #include <unistd.h>
  39 #include <signal.h>
  40 #include <string.h>
  41 #include <errno.h>
  42 #include <sys/types.h>                  /* Needed by in.h on Ultrix */
  43 #include <netinet/in.h>
  44
  45 #include "postgres.h"
  46
  47 #include "access/heapam.h"
  48 #include "access/relscan.h"
  49 #include "access/xact.h"
  50 #include "catalog/catname.h"
  51 #include "catalog/pg_listener.h"
  52 #include "commands/async.h"
  53 #include "fmgr.h"
  54 #include "lib/dllist.h"
  55 #include "libpq/libpq.h"
  56 #include "miscadmin.h"
  57 #include "nodes/memnodes.h"
  58 #include "storage/bufmgr.h"
  59 #include "storage/lmgr.h"
  60 #include "tcop/dest.h"
  61 #include "utils/mcxt.h"
  62 #include "utils/syscache.h"
  63 #include <utils/trace.h>
  64 #include <utils/ps_status.h>
  65
  66 #define NotifyUnlock pg_options[OPT_NOTIFYUNLOCK]
  67 #define NotifyHack       pg_options[OPT_NOTIFYHACK]
  68
  69 extern TransactionState CurrentTransactionState;
  70 extern CommandDest whereToSendOutput;
  71
  72 GlobalMemory notifyContext = NULL;
  73
  74 static int      notifyFrontEndPending = 0;
  75 static int      notifyIssued = 0;
  76 static Dllist *pendingNotifies = NULL;
  77
  78 static int      AsyncExistsPendingNotify(char *);
  79 static void ClearPendingNotify(void);
  80 static void Async_NotifyFrontEnd(void);
  81 static void Async_NotifyFrontEnd_Aux(void);
  82 void            Async_Unlisten(char *relname, int pid);
  83 static void Async_UnlistenOnExit(int code, char *relname);
  84 static void Async_UnlistenAll(void);
  85
  86 /*
  87  *--------------------------------------------------------------
  88  * Async_NotifyHandler --
  89  *
  90  *              This is the signal handler for SIGUSR2.  When the backend
  91  *              is signaled, the backend can be in two states.
  92  *              1. If the backend is in the middle of another transaction,
  93  *                 we set the flag, notifyFrontEndPending, and wait until
  94  *                 the end of the transaction to notify the front end.
  95  *              2. If the backend is not in the middle of another transaction,
  96  *                 we notify the front end immediately.
  97  *
  98  *              -- jw, 12/28/93
  99  * Results:
 100  *              none
 101  *
 102  * Side effects:
 103  *              none
 104  */
 105 void
 106 Async_NotifyHandler(SIGNAL_ARGS)
 107 {
 108         TPRINTF(TRACE_NOTIFY, "Async_NotifyHandler");
 109
 110         if ((CurrentTransactionState->state == TRANS_DEFAULT) &&
 111                 (CurrentTransactionState->blockState == TRANS_DEFAULT))
 112         {
 113                 TPRINTF(TRACE_NOTIFY, "Async_NotifyHandler: "
 114                                 "waking up sleeping backend process");
 115                 PS_SET_STATUS("async_notify");
 116                 Async_NotifyFrontEnd();
 117                 PS_SET_STATUS("idle");
 118         }
 119         else
 120         {
 121                 TPRINTF(TRACE_NOTIFY, "Async_NotifyHandler: "
 122                          "process in middle of transaction, state=%d, blockstate=%d",
 123                                 CurrentTransactionState->state,
 124                                 CurrentTransactionState->blockState);
 125                 notifyFrontEndPending = 1;
 126                 TPRINTF(TRACE_NOTIFY, "Async_NotifyHandler: notify frontend pending");
 127         }
 128
 129         TPRINTF(TRACE_NOTIFY, "Async_NotifyHandler: done");
 130 }
 131
 132 /*
 133  *--------------------------------------------------------------
 134  * Async_Notify --
 135  *
 136  *              This is executed by the SQL notify command.
 137  *
 138  *              Adds the relation to the list of pending notifies.
 139  *              All notification happens at end of commit.
 140  *              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 141  *
 142  *              All notification of backend processes happens here,
 143  *              then each backend notifies its corresponding front end at
 144  *              the end of commit.
 145  *
 146  *              -- jw, 12/28/93
 147  *
 148  * Results:
 149  *              XXX
 150  *
 151  * Side effects:
 152  *              All tuples for relname in pg_listener are updated.
 153  *
 154  *--------------------------------------------------------------
 155  */
 156 void
 157 Async_Notify(char *relname)
 158 {
 159
 160         HeapTuple       lTuple,
 161                                 rTuple;
 162         Relation        lRel;
 163         HeapScanDesc sRel;
 164         TupleDesc       tdesc;
 165         ScanKeyData key;
 166         Datum           d,
 167                                 value[3];
 168         bool            isnull;
 169         char            repl[3],
 170                                 nulls[3];
 171
 172         char       *notifyName;
 173
 174         TPRINTF(TRACE_NOTIFY, "Async_Notify: %s", relname);
 175
 176         if (!pendingNotifies)
 177                 pendingNotifies = DLNewList();
 178
 179         /*
 180          * Allocate memory from the global malloc pool because it needs to be
 181          * referenced also when the transaction is finished.  DZ - 26-08-1996
 182          */
 183         notifyName = strdup(relname);
 184         DLAddHead(pendingNotifies, DLNewElem(notifyName));
 185
 186         ScanKeyEntryInitialize(&key, 0,
 187                                                    Anum_pg_listener_relname,
 188                                                    F_NAMEEQ,
 189                                                    PointerGetDatum(notifyName));
 190
 191         lRel = heap_openr(ListenerRelationName);
 192         tdesc = RelationGetDescr(lRel);
 193         RelationSetLockForWrite(lRel);
 194         sRel = heap_beginscan(lRel, 0, SnapshotNow, 1, &key);
 195
 196         nulls[0] = nulls[1] = nulls[2] = ' ';
 197         repl[0] = repl[1] = repl[2] = ' ';
 198         repl[Anum_pg_listener_notify - 1] = 'r';
 199         value[0] = value[1] = value[2] = (Datum) 0;
 200         value[Anum_pg_listener_notify - 1] = Int32GetDatum(1);
 201
 202         while (HeapTupleIsValid(lTuple = heap_getnext(sRel, 0)))
 203         {
 204                 d = heap_getattr(lTuple, Anum_pg_listener_notify, tdesc, &isnull);
 205                 if (!DatumGetInt32(d))
 206                 {
 207                         rTuple = heap_modifytuple(lTuple, lRel, value, nulls, repl);
 208                         heap_replace(lRel, &lTuple->t_ctid, rTuple);
 209                         /* notify is really issued only if a tuple has been changed */
 210                         notifyIssued = 1;
 211                 }
 212         }
 213         heap_endscan(sRel);
 214
 215         /*
 216          * Note: if the write lock is unset we can get multiple tuples with
 217          * same oid if other backends notify the same relation. Use this
 218          * option at your own risk.
 219          */
 220         if (NotifyUnlock)
 221                 RelationUnsetLockForWrite(lRel);
 222
 223         heap_close(lRel);
 224
 225         TPRINTF(TRACE_NOTIFY, "Async_Notify: done %s", relname);
 226 }
 227
 228 /*
 229  *--------------------------------------------------------------
 230  * Async_NotifyAtCommit --
 231  *
 232  *              This is called at transaction commit.
 233  *
 234  *              Signal our corresponding frontend process on relations that
 235  *              were notified.  Signal all other backend process that
 236  *              are listening also.
 237  *
 238  *              -- jw, 12/28/93
 239  *
 240  * Results:
 241  *              XXX
 242  *
 243  * Side effects:
 244  *              Tuples in pg_listener that has our listenerpid are updated so
 245  *              that the notification is 0.  We do not want to notify frontend
 246  *              more than once.
 247  *
 248  *              -- jw, 12/28/93
 249  *
 250  *--------------------------------------------------------------
 251  */
 252 void
 253 Async_NotifyAtCommit()
 254 {
 255         HeapTuple       lTuple;
 256         Relation        lRel;
 257         HeapScanDesc sRel;
 258         TupleDesc       tdesc;
 259         ScanKeyData key;
 260         Datum           d;
 261         bool            isnull;
 262         extern TransactionState CurrentTransactionState;
 263
 264         if (!pendingNotifies)
 265                 pendingNotifies = DLNewList();
 266
 267         if ((CurrentTransactionState->state == TRANS_DEFAULT) &&
 268                 (CurrentTransactionState->blockState == TRANS_DEFAULT))
 269         {
 270                 if (notifyIssued)
 271                 {
 272                         /* 'notify <relname>' issued by us */
 273                         notifyIssued = 0;
 274                         StartTransactionCommand();
 275                         TPRINTF(TRACE_NOTIFY, "Async_NotifyAtCommit");
 276                         ScanKeyEntryInitialize(&key, 0,
 277                                                                    Anum_pg_listener_notify,
 278                                                                    F_INT4EQ,
 279                                                                    Int32GetDatum(1));
 280                         lRel = heap_openr(ListenerRelationName);
 281                         RelationSetLockForWrite(lRel);
 282                         sRel = heap_beginscan(lRel, 0, SnapshotNow, 1, &key);
 283                         tdesc = RelationGetDescr(lRel);
 284
 285                         while (HeapTupleIsValid(lTuple = heap_getnext(sRel, 0)))
 286                         {
 287                                 d = heap_getattr(lTuple, Anum_pg_listener_relname,
 288                                                                  tdesc, &isnull);
 289
 290                                 if (AsyncExistsPendingNotify((char *) DatumGetPointer(d)))
 291                                 {
 292                                         d = heap_getattr(lTuple, Anum_pg_listener_pid,
 293                                                                          tdesc, &isnull);
 294
 295                                         if (MyProcPid == DatumGetInt32(d))
 296                                         {
 297                                                 notifyFrontEndPending = 1;
 298                                                 TPRINTF(TRACE_NOTIFY,
 299                                                                 "Async_NotifyAtCommit: notifying self");
 300                                         }
 301                                         else
 302                                         {
 303                                                 TPRINTF(TRACE_NOTIFY,
 304                                                                 "Async_NotifyAtCommit: notifying pid %d",
 305                                                                 DatumGetInt32(d));
 306 #ifdef HAVE_KILL
 307                                                 if (kill(DatumGetInt32(d), SIGUSR2) < 0)
 308                                                 {
 309                                                         if (errno == ESRCH)
 310                                                                 heap_delete(lRel, &lTuple->t_ctid);
 311                                                 }
 312 #endif
 313                                         }
 314                                 }
 315                         }
 316                         heap_endscan(sRel);
 317                         heap_close(lRel);
 318
 319                         /*
 320                          * Notify the frontend inside the current transaction while we
 321                          * still have a valid write lock on pg_listeners. This avoid
 322                          * waiting until all other backends have finished with
 323                          * pg_listener.
 324                          */
 325                         if (notifyFrontEndPending)
 326                         {
 327                                 /* The aux version is called inside transaction */
 328                                 Async_NotifyFrontEnd_Aux();
 329                         }
 330
 331                         TPRINTF(TRACE_NOTIFY, "Async_NotifyAtCommit: done");
 332                         CommitTransactionCommand();
 333                 }
 334                 else
 335                 {
 336
 337                         /*
 338                          * No notifies issued by us. If notifyFrontEndPending has been
 339                          * set by Async_NotifyHandler notify the frontend of pending
 340                          * notifies from other backends.
 341                          */
 342                         if (notifyFrontEndPending)
 343                                 Async_NotifyFrontEnd();
 344                 }
 345
 346                 ClearPendingNotify();
 347         }
 348 }
 349
 350 /*
 351  *--------------------------------------------------------------
 352  * Async_NotifyAtAbort --
 353  *
 354  *              This is called at transaction commit.
 355  *
 356  *              Gets rid of pending notifies.  List elements are automatically
 357  *              freed through memory context.
 358  *
 359  *
 360  * Results:
 361  *              XXX
 362  *
 363  * Side effects:
 364  *              XXX
 365  *
 366  *--------------------------------------------------------------
 367  */
 368 void
 369 Async_NotifyAtAbort()
 370 {
 371         if (pendingNotifies)
 372         {
 373                 ClearPendingNotify();
 374                 DLFreeList(pendingNotifies);
 375         }
 376         pendingNotifies = DLNewList();
 377         notifyIssued = 0;
 378
 379         if ((CurrentTransactionState->state == TRANS_DEFAULT) &&
 380                 (CurrentTransactionState->blockState == TRANS_DEFAULT))
 381         {
 382                 /* don't forget to notify front end */
 383                 if (notifyFrontEndPending)
 384                         Async_NotifyFrontEnd();
 385         }
 386 }
 387
 388 /*
 389  *--------------------------------------------------------------
 390  * Async_Listen --
 391  *
 392  *              This is executed by the SQL listen command.
 393  *
 394  *              Register a backend (identified by its Unix PID) as listening
 395  *              on the specified relation.
 396  *
 397  *              One listener per relation, pg_listener relation is keyed
 398  *              on (relname,pid) to provide multiple listeners in future.
 399  *
 400  * Results:
 401  *              pg_listeners is updated.
 402  *
 403  * Side effects:
 404  *              XXX
 405  *
 406  *--------------------------------------------------------------
 407  */
 408 void
 409 Async_Listen(char *relname, int pid)
 410 {
 411         Datum           values[Natts_pg_listener];
 412         char            nulls[Natts_pg_listener];
 413         TupleDesc       tdesc;
 414         HeapScanDesc scan;
 415         HeapTuple       tuple,
 416                                 newtup;
 417         Relation        lDesc;
 418         Datum           d;
 419         int                     i;
 420         bool            isnull;
 421         int                     alreadyListener = 0;
 422         char       *relnamei;
 423         TupleDesc       tupDesc;
 424
 425         if (whereToSendOutput != Remote)
 426         {
 427                 elog(NOTICE, "Async_Listen: "
 428                          "listen not available on interactive sessions");
 429                 return;
 430         }
 431
 432         TPRINTF(TRACE_NOTIFY, "Async_Listen: %s", relname);
 433         for (i = 0; i < Natts_pg_listener; i++)
 434         {
 435                 nulls[i] = ' ';
 436                 values[i] = PointerGetDatum(NULL);
 437         }
 438
 439         i = 0;
 440         values[i++] = (Datum) relname;
 441         values[i++] = (Datum) pid;
 442         values[i++] = (Datum) 0;        /* no notifies pending */
 443
 444         lDesc = heap_openr(ListenerRelationName);
 445         RelationSetLockForWrite(lDesc);
 446
 447         /* is someone already listening.  One listener per relation */
 448         tdesc = RelationGetDescr(lDesc);
 449         scan = heap_beginscan(lDesc, 0, SnapshotNow, 0, (ScanKey) NULL);
 450         while (HeapTupleIsValid(tuple = heap_getnext(scan, 0)))
 451         {
 452                 d = heap_getattr(tuple, Anum_pg_listener_relname, tdesc,
 453                                                  &isnull);
 454                 relnamei = DatumGetPointer(d);
 455                 if (!strncmp(relnamei, relname, NAMEDATALEN))
 456                 {
 457                         d = heap_getattr(tuple, Anum_pg_listener_pid, tdesc, &isnull);
 458                         pid = DatumGetInt32(d);
 459                         if (pid == MyProcPid)
 460                                 alreadyListener = 1;
 461                 }
 462                 if (alreadyListener)
 463                 {
 464                         /* No need to scan the rest of the table */
 465                         break;
 466                 }
 467         }
 468         heap_endscan(scan);
 469
 470         if (alreadyListener)
 471         {
 472                 elog(NOTICE, "Async_Listen: We are already listening on %s",
 473                          relname);
 474                 RelationUnsetLockForWrite(lDesc);
 475                 heap_close(lDesc);
 476                 return;
 477         }
 478
 479         tupDesc = lDesc->rd_att;
 480         newtup = heap_formtuple(tupDesc, values, nulls);
 481         heap_insert(lDesc, newtup);
 482         pfree(newtup);
 483
 484         /*
 485          * if (alreadyListener) { elog(NOTICE,"Async_Listen: already one
 486          * listener on %s (possibly dead)",relname); }
 487          */
 488
 489         RelationUnsetLockForWrite(lDesc);
 490         heap_close(lDesc);
 491
 492         /*
 493          * now that we are listening, we should make a note to ourselves to
 494          * unlisten prior to dying.
 495          */
 496         relnamei = malloc(NAMEDATALEN);         /* persists to process exit */
 497         StrNCpy(relnamei, relname, NAMEDATALEN);
 498         on_shmem_exit(Async_UnlistenOnExit, (caddr_t) relnamei);
 499 }
 500
 501 /*
 502  *--------------------------------------------------------------
 503  * Async_Unlisten --
 504  *
 505  *              This is executed by the SQL unlisten command.
 506  *
 507  *              Remove the backend from the list of listening backends
 508  *              for the specified relation.
 509  *
 510  * Results:
 511  *              pg_listeners is updated.
 512  *
 513  * Side effects:
 514  *              XXX
 515  *
 516  *--------------------------------------------------------------
 517  */
 518 void
 519 Async_Unlisten(char *relname, int pid)
 520 {
 521         Relation        lDesc;
 522         HeapTuple       lTuple;
 523
 524         /* Handle specially the `unlisten "*"' command */
 525         if ((!relname) || (*relname == '\0') || (strcmp(relname, "*") == 0))
 526         {
 527                 Async_UnlistenAll();
 528                 return;
 529         }
 530
 531         TPRINTF(TRACE_NOTIFY, "Async_Unlisten %s", relname);
 532
 533         lTuple = SearchSysCacheTuple(LISTENREL, PointerGetDatum(relname),
 534                                                                  Int32GetDatum(pid),
 535                                                                  0, 0);
 536         if (lTuple != NULL)
 537         {
 538                 lDesc = heap_openr(ListenerRelationName);
 539                 RelationSetLockForWrite(lDesc);
 540                 heap_delete(lDesc, &lTuple->t_ctid);
 541                 RelationUnsetLockForWrite(lDesc);
 542                 heap_close(lDesc);
 543         }
 544 }
 545
 546 /*
 547  *--------------------------------------------------------------
 548  * Async_UnlistenAll --
 549  *
 550  *              Unlisten all relations for this backend.
 551  *
 552  * Results:
 553  *              pg_listeners is updated.
 554  *
 555  * Side effects:
 556  *              XXX
 557  *
 558  *--------------------------------------------------------------
 559  */
 560 static void
 561 Async_UnlistenAll()
 562 {
 563         HeapTuple       lTuple;
 564         Relation        lRel;
 565         HeapScanDesc sRel;
 566         TupleDesc       tdesc;
 567         ScanKeyData key[1];
 568
 569         TPRINTF(TRACE_NOTIFY, "Async_UnlistenAll");
 570         ScanKeyEntryInitialize(&key[0], 0,
 571                                                    Anum_pg_listener_pid,
 572                                                    F_INT4EQ,
 573                                                    Int32GetDatum(MyProcPid));
 574         lRel = heap_openr(ListenerRelationName);
 575         RelationSetLockForWrite(lRel);
 576         tdesc = RelationGetDescr(lRel);
 577         sRel = heap_beginscan(lRel, 0, SnapshotNow, 1, key);
 578
 579         while (HeapTupleIsValid(lTuple = heap_getnext(sRel, 0)))
 580                 heap_delete(lRel, &lTuple->t_ctid);
 581         heap_endscan(sRel);
 582         RelationUnsetLockForWrite(lRel);
 583         heap_close(lRel);
 584         TPRINTF(TRACE_NOTIFY, "Async_UnlistenAll: done");
 585 }
 586
 587 /*
 588  * --------------------------------------------------------------
 589  * Async_UnlistenOnExit --
 590  *
 591  *              This is called at backend exit for each registered listen.
 592  *
 593  * Results:
 594  *              XXX
 595  *
 596  * --------------------------------------------------------------
 597  */
 598 static void
 599 Async_UnlistenOnExit(int code,  /* from exitpg */
 600                                          char *relname)
 601 {
 602         Async_Unlisten((char *) relname, MyProcPid);
 603 }
 604
 605 /*
 606  * --------------------------------------------------------------
 607  * Async_NotifyFrontEnd --
 608  *
 609  *              This is called outside transactions. The real work is done
 610  *              by Async_NotifyFrontEnd_Aux().
 611  *
 612  * --------------------------------------------------------------
 613  */
 614 static void
 615 Async_NotifyFrontEnd()
 616 {
 617         StartTransactionCommand();
 618         Async_NotifyFrontEnd_Aux();
 619         CommitTransactionCommand();
 620 }
 621
 622 /*
 623  * --------------------------------------------------------------
 624  * Async_NotifyFrontEnd_Aux --
 625  *
 626  *              This must be called inside a transaction block.
 627  *
 628  *              Perform an asynchronous notification to front end over
 629  *              portal comm channel.  The name of the relation which contains the
 630  *              data is sent to the front end.
 631  *
 632  *              We remove the notification flag from the pg_listener tuple
 633  *              associated with our process.
 634  *
 635  * Results:
 636  *              XXX
 637  *
 638  * --------------------------------------------------------------
 639  */
 640 static void
 641 Async_NotifyFrontEnd_Aux()
 642 {
 643         HeapTuple       lTuple,
 644                                 rTuple;
 645         Relation        lRel;
 646         HeapScanDesc sRel;
 647         TupleDesc       tdesc;
 648         ScanKeyData key[2];
 649         Datum           d,
 650                                 value[3];
 651         char            repl[3],
 652                                 nulls[3];
 653         bool            isnull;
 654
 655 #define MAX_DONE 64
 656
 657         char       *done[MAX_DONE];
 658         int                     ndone = 0;
 659         int                     i;
 660
 661         notifyFrontEndPending = 0;
 662
 663         TPRINTF(TRACE_NOTIFY, "Async_NotifyFrontEnd");
 664         StartTransactionCommand();
 665         ScanKeyEntryInitialize(&key[0], 0,
 666                                                    Anum_pg_listener_notify,
 667                                                    F_INT4EQ,
 668                                                    Int32GetDatum(1));
 669         ScanKeyEntryInitialize(&key[1], 0,
 670                                                    Anum_pg_listener_pid,
 671                                                    F_INT4EQ,
 672                                                    Int32GetDatum(MyProcPid));
 673         lRel = heap_openr(ListenerRelationName);
 674         RelationSetLockForWrite(lRel);
 675         tdesc = RelationGetDescr(lRel);
 676         sRel = heap_beginscan(lRel, 0, SnapshotNow, 2, key);
 677
 678         nulls[0] = nulls[1] = nulls[2] = ' ';
 679         repl[0] = repl[1] = repl[2] = ' ';
 680         repl[Anum_pg_listener_notify - 1] = 'r';
 681         value[0] = value[1] = value[2] = (Datum) 0;
 682         value[Anum_pg_listener_notify - 1] = Int32GetDatum(0);
 683
 684         while (HeapTupleIsValid(lTuple = heap_getnext(sRel, 0)))
 685         {
 686                 d = heap_getattr(lTuple, Anum_pg_listener_relname, tdesc,
 687                                                  &isnull);
 688
 689                 /*
 690                  * This hack deletes duplicate tuples which can be left in the
 691                  * table if the NotifyUnlock option is set. I'm further
 692                  * investigating this.  -- dz
 693                  */
 694                 if (NotifyHack)
 695                 {
 696                         for (i = 0; i < ndone; i++)
 697                         {
 698                                 if (strcmp(DatumGetName(d)->data, done[i]) == 0)
 699                                 {
 700                                         TPRINTF(TRACE_NOTIFY,
 701                                                         "Async_NotifyFrontEnd: duplicate %s",
 702                                                         DatumGetName(d)->data);
 703                                         heap_delete(lRel, &lTuple->t_ctid);
 704                                         continue;
 705                                 }
 706                         }
 707                         if (ndone < MAX_DONE)
 708                                 done[ndone++] = pstrdup(DatumGetName(d)->data);
 709                 }
 710
 711                 rTuple = heap_modifytuple(lTuple, lRel, value, nulls, repl);
 712                 heap_replace(lRel, &lTuple->t_ctid, rTuple);
 713
 714                 /* notifying the front end */
 715                 TPRINTF(TRACE_NOTIFY, "Async_NotifyFrontEnd: notifying %s",
 716                                 DatumGetName(d)->data);
 717
 718                 if (whereToSendOutput == Remote)
 719                 {
 720                         pq_putnchar("A", 1);
 721                         pq_putint((int32) MyProcPid, sizeof(int32));
 722                         pq_putstr(DatumGetName(d)->data);
 723                         pq_flush();
 724                 }
 725         }
 726         heap_endscan(sRel);
 727         RelationUnsetLockForWrite(lRel);
 728         heap_close(lRel);
 729
 730         TPRINTF(TRACE_NOTIFY, "Async_NotifyFrontEnd: done");
 731 }
 732
 733 static int
 734 AsyncExistsPendingNotify(char *relname)
 735 {
 736         Dlelem     *p;
 737
 738         for (p = DLGetHead(pendingNotifies);
 739                  p != NULL;
 740                  p = DLGetSucc(p))
 741         {
 742                 /* Use NAMEDATALEN for relname comparison.        DZ - 26-08-1996 */
 743                 if (!strncmp((const char *) DLE_VAL(p), relname, NAMEDATALEN))
 744                         return 1;
 745         }
 746
 747         return 0;
 748 }
 749
 750 static void
 751 ClearPendingNotify()
 752 {
 753         Dlelem     *p;
 754
 755         while ((p = DLRemHead(pendingNotifies)) != NULL)
 756                 free(DLE_VAL(p));
 757 }