PG主程序模块（PostMaster）——循环等待客户连接请求ServerLoop

缺乏、安全感 2022-11-12 04:25 160阅读 0赞

PostgreSQL系统采用客户端/服务器模式，系统为每个客户端分配一个后台服务进程Postgres。一次PostgreSQL会话由下列相关的进程（程序）组成：

*  服务进程叫Postgres，它接受来自客户端应用于数据库的连接，并且代表客户端在数据库上执行操作
 *  客户端进程，是指运行在客户端与Postmaster或Postgres进程交互的进程，比如在客户端运行的psql进程

## 架构 ##

和典型的CS应用一样，这些客户端和服务器可以在不同的主机上。这时它们通过网络连接通信。当客户端欲访问某一数据库时，就调用接口库（比如libpq、ODBC）把用户的请求通过网络连接发给守护进程Postmaster。Postmaster将启动一个新的服务进程Postgres为用户服务，此后客户端进程和服务进程不再通过Postmaster而是直接进行通信。Postmaster总是监听用户连接请求并为用户分配服务进程Postgres，而Postgres则负责为客户端执行各种命令。  
ServerLoop函数是postmaster（src/backend/postmaster/postmaster.c 1636行）的主循环。Postmaster将调用ServerLoop函数来循环等待客户端的连接请求，该函数的主体是一个死循环，它的主要功能就是在监听到用户的连接请求后建立与该用户的连接，然后通过调用fork函数复制出一个Postgres进程为该用户服务。

首先调用initMasks函数初始化Postmaster所关心的读文件描述符集（通过扫描ListenSocket数组将里面的监听套接字描述符加入到读文件描述符集中）。

static int ServerLoop(void) { 
    	fd_set		readmask;
    	int			nSockets;
    	time_t		last_lockfile_recheck_time, last_touch_time;
    	last_lockfile_recheck_time = last_touch_time = time(NULL);
    	nSockets = initMasks(&readmask);

接下来进入到一个死循环中，循环中首先调用select系统函数等待客户端提出连接请求。除了sleeping，其他处理逻辑都需要阻塞所有信号。当执行逻辑时，再次阻塞所有的信号对于信号处理函数才安全。如果处于PM\_WAIT\_DEAD\_END状态（PM\_WAIT\_DEAD\_END waiting for dead\_end children to exit 等待dead\_end子进程退出），数据库不想接受任何新连接，所以不调用select函数，仅sleep。

for (;;){ 
    		fd_set		rmask;
    		int			selres;
    		time_t		now;
    		/* Wait for a connection request to arrive. * We block all signals except while sleeping. That makes it safe for * signal handlers, which again block all signals while executing, to * do nontrivial work. If we are in PM_WAIT_DEAD_END state, then we don't want to accept * any new connections, so we don't call select(), and just sleep. */
    		memcpy((char *) &rmask, (char *) &readmask, sizeof(fd_set));
    		if (pmState == PM_WAIT_DEAD_END){ 
    			PG_SETMASK(&UnBlockSig);
    			pg_usleep(100000L); /* 100 msec seems reasonable */
    			selres = 0;
    			PG_SETMASK(&BlockSig);
    		} else { 
    			/* must set timeout each time; some OSes change it! */
    			struct timeval timeout;
    			/* Needs to run with blocked signals! */
    			DetermineSleepTime(&timeout);
    			PG_SETMASK(&UnBlockSig);
    			selres = select(nSockets, &rmask, NULL, NULL, &timeout);
    			PG_SETMASK(&BlockSig);
    		}

多路IO复用，有时也称为事件驱动IO。它的基本原理就是有个函数(如select)会不断地轮询所负责的所有socket，当某个socket有数据到达，就通知用户进程。`int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout);`当用户进程调用select，那么整个进程会被阻塞，而同时，内核会监视所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从内核拷贝到用户进程。在多路复用IO模型中，对于每个socket，一般都设置为非阻塞，整个用户的进程其实是一直被阻塞的。只不过进程是被select函数阻塞，而不是被socket IO阻塞。

当select函数非正常返回，以下为处理逻辑。

/* Now check the select() result */
    		if (selres < 0){ 
    			if (errno != EINTR && errno != EWOULDBLOCK){ 
    				ereport(LOG,(errcode_for_socket_access(),errmsg("select() failed in postmaster: %m")));
    				return STATUS_ERROR;
    			}
    		}

当select函数正常返回后，扫描ListenSocket数组，检查数组中的每一个监听套接字描述符（FD\_ISSET(int fd, fd\_set\* fds); //如果fd在fds中则为真），如果发现该套接字上有用户提出连接请求，那么将**调用ConnCreate函数来创建一个Port结构体**。

/* New connection pending on any of our sockets? If so, fork a child process to deal with it. */
    		if (selres > 0){ 
    			int			i;
    			for (i = 0; i < MAXLISTEN; i++){ 
    				if (ListenSocket[i] == PGINVALID_SOCKET)
    					break;
    				if (FD_ISSET(ListenSocket[i], &rmask)){ 
    					Port	   *port;
    					port = ConnCreate(ListenSocket[i]);
    					if (port){ 
    						BackendStartup(port);
    						/* We no longer need the open socket or port structure in this process */
    						StreamClose(port->sock);
    						ConnFree(port);
    					}
    				}
    			}
    		}

成功创建Port结构体之后，调用BackendStartup启动后台进程接替Postmaster与用户进行连接，然后Postmaster关闭与用户的连接，释放Port结构体。BackendStartup函数将根据Port结构体来启动一个Postgres服务进程，为用户提供服务。函数首先调用PostmasterRandom函数为即将创建的子进程生成一个随机数，并分配一个Backend结构。然后调用fork函数复制出一个Postgres子进程。

/* BackendStartup -- start backend process * returns: STATUS_ERROR if the fork failed, STATUS_OK otherwise. * Note: if you change this code, also consider StartAutovacuumWorker. */
    static int BackendStartup(Port *port){ 
    	Backend    *bn;				/* for backend cleanup */
    	pid_t		pid;
    	/* Create backend data structure. Better before the fork() so we can handle failure cleanly. */
    	bn = (Backend *) malloc(sizeof(Backend));
    	if (!bn){ 
    		ereport(LOG,(errcode(ERRCODE_OUT_OF_MEMORY), errmsg("out of memory")));
    		return STATUS_ERROR;
    	}

活动后端的列表（或子进程；无论如何，我们实际上不知道给定的子进程是否已成为后端或仍处于授权阶段）。 这主要用于跟踪我们有多少孩子进行，并在必要时向他们发送适当的信号。 “特殊”子项（例如启动，bgwriter和autovacuum启动器任务）不在此列表中。 Autovacuum工人和walsender都在其中。 此外，其中还包含“ dead\_end”子项：这些子项是为了向潜在客户发送友好的拒绝消息而启动的。 我们必须跟踪它们，因为它们已附加到共享内存中，但是我们知道它们永远不会成为实时后端。 没有为dead\_end子项分配PMChildSlot。 后台工作者也在此列表中。

/* List of active backends (or child processes anyway; we don't actually know whether a given child has become a backend or is still in the authorization phase). This is used mainly to keep track of how many children we have and send them appropriate signals when necessary. "Special" children such as the startup, bgwriter and autovacuum launcher tasks are not in this list. Autovacuum worker and walsender are in it. Also, "dead_end" children are in it: these are children launched just for the purpose of sending a friendly rejection message to a would-be client. We must track them because they are attached to shared memory, but we know they will never become live backends. dead_end children are not assigned a PMChildSlot. Background workers are in this list, too. */
    typedef struct bkend{ 
    	pid_t		pid;			/* process id of backend */                      // 后端进程的pid
    	int32		cancel_key;		/* cancel key for cancels for this backend */    
    	int			child_slot;		/* PMChildSlot for this backend, if any */       // 后端的PMChildSlot
    	/* Flavor of backend or auxiliary process. Note that BACKEND_TYPE_WALSND backends initially announce themselves as BACKEND_TYPE_NORMAL, so if bkend_type is normal, you should check for a recent transition. */
    	int			bkend_type;
    	bool		dead_end;		/* is it going to send an error and quit? */
    	bool		bgworker_notify;	/* gets bgworker start/stop notifications */
    	dlist_node	elem;			/* list link in BackendList */
    } Backend;

调用PostmasterRandom函数**为即将创建的子进程生成一个随机数MyCancelKey**，并分配一个Backend结构，并将其赋给Backend的cancel\_key成员。对Backend结构的child\_slot成员赋值AssignPostmasterChildSlot函数返回值。

/* Compute the cancel key that will be assigned to this backend. The backend will have its own copy in the forked-off process' value of MyCancelKey, so that it can transmit the key to the frontend. */
    	if (!RandomCancelKey(&MyCancelKey)){ 
    		free(bn);
    		ereport(LOG,(errcode(ERRCODE_INTERNAL_ERROR),errmsg("could not generate random cancel key")));
    		return STATUS_ERROR;
    	}
    	bn->cancel_key = MyCancelKey;
    	/* Pass down canAcceptConnections state */
    	port->canAcceptConnections = canAcceptConnections(BACKEND_TYPE_NORMAL);
    	bn->dead_end = (port->canAcceptConnections != CAC_OK && port->canAcceptConnections != CAC_SUPERUSER);
    	/* Unless it's a dead_end child, assign it a child slot number */
    	if (!bn->dead_end)
    		bn->child_slot = MyPMChildSlot = AssignPostmasterChildSlot();
    	else
    		bn->child_slot = 0;
    	/* Hasn't asked to be notified about any bgworkers yet */
    	bn->bgworker_notify = false;

以下代码创建子进程，启用宏EXEC\_BACKEND，则使用backend\_forkexec(port)，否则使用fork\_process函数。

#ifdef EXEC_BACKEND
    	pid = backend_forkexec(port);
    #else /* !EXEC_BACKEND */
    	pid = fork_process();
    	if (pid == 0)				/* child */{ 
    		free(bn);
    		/* Detangle from postmaster */
    		InitPostmasterChild();
    		/* Close the postmaster's sockets */
    		ClosePostmasterPorts(false);
    		/* Perform additional initialization and collect startup packet */
    		BackendInitialize(port);
    		/* And run the backend */
    		BackendRun(port);
    	}
    #endif /* EXEC_BACKEND */

以下代码处理子进程fork失败，父进程需要处理的代码

if (pid < 0){ 
    		/* in parent, fork failed */
    		int			save_errno = errno;
    		if (!bn->dead_end)
    			(void) ReleasePostmasterChildSlot(bn->child_slot);
    		free(bn);
    		errno = save_errno;
    		ereport(LOG,(errmsg("could not fork new process for connection: %m")));
    		report_fork_failure_to_client(port, save_errno);
    		return STATUS_ERROR;
    	}

以下代码处理子进程fork成功，父进程处理的代码，将子进程的Backend结构的dlist\_node成员elem加入到BackendList中，以方便Postmaster对子进程的管理。

/* in parent, successful fork */
    	ereport(DEBUG2,(errmsg_internal("forked new backend, pid=%d socket=%d", (int) pid, (int) port->sock)));
    	/* Everything's been successful, it's safe to add this backend to our list of backends. */
    	bn->pid = pid;
    	bn->bkend_type = BACKEND_TYPE_NORMAL;	/* Can change later to WALSND */
    	dlist_push_head(&BackendList, &bn->elem);
    #ifdef EXEC_BACKEND
    	if (!bn->dead_end)
    		ShmemBackendArrayAdd(bn);
    #endif
    	return STATUS_OK;
    }

如果设置了log collector，且log collector失活，尝试启动新的

if (SysLoggerPID == 0 && Logging_collector)
    			SysLoggerPID = SysLogger_Start();

### 状态流转 ###

我们使用一个简单的状态机来控制启动，关闭和崩溃恢复（这就像先关闭再启动一样）。完成所有postmaster的初始化工作后，我们进入PM\_STARTUP状态并启动startup过程。启动过程通过读取控制文件和其他初步初始化步骤开始。在正常启动中或崩溃恢复之后，启动过程将以退出代码0退出，并且我们切换到PM\_RUN状态。但是，归档恢复是专门处理的，因为它需要更长的时间，并且我们希望在归档恢复期间支持热备份。  
当启动过程准备好开始存档恢复时，它会向postmaster发出信号，然后我们切换到PM\_RECOVERY状态。启动后台编写器和检查指针，而启动过程继续应用WAL。如果启用了热备，则在WAL重做中达到一致点后，启动过程会再次向我们发出信号，然后我们切换到PM\_HOT\_STANDBY状态并开始接受连接以执行只读查询。归档恢复完成后，启动过程将以退出代码0退出，并且我们切换到PM\_RUN状态。  
普通的子后端只能在我们处于PM\_RUN或PM\_HOT\_STANDBY状态时启动。 （connsAllowed也可能限制启动。）在其他状态下，我们通过启动“ dead\_end”子进程来处理连接请求，该子进程只会向客户端发送错误消息并退出。 （我们在BackendList中跟踪它们，以便我们知道它们什么时候消失了；这很重要，因为它们仍然连接到共享内存，并且会干扰破坏shmem段的尝试，当我们尝试制作一个新的。）在PM\_WAIT\_DEAD\_END状态下，我们正在等待所有dead\_end子节点耗尽系统，因此完全停止接受连接请求，直到最后一个现有子节点退出为止（希望不会太长）。 。  
请注意，这个状态变量并不能区分为什么我们在PM\_RUN之后进入状态-为什么必须咨询Shutdown和FatalError才能发现这一点。在PM\_RECOVERY，PM\_HOT\_STANDBY或PM\_RUN状态下，或者在PM\_SHUTDOWN状态下，FatalError永远不会为真（因为尝试从崩溃中恢复时不会输入这些状态）。在PM\_STARTUP状态下可能是正确的，因为在成功启动WAL重做之前，我们不会清除它。

/* We use a simple state machine to control startup, shutdown, and crash recovery (which is rather like shutdown followed by startup). After doing all the postmaster initialization work, we enter PM_STARTUP state and the startup process is launched. The startup process begins by reading the control file and other preliminary initialization steps. In a normal startup, or after crash recovery, the startup process exits with exit code 0 and we switch to PM_RUN state. However, archive recovery is handled specially since it takes much longer and we would like to support hot standby during archive recovery. When the startup process is ready to start archive recovery, it signals the postmaster, and we switch to PM_RECOVERY state. The background writer and checkpointer are launched, while the startup process continues applying WAL. If Hot Standby is enabled, then, after reaching a consistent point in WAL redo, startup process signals us again, and we switch to PM_HOT_STANDBY state and begin accepting connections to perform read-only queries. When archive recovery is finished, the startup process exits with exit code 0 and we switch to PM_RUN state. Normal child backends can only be launched when we are in PM_RUN or PM_HOT_STANDBY state. (connsAllowed can also restrict launching.) In other states we handle connection requests by launching "dead_end" child processes, which will simply send the client an error message and quit. (We track these in the BackendList so that we can know when they are all gone; this is important because they're still connected to shared memory, and would interfere with an attempt to destroy the shmem segment, possibly leading to SHMALL failure when we try to make a new one.) In PM_WAIT_DEAD_END state we are waiting for all the dead_end children to drain out of the system, and therefore stop accepting connection requests at all until the last existing child has quit (which hopefully will not be very long). Notice that this state variable does not distinguish *why* we entered states later than PM_RUN --- Shutdown and FatalError must be consulted to find that out. FatalError is never true in PM_RECOVERY, PM_HOT_STANDBY, or PM_RUN states, nor in PM_SHUTDOWN states (because we don't enter those states when trying to recover from a crash). It can be true in PM_STARTUP state, because we don't clear it until we've successfully started WAL redo. */
    typedef enum{ 
    	PM_INIT,					/* postmaster starting */
    	PM_STARTUP,					/* waiting for startup subprocess */
    	PM_RECOVERY,				/* in archive recovery mode */
    	PM_HOT_STANDBY,				/* in hot standby mode */
    	PM_RUN,						/* normal "database is alive" state */
    	PM_STOP_BACKENDS,			/* need to stop remaining backends */
    	PM_WAIT_BACKENDS,			/* waiting for live backends to exit */
    	PM_SHUTDOWN,				/* waiting for checkpointer to do shutdown ckpt */
    	PM_SHUTDOWN_2,				/* waiting for archiver and walsenders to finish */
    	PM_WAIT_DEAD_END,			/* waiting for dead_end children to exit */
    	PM_NO_CHILDREN				/* all important children have exited */
    } PMState;
    static PMState pmState = PM_INIT;

如果处于PM\_RUN或PM\_RECOVERY或PM\_HOT\_STANDBY状态，根据情况启动CheckPointer和BgWriterPID。

/* If no background writer process is running, and we are not in a state that prevents it, start one. It doesn't matter if this fails, we'll just try again later. Likewise for the checkpointer. */
    		if (pmState == PM_RUN || pmState == PM_RECOVERY || pmState == PM_HOT_STANDBY){ 
    			if (CheckpointerPID == 0)
    				CheckpointerPID = StartCheckpointer();
    			if (BgWriterPID == 0)
    				BgWriterPID = StartBackgroundWriter();
    		}
    		/* Likewise, if we have lost the walwriter process, try to start a new one. But this is needed only in normal operation (else we cannot be writing any new WAL). */
    		if (WalWriterPID == 0 && pmState == PM_RUN)
    			WalWriterPID = StartWalWriter();
    		/* If we have lost the autovacuum launcher, try to start a new one. We don't want autovacuum to run in binary upgrade mode because autovacuum might update relfrozenxid for empty tables before the physical files are put in place. */
    		if (!IsBinaryUpgrade && AutoVacPID == 0 && (AutoVacuumingActive() || start_autovac_launcher) && pmState == PM_RUN){ 
    			AutoVacPID = StartAutoVacLauncher();
    			if (AutoVacPID != 0)
    				start_autovac_launcher = false; /* signal processed */
    		}
    		/* If we have lost the stats collector, try to start a new one */
    		if (PgStatPID == 0 && (pmState == PM_RUN || pmState == PM_HOT_STANDBY))
    			PgStatPID = pgstat_start();
    		/* If we have lost the archiver, try to start a new one. */
    		if (PgArchPID == 0 && PgArchStartupAllowed())
    			PgArchPID = pgarch_start();
    		/* If we need to signal the autovacuum launcher, do so now */
    		if (avlauncher_needs_signal){ 
    			avlauncher_needs_signal = false;
    			if (AutoVacPID != 0)
    				kill(AutoVacPID, SIGUSR2);
    		}
    		/* If we need to start a WAL receiver, try to do that now */
    		if (WalReceiverRequested)
    			MaybeStartWalReceiver();
    		/* Get other worker processes running, if needed */
    		if (StartWorkerNeeded || HaveCrashedWorker)
    			maybe_start_bgworkers();
    #ifdef HAVE_PTHREAD_IS_THREADED_NP
    		/* With assertions enabled, check regularly for appearance of additional threads. All builds check at start and exit. */
    		Assert(pthread_is_threaded_np() == 0);
    #endif

### 扫尾工作 ###

/* * Lastly, check to see if it's time to do some things that we don't * want to do every single time through the loop, because they're a * bit expensive. Note that there's up to a minute of slop in when * these tasks will be performed, since DetermineSleepTime() will let * us sleep at most that long; except for SIGKILL timeout which has * special-case logic there. */
    		now = time(NULL);
    
    		/* * If we already sent SIGQUIT to children and they are slow to shut * down, it's time to send them SIGKILL. This doesn't happen * normally, but under certain conditions backends can get stuck while * shutting down. This is a last measure to get them unwedged. * * Note we also do this during recovery from a process crash. */
    		if ((Shutdown >= ImmediateShutdown || (FatalError && !SendStop)) &&
    			AbortStartTime != 0 &&
    			(now - AbortStartTime) >= SIGKILL_CHILDREN_AFTER_SECS)
    		{ 
    			/* We were gentle with them before. Not anymore */
    			TerminateChildren(SIGKILL);
    			/* reset flag so we don't SIGKILL again */
    			AbortStartTime = 0;
    		}
    
    		/* * Once a minute, verify that postmaster.pid hasn't been removed or * overwritten. If it has, we force a shutdown. This avoids having * postmasters and child processes hanging around after their database * is gone, and maybe causing problems if a new database cluster is * created in the same place. It also provides some protection * against a DBA foolishly removing postmaster.pid and manually * starting a new postmaster. Data corruption is likely to ensue from * that anyway, but we can minimize the damage by aborting ASAP. */
    		if (now - last_lockfile_recheck_time >= 1 * SECS_PER_MINUTE)
    		{ 
    			if (!RecheckDataDirLockFile())
    			{ 
    				ereport(LOG,
    						(errmsg("performing immediate shutdown because data directory lock file is invalid")));
    				kill(MyProcPid, SIGQUIT);
    			}
    			last_lockfile_recheck_time = now;
    		}
    
    		/* * Touch Unix socket and lock files every 58 minutes, to ensure that * they are not removed by overzealous /tmp-cleaning tasks. We assume * no one runs cleaners with cutoff times of less than an hour ... */
    		if (now - last_touch_time >= 58 * SECS_PER_MINUTE)
    		{ 
    			TouchSocketFiles();
    			TouchSocketLockFiles();
    			last_touch_time = now;
    		}
    	}

## 函数解析 ##

PostgreSQL允许通过网络（TCP、Bonjour等）或UNIX本地机来访问数据库。**整型数组ListenSocket\[MAXLISTEM\]用来保存与服务器上某个IP地址（一个服务器可能有多个网卡，因此会设置多个IP地址）绑定的监听套接字描述符**，初始时全为-1。  
initMasks函数为select函数监听的套接字的mask，返回监听端口的数量。  
FD\_ZERO(fd\_set\* fds); //将set清零  
FD\_SET(int fd, fd\_set\* fds); //将fd加入fds  
FD\_CLR(int fd, fd\_set\* fds); //将fd从fds中清除

/* Initialise the masks for select() for the ports we are listening on. * Return the number of sockets to listen on. */
    static int initMasks(fd_set *rmask) { 
    	int			maxsock = -1;
    	int			i;
    	FD_ZERO(rmask);
    	for (i = 0; i < MAXLISTEN; i++){ 
    		int			fd = ListenSocket[i];
    		if (fd == PGINVALID_SOCKET)
    			break;
    		FD_SET(fd, rmask);
    		if (fd > maxsock)
    			maxsock = fd;
    	}
    	return maxsock + 1;
    }

Port结构体

typedef struct Port{ 
        int     sock;  //已连接的套接字描述符
        SockAddr laddr;//服务器端的地址信息
        SockAddr raddr;//客户端的地址信息
        char *database_name;//要连接的数据库名
        char *user_name;//连接的用户名
        ...
    }

ConnCreate函数首先从CurrentMemoryContext中创建一个Port结构体，然后调用Accept系统函数接受由该监听套接字所维护的用户连接请求队列中的第一个连接请求，将监听套接字转化成一个已连接的套接字，并将返回的已连接套接字和用户端套接字地址结构都填入到Port的相应的属性中。对于TCP连接，还要为连接设置超时等连接选项。

/* ConnCreate -- create a local connection data structure * Returns NULL on failure, other than out-of-memory which is fatal. */
    static Port *ConnCreate(int serverFd){ 
    	Port	   *port;
    	if (!(port = (Port *) calloc(1, sizeof(Port)))){ 
    		ereport(LOG,(errcode(ERRCODE_OUT_OF_MEMORY),errmsg("out of memory")));
    		ExitPostmaster(1);
    	}
    	if (StreamConnection(serverFd, port) != STATUS_OK){ 
    		if (port->sock != PGINVALID_SOCKET)
    			StreamClose(port->sock);
    		ConnFree(port);
    		return NULL;
    	}
    	return port;
    }

StreamConnection函数使用server端口与客户端创建链接，将post->sock设置为新连接的FD，返回STATUS\_OK或STATUS\_ERROR。ASSUME: that this doesn’t need to be non-blocking because the Postmaster uses select() to tell when the server master socket is ready for accept().

int StreamConnection(pgsocket server_fd, Port *port){ 
    	/* accept connection and fill in the client (remote) address */
    	port->raddr.salen = sizeof(port->raddr.addr);
    	if ((port->sock = accept(server_fd, (struct sockaddr *) &port->raddr.addr, &port->raddr.salen)) == PGINVALID_SOCKET){ 
    		ereport(LOG, (errcode_for_socket_access(), errmsg("could not accept new connection: %m")));
    		/* If accept() fails then postmaster.c will still see the server socket as read-ready, and will immediately try again. To avoid uselessly sucking lots of CPU, delay a bit before trying again. (The most likely reason for failure is being out of kernel file table slots; we can do little except hope some will get freed up.) */
    		pg_usleep(100000L);		/* wait 0.1 sec */
    		return STATUS_ERROR;
    	}
    	/* fill in the server (local) address */
    	port->laddr.salen = sizeof(port->laddr.addr);
    	if (getsockname(port->sock,(struct sockaddr *) &port->laddr.addr,&port->laddr.salen) < 0){ 
    		elog(LOG, "getsockname() failed: %m");
    		return STATUS_ERROR;
    	}
    	/* select NODELAY and KEEPALIVE options if it's a TCP connection */
    	if (!IS_AF_UNIX(port->laddr.addr.ss_family)){ 
    		int			on;
    #ifdef TCP_NODELAY
    		on = 1;
    		if (setsockopt(port->sock, IPPROTO_TCP, TCP_NODELAY, (char *) &on, sizeof(on)) < 0){ 
    			elog(LOG, "setsockopt(%s) failed: %m", "TCP_NODELAY");
    			return STATUS_ERROR;
    		}
    #endif
    		on = 1;
    		if (setsockopt(port->sock, SOL_SOCKET, SO_KEEPALIVE, (char *) &on, sizeof(on)) < 0){ 
    			elog(LOG, "setsockopt(%s) failed: %m", "SO_KEEPALIVE");
    			return STATUS_ERROR;
    		}
    
    		/* Also apply the current keepalive parameters. If we fail to set a parameter, don't error out, because these aren't universally supported. (Note: you might think we need to reset the GUC variables to 0 in such a case, but it's not necessary because the show hooks for these variables report the truth anyway.) */
    		(void) pq_setkeepalivesidle(tcp_keepalives_idle, port);
    		(void) pq_setkeepalivesinterval(tcp_keepalives_interval, port);
    		(void) pq_setkeepalivescount(tcp_keepalives_count, port);
    		(void) pq_settcpusertimeout(tcp_user_timeout, port);
    	}
    	return STATUS_OK;
    }