LCOV - code coverage report
Current view: top level - src/backend/libpq - pqmq.c (source / functions) Hit Total Coverage
Test: PostgreSQL 16beta1 Lines: 87 137 63.5 %
Date: 2023-05-30 23:12:14 Functions: 6 10 60.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * pqmq.c
       4             :  *    Use the frontend/backend protocol for communication over a shm_mq
       5             :  *
       6             :  * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  *  src/backend/libpq/pqmq.c
      10             :  *
      11             :  *-------------------------------------------------------------------------
      12             :  */
      13             : 
      14             : #include "postgres.h"
      15             : 
      16             : #include "access/parallel.h"
      17             : #include "libpq/libpq.h"
      18             : #include "libpq/pqformat.h"
      19             : #include "libpq/pqmq.h"
      20             : #include "miscadmin.h"
      21             : #include "pgstat.h"
      22             : #include "replication/logicalworker.h"
      23             : #include "tcop/tcopprot.h"
      24             : #include "utils/builtins.h"
      25             : 
      26             : static shm_mq_handle *pq_mq_handle;
      27             : static bool pq_mq_busy = false;
      28             : static pid_t pq_mq_parallel_leader_pid = 0;
      29             : static pid_t pq_mq_parallel_leader_backend_id = InvalidBackendId;
      30             : 
      31             : static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg);
      32             : static void mq_comm_reset(void);
      33             : static int  mq_flush(void);
      34             : static int  mq_flush_if_writable(void);
      35             : static bool mq_is_send_pending(void);
      36             : static int  mq_putmessage(char msgtype, const char *s, size_t len);
      37             : static void mq_putmessage_noblock(char msgtype, const char *s, size_t len);
      38             : 
      39             : static const PQcommMethods PqCommMqMethods = {
      40             :     mq_comm_reset,
      41             :     mq_flush,
      42             :     mq_flush_if_writable,
      43             :     mq_is_send_pending,
      44             :     mq_putmessage,
      45             :     mq_putmessage_noblock
      46             : };
      47             : 
      48             : /*
      49             :  * Arrange to redirect frontend/backend protocol messages to a shared-memory
      50             :  * message queue.
      51             :  */
      52             : void
      53        2624 : pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
      54             : {
      55        2624 :     PqCommMethods = &PqCommMqMethods;
      56        2624 :     pq_mq_handle = mqh;
      57        2624 :     whereToSendOutput = DestRemote;
      58        2624 :     FrontendProtocol = PG_PROTOCOL_LATEST;
      59        2624 :     on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
      60        2624 : }
      61             : 
      62             : /*
      63             :  * When the DSM that contains our shm_mq goes away, we need to stop sending
      64             :  * messages to it.
      65             :  */
      66             : static void
      67        2624 : pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
      68             : {
      69        2624 :     pq_mq_handle = NULL;
      70        2624 :     whereToSendOutput = DestNone;
      71        2624 : }
      72             : 
      73             : /*
      74             :  * Arrange to SendProcSignal() to the parallel leader each time we transmit
      75             :  * message data via the shm_mq.
      76             :  */
      77             : void
      78        2624 : pq_set_parallel_leader(pid_t pid, BackendId backend_id)
      79             : {
      80             :     Assert(PqCommMethods == &PqCommMqMethods);
      81        2624 :     pq_mq_parallel_leader_pid = pid;
      82        2624 :     pq_mq_parallel_leader_backend_id = backend_id;
      83        2624 : }
      84             : 
      85             : static void
      86           0 : mq_comm_reset(void)
      87             : {
      88             :     /* Nothing to do. */
      89           0 : }
      90             : 
      91             : static int
      92          16 : mq_flush(void)
      93             : {
      94             :     /* Nothing to do. */
      95          16 :     return 0;
      96             : }
      97             : 
      98             : static int
      99           0 : mq_flush_if_writable(void)
     100             : {
     101             :     /* Nothing to do. */
     102           0 :     return 0;
     103             : }
     104             : 
     105             : static bool
     106           0 : mq_is_send_pending(void)
     107             : {
     108             :     /* There's never anything pending. */
     109           0 :     return 0;
     110             : }
     111             : 
     112             : /*
     113             :  * Transmit a libpq protocol message to the shared memory message queue
     114             :  * selected via pq_mq_handle.  We don't include a length word, because the
     115             :  * receiver will know the length of the message from shm_mq_receive().
     116             :  */
     117             : static int
     118        5218 : mq_putmessage(char msgtype, const char *s, size_t len)
     119             : {
     120             :     shm_mq_iovec iov[2];
     121             :     shm_mq_result result;
     122             : 
     123             :     /*
     124             :      * If we're sending a message, and we have to wait because the queue is
     125             :      * full, and then we get interrupted, and that interrupt results in trying
     126             :      * to send another message, we respond by detaching the queue.  There's no
     127             :      * way to return to the original context, but even if there were, just
     128             :      * queueing the message would amount to indefinitely postponing the
     129             :      * response to the interrupt.  So we do this instead.
     130             :      */
     131        5218 :     if (pq_mq_busy)
     132             :     {
     133           0 :         if (pq_mq_handle != NULL)
     134           0 :             shm_mq_detach(pq_mq_handle);
     135           0 :         pq_mq_handle = NULL;
     136           0 :         return EOF;
     137             :     }
     138             : 
     139             :     /*
     140             :      * If the message queue is already gone, just ignore the message. This
     141             :      * doesn't necessarily indicate a problem; for example, DEBUG messages can
     142             :      * be generated late in the shutdown sequence, after all DSMs have already
     143             :      * been detached.
     144             :      */
     145        5218 :     if (pq_mq_handle == NULL)
     146           0 :         return 0;
     147             : 
     148        5218 :     pq_mq_busy = true;
     149             : 
     150        5218 :     iov[0].data = &msgtype;
     151        5218 :     iov[0].len = 1;
     152        5218 :     iov[1].data = s;
     153        5218 :     iov[1].len = len;
     154             : 
     155             :     Assert(pq_mq_handle != NULL);
     156             : 
     157             :     for (;;)
     158             :     {
     159             :         /*
     160             :          * Immediately notify the receiver by passing force_flush as true so
     161             :          * that the shared memory value is updated before we send the parallel
     162             :          * message signal right after this.
     163             :          */
     164        5228 :         result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);
     165             : 
     166        5228 :         if (pq_mq_parallel_leader_pid != 0)
     167             :         {
     168        5228 :             if (IsLogicalParallelApplyWorker())
     169          10 :                 SendProcSignal(pq_mq_parallel_leader_pid,
     170             :                                PROCSIG_PARALLEL_APPLY_MESSAGE,
     171             :                                pq_mq_parallel_leader_backend_id);
     172             :             else
     173             :             {
     174             :                 Assert(IsParallelWorker());
     175        5218 :                 SendProcSignal(pq_mq_parallel_leader_pid,
     176             :                                PROCSIG_PARALLEL_MESSAGE,
     177             :                                pq_mq_parallel_leader_backend_id);
     178             :             }
     179             :         }
     180             : 
     181        5228 :         if (result != SHM_MQ_WOULD_BLOCK)
     182        5218 :             break;
     183             : 
     184          10 :         (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
     185             :                          WAIT_EVENT_MQ_PUT_MESSAGE);
     186          10 :         ResetLatch(MyLatch);
     187          10 :         CHECK_FOR_INTERRUPTS();
     188             :     }
     189             : 
     190        5218 :     pq_mq_busy = false;
     191             : 
     192             :     Assert(result == SHM_MQ_SUCCESS || result == SHM_MQ_DETACHED);
     193        5218 :     if (result != SHM_MQ_SUCCESS)
     194           4 :         return EOF;
     195        5214 :     return 0;
     196             : }
     197             : 
     198             : static void
     199           0 : mq_putmessage_noblock(char msgtype, const char *s, size_t len)
     200             : {
     201             :     /*
     202             :      * While the shm_mq machinery does support sending a message in
     203             :      * non-blocking mode, there's currently no way to try sending beginning to
     204             :      * send the message that doesn't also commit us to completing the
     205             :      * transmission.  This could be improved in the future, but for now we
     206             :      * don't need it.
     207             :      */
     208           0 :     elog(ERROR, "not currently supported");
     209             : }
     210             : 
     211             : /*
     212             :  * Parse an ErrorResponse or NoticeResponse payload and populate an ErrorData
     213             :  * structure with the results.
     214             :  */
     215             : void
     216          10 : pq_parse_errornotice(StringInfo msg, ErrorData *edata)
     217             : {
     218             :     /* Initialize edata with reasonable defaults. */
     219         240 :     MemSet(edata, 0, sizeof(ErrorData));
     220          10 :     edata->elevel = ERROR;
     221          10 :     edata->assoc_context = CurrentMemoryContext;
     222             : 
     223             :     /* Loop over fields and extract each one. */
     224             :     for (;;)
     225          82 :     {
     226          92 :         char        code = pq_getmsgbyte(msg);
     227             :         const char *value;
     228             : 
     229          92 :         if (code == '\0')
     230             :         {
     231          10 :             pq_getmsgend(msg);
     232          10 :             break;
     233             :         }
     234          82 :         value = pq_getmsgrawstring(msg);
     235             : 
     236          82 :         switch (code)
     237             :         {
     238          10 :             case PG_DIAG_SEVERITY:
     239             :                 /* ignore, trusting we'll get a nonlocalized version */
     240          10 :                 break;
     241          10 :             case PG_DIAG_SEVERITY_NONLOCALIZED:
     242          10 :                 if (strcmp(value, "DEBUG") == 0)
     243             :                 {
     244             :                     /*
     245             :                      * We can't reconstruct the exact DEBUG level, but
     246             :                      * presumably it was >= client_min_messages, so select
     247             :                      * DEBUG1 to ensure we'll pass it on to the client.
     248             :                      */
     249           0 :                     edata->elevel = DEBUG1;
     250             :                 }
     251          10 :                 else if (strcmp(value, "LOG") == 0)
     252             :                 {
     253             :                     /*
     254             :                      * It can't be LOG_SERVER_ONLY, or the worker wouldn't
     255             :                      * have sent it to us; so LOG is the correct value.
     256             :                      */
     257           0 :                     edata->elevel = LOG;
     258             :                 }
     259          10 :                 else if (strcmp(value, "INFO") == 0)
     260           0 :                     edata->elevel = INFO;
     261          10 :                 else if (strcmp(value, "NOTICE") == 0)
     262           0 :                     edata->elevel = NOTICE;
     263          10 :                 else if (strcmp(value, "WARNING") == 0)
     264           0 :                     edata->elevel = WARNING;
     265          10 :                 else if (strcmp(value, "ERROR") == 0)
     266          10 :                     edata->elevel = ERROR;
     267           0 :                 else if (strcmp(value, "FATAL") == 0)
     268           0 :                     edata->elevel = FATAL;
     269           0 :                 else if (strcmp(value, "PANIC") == 0)
     270           0 :                     edata->elevel = PANIC;
     271             :                 else
     272           0 :                     elog(ERROR, "unrecognized error severity: \"%s\"", value);
     273          10 :                 break;
     274          10 :             case PG_DIAG_SQLSTATE:
     275          10 :                 if (strlen(value) != 5)
     276           0 :                     elog(ERROR, "invalid SQLSTATE: \"%s\"", value);
     277          10 :                 edata->sqlerrcode = MAKE_SQLSTATE(value[0], value[1], value[2],
     278             :                                                   value[3], value[4]);
     279          10 :                 break;
     280          10 :             case PG_DIAG_MESSAGE_PRIMARY:
     281          10 :                 edata->message = pstrdup(value);
     282          10 :                 break;
     283           4 :             case PG_DIAG_MESSAGE_DETAIL:
     284           4 :                 edata->detail = pstrdup(value);
     285           4 :                 break;
     286           4 :             case PG_DIAG_MESSAGE_HINT:
     287           4 :                 edata->hint = pstrdup(value);
     288           4 :                 break;
     289           0 :             case PG_DIAG_STATEMENT_POSITION:
     290           0 :                 edata->cursorpos = pg_strtoint32(value);
     291           0 :                 break;
     292           0 :             case PG_DIAG_INTERNAL_POSITION:
     293           0 :                 edata->internalpos = pg_strtoint32(value);
     294           0 :                 break;
     295           0 :             case PG_DIAG_INTERNAL_QUERY:
     296           0 :                 edata->internalquery = pstrdup(value);
     297           0 :                 break;
     298           4 :             case PG_DIAG_CONTEXT:
     299           4 :                 edata->context = pstrdup(value);
     300           4 :                 break;
     301           0 :             case PG_DIAG_SCHEMA_NAME:
     302           0 :                 edata->schema_name = pstrdup(value);
     303           0 :                 break;
     304           0 :             case PG_DIAG_TABLE_NAME:
     305           0 :                 edata->table_name = pstrdup(value);
     306           0 :                 break;
     307           0 :             case PG_DIAG_COLUMN_NAME:
     308           0 :                 edata->column_name = pstrdup(value);
     309           0 :                 break;
     310           0 :             case PG_DIAG_DATATYPE_NAME:
     311           0 :                 edata->datatype_name = pstrdup(value);
     312           0 :                 break;
     313           0 :             case PG_DIAG_CONSTRAINT_NAME:
     314           0 :                 edata->constraint_name = pstrdup(value);
     315           0 :                 break;
     316          10 :             case PG_DIAG_SOURCE_FILE:
     317          10 :                 edata->filename = pstrdup(value);
     318          10 :                 break;
     319          10 :             case PG_DIAG_SOURCE_LINE:
     320          10 :                 edata->lineno = pg_strtoint32(value);
     321          10 :                 break;
     322          10 :             case PG_DIAG_SOURCE_FUNCTION:
     323          10 :                 edata->funcname = pstrdup(value);
     324          10 :                 break;
     325           0 :             default:
     326           0 :                 elog(ERROR, "unrecognized error field code: %d", (int) code);
     327             :                 break;
     328             :         }
     329             :     }
     330          10 : }

Generated by: LCOV version 1.14