granicus.if.org Git - postgresql/blob - src/backend/libpq/pqcomm.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * pqcomm.c
   4  *        Communication functions between the Frontend and the Backend
   5  *
   6  * These routines handle the low-level details of communication between
   7  * frontend and backend.  They just shove data across the communication
   8  * channel, and are ignorant of the semantics of the data --- or would be,
   9  * except for major brain damage in the design of the old COPY OUT protocol.
  10  * Unfortunately, COPY OUT was designed to commandeer the communication
  11  * channel (it just transfers data without wrapping it into messages).
  12  * No other messages can be sent while COPY OUT is in progress; and if the
  13  * copy is aborted by an ereport(ERROR), we need to close out the copy so that
  14  * the frontend gets back into sync.  Therefore, these routines have to be
  15  * aware of COPY OUT state.  (New COPY-OUT is message-based and does *not*
  16  * set the DoingCopyOut flag.)
  17  *
  18  * NOTE: generally, it's a bad idea to emit outgoing messages directly with
  19  * pq_putbytes(), especially if the message would require multiple calls
  20  * to send.  Instead, use the routines in pqformat.c to construct the message
  21  * in a buffer and then emit it in one call to pq_putmessage.  This ensures
  22  * that the channel will not be clogged by an incomplete message if execution
  23  * is aborted by ereport(ERROR) partway through the message.  The only
  24  * non-libpq code that should call pq_putbytes directly is old-style COPY OUT.
  25  *
  26  * At one time, libpq was shared between frontend and backend, but now
  27  * the backend's "backend/libpq" is quite separate from "interfaces/libpq".
  28  * All that remains is similarities of names to trap the unwary...
  29  *
  30  * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
  31  * Portions Copyright (c) 1994, Regents of the University of California
  32  *
  33  *      src/backend/libpq/pqcomm.c
  34  *
  35  *-------------------------------------------------------------------------
  36  */
  37
  38 /*------------------------
  39  * INTERFACE ROUTINES
  40  *
  41  * setup/teardown:
  42  *              StreamServerPort        - Open postmaster's server port
  43  *              StreamConnection        - Create new connection with client
  44  *              StreamClose                     - Close a client/backend connection
  45  *              TouchSocketFiles        - Protect socket files against /tmp cleaners
  46  *              pq_init                 - initialize libpq at backend startup
  47  *              pq_comm_reset   - reset libpq during error recovery
  48  *              pq_close                - shutdown libpq at backend exit
  49  *
  50  * low-level I/O:
  51  *              pq_getbytes             - get a known number of bytes from connection
  52  *              pq_getstring    - get a null terminated string from connection
  53  *              pq_getmessage   - get a message with length word from connection
  54  *              pq_getbyte              - get next byte from connection
  55  *              pq_peekbyte             - peek at next byte from connection
  56  *              pq_putbytes             - send bytes to connection (not flushed until pq_flush)
  57  *              pq_flush                - flush pending output
  58  *              pq_flush_if_writable - flush pending output if writable without blocking
  59  *              pq_getbyte_if_available - get a byte if available without blocking
  60  *
  61  * message-level I/O (and old-style-COPY-OUT cruft):
  62  *              pq_putmessage   - send a normal message (suppressed in COPY OUT mode)
  63  *              pq_putmessage_noblock - buffer a normal message (suppressed in COPY OUT)
  64  *              pq_startcopyout - inform libpq that a COPY OUT transfer is beginning
  65  *              pq_endcopyout   - end a COPY OUT transfer
  66  *
  67  *------------------------
  68  */
  69 #include "postgres.h"
  70
  71 #include <signal.h>
  72 #include <fcntl.h>
  73 #include <grp.h>
  74 #include <unistd.h>
  75 #include <sys/file.h>
  76 #include <sys/socket.h>
  77 #include <sys/stat.h>
  78 #include <sys/time.h>
  79 #include <netdb.h>
  80 #include <netinet/in.h>
  81 #ifdef HAVE_NETINET_TCP_H
  82 #include <netinet/tcp.h>
  83 #endif
  84 #include <arpa/inet.h>
  85 #ifdef HAVE_UTIME_H
  86 #include <utime.h>
  87 #endif
  88 #ifdef WIN32_ONLY_COMPILER              /* mstcpip.h is missing on mingw */
  89 #include <mstcpip.h>
  90 #endif
  91
  92 #include "libpq/ip.h"
  93 #include "libpq/libpq.h"
  94 #include "miscadmin.h"
  95 #include "storage/ipc.h"
  96 #include "utils/guc.h"
  97 #include "utils/memutils.h"
  98
  99 /*
 100  * Configuration options
 101  */
 102 int                     Unix_socket_permissions;
 103 char       *Unix_socket_group;
 104
 105 /* Where the Unix socket files are (list of palloc'd strings) */
 106 static List *sock_paths = NIL;
 107
 108 PQcommMethods *PqCommMethods;
 109
 110
 111 /*
 112  * Buffers for low-level I/O.
 113  *
 114  * The receive buffer is fixed size. Send buffer is usually 8k, but can be
 115  * enlarged by pq_putmessage_noblock() if the message doesn't fit otherwise.
 116  */
 117
 118 #define PQ_SEND_BUFFER_SIZE 8192
 119 #define PQ_RECV_BUFFER_SIZE 8192
 120
 121 static char *PqSendBuffer;
 122 static int      PqSendBufferSize;       /* Size send buffer */
 123 static int      PqSendPointer;          /* Next index to store a byte in PqSendBuffer */
 124 static int      PqSendStart;            /* Next index to send a byte in PqSendBuffer */
 125
 126 static char PqRecvBuffer[PQ_RECV_BUFFER_SIZE];
 127 static int      PqRecvPointer;          /* Next index to read a byte from PqRecvBuffer */
 128 static int      PqRecvLength;           /* End of data available in PqRecvBuffer */
 129
 130 /*
 131  * Message status
 132  */
 133 static bool PqCommBusy;
 134 static bool DoingCopyOut;
 135
 136
 137 /* Internal functions */
 138 static void socket_comm_reset(void);
 139 static void socket_close(int code, Datum arg);
 140 static void socket_set_nonblocking(bool nonblocking);
 141 static int      socket_flush(void);
 142 static int      socket_flush_if_writable(void);
 143 static bool socket_is_send_pending(void);
 144 static int      socket_putmessage(char msgtype, const char *s, size_t len);
 145 static void socket_putmessage_noblock(char msgtype, const char *s, size_t len);
 146 static void socket_startcopyout(void);
 147 static void socket_endcopyout(bool errorAbort);
 148 static int      internal_putbytes(const char *s, size_t len);
 149 static int      internal_flush(void);
 150 static void socket_set_nonblocking(bool nonblocking);
 151
 152 #ifdef HAVE_UNIX_SOCKETS
 153 static int      Lock_AF_UNIX(char *unixSocketDir, char *unixSocketPath);
 154 static int      Setup_AF_UNIX(char *sock_path);
 155 #endif   /* HAVE_UNIX_SOCKETS */
 156
 157 PQcommMethods PQcommSocketMethods;
 158
 159 static PQcommMethods PqCommSocketMethods = {
 160         socket_comm_reset,
 161         socket_flush,
 162         socket_flush_if_writable,
 163         socket_is_send_pending,
 164         socket_putmessage,
 165         socket_putmessage_noblock,
 166         socket_startcopyout,
 167         socket_endcopyout
 168 };
 169
 170
 171 /* --------------------------------
 172  *              pq_init - initialize libpq at backend startup
 173  * --------------------------------
 174  */
 175 void
 176 pq_init(void)
 177 {
 178         PqCommMethods = &PqCommSocketMethods;
 179         PqSendBufferSize = PQ_SEND_BUFFER_SIZE;
 180         PqSendBuffer = MemoryContextAlloc(TopMemoryContext, PqSendBufferSize);
 181         PqSendPointer = PqSendStart = PqRecvPointer = PqRecvLength = 0;
 182         PqCommBusy = false;
 183         DoingCopyOut = false;
 184         on_proc_exit(socket_close, 0);
 185 }
 186
 187 /* --------------------------------
 188  *              socket_comm_reset - reset libpq during error recovery
 189  *
 190  * This is called from error recovery at the outer idle loop.  It's
 191  * just to get us out of trouble if we somehow manage to elog() from
 192  * inside a pqcomm.c routine (which ideally will never happen, but...)
 193  * --------------------------------
 194  */
 195 static void
 196 socket_comm_reset(void)
 197 {
 198         /* Do not throw away pending data, but do reset the busy flag */
 199         PqCommBusy = false;
 200         /* We can abort any old-style COPY OUT, too */
 201         pq_endcopyout(true);
 202 }
 203
 204 /* --------------------------------
 205  *              socket_close - shutdown libpq at backend exit
 206  *
 207  * Note: in a standalone backend MyProcPort will be null,
 208  * don't crash during exit...
 209  * --------------------------------
 210  */
 211 static void
 212 socket_close(int code, Datum arg)
 213 {
 214         if (MyProcPort != NULL)
 215         {
 216 #if defined(ENABLE_GSS) || defined(ENABLE_SSPI)
 217 #ifdef ENABLE_GSS
 218                 OM_uint32       min_s;
 219
 220                 /* Shutdown GSSAPI layer */
 221                 if (MyProcPort->gss->ctx != GSS_C_NO_CONTEXT)
 222                         gss_delete_sec_context(&min_s, &MyProcPort->gss->ctx, NULL);
 223
 224                 if (MyProcPort->gss->cred != GSS_C_NO_CREDENTIAL)
 225                         gss_release_cred(&min_s, &MyProcPort->gss->cred);
 226 #endif   /* ENABLE_GSS */
 227                 /* GSS and SSPI share the port->gss struct */
 228
 229                 free(MyProcPort->gss);
 230 #endif   /* ENABLE_GSS || ENABLE_SSPI */
 231
 232                 /* Cleanly shut down SSL layer */
 233                 secure_close(MyProcPort);
 234
 235                 /*
 236                  * Formerly we did an explicit close() here, but it seems better to
 237                  * leave the socket open until the process dies.  This allows clients
 238                  * to perform a "synchronous close" if they care --- wait till the
 239                  * transport layer reports connection closure, and you can be sure the
 240                  * backend has exited.
 241                  *
 242                  * We do set sock to PGINVALID_SOCKET to prevent any further I/O,
 243                  * though.
 244                  */
 245                 MyProcPort->sock = PGINVALID_SOCKET;
 246         }
 247 }
 248
 249
 250
 251 /*
 252  * Streams -- wrapper around Unix socket system calls
 253  *
 254  *
 255  *              Stream functions are used for vanilla TCP connection protocol.
 256  */
 257
 258
 259 /* StreamDoUnlink()
 260  * Shutdown routine for backend connection
 261  * If any Unix sockets are used for communication, explicitly close them.
 262  */
 263 #ifdef HAVE_UNIX_SOCKETS
 264 static void
 265 StreamDoUnlink(int code, Datum arg)
 266 {
 267         ListCell   *l;
 268
 269         /* Loop through all created sockets... */
 270         foreach(l, sock_paths)
 271         {
 272                 char       *sock_path = (char *) lfirst(l);
 273
 274                 unlink(sock_path);
 275         }
 276         /* Since we're about to exit, no need to reclaim storage */
 277         sock_paths = NIL;
 278 }
 279 #endif   /* HAVE_UNIX_SOCKETS */
 280
 281 /*
 282  * StreamServerPort -- open a "listening" port to accept connections.
 283  *
 284  * family should be AF_UNIX or AF_UNSPEC; portNumber is the port number.
 285  * For AF_UNIX ports, hostName should be NULL and unixSocketDir must be
 286  * specified.  For TCP ports, hostName is either NULL for all interfaces or
 287  * the interface to listen on, and unixSocketDir is ignored (can be NULL).
 288  *
 289  * Successfully opened sockets are added to the ListenSocket[] array (of
 290  * length MaxListen), at the first position that isn't PGINVALID_SOCKET.
 291  *
 292  * RETURNS: STATUS_OK or STATUS_ERROR
 293  */
 294
 295 int
 296 StreamServerPort(int family, char *hostName, unsigned short portNumber,
 297                                  char *unixSocketDir,
 298                                  pgsocket ListenSocket[], int MaxListen)
 299 {
 300         pgsocket        fd;
 301         int                     err;
 302         int                     maxconn;
 303         int                     ret;
 304         char            portNumberStr[32];
 305         const char *familyDesc;
 306         char            familyDescBuf[64];
 307         char       *service;
 308         struct addrinfo *addrs = NULL,
 309                            *addr;
 310         struct addrinfo hint;
 311         int                     listen_index = 0;
 312         int                     added = 0;
 313
 314 #ifdef HAVE_UNIX_SOCKETS
 315         char            unixSocketPath[MAXPGPATH];
 316 #endif
 317 #if !defined(WIN32) || defined(IPV6_V6ONLY)
 318         int                     one = 1;
 319 #endif
 320
 321         /* Initialize hint structure */
 322         MemSet(&hint, 0, sizeof(hint));
 323         hint.ai_family = family;
 324         hint.ai_flags = AI_PASSIVE;
 325         hint.ai_socktype = SOCK_STREAM;
 326
 327 #ifdef HAVE_UNIX_SOCKETS
 328         if (family == AF_UNIX)
 329         {
 330                 /*
 331                  * Create unixSocketPath from portNumber and unixSocketDir and lock
 332                  * that file path
 333                  */
 334                 UNIXSOCK_PATH(unixSocketPath, portNumber, unixSocketDir);
 335                 if (strlen(unixSocketPath) >= UNIXSOCK_PATH_BUFLEN)
 336                 {
 337                         ereport(LOG,
 338                                         (errmsg("Unix-domain socket path \"%s\" is too long (maximum %d bytes)",
 339                                                         unixSocketPath,
 340                                                         (int) (UNIXSOCK_PATH_BUFLEN - 1))));
 341                         return STATUS_ERROR;
 342                 }
 343                 if (Lock_AF_UNIX(unixSocketDir, unixSocketPath) != STATUS_OK)
 344                         return STATUS_ERROR;
 345                 service = unixSocketPath;
 346         }
 347         else
 348 #endif   /* HAVE_UNIX_SOCKETS */
 349         {
 350                 snprintf(portNumberStr, sizeof(portNumberStr), "%d", portNumber);
 351                 service = portNumberStr;
 352         }
 353
 354         ret = pg_getaddrinfo_all(hostName, service, &hint, &addrs);
 355         if (ret || !addrs)
 356         {
 357                 if (hostName)
 358                         ereport(LOG,
 359                                         (errmsg("could not translate host name \"%s\", service \"%s\" to address: %s",
 360                                                         hostName, service, gai_strerror(ret))));
 361                 else
 362                         ereport(LOG,
 363                                  (errmsg("could not translate service \"%s\" to address: %s",
 364                                                  service, gai_strerror(ret))));
 365                 if (addrs)
 366                         pg_freeaddrinfo_all(hint.ai_family, addrs);
 367                 return STATUS_ERROR;
 368         }
 369
 370         for (addr = addrs; addr; addr = addr->ai_next)
 371         {
 372                 if (!IS_AF_UNIX(family) && IS_AF_UNIX(addr->ai_family))
 373                 {
 374                         /*
 375                          * Only set up a unix domain socket when they really asked for it.
 376                          * The service/port is different in that case.
 377                          */
 378                         continue;
 379                 }
 380
 381                 /* See if there is still room to add 1 more socket. */
 382                 for (; listen_index < MaxListen; listen_index++)
 383                 {
 384                         if (ListenSocket[listen_index] == PGINVALID_SOCKET)
 385                                 break;
 386                 }
 387                 if (listen_index >= MaxListen)
 388                 {
 389                         ereport(LOG,
 390                                         (errmsg("could not bind to all requested addresses: MAXLISTEN (%d) exceeded",
 391                                                         MaxListen)));
 392                         break;
 393                 }
 394
 395                 /* set up family name for possible error messages */
 396                 switch (addr->ai_family)
 397                 {
 398                         case AF_INET:
 399                                 familyDesc = _("IPv4");
 400                                 break;
 401 #ifdef HAVE_IPV6
 402                         case AF_INET6:
 403                                 familyDesc = _("IPv6");
 404                                 break;
 405 #endif
 406 #ifdef HAVE_UNIX_SOCKETS
 407                         case AF_UNIX:
 408                                 familyDesc = _("Unix");
 409                                 break;
 410 #endif
 411                         default:
 412                                 snprintf(familyDescBuf, sizeof(familyDescBuf),
 413                                                  _("unrecognized address family %d"),
 414                                                  addr->ai_family);
 415                                 familyDesc = familyDescBuf;
 416                                 break;
 417                 }
 418
 419                 if ((fd = socket(addr->ai_family, SOCK_STREAM, 0)) == PGINVALID_SOCKET)
 420                 {
 421                         ereport(LOG,
 422                                         (errcode_for_socket_access(),
 423                         /* translator: %s is IPv4, IPv6, or Unix */
 424                                          errmsg("could not create %s socket: %m",
 425                                                         familyDesc)));
 426                         continue;
 427                 }
 428
 429 #ifndef WIN32
 430
 431                 /*
 432                  * Without the SO_REUSEADDR flag, a new postmaster can't be started
 433                  * right away after a stop or crash, giving "address already in use"
 434                  * error on TCP ports.
 435                  *
 436                  * On win32, however, this behavior only happens if the
 437                  * SO_EXLUSIVEADDRUSE is set. With SO_REUSEADDR, win32 allows multiple
 438                  * servers to listen on the same address, resulting in unpredictable
 439                  * behavior. With no flags at all, win32 behaves as Unix with
 440                  * SO_REUSEADDR.
 441                  */
 442                 if (!IS_AF_UNIX(addr->ai_family))
 443                 {
 444                         if ((setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
 445                                                         (char *) &one, sizeof(one))) == -1)
 446                         {
 447                                 ereport(LOG,
 448                                                 (errcode_for_socket_access(),
 449                                                  errmsg("setsockopt(SO_REUSEADDR) failed: %m")));
 450                                 closesocket(fd);
 451                                 continue;
 452                         }
 453                 }
 454 #endif
 455
 456 #ifdef IPV6_V6ONLY
 457                 if (addr->ai_family == AF_INET6)
 458                 {
 459                         if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
 460                                                    (char *) &one, sizeof(one)) == -1)
 461                         {
 462                                 ereport(LOG,
 463                                                 (errcode_for_socket_access(),
 464                                                  errmsg("setsockopt(IPV6_V6ONLY) failed: %m")));
 465                                 closesocket(fd);
 466                                 continue;
 467                         }
 468                 }
 469 #endif
 470
 471                 /*
 472                  * Note: This might fail on some OS's, like Linux older than
 473                  * 2.4.21-pre3, that don't have the IPV6_V6ONLY socket option, and map
 474                  * ipv4 addresses to ipv6.  It will show ::ffff:ipv4 for all ipv4
 475                  * connections.
 476                  */
 477                 err = bind(fd, addr->ai_addr, addr->ai_addrlen);
 478                 if (err < 0)
 479                 {
 480                         ereport(LOG,
 481                                         (errcode_for_socket_access(),
 482                         /* translator: %s is IPv4, IPv6, or Unix */
 483                                          errmsg("could not bind %s socket: %m",
 484                                                         familyDesc),
 485                                          (IS_AF_UNIX(addr->ai_family)) ?
 486                                   errhint("Is another postmaster already running on port %d?"
 487                                                   " If not, remove socket file \"%s\" and retry.",
 488                                                   (int) portNumber, service) :
 489                                   errhint("Is another postmaster already running on port %d?"
 490                                                   " If not, wait a few seconds and retry.",
 491                                                   (int) portNumber)));
 492                         closesocket(fd);
 493                         continue;
 494                 }
 495
 496 #ifdef HAVE_UNIX_SOCKETS
 497                 if (addr->ai_family == AF_UNIX)
 498                 {
 499                         if (Setup_AF_UNIX(service) != STATUS_OK)
 500                         {
 501                                 closesocket(fd);
 502                                 break;
 503                         }
 504                 }
 505 #endif
 506
 507                 /*
 508                  * Select appropriate accept-queue length limit.  PG_SOMAXCONN is only
 509                  * intended to provide a clamp on the request on platforms where an
 510                  * overly large request provokes a kernel error (are there any?).
 511                  */
 512                 maxconn = MaxBackends * 2;
 513                 if (maxconn > PG_SOMAXCONN)
 514                         maxconn = PG_SOMAXCONN;
 515
 516                 err = listen(fd, maxconn);
 517                 if (err < 0)
 518                 {
 519                         ereport(LOG,
 520                                         (errcode_for_socket_access(),
 521                         /* translator: %s is IPv4, IPv6, or Unix */
 522                                          errmsg("could not listen on %s socket: %m",
 523                                                         familyDesc)));
 524                         closesocket(fd);
 525                         continue;
 526                 }
 527                 ListenSocket[listen_index] = fd;
 528                 added++;
 529         }
 530
 531         pg_freeaddrinfo_all(hint.ai_family, addrs);
 532
 533         if (!added)
 534                 return STATUS_ERROR;
 535
 536         return STATUS_OK;
 537 }
 538
 539
 540 #ifdef HAVE_UNIX_SOCKETS
 541
 542 /*
 543  * Lock_AF_UNIX -- configure unix socket file path
 544  */
 545 static int
 546 Lock_AF_UNIX(char *unixSocketDir, char *unixSocketPath)
 547 {
 548         /*
 549          * Grab an interlock file associated with the socket file.
 550          *
 551          * Note: there are two reasons for using a socket lock file, rather than
 552          * trying to interlock directly on the socket itself.  First, it's a lot
 553          * more portable, and second, it lets us remove any pre-existing socket
 554          * file without race conditions.
 555          */
 556         CreateSocketLockFile(unixSocketPath, true, unixSocketDir);
 557
 558         /*
 559          * Once we have the interlock, we can safely delete any pre-existing
 560          * socket file to avoid failure at bind() time.
 561          */
 562         unlink(unixSocketPath);
 563
 564         /*
 565          * Arrange to unlink the socket file(s) at proc_exit.  If this is the
 566          * first one, set up the on_proc_exit function to do it; then add this
 567          * socket file to the list of files to unlink.
 568          */
 569         if (sock_paths == NIL)
 570                 on_proc_exit(StreamDoUnlink, 0);
 571
 572         sock_paths = lappend(sock_paths, pstrdup(unixSocketPath));
 573
 574         return STATUS_OK;
 575 }
 576
 577
 578 /*
 579  * Setup_AF_UNIX -- configure unix socket permissions
 580  */
 581 static int
 582 Setup_AF_UNIX(char *sock_path)
 583 {
 584         /*
 585          * Fix socket ownership/permission if requested.  Note we must do this
 586          * before we listen() to avoid a window where unwanted connections could
 587          * get accepted.
 588          */
 589         Assert(Unix_socket_group);
 590         if (Unix_socket_group[0] != '\0')
 591         {
 592 #ifdef WIN32
 593                 elog(WARNING, "configuration item unix_socket_group is not supported on this platform");
 594 #else
 595                 char       *endptr;
 596                 unsigned long val;
 597                 gid_t           gid;
 598
 599                 val = strtoul(Unix_socket_group, &endptr, 10);
 600                 if (*endptr == '\0')
 601                 {                                               /* numeric group id */
 602                         gid = val;
 603                 }
 604                 else
 605                 {                                               /* convert group name to id */
 606                         struct group *gr;
 607
 608                         gr = getgrnam(Unix_socket_group);
 609                         if (!gr)
 610                         {
 611                                 ereport(LOG,
 612                                                 (errmsg("group \"%s\" does not exist",
 613                                                                 Unix_socket_group)));
 614                                 return STATUS_ERROR;
 615                         }
 616                         gid = gr->gr_gid;
 617                 }
 618                 if (chown(sock_path, -1, gid) == -1)
 619                 {
 620                         ereport(LOG,
 621                                         (errcode_for_file_access(),
 622                                          errmsg("could not set group of file \"%s\": %m",
 623                                                         sock_path)));
 624                         return STATUS_ERROR;
 625                 }
 626 #endif
 627         }
 628
 629         if (chmod(sock_path, Unix_socket_permissions) == -1)
 630         {
 631                 ereport(LOG,
 632                                 (errcode_for_file_access(),
 633                                  errmsg("could not set permissions of file \"%s\": %m",
 634                                                 sock_path)));
 635                 return STATUS_ERROR;
 636         }
 637         return STATUS_OK;
 638 }
 639 #endif   /* HAVE_UNIX_SOCKETS */
 640
 641
 642 /*
 643  * StreamConnection -- create a new connection with client using
 644  *              server port.  Set port->sock to the FD of the new connection.
 645  *
 646  * ASSUME: that this doesn't need to be non-blocking because
 647  *              the Postmaster uses select() to tell when the server master
 648  *              socket is ready for accept().
 649  *
 650  * RETURNS: STATUS_OK or STATUS_ERROR
 651  */
 652 int
 653 StreamConnection(pgsocket server_fd, Port *port)
 654 {
 655         /* accept connection and fill in the client (remote) address */
 656         port->raddr.salen = sizeof(port->raddr.addr);
 657         if ((port->sock = accept(server_fd,
 658                                                          (struct sockaddr *) & port->raddr.addr,
 659                                                          &port->raddr.salen)) == PGINVALID_SOCKET)
 660         {
 661                 ereport(LOG,
 662                                 (errcode_for_socket_access(),
 663                                  errmsg("could not accept new connection: %m")));
 664
 665                 /*
 666                  * If accept() fails then postmaster.c will still see the server
 667                  * socket as read-ready, and will immediately try again.  To avoid
 668                  * uselessly sucking lots of CPU, delay a bit before trying again.
 669                  * (The most likely reason for failure is being out of kernel file
 670                  * table slots; we can do little except hope some will get freed up.)
 671                  */
 672                 pg_usleep(100000L);             /* wait 0.1 sec */
 673                 return STATUS_ERROR;
 674         }
 675
 676 #ifdef SCO_ACCEPT_BUG
 677
 678         /*
 679          * UnixWare 7+ and OpenServer 5.0.4 are known to have this bug, but it
 680          * shouldn't hurt to catch it for all versions of those platforms.
 681          */
 682         if (port->raddr.addr.ss_family == 0)
 683                 port->raddr.addr.ss_family = AF_UNIX;
 684 #endif
 685
 686         /* fill in the server (local) address */
 687         port->laddr.salen = sizeof(port->laddr.addr);
 688         if (getsockname(port->sock,
 689                                         (struct sockaddr *) & port->laddr.addr,
 690                                         &port->laddr.salen) < 0)
 691         {
 692                 elog(LOG, "getsockname() failed: %m");
 693                 return STATUS_ERROR;
 694         }
 695
 696         /* select NODELAY and KEEPALIVE options if it's a TCP connection */
 697         if (!IS_AF_UNIX(port->laddr.addr.ss_family))
 698         {
 699                 int                     on;
 700
 701 #ifdef  TCP_NODELAY
 702                 on = 1;
 703                 if (setsockopt(port->sock, IPPROTO_TCP, TCP_NODELAY,
 704                                            (char *) &on, sizeof(on)) < 0)
 705                 {
 706                         elog(LOG, "setsockopt(TCP_NODELAY) failed: %m");
 707                         return STATUS_ERROR;
 708                 }
 709 #endif
 710                 on = 1;
 711                 if (setsockopt(port->sock, SOL_SOCKET, SO_KEEPALIVE,
 712                                            (char *) &on, sizeof(on)) < 0)
 713                 {
 714                         elog(LOG, "setsockopt(SO_KEEPALIVE) failed: %m");
 715                         return STATUS_ERROR;
 716                 }
 717
 718 #ifdef WIN32
 719
 720                 /*
 721                  * This is a Win32 socket optimization.  The ideal size is 32k.
 722                  * http://support.microsoft.com/kb/823764/EN-US/
 723                  */
 724                 on = PQ_SEND_BUFFER_SIZE * 4;
 725                 if (setsockopt(port->sock, SOL_SOCKET, SO_SNDBUF, (char *) &on,
 726                                            sizeof(on)) < 0)
 727                 {
 728                         elog(LOG, "setsockopt(SO_SNDBUF) failed: %m");
 729                         return STATUS_ERROR;
 730                 }
 731 #endif
 732
 733                 /*
 734                  * Also apply the current keepalive parameters.  If we fail to set a
 735                  * parameter, don't error out, because these aren't universally
 736                  * supported.  (Note: you might think we need to reset the GUC
 737                  * variables to 0 in such a case, but it's not necessary because the
 738                  * show hooks for these variables report the truth anyway.)
 739                  */
 740                 (void) pq_setkeepalivesidle(tcp_keepalives_idle, port);
 741                 (void) pq_setkeepalivesinterval(tcp_keepalives_interval, port);
 742                 (void) pq_setkeepalivescount(tcp_keepalives_count, port);
 743         }
 744
 745         return STATUS_OK;
 746 }
 747
 748 /*
 749  * StreamClose -- close a client/backend connection
 750  *
 751  * NOTE: this is NOT used to terminate a session; it is just used to release
 752  * the file descriptor in a process that should no longer have the socket
 753  * open.  (For example, the postmaster calls this after passing ownership
 754  * of the connection to a child process.)  It is expected that someone else
 755  * still has the socket open.  So, we only want to close the descriptor,
 756  * we do NOT want to send anything to the far end.
 757  */
 758 void
 759 StreamClose(pgsocket sock)
 760 {
 761         closesocket(sock);
 762 }
 763
 764 /*
 765  * TouchSocketFiles -- mark socket files as recently accessed
 766  *
 767  * This routine should be called every so often to ensure that the socket
 768  * files have a recent mod date (ordinary operations on sockets usually won't
 769  * change the mod date).  That saves them from being removed by
 770  * overenthusiastic /tmp-directory-cleaner daemons.  (Another reason we should
 771  * never have put the socket file in /tmp...)
 772  */
 773 void
 774 TouchSocketFiles(void)
 775 {
 776         ListCell   *l;
 777
 778         /* Loop through all created sockets... */
 779         foreach(l, sock_paths)
 780         {
 781                 char       *sock_path = (char *) lfirst(l);
 782
 783                 /*
 784                  * utime() is POSIX standard, utimes() is a common alternative. If we
 785                  * have neither, there's no way to affect the mod or access time of
 786                  * the socket :-(
 787                  *
 788                  * In either path, we ignore errors; there's no point in complaining.
 789                  */
 790 #ifdef HAVE_UTIME
 791                 utime(sock_path, NULL);
 792 #else                                                   /* !HAVE_UTIME */
 793 #ifdef HAVE_UTIMES
 794                 utimes(sock_path, NULL);
 795 #endif   /* HAVE_UTIMES */
 796 #endif   /* HAVE_UTIME */
 797         }
 798 }
 799
 800
 801 /* --------------------------------
 802  * Low-level I/O routines begin here.
 803  *
 804  * These routines communicate with a frontend client across a connection
 805  * already established by the preceding routines.
 806  * --------------------------------
 807  */
 808
 809 /* --------------------------------
 810  *                        socket_set_nonblocking - set socket blocking/non-blocking
 811  *
 812  * Sets the socket non-blocking if nonblocking is TRUE, or sets it
 813  * blocking otherwise.
 814  * --------------------------------
 815  */
 816 static void
 817 socket_set_nonblocking(bool nonblocking)
 818 {
 819         if (MyProcPort == NULL)
 820                 ereport(ERROR,
 821                                 (errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
 822                                  errmsg("there is no client connection")));
 823
 824         if (MyProcPort->noblock == nonblocking)
 825                 return;
 826
 827 #ifdef WIN32
 828         pgwin32_noblock = nonblocking ? 1 : 0;
 829 #else
 830
 831         /*
 832          * Use COMMERROR on failure, because ERROR would try to send the error to
 833          * the client, which might require changing the mode again, leading to
 834          * infinite recursion.
 835          */
 836         if (nonblocking)
 837         {
 838                 if (!pg_set_noblock(MyProcPort->sock))
 839                         ereport(COMMERROR,
 840                                         (errmsg("could not set socket to nonblocking mode: %m")));
 841         }
 842         else
 843         {
 844                 if (!pg_set_block(MyProcPort->sock))
 845                         ereport(COMMERROR,
 846                                         (errmsg("could not set socket to blocking mode: %m")));
 847         }
 848 #endif
 849         MyProcPort->noblock = nonblocking;
 850 }
 851
 852 /* --------------------------------
 853  *              pq_recvbuf - load some bytes into the input buffer
 854  *
 855  *              returns 0 if OK, EOF if trouble
 856  * --------------------------------
 857  */
 858 static int
 859 pq_recvbuf(void)
 860 {
 861         if (PqRecvPointer > 0)
 862         {
 863                 if (PqRecvLength > PqRecvPointer)
 864                 {
 865                         /* still some unread data, left-justify it in the buffer */
 866                         memmove(PqRecvBuffer, PqRecvBuffer + PqRecvPointer,
 867                                         PqRecvLength - PqRecvPointer);
 868                         PqRecvLength -= PqRecvPointer;
 869                         PqRecvPointer = 0;
 870                 }
 871                 else
 872                         PqRecvLength = PqRecvPointer = 0;
 873         }
 874
 875         /* Ensure that we're in blocking mode */
 876         socket_set_nonblocking(false);
 877
 878         /* Can fill buffer from PqRecvLength and upwards */
 879         for (;;)
 880         {
 881                 int                     r;
 882
 883                 r = secure_read(MyProcPort, PqRecvBuffer + PqRecvLength,
 884                                                 PQ_RECV_BUFFER_SIZE - PqRecvLength);
 885
 886                 if (r < 0)
 887                 {
 888                         if (errno == EINTR)
 889                                 continue;               /* Ok if interrupted */
 890
 891                         /*
 892                          * Careful: an ereport() that tries to write to the client would
 893                          * cause recursion to here, leading to stack overflow and core
 894                          * dump!  This message must go *only* to the postmaster log.
 895                          */
 896                         ereport(COMMERROR,
 897                                         (errcode_for_socket_access(),
 898                                          errmsg("could not receive data from client: %m")));
 899                         return EOF;
 900                 }
 901                 if (r == 0)
 902                 {
 903                         /*
 904                          * EOF detected.  We used to write a log message here, but it's
 905                          * better to expect the ultimate caller to do that.
 906                          */
 907                         return EOF;
 908                 }
 909                 /* r contains number of bytes read, so just incr length */
 910                 PqRecvLength += r;
 911                 return 0;
 912         }
 913 }
 914
 915 /* --------------------------------
 916  *              pq_getbyte      - get a single byte from connection, or return EOF
 917  * --------------------------------
 918  */
 919 int
 920 pq_getbyte(void)
 921 {
 922         while (PqRecvPointer >= PqRecvLength)
 923         {
 924                 if (pq_recvbuf())               /* If nothing in buffer, then recv some */
 925                         return EOF;                     /* Failed to recv data */
 926         }
 927         return (unsigned char) PqRecvBuffer[PqRecvPointer++];
 928 }
 929
 930 /* --------------------------------
 931  *              pq_peekbyte             - peek at next byte from connection
 932  *
 933  *       Same as pq_getbyte() except we don't advance the pointer.
 934  * --------------------------------
 935  */
 936 int
 937 pq_peekbyte(void)
 938 {
 939         while (PqRecvPointer >= PqRecvLength)
 940         {
 941                 if (pq_recvbuf())               /* If nothing in buffer, then recv some */
 942                         return EOF;                     /* Failed to recv data */
 943         }
 944         return (unsigned char) PqRecvBuffer[PqRecvPointer];
 945 }
 946
 947 /* --------------------------------
 948  *              pq_getbyte_if_available - get a single byte from connection,
 949  *                      if available
 950  *
 951  * The received byte is stored in *c. Returns 1 if a byte was read,
 952  * 0 if no data was available, or EOF if trouble.
 953  * --------------------------------
 954  */
 955 int
 956 pq_getbyte_if_available(unsigned char *c)
 957 {
 958         int                     r;
 959
 960         if (PqRecvPointer < PqRecvLength)
 961         {
 962                 *c = PqRecvBuffer[PqRecvPointer++];
 963                 return 1;
 964         }
 965
 966         /* Put the socket into non-blocking mode */
 967         socket_set_nonblocking(true);
 968
 969         r = secure_read(MyProcPort, c, 1);
 970         if (r < 0)
 971         {
 972                 /*
 973                  * Ok if no data available without blocking or interrupted (though
 974                  * EINTR really shouldn't happen with a non-blocking socket). Report
 975                  * other errors.
 976                  */
 977                 if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
 978                         r = 0;
 979                 else
 980                 {
 981                         /*
 982                          * Careful: an ereport() that tries to write to the client would
 983                          * cause recursion to here, leading to stack overflow and core
 984                          * dump!  This message must go *only* to the postmaster log.
 985                          */
 986                         ereport(COMMERROR,
 987                                         (errcode_for_socket_access(),
 988                                          errmsg("could not receive data from client: %m")));
 989                         r = EOF;
 990                 }
 991         }
 992         else if (r == 0)
 993         {
 994                 /* EOF detected */
 995                 r = EOF;
 996         }
 997
 998         return r;
 999 }
1000
1001 /* --------------------------------
1002  *              pq_getbytes             - get a known number of bytes from connection
1003  *
1004  *              returns 0 if OK, EOF if trouble
1005  * --------------------------------
1006  */
1007 int
1008 pq_getbytes(char *s, size_t len)
1009 {
1010         size_t          amount;
1011
1012         while (len > 0)
1013         {
1014                 while (PqRecvPointer >= PqRecvLength)
1015                 {
1016                         if (pq_recvbuf())       /* If nothing in buffer, then recv some */
1017                                 return EOF;             /* Failed to recv data */
1018                 }
1019                 amount = PqRecvLength - PqRecvPointer;
1020                 if (amount > len)
1021                         amount = len;
1022                 memcpy(s, PqRecvBuffer + PqRecvPointer, amount);
1023                 PqRecvPointer += amount;
1024                 s += amount;
1025                 len -= amount;
1026         }
1027         return 0;
1028 }
1029
1030 /* --------------------------------
1031  *              pq_discardbytes         - throw away a known number of bytes
1032  *
1033  *              same as pq_getbytes except we do not copy the data to anyplace.
1034  *              this is used for resynchronizing after read errors.
1035  *
1036  *              returns 0 if OK, EOF if trouble
1037  * --------------------------------
1038  */
1039 static int
1040 pq_discardbytes(size_t len)
1041 {
1042         size_t          amount;
1043
1044         while (len > 0)
1045         {
1046                 while (PqRecvPointer >= PqRecvLength)
1047                 {
1048                         if (pq_recvbuf())       /* If nothing in buffer, then recv some */
1049                                 return EOF;             /* Failed to recv data */
1050                 }
1051                 amount = PqRecvLength - PqRecvPointer;
1052                 if (amount > len)
1053                         amount = len;
1054                 PqRecvPointer += amount;
1055                 len -= amount;
1056         }
1057         return 0;
1058 }
1059
1060 /* --------------------------------
1061  *              pq_getstring    - get a null terminated string from connection
1062  *
1063  *              The return value is placed in an expansible StringInfo, which has
1064  *              already been initialized by the caller.
1065  *
1066  *              This is used only for dealing with old-protocol clients.  The idea
1067  *              is to produce a StringInfo that looks the same as we would get from
1068  *              pq_getmessage() with a newer client; we will then process it with
1069  *              pq_getmsgstring.  Therefore, no character set conversion is done here,
1070  *              even though this is presumably useful only for text.
1071  *
1072  *              returns 0 if OK, EOF if trouble
1073  * --------------------------------
1074  */
1075 int
1076 pq_getstring(StringInfo s)
1077 {
1078         int                     i;
1079
1080         resetStringInfo(s);
1081
1082         /* Read until we get the terminating '\0' */
1083         for (;;)
1084         {
1085                 while (PqRecvPointer >= PqRecvLength)
1086                 {
1087                         if (pq_recvbuf())       /* If nothing in buffer, then recv some */
1088                                 return EOF;             /* Failed to recv data */
1089                 }
1090
1091                 for (i = PqRecvPointer; i < PqRecvLength; i++)
1092                 {
1093                         if (PqRecvBuffer[i] == '\0')
1094                         {
1095                                 /* include the '\0' in the copy */
1096                                 appendBinaryStringInfo(s, PqRecvBuffer + PqRecvPointer,
1097                                                                            i - PqRecvPointer + 1);
1098                                 PqRecvPointer = i + 1;  /* advance past \0 */
1099                                 return 0;
1100                         }
1101                 }
1102
1103                 /* If we're here we haven't got the \0 in the buffer yet. */
1104                 appendBinaryStringInfo(s, PqRecvBuffer + PqRecvPointer,
1105                                                            PqRecvLength - PqRecvPointer);
1106                 PqRecvPointer = PqRecvLength;
1107         }
1108 }
1109
1110
1111 /* --------------------------------
1112  *              pq_getmessage   - get a message with length word from connection
1113  *
1114  *              The return value is placed in an expansible StringInfo, which has
1115  *              already been initialized by the caller.
1116  *              Only the message body is placed in the StringInfo; the length word
1117  *              is removed.  Also, s->cursor is initialized to zero for convenience
1118  *              in scanning the message contents.
1119  *
1120  *              If maxlen is not zero, it is an upper limit on the length of the
1121  *              message we are willing to accept.  We abort the connection (by
1122  *              returning EOF) if client tries to send more than that.
1123  *
1124  *              returns 0 if OK, EOF if trouble
1125  * --------------------------------
1126  */
1127 int
1128 pq_getmessage(StringInfo s, int maxlen)
1129 {
1130         int32           len;
1131
1132         resetStringInfo(s);
1133
1134         /* Read message length word */
1135         if (pq_getbytes((char *) &len, 4) == EOF)
1136         {
1137                 ereport(COMMERROR,
1138                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
1139                                  errmsg("unexpected EOF within message length word")));
1140                 return EOF;
1141         }
1142
1143         len = ntohl(len);
1144
1145         if (len < 4 ||
1146                 (maxlen > 0 && len > maxlen))
1147         {
1148                 ereport(COMMERROR,
1149                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
1150                                  errmsg("invalid message length")));
1151                 return EOF;
1152         }
1153
1154         len -= 4;                                       /* discount length itself */
1155
1156         if (len > 0)
1157         {
1158                 /*
1159                  * Allocate space for message.  If we run out of room (ridiculously
1160                  * large message), we will elog(ERROR), but we want to discard the
1161                  * message body so as not to lose communication sync.
1162                  */
1163                 PG_TRY();
1164                 {
1165                         enlargeStringInfo(s, len);
1166                 }
1167                 PG_CATCH();
1168                 {
1169                         if (pq_discardbytes(len) == EOF)
1170                                 ereport(COMMERROR,
1171                                                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
1172                                                  errmsg("incomplete message from client")));
1173                         PG_RE_THROW();
1174                 }
1175                 PG_END_TRY();
1176
1177                 /* And grab the message */
1178                 if (pq_getbytes(s->data, len) == EOF)
1179                 {
1180                         ereport(COMMERROR,
1181                                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
1182                                          errmsg("incomplete message from client")));
1183                         return EOF;
1184                 }
1185                 s->len = len;
1186                 /* Place a trailing null per StringInfo convention */
1187                 s->data[len] = '\0';
1188         }
1189
1190         return 0;
1191 }
1192
1193
1194 /* --------------------------------
1195  *              pq_putbytes             - send bytes to connection (not flushed until pq_flush)
1196  *
1197  *              returns 0 if OK, EOF if trouble
1198  * --------------------------------
1199  */
1200 int
1201 pq_putbytes(const char *s, size_t len)
1202 {
1203         int                     res;
1204
1205         /* Should only be called by old-style COPY OUT */
1206         Assert(DoingCopyOut);
1207         /* No-op if reentrant call */
1208         if (PqCommBusy)
1209                 return 0;
1210         PqCommBusy = true;
1211         res = internal_putbytes(s, len);
1212         PqCommBusy = false;
1213         return res;
1214 }
1215
1216 static int
1217 internal_putbytes(const char *s, size_t len)
1218 {
1219         size_t          amount;
1220
1221         while (len > 0)
1222         {
1223                 /* If buffer is full, then flush it out */
1224                 if (PqSendPointer >= PqSendBufferSize)
1225                 {
1226                         socket_set_nonblocking(false);
1227                         if (internal_flush())
1228                                 return EOF;
1229                 }
1230                 amount = PqSendBufferSize - PqSendPointer;
1231                 if (amount > len)
1232                         amount = len;
1233                 memcpy(PqSendBuffer + PqSendPointer, s, amount);
1234                 PqSendPointer += amount;
1235                 s += amount;
1236                 len -= amount;
1237         }
1238         return 0;
1239 }
1240
1241 /* --------------------------------
1242  *              socket_flush            - flush pending output
1243  *
1244  *              returns 0 if OK, EOF if trouble
1245  * --------------------------------
1246  */
1247 static int
1248 socket_flush(void)
1249 {
1250         int                     res;
1251
1252         /* No-op if reentrant call */
1253         if (PqCommBusy)
1254                 return 0;
1255         PqCommBusy = true;
1256         socket_set_nonblocking(false);
1257         res = internal_flush();
1258         PqCommBusy = false;
1259         return res;
1260 }
1261
1262 /* --------------------------------
1263  *              internal_flush - flush pending output
1264  *
1265  * Returns 0 if OK (meaning everything was sent, or operation would block
1266  * and the socket is in non-blocking mode), or EOF if trouble.
1267  * --------------------------------
1268  */
1269 static int
1270 internal_flush(void)
1271 {
1272         static int      last_reported_send_errno = 0;
1273
1274         char       *bufptr = PqSendBuffer + PqSendStart;
1275         char       *bufend = PqSendBuffer + PqSendPointer;
1276
1277         while (bufptr < bufend)
1278         {
1279                 int                     r;
1280
1281                 r = secure_write(MyProcPort, bufptr, bufend - bufptr);
1282
1283                 if (r <= 0)
1284                 {
1285                         if (errno == EINTR)
1286                                 continue;               /* Ok if we were interrupted */
1287
1288                         /*
1289                          * Ok if no data writable without blocking, and the socket is in
1290                          * non-blocking mode.
1291                          */
1292                         if (errno == EAGAIN ||
1293                                 errno == EWOULDBLOCK)
1294                         {
1295                                 return 0;
1296                         }
1297
1298                         /*
1299                          * Careful: an ereport() that tries to write to the client would
1300                          * cause recursion to here, leading to stack overflow and core
1301                          * dump!  This message must go *only* to the postmaster log.
1302                          *
1303                          * If a client disconnects while we're in the midst of output, we
1304                          * might write quite a bit of data before we get to a safe query
1305                          * abort point.  So, suppress duplicate log messages.
1306                          */
1307                         if (errno != last_reported_send_errno)
1308                         {
1309                                 last_reported_send_errno = errno;
1310                                 ereport(COMMERROR,
1311                                                 (errcode_for_socket_access(),
1312                                                  errmsg("could not send data to client: %m")));
1313                         }
1314
1315                         /*
1316                          * We drop the buffered data anyway so that processing can
1317                          * continue, even though we'll probably quit soon. We also set a
1318                          * flag that'll cause the next CHECK_FOR_INTERRUPTS to terminate
1319                          * the connection.
1320                          */
1321                         PqSendStart = PqSendPointer = 0;
1322                         ClientConnectionLost = 1;
1323                         InterruptPending = 1;
1324                         return EOF;
1325                 }
1326
1327                 last_reported_send_errno = 0;   /* reset after any successful send */
1328                 bufptr += r;
1329                 PqSendStart += r;
1330         }
1331
1332         PqSendStart = PqSendPointer = 0;
1333         return 0;
1334 }
1335
1336 /* --------------------------------
1337  *              pq_flush_if_writable - flush pending output if writable without blocking
1338  *
1339  * Returns 0 if OK, or EOF if trouble.
1340  * --------------------------------
1341  */
1342 static int
1343 socket_flush_if_writable(void)
1344 {
1345         int                     res;
1346
1347         /* Quick exit if nothing to do */
1348         if (PqSendPointer == PqSendStart)
1349                 return 0;
1350
1351         /* No-op if reentrant call */
1352         if (PqCommBusy)
1353                 return 0;
1354
1355         /* Temporarily put the socket into non-blocking mode */
1356         socket_set_nonblocking(true);
1357
1358         PqCommBusy = true;
1359         res = internal_flush();
1360         PqCommBusy = false;
1361         return res;
1362 }
1363
1364 /* --------------------------------
1365  *      socket_is_send_pending  - is there any pending data in the output buffer?
1366  * --------------------------------
1367  */
1368 static bool
1369 socket_is_send_pending(void)
1370 {
1371         return (PqSendStart < PqSendPointer);
1372 }
1373
1374 /* --------------------------------
1375  * Message-level I/O routines begin here.
1376  *
1377  * These routines understand about the old-style COPY OUT protocol.
1378  * --------------------------------
1379  */
1380
1381
1382 /* --------------------------------
1383  *              socket_putmessage - send a normal message (suppressed in COPY OUT mode)
1384  *
1385  *              If msgtype is not '\0', it is a message type code to place before
1386  *              the message body.  If msgtype is '\0', then the message has no type
1387  *              code (this is only valid in pre-3.0 protocols).
1388  *
1389  *              len is the length of the message body data at *s.  In protocol 3.0
1390  *              and later, a message length word (equal to len+4 because it counts
1391  *              itself too) is inserted by this routine.
1392  *
1393  *              All normal messages are suppressed while old-style COPY OUT is in
1394  *              progress.  (In practice only a few notice messages might get emitted
1395  *              then; dropping them is annoying, but at least they will still appear
1396  *              in the postmaster log.)
1397  *
1398  *              We also suppress messages generated while pqcomm.c is busy.  This
1399  *              avoids any possibility of messages being inserted within other
1400  *              messages.  The only known trouble case arises if SIGQUIT occurs
1401  *              during a pqcomm.c routine --- quickdie() will try to send a warning
1402  *              message, and the most reasonable approach seems to be to drop it.
1403  *
1404  *              returns 0 if OK, EOF if trouble
1405  * --------------------------------
1406  */
1407 static int
1408 socket_putmessage(char msgtype, const char *s, size_t len)
1409 {
1410         if (DoingCopyOut || PqCommBusy)
1411                 return 0;
1412         PqCommBusy = true;
1413         if (msgtype)
1414                 if (internal_putbytes(&msgtype, 1))
1415                         goto fail;
1416         if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
1417         {
1418                 uint32          n32;
1419
1420                 n32 = htonl((uint32) (len + 4));
1421                 if (internal_putbytes((char *) &n32, 4))
1422                         goto fail;
1423         }
1424         if (internal_putbytes(s, len))
1425                 goto fail;
1426         PqCommBusy = false;
1427         return 0;
1428
1429 fail:
1430         PqCommBusy = false;
1431         return EOF;
1432 }
1433
1434 /* --------------------------------
1435  *              pq_putmessage_noblock   - like pq_putmessage, but never blocks
1436  *
1437  *              If the output buffer is too small to hold the message, the buffer
1438  *              is enlarged.
1439  */
1440 static void
1441 socket_putmessage_noblock(char msgtype, const char *s, size_t len)
1442 {
1443         int res         PG_USED_FOR_ASSERTS_ONLY;
1444         int                     required;
1445
1446         /*
1447          * Ensure we have enough space in the output buffer for the message header
1448          * as well as the message itself.
1449          */
1450         required = PqSendPointer + 1 + 4 + len;
1451         if (required > PqSendBufferSize)
1452         {
1453                 PqSendBuffer = repalloc(PqSendBuffer, required);
1454                 PqSendBufferSize = required;
1455         }
1456         res = pq_putmessage(msgtype, s, len);
1457         Assert(res == 0);                       /* should not fail when the message fits in
1458                                                                  * buffer */
1459 }
1460
1461
1462 /* --------------------------------
1463  *              socket_startcopyout - inform libpq that an old-style COPY OUT transfer
1464  *                      is beginning
1465  * --------------------------------
1466  */
1467 static void
1468 socket_startcopyout(void)
1469 {
1470         DoingCopyOut = true;
1471 }
1472
1473 /* --------------------------------
1474  *              socket_endcopyout       - end an old-style COPY OUT transfer
1475  *
1476  *              If errorAbort is indicated, we are aborting a COPY OUT due to an error,
1477  *              and must send a terminator line.  Since a partial data line might have
1478  *              been emitted, send a couple of newlines first (the first one could
1479  *              get absorbed by a backslash...)  Note that old-style COPY OUT does
1480  *              not allow binary transfers, so a textual terminator is always correct.
1481  * --------------------------------
1482  */
1483 static void
1484 socket_endcopyout(bool errorAbort)
1485 {
1486         if (!DoingCopyOut)
1487                 return;
1488         if (errorAbort)
1489                 pq_putbytes("\n\n\\.\n", 5);
1490         /* in non-error case, copy.c will have emitted the terminator line */
1491         DoingCopyOut = false;
1492 }
1493
1494 /*
1495  * Support for TCP Keepalive parameters
1496  */
1497
1498 /*
1499  * On Windows, we need to set both idle and interval at the same time.
1500  * We also cannot reset them to the default (setting to zero will
1501  * actually set them to zero, not default), therefor we fallback to
1502  * the out-of-the-box default instead.
1503  */
1504 #if defined(WIN32) && defined(SIO_KEEPALIVE_VALS)
1505 static int
1506 pq_setkeepaliveswin32(Port *port, int idle, int interval)
1507 {
1508         struct tcp_keepalive ka;
1509         DWORD           retsize;
1510
1511         if (idle <= 0)
1512                 idle = 2 * 60 * 60;             /* default = 2 hours */
1513         if (interval <= 0)
1514                 interval = 1;                   /* default = 1 second */
1515
1516         ka.onoff = 1;
1517         ka.keepalivetime = idle * 1000;
1518         ka.keepaliveinterval = interval * 1000;
1519
1520         if (WSAIoctl(port->sock,
1521                                  SIO_KEEPALIVE_VALS,
1522                                  (LPVOID) &ka,
1523                                  sizeof(ka),
1524                                  NULL,
1525                                  0,
1526                                  &retsize,
1527                                  NULL,
1528                                  NULL)
1529                 != 0)
1530         {
1531                 elog(LOG, "WSAIoctl(SIO_KEEPALIVE_VALS) failed: %ui",
1532                          WSAGetLastError());
1533                 return STATUS_ERROR;
1534         }
1535         if (port->keepalives_idle != idle)
1536                 port->keepalives_idle = idle;
1537         if (port->keepalives_interval != interval)
1538                 port->keepalives_interval = interval;
1539         return STATUS_OK;
1540 }
1541 #endif
1542
1543 int
1544 pq_getkeepalivesidle(Port *port)
1545 {
1546 #if defined(TCP_KEEPIDLE) || defined(TCP_KEEPALIVE) || defined(WIN32)
1547         if (port == NULL || IS_AF_UNIX(port->laddr.addr.ss_family))
1548                 return 0;
1549
1550         if (port->keepalives_idle != 0)
1551                 return port->keepalives_idle;
1552
1553         if (port->default_keepalives_idle == 0)
1554         {
1555 #ifndef WIN32
1556                 ACCEPT_TYPE_ARG3 size = sizeof(port->default_keepalives_idle);
1557
1558 #ifdef TCP_KEEPIDLE
1559                 if (getsockopt(port->sock, IPPROTO_TCP, TCP_KEEPIDLE,
1560                                            (char *) &port->default_keepalives_idle,
1561                                            &size) < 0)
1562                 {
1563                         elog(LOG, "getsockopt(TCP_KEEPIDLE) failed: %m");
1564                         port->default_keepalives_idle = -1; /* don't know */
1565                 }
1566 #else
1567                 if (getsockopt(port->sock, IPPROTO_TCP, TCP_KEEPALIVE,
1568                                            (char *) &port->default_keepalives_idle,
1569                                            &size) < 0)
1570                 {
1571                         elog(LOG, "getsockopt(TCP_KEEPALIVE) failed: %m");
1572                         port->default_keepalives_idle = -1; /* don't know */
1573                 }
1574 #endif   /* TCP_KEEPIDLE */
1575 #else                                                   /* WIN32 */
1576                 /* We can't get the defaults on Windows, so return "don't know" */
1577                 port->default_keepalives_idle = -1;
1578 #endif   /* WIN32 */
1579         }
1580
1581         return port->default_keepalives_idle;
1582 #else
1583         return 0;
1584 #endif
1585 }
1586
1587 int
1588 pq_setkeepalivesidle(int idle, Port *port)
1589 {
1590         if (port == NULL || IS_AF_UNIX(port->laddr.addr.ss_family))
1591                 return STATUS_OK;
1592
1593 #if defined(TCP_KEEPIDLE) || defined(TCP_KEEPALIVE) || defined(SIO_KEEPALIVE_VALS)
1594         if (idle == port->keepalives_idle)
1595                 return STATUS_OK;
1596
1597 #ifndef WIN32
1598         if (port->default_keepalives_idle <= 0)
1599         {
1600                 if (pq_getkeepalivesidle(port) < 0)
1601                 {
1602                         if (idle == 0)
1603                                 return STATUS_OK;               /* default is set but unknown */
1604                         else
1605                                 return STATUS_ERROR;
1606                 }
1607         }
1608
1609         if (idle == 0)
1610                 idle = port->default_keepalives_idle;
1611
1612 #ifdef TCP_KEEPIDLE
1613         if (setsockopt(port->sock, IPPROTO_TCP, TCP_KEEPIDLE,
1614                                    (char *) &idle, sizeof(idle)) < 0)
1615         {
1616                 elog(LOG, "setsockopt(TCP_KEEPIDLE) failed: %m");
1617                 return STATUS_ERROR;
1618         }
1619 #else
1620         if (setsockopt(port->sock, IPPROTO_TCP, TCP_KEEPALIVE,
1621                                    (char *) &idle, sizeof(idle)) < 0)
1622         {
1623                 elog(LOG, "setsockopt(TCP_KEEPALIVE) failed: %m");
1624                 return STATUS_ERROR;
1625         }
1626 #endif
1627
1628         port->keepalives_idle = idle;
1629 #else                                                   /* WIN32 */
1630         return pq_setkeepaliveswin32(port, idle, port->keepalives_interval);
1631 #endif
1632 #else                                                   /* TCP_KEEPIDLE || SIO_KEEPALIVE_VALS */
1633         if (idle != 0)
1634         {
1635                 elog(LOG, "setting the keepalive idle time is not supported");
1636                 return STATUS_ERROR;
1637         }
1638 #endif
1639         return STATUS_OK;
1640 }
1641
1642 int
1643 pq_getkeepalivesinterval(Port *port)
1644 {
1645 #if defined(TCP_KEEPINTVL) || defined(SIO_KEEPALIVE_VALS)
1646         if (port == NULL || IS_AF_UNIX(port->laddr.addr.ss_family))
1647                 return 0;
1648
1649         if (port->keepalives_interval != 0)
1650                 return port->keepalives_interval;
1651
1652         if (port->default_keepalives_interval == 0)
1653         {
1654 #ifndef WIN32
1655                 ACCEPT_TYPE_ARG3 size = sizeof(port->default_keepalives_interval);
1656
1657                 if (getsockopt(port->sock, IPPROTO_TCP, TCP_KEEPINTVL,
1658                                            (char *) &port->default_keepalives_interval,
1659                                            &size) < 0)
1660                 {
1661                         elog(LOG, "getsockopt(TCP_KEEPINTVL) failed: %m");
1662                         port->default_keepalives_interval = -1;         /* don't know */
1663                 }
1664 #else
1665                 /* We can't get the defaults on Windows, so return "don't know" */
1666                 port->default_keepalives_interval = -1;
1667 #endif   /* WIN32 */
1668         }
1669
1670         return port->default_keepalives_interval;
1671 #else
1672         return 0;
1673 #endif
1674 }
1675
1676 int
1677 pq_setkeepalivesinterval(int interval, Port *port)
1678 {
1679         if (port == NULL || IS_AF_UNIX(port->laddr.addr.ss_family))
1680                 return STATUS_OK;
1681
1682 #if defined(TCP_KEEPINTVL) || defined (SIO_KEEPALIVE_VALS)
1683         if (interval == port->keepalives_interval)
1684                 return STATUS_OK;
1685
1686 #ifndef WIN32
1687         if (port->default_keepalives_interval <= 0)
1688         {
1689                 if (pq_getkeepalivesinterval(port) < 0)
1690                 {
1691                         if (interval == 0)
1692                                 return STATUS_OK;               /* default is set but unknown */
1693                         else
1694                                 return STATUS_ERROR;
1695                 }
1696         }
1697
1698         if (interval == 0)
1699                 interval = port->default_keepalives_interval;
1700
1701         if (setsockopt(port->sock, IPPROTO_TCP, TCP_KEEPINTVL,
1702                                    (char *) &interval, sizeof(interval)) < 0)
1703         {
1704                 elog(LOG, "setsockopt(TCP_KEEPINTVL) failed: %m");
1705                 return STATUS_ERROR;
1706         }
1707
1708         port->keepalives_interval = interval;
1709 #else                                                   /* WIN32 */
1710         return pq_setkeepaliveswin32(port, port->keepalives_idle, interval);
1711 #endif
1712 #else
1713         if (interval != 0)
1714         {
1715                 elog(LOG, "setsockopt(TCP_KEEPINTVL) not supported");
1716                 return STATUS_ERROR;
1717         }
1718 #endif
1719
1720         return STATUS_OK;
1721 }
1722
1723 int
1724 pq_getkeepalivescount(Port *port)
1725 {
1726 #ifdef TCP_KEEPCNT
1727         if (port == NULL || IS_AF_UNIX(port->laddr.addr.ss_family))
1728                 return 0;
1729
1730         if (port->keepalives_count != 0)
1731                 return port->keepalives_count;
1732
1733         if (port->default_keepalives_count == 0)
1734         {
1735                 ACCEPT_TYPE_ARG3 size = sizeof(port->default_keepalives_count);
1736
1737                 if (getsockopt(port->sock, IPPROTO_TCP, TCP_KEEPCNT,
1738                                            (char *) &port->default_keepalives_count,
1739                                            &size) < 0)
1740                 {
1741                         elog(LOG, "getsockopt(TCP_KEEPCNT) failed: %m");
1742                         port->default_keepalives_count = -1;            /* don't know */
1743                 }
1744         }
1745
1746         return port->default_keepalives_count;
1747 #else
1748         return 0;
1749 #endif
1750 }
1751
1752 int
1753 pq_setkeepalivescount(int count, Port *port)
1754 {
1755         if (port == NULL || IS_AF_UNIX(port->laddr.addr.ss_family))
1756                 return STATUS_OK;
1757
1758 #ifdef TCP_KEEPCNT
1759         if (count == port->keepalives_count)
1760                 return STATUS_OK;
1761
1762         if (port->default_keepalives_count <= 0)
1763         {
1764                 if (pq_getkeepalivescount(port) < 0)
1765                 {
1766                         if (count == 0)
1767                                 return STATUS_OK;               /* default is set but unknown */
1768                         else
1769                                 return STATUS_ERROR;
1770                 }
1771         }
1772
1773         if (count == 0)
1774                 count = port->default_keepalives_count;
1775
1776         if (setsockopt(port->sock, IPPROTO_TCP, TCP_KEEPCNT,
1777                                    (char *) &count, sizeof(count)) < 0)
1778         {
1779                 elog(LOG, "setsockopt(TCP_KEEPCNT) failed: %m");
1780                 return STATUS_ERROR;
1781         }
1782
1783         port->keepalives_count = count;
1784 #else
1785         if (count != 0)
1786         {
1787                 elog(LOG, "setsockopt(TCP_KEEPCNT) not supported");
1788                 return STATUS_ERROR;
1789         }
1790 #endif
1791
1792         return STATUS_OK;
1793 }