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