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

Generated by: LCOV version 1.13