src/slurmdbd/rpc_mgr.c

/*****************************************************************************\
 *  rpc_mgr.c - functions for processing RPCs.
 *****************************************************************************
 *  Copyright (C) 2002-2007 The Regents of the University of California.
 *  Copyright (C) 2008-2009 Lawrence Livermore National Security.
 *  Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
 *  Written by Morris Jette <jette1@llnl.gov>
 *  CODE-OCEC-09-009. All rights reserved.
 *
 *  This file is part of Slurm, a resource management program.
 *  For details, see <https://slurm.schedmd.com/>.
 *  Please also read the included file: DISCLAIMER.
 *
 *  Slurm is free software; you can redistribute it and/or modify it under
 *  the terms of the GNU General Public License as published by the Free
 *  Software Foundation; either version 2 of the License, or (at your option)
 *  any later version.
 *
 *  In addition, as a special exception, the copyright holders give permission
 *  to link the code of portions of this program with the OpenSSL library under
 *  certain conditions as described in each individual source file, and
 *  distribute linked combinations including the two. You must obey the GNU
 *  General Public License in all respects for all of the code used other than
 *  OpenSSL. If you modify file(s) with this exception, you may extend this
 *  exception to your version of the file(s), but you are not obligated to do
 *  so. If you do not wish to do so, delete this exception statement from your
 *  version.  If you delete this exception statement from all source files in
 *  the program, then also delete it here.
 *
 *  Slurm is distributed in the hope that it will be useful, but WITHOUT ANY
 *  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 *  FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
 *  details.
 *
 *  You should have received a copy of the GNU General Public License along
 *  with Slurm; if not, write to the Free Software Foundation, Inc.,
 *  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA.
\*****************************************************************************/

#include <arpa/inet.h>
#include <poll.h>
#include <pthread.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include "src/common/fd.h"
#include "src/common/log.h"
#include "src/common/macros.h"
#include "src/common/slurm_protocol_api.h"
#include "src/common/slurmdbd_defs.h"
#include "src/common/xmalloc.h"

#include "src/interfaces/accounting_storage.h"

#include "src/slurmdbd/proc_req.h"
#include "src/slurmdbd/read_config.h"
#include "src/slurmdbd/rpc_mgr.h"
#include "src/slurmdbd/slurmdbd.h"

/* Local functions */
static void _connection_fini_callback(void *arg);

/* Local variables */
static pthread_t       master_thread_id = 0;

/* Process incoming RPCs. Meant to execute as a pthread */
extern void *rpc_mgr(void *no_data)
{
	int sockfd, newsockfd;
	slurm_addr_t cli_addr;
	slurmdbd_conn_t *dbd_conn = NULL;

	master_thread_id = pthread_self();

	/* initialize port for RPCs */
	if ((sockfd = slurm_init_msg_engine_port(slurmdbd_conf->dbd_port))
	    == SLURM_ERROR)
		fatal("slurm_init_msg_engine_port error %m");

	slurm_persist_conn_recv_server_init();

	/*
	 * Process incoming RPCs until told to shutdown
	 */
	while (!shutdown_time) {
		/*
		 * accept needed for stream implementation is a no-op in
		 * message implementation that just passes sockfd to newsockfd
		 */
		if ((newsockfd = slurm_accept_conn(sockfd, &cli_addr)) ==
		    SLURM_ERROR) {
			if (errno != EINTR)
				error("slurm_accept_conn: %m");
			continue;
		}
		fd_set_nonblocking(newsockfd);

		dbd_conn = xmalloc(sizeof(slurmdbd_conn_t));
		dbd_conn->pcon = xmalloc(sizeof(persist_conn_t));
		dbd_conn->pcon->flags = PERSIST_FLAG_DBD;
		dbd_conn->pcon->callback_proc = proc_req;
		dbd_conn->pcon->callback_fini = _connection_fini_callback;
		dbd_conn->pcon->shutdown = &shutdown_time;
		dbd_conn->pcon->version = SLURM_MIN_PROTOCOL_VERSION;
		dbd_conn->pcon->rem_host = xmalloc(INET6_ADDRSTRLEN);
		/* Don't fill in the rem_port here.  It will be filled in
		 * later if it is a slurmctld connection. */
		slurm_get_ip_str(&cli_addr, dbd_conn->pcon->rem_host,
				 INET6_ADDRSTRLEN);

		slurm_persist_conn_recv_thread_init(dbd_conn->pcon, newsockfd,
						    -1, dbd_conn);
	}

	debug("rpc_mgr shutting down");
	close(sockfd);
	return NULL;
}

/* Wake up the RPC manager and all spawned threads so they can exit */
extern void rpc_mgr_wake(void)
{
	if (master_thread_id)
		pthread_kill(master_thread_id, SIGUSR1);
	slurm_persist_conn_recv_server_fini();
}

static void _connection_fini_callback(void *arg)
{
	slurmdbd_conn_t *dbd_conn = (slurmdbd_conn_t *) arg;
	bool stay_locked = false;
	int tries = 0;

	/*
	 * If we are sending updates to this ctld, it is holding
	 * pcon_send_lock and is in slurm_persist_conn_open() -- either
	 * still inside connect() (pcon_send->last_fd not yet published)
	 * or already past it and sitting in poll() on the pcon_send
	 * socket. The ctld has disconnected (that is why we are here)
	 * so the socket will never get a reply. Wake the push by
	 * shutting down the published fd. Clearing PERSIST_FLAG_RECONNECT
	 * first prevents _slurm_persist_recv_msg() from reopening the
	 * connection and re-wedging on a fresh fd. The push errors out
	 * of slurm_persist_conn_open(), releases pcon_send_lock, and we
	 * proceed with destroy.
	 *
	 * Reading pcon_send->last_fd lock-free is safe because it is a
	 * scalar published by _open_persist_conn() and cleared to -1 by
	 * _conn_destroy() before the conn struct is freed -- so unlike
	 * dereferencing pcon_send->conn, this cannot land on a freed
	 * TLS conn struct.
	 *
	 * Loop while we cannot acquire the lock: last_fd is only valid
	 * once the socket has been created, so during connect() it may
	 * be -1 for up to a TCP SYN timeout. Re-attempt the shutdown
	 * each pass so we wake the push as soon as the fd is published.
	 */
	if (dbd_conn->pcon_send)
		dbd_conn->pcon_send->flags &= ~PERSIST_FLAG_RECONNECT;
	while (pthread_mutex_trylock(&dbd_conn->pcon_send_lock) == EBUSY) {
		if (dbd_conn->pcon_send) {
			int fd = dbd_conn->pcon_send->last_fd;
			if (fd > 0) {
				debug("Terminating send to the slurmctld on cluster %s. It is restarting or shutting down.",
				      dbd_conn->pcon->cluster_name);
				shutdown(fd, SHUT_RDWR);
			}
		}
		if (tries++ >= 100) {
			slurm_mutex_lock(&dbd_conn->pcon_send_lock);
			break;
		}
		usleep(10000); /* 10ms */
	}

	slurm_persist_conn_destroy(dbd_conn->pcon_send);
	dbd_conn->pcon_send = NULL;
	slurm_mutex_unlock(&dbd_conn->pcon_send_lock);

	if (dbd_conn->pcon->rem_port) {
		if (!shutdown_time) {
			slurmdb_cluster_rec_t cluster_rec;
			memset(&cluster_rec, 0, sizeof(slurmdb_cluster_rec_t));
			cluster_rec.name = dbd_conn->pcon->cluster_name;
			cluster_rec.control_host = dbd_conn->pcon->rem_host;
			cluster_rec.control_port = dbd_conn->pcon->rem_port;
			cluster_rec.rpc_version = dbd_conn->pcon->version;
			cluster_rec.tres_str = dbd_conn->tres_str;
			if (dbd_conn->pcon->flags & PERSIST_FLAG_EXT_DBD)
				cluster_rec.flags = CLUSTER_FLAG_EXT;
			debug("cluster %s has disconnected",
			      dbd_conn->pcon->cluster_name);

			clusteracct_storage_g_fini_ctld(
				dbd_conn->db_conn, &cluster_rec);
		} else if (slurmdbd_conf->commit_delay)
			stay_locked = true;

		/*
		 * On connection close, remove from the list of registered
		 * clusters. The List ensures acct_storage_g_commit() is run
		 * every CommitDelay interval, but the final commit is handled
		 * below.
		 */
		slurm_mutex_lock(&registered_lock);
		if (registered_clusters)
			list_delete_ptr(registered_clusters, dbd_conn);
		if (!stay_locked)
			slurm_mutex_unlock(&registered_lock);

		/* needs to be the last thing done */
		acct_storage_g_commit(dbd_conn->db_conn, 1);
	}

	acct_storage_g_close_connection(&dbd_conn->db_conn);

	if (stay_locked)
		slurm_mutex_unlock(&registered_lock);
	/* handled directly in the internal persist_conn code */
	//slurm_persist_conn_members_destroy(&dbd_conn->pcon);
	slurm_mutex_destroy(&dbd_conn->pcon_send_lock);
	xfree(dbd_conn->tres_str);
	xfree(dbd_conn);
}