/*****************************************************************************\
* agent.c - parallel background communication functions. This is where
* logic could be placed for broadcast communications.
*****************************************************************************
* Copyright (C) 2002 The Regents of the University of California.
* Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
* Written by Moe Jette <jette@llnl.gov>, et. al.
* Derived from pdsh written by Jim Garlick <garlick1@llnl.gov>
* UCRL-CODE-2002-040.
*
* This file is part of SLURM, a resource management program.
* For details, see <http://www.llnl.gov/linux/slurm/>.
*
* 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.
*
* 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.,
* 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
*****************************************************************************
* Theory of operation:
*
* The functions below permit slurm to initiate parallel tasks as a
* detached thread and let the functions below make sure the work happens.
* For example, when a job's time limit is to be changed slurmctld needs
* to notify the slurmd on every node to which the job was allocated.
* We don't want to hang slurmctld's primary function (the job update RPC)
* to perform this work, so it just initiates an agent to perform the work.
* The agent is passed all details required to perform the work, so it will
* be possible to execute the agent as an pthread, process, or even a daemon
* on some other computer.
*
* The main agent thread creates a separate thread for each node to be
* communicated with up to AGENT_THREAD_COUNT. A special watchdog thread
* sends SIGLARM to any threads that have been active (in DSH_ACTIVE state)
* for more than COMMAND_TIMEOUT seconds.
* The agent responds to slurmctld via a function call or an RPC as required.
* For example, informing slurmctld that some node is not responding.
*
* All the state for each thread is maintained in thd_t struct, which is
* used by the watchdog thread as well as the communication threads.
*
* NOTE: REQUEST_KILL_JOB and REQUEST_KILL_TIMELIMIT responses are
* handled immediately rather than in bulk upon completion of the RPC
* to all nodes
\*****************************************************************************/
#ifdef HAVE_CONFIG_H
# include "config.h"
#endif
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <string.h>
#include <unistd.h>
#include "src/common/list.h"
#include "src/common/log.h"
#include "src/common/macros.h"
#include "src/common/xsignal.h"
#include "src/common/xassert.h"
#include "src/common/xmalloc.h"
#include "src/common/xstring.h"
#include "src/common/slurm_protocol_api.h"
#include "src/slurmctld/agent.h"
#include "src/slurmctld/locks.h"
#include "src/slurmctld/ping_nodes.h"
#include "src/slurmctld/slurmctld.h"
#if COMMAND_TIMEOUT == 1
# define WDOG_POLL 1 /* secs */
#else
# define WDOG_POLL 2 /* secs */
#endif
#define MAX_RETRIES 10
typedef enum {
DSH_NEW, /* Request not yet started */
DSH_ACTIVE, /* Request in progress */
DSH_DONE, /* Request completed normally */
DSH_NO_RESP, /* Request timed out */
DSH_FAILED, /* Request resulted in error */
DSH_JOB_HUNG /* Job has non-killable processes */
} state_t;
typedef struct thd {
pthread_t thread; /* thread ID */
pthread_attr_t attr; /* thread attributes */
state_t state; /* thread state */
time_t start_time; /* start time */
time_t end_time; /* end time or delta time
* upon termination */
struct sockaddr_in slurm_addr; /* network address */
char node_name[MAX_NAME_LEN]; /* node's name */
} thd_t;
typedef struct agent_info {
pthread_mutex_t thread_mutex; /* agent specific mutex */
pthread_cond_t thread_cond; /* agent specific condition */
uint32_t thread_count; /* number of threads records */
uint32_t threads_active; /* currently active threads */
uint16_t retry; /* if set, keep trying */
thd_t *thread_struct; /* thread structures */
bool get_reply; /* flag if reply expected */
slurm_msg_type_t msg_type; /* RPC to be issued */
void **msg_args_pptr; /* RPC data to be used */
} agent_info_t;
typedef struct task_info {
pthread_mutex_t *thread_mutex_ptr; /* pointer to agent specific
* mutex */
pthread_cond_t *thread_cond_ptr;/* pointer to agent specific
* condition */
uint32_t *threads_active_ptr; /* currently active thread ptr */
thd_t *thread_struct_ptr; /* thread structures ptr */
bool get_reply; /* flag if reply expected */
slurm_msg_type_t msg_type; /* RPC to be issued */
void *msg_args_ptr; /* ptr to RPC data to be used */
} task_info_t;
static void _alarm_handler(int dummy);
static inline void _comm_err(char *node_name);
static void _list_delete_retry(void *retry_entry);
static agent_info_t *_make_agent_info(agent_arg_t *agent_arg_ptr);
static task_info_t *_make_task_data(agent_info_t *agent_info_ptr, int inx);
static void _purge_agent_args(agent_arg_t *agent_arg_ptr);
static void _queue_agent_retry(agent_info_t * agent_info_ptr, int count);
static void _slurmctld_free_job_launch_msg(batch_job_launch_msg_t * msg);
static void _spawn_retry_agent(agent_arg_t * agent_arg_ptr);
static void *_thread_per_node_rpc(void *args);
static int _valid_agent_arg(agent_arg_t *agent_arg_ptr);
static void *_wdog(void *args);
static pthread_mutex_t retry_mutex = PTHREAD_MUTEX_INITIALIZER;
static List retry_list = NULL; /* agent_arg_t list for retry */
/*
* agent - party responsible for transmitting an common RPC in parallel
* across a set of nodes
* IN pointer to agent_arg_t, which is xfree'd (including slurm_addr,
* node_names and msg_args) upon completion if AGENT_IS_THREAD is set
* RET always NULL (function format just for use as pthread)
*/
void *agent(void *args)
{
int i, rc, retries = 0;
pthread_attr_t attr_wdog;
pthread_t thread_wdog;
agent_arg_t *agent_arg_ptr = args;
agent_info_t *agent_info_ptr = NULL;
thd_t *thread_ptr;
task_info_t *task_specific_ptr;
/* basic argument value tests */
if (_valid_agent_arg(agent_arg_ptr))
goto cleanup;
xsignal(SIGALRM, _alarm_handler);
/* initialize the agent data structures */
agent_info_ptr = _make_agent_info(agent_arg_ptr);
thread_ptr = agent_info_ptr->thread_struct;
/* start the watchdog thread */
if (pthread_attr_init(&attr_wdog))
fatal("pthread_attr_init error %m");
if (pthread_attr_setdetachstate
(&attr_wdog, PTHREAD_CREATE_JOINABLE))
error("pthread_attr_setdetachstate error %m");
#ifdef PTHREAD_SCOPE_SYSTEM
if (pthread_attr_setscope(&attr_wdog, PTHREAD_SCOPE_SYSTEM))
error("pthread_attr_setscope error %m");
#endif
while (pthread_create(&thread_wdog, &attr_wdog, _wdog,
(void *) agent_info_ptr)) {
error("pthread_create error %m");
if (++retries > MAX_RETRIES)
fatal("Can't create pthread");
sleep(1); /* sleep and again */
}
#if AGENT_THREAD_COUNT < 1
fatal("AGENT_THREAD_COUNT value is invalid");
#endif
/* start all the other threads (up to AGENT_THREAD_COUNT active) */
for (i = 0; i < agent_info_ptr->thread_count; i++) {
/* wait until "room" for another thread */
slurm_mutex_lock(&agent_info_ptr->thread_mutex);
while (agent_info_ptr->threads_active >=
AGENT_THREAD_COUNT) {
pthread_cond_wait(&agent_info_ptr->thread_cond,
&agent_info_ptr->thread_mutex);
}
/* create thread specific data, NOTE: freed from
* _thread_per_node_rpc() */
task_specific_ptr = _make_task_data(agent_info_ptr, i);
if (pthread_attr_init(&thread_ptr[i].attr))
fatal("pthread_attr_init error %m");
if (pthread_attr_setdetachstate(&thread_ptr[i].attr,
PTHREAD_CREATE_DETACHED))
error("pthread_attr_setdetachstate error %m");
#ifdef PTHREAD_SCOPE_SYSTEM
if (pthread_attr_setscope(&thread_ptr[i].attr,
PTHREAD_SCOPE_SYSTEM))
error("pthread_attr_setscope error %m");
#endif
while ((rc = pthread_create(&thread_ptr[i].thread,
&thread_ptr[i].attr,
_thread_per_node_rpc,
(void *) task_specific_ptr))) {
error("pthread_create error %m");
if (agent_info_ptr->threads_active)
pthread_cond_wait(&agent_info_ptr->
thread_cond,
&agent_info_ptr->
thread_mutex);
else {
slurm_mutex_unlock(&agent_info_ptr->
thread_mutex);
sleep(1);
slurm_mutex_lock(&agent_info_ptr->
thread_mutex);
}
}
agent_info_ptr->threads_active++;
slurm_mutex_unlock(&agent_info_ptr->thread_mutex);
}
/* wait for termination of remaining threads */
pthread_join(thread_wdog, NULL);
cleanup:
#if AGENT_IS_THREAD
_purge_agent_args(agent_arg_ptr);
#endif
if (agent_info_ptr) {
xfree(agent_info_ptr->thread_struct);
xfree(agent_info_ptr);
}
return NULL;
}
/* Basic validity test of agent argument */
static int _valid_agent_arg(agent_arg_t *agent_arg_ptr)
{
xassert(agent_arg_ptr);
xassert(agent_arg_ptr->slurm_addr);
xassert(agent_arg_ptr->node_names);
xassert((agent_arg_ptr->msg_type == REQUEST_KILL_JOB) ||
(agent_arg_ptr->msg_type == REQUEST_KILL_TIMELIMIT) ||
(agent_arg_ptr->msg_type == REQUEST_UPDATE_JOB_TIME) ||
(agent_arg_ptr->msg_type == REQUEST_KILL_TASKS) ||
(agent_arg_ptr->msg_type == REQUEST_PING) ||
(agent_arg_ptr->msg_type == REQUEST_BATCH_JOB_LAUNCH) ||
(agent_arg_ptr->msg_type == REQUEST_SHUTDOWN) ||
(agent_arg_ptr->msg_type == REQUEST_RECONFIGURE) ||
(agent_arg_ptr->msg_type == REQUEST_NODE_REGISTRATION_STATUS));
if (agent_arg_ptr->node_count == 0)
return SLURM_FAILURE; /* no messages to be sent */
return SLURM_SUCCESS;
}
static agent_info_t *_make_agent_info(agent_arg_t *agent_arg_ptr)
{
int i;
agent_info_t *agent_info_ptr;
thd_t *thread_ptr;
agent_info_ptr = xmalloc(sizeof(agent_info_t));
slurm_mutex_init(&agent_info_ptr->thread_mutex);
if (pthread_cond_init(&agent_info_ptr->thread_cond, NULL))
fatal("pthread_cond_init error %m");
agent_info_ptr->thread_count = agent_arg_ptr->node_count;
agent_info_ptr->retry = agent_arg_ptr->retry;
agent_info_ptr->threads_active = 0;
thread_ptr = xmalloc(agent_arg_ptr->node_count * sizeof(thd_t));
agent_info_ptr->thread_struct = thread_ptr;
agent_info_ptr->msg_type = agent_arg_ptr->msg_type;
agent_info_ptr->msg_args_pptr = &agent_arg_ptr->msg_args;
if ((agent_arg_ptr->msg_type != REQUEST_SHUTDOWN) &&
(agent_arg_ptr->msg_type != REQUEST_RECONFIGURE))
agent_info_ptr->get_reply = true;
for (i = 0; i < agent_info_ptr->thread_count; i++) {
thread_ptr[i].state = DSH_NEW;
thread_ptr[i].slurm_addr = agent_arg_ptr->slurm_addr[i];
strncpy(thread_ptr[i].node_name,
&agent_arg_ptr->node_names[i * MAX_NAME_LEN],
MAX_NAME_LEN);
}
return agent_info_ptr;
}
static task_info_t *_make_task_data(agent_info_t *agent_info_ptr, int inx)
{
task_info_t *task_info_ptr;
task_info_ptr = xmalloc(sizeof(task_info_t));
task_info_ptr->thread_mutex_ptr = &agent_info_ptr->thread_mutex;
task_info_ptr->thread_cond_ptr = &agent_info_ptr->thread_cond;
task_info_ptr->threads_active_ptr= &agent_info_ptr->threads_active;
task_info_ptr->thread_struct_ptr = &agent_info_ptr->thread_struct[inx];
task_info_ptr->get_reply = agent_info_ptr->get_reply;
task_info_ptr->msg_type = agent_info_ptr->msg_type;
task_info_ptr->msg_args_ptr = *agent_info_ptr->msg_args_pptr;
return task_info_ptr;
}
/*
* _wdog - Watchdog thread. Send SIGALRM to threads which have been active
* for too long.
* IN args - pointer to agent_info_t with info on threads to watch
* Sleep for WDOG_POLL seconds between polls.
*/
static void *_wdog(void *args)
{
int fail_cnt, no_resp_cnt, retry_cnt;
bool work_done;
int i, max_delay = 0;
agent_info_t *agent_ptr = (agent_info_t *) args;
thd_t *thread_ptr = agent_ptr->thread_struct;
time_t now;
#if AGENT_IS_THREAD
/* Locks: Write job and write node */
slurmctld_lock_t node_write_lock =
{ NO_LOCK, WRITE_LOCK, WRITE_LOCK, NO_LOCK };
#else
int done_cnt;
char *slurm_names;
#endif
while (1) {
work_done = true; /* assume all threads complete */
fail_cnt = 0; /* assume no threads failures */
no_resp_cnt = 0; /* assume all threads respond */
retry_cnt = 0; /* assume no required retries */
sleep(WDOG_POLL);
now = time(NULL);
slurm_mutex_lock(&agent_ptr->thread_mutex);
for (i = 0; i < agent_ptr->thread_count; i++) {
switch (thread_ptr[i].state) {
case DSH_ACTIVE:
work_done = false;
if (thread_ptr[i].end_time <= now) {
debug3("thd %d timed out\n",
thread_ptr[i].thread);
pthread_kill(thread_ptr[i].thread,
SIGALRM);
}
break;
case DSH_NEW:
work_done = false;
break;
case DSH_DONE:
if (max_delay < (int)thread_ptr[i].end_time)
max_delay = (int)thread_ptr[i].end_time;
break;
case DSH_NO_RESP:
no_resp_cnt++;
retry_cnt++;
break;
case DSH_FAILED:
fail_cnt++;
break;
case DSH_JOB_HUNG:
retry_cnt++;
break;
}
}
if (work_done)
break;
slurm_mutex_unlock(&agent_ptr->thread_mutex);
}
/* Notify slurmctld of non-responding nodes */
if (no_resp_cnt) {
#if AGENT_IS_THREAD
/* Update node table data for non-responding nodes */
lock_slurmctld(node_write_lock);
for (i = 0; i < agent_ptr->thread_count; i++) {
if (thread_ptr[i].state == DSH_NO_RESP)
node_not_resp(thread_ptr[i].node_name,
thread_ptr[i].start_time);
}
if (agent_ptr->msg_type == REQUEST_BATCH_JOB_LAUNCH) {
/* Requeue the request */
batch_job_launch_msg_t *launch_msg_ptr =
*agent_ptr->msg_args_pptr;
uint32_t job_id = launch_msg_ptr->job_id;
info("Non-responding node, requeue JobId=%u", job_id);
job_complete(job_id, 0, true, 0);
}
unlock_slurmctld(node_write_lock);
#else
/* Build a list of all non-responding nodes and send
* it to slurmctld */
slurm_names = xmalloc(fail_cnt * MAX_NAME_LEN);
fail_cnt = 0;
for (i = 0; i < agent_ptr->thread_count; i++) {
if (thread_ptr[i].state == DSH_NO_RESP) {
strncpy(&slurm_names
[MAX_NAME_LEN * fail_cnt],
thread_ptr[i].node_name,
MAX_NAME_LEN);
error
("agent/_wdog: node %s failed to respond",
thread_ptr[i].node_name);
fail_cnt++;
}
}
/* send RPC */
fatal("Code development needed here if agent is not thread");
xfree(slurm_names);
#endif
}
if (retry_cnt && agent_ptr->retry)
_queue_agent_retry(agent_ptr, retry_cnt);
#if AGENT_IS_THREAD
/* Update last_response on responding nodes */
lock_slurmctld(node_write_lock);
for (i = 0; i < agent_ptr->thread_count; i++) {
if (thread_ptr[i].state == DSH_FAILED)
set_node_down(thread_ptr[i].node_name,
"Prolog/epilog failure");
if ((thread_ptr[i].state == DSH_DONE) ||
(thread_ptr[i].state == DSH_JOB_HUNG))
node_did_resp(thread_ptr[i].node_name);
}
unlock_slurmctld(node_write_lock);
/* attempt to schedule when all nodes registered, not
* after each node, the overhead would be too high */
if ((agent_ptr->msg_type == REQUEST_KILL_TIMELIMIT) ||
(agent_ptr->msg_type == REQUEST_KILL_JOB))
schedule();
else if ((agent_ptr->msg_type == REQUEST_PING) ||
(agent_ptr->msg_type == REQUEST_NODE_REGISTRATION_STATUS))
ping_end();
#else
/* Build a list of all responding nodes and send it to slurmctld to
* update time stamps */
done_cnt = agent_ptr->thread_count - fail_cnt - no_resp_cnt;
slurm_names = xmalloc(done_cnt * MAX_NAME_LEN);
done_cnt = 0;
for (i = 0; i < agent_ptr->thread_count; i++) {
if (thread_ptr[i].state == DSH_DONE)
strncpy(&slurm_names[MAX_NAME_LEN * done_cnt],
thread_ptr[i].node_name, MAX_NAME_LEN);
done_cnt++;
}
}
/* need support for node failures here too */
/* send RPC */
fatal("Code development needed here if agent is not thread");
xfree(slurm_addr);
#endif
if (max_delay)
debug2("agent maximum delay %d seconds", max_delay);
slurm_mutex_unlock(&agent_ptr->thread_mutex);
return (void *) NULL;
}
/* Report a communications error for specified node */
static inline void _comm_err(char *node_name)
{
#if AGENT_IS_THREAD
if (is_node_resp (node_name))
#endif
error("agent/send_recv_msg: %s: %m", node_name);
}
/*
* _thread_per_node_rpc - thread to issue an RPC on a collection of nodes
* IN/OUT args - pointer to task_info_t, xfree'd on completion
*/
static void *_thread_per_node_rpc(void *args)
{
int rc = SLURM_SUCCESS, timeout = 0;
slurm_msg_t msg;
task_info_t *task_ptr = (task_info_t *) args;
thd_t *thread_ptr = task_ptr->thread_struct_ptr;
state_t thread_state = DSH_NO_RESP;
slurm_msg_type_t msg_type = task_ptr->msg_type;
bool is_kill_msg;
#if AGENT_IS_THREAD
struct node_record *node_ptr;
/* Locks: Write write node */
slurmctld_lock_t node_write_lock =
{ NO_LOCK, NO_LOCK, WRITE_LOCK, NO_LOCK };
/* Locks: Write job, write node */
slurmctld_lock_t job_write_lock = {
NO_LOCK, WRITE_LOCK, WRITE_LOCK, NO_LOCK };
#endif
xassert(args != NULL);
slurm_mutex_lock(task_ptr->thread_mutex_ptr);
thread_ptr->state = DSH_ACTIVE;
thread_ptr->start_time = time(NULL);
slurm_mutex_unlock(task_ptr->thread_mutex_ptr);
is_kill_msg = ( (msg_type == REQUEST_KILL_TIMELIMIT)
|| (msg_type == REQUEST_KILL_JOB) );
/* send request message */
msg.address = thread_ptr->slurm_addr;
msg.msg_type = msg_type;
msg.data = task_ptr->msg_args_ptr;
if (is_kill_msg) {
/* Extend time that this thread is allowed to exist. */
timeout = slurmctld_conf.kill_wait + 2;
thread_ptr->end_time = thread_ptr->start_time + timeout;
} else
thread_ptr->end_time = thread_ptr->start_time +
COMMAND_TIMEOUT;
if (task_ptr->get_reply) {
if (slurm_send_recv_rc_msg(&msg, &rc, timeout) < 0) {
_comm_err(thread_ptr->node_name);
goto cleanup;
}
} else {
if (slurm_send_only_node_msg(&msg) < 0)
_comm_err(thread_ptr->node_name);
else
thread_state = DSH_DONE;
goto cleanup;
}
#if AGENT_IS_THREAD
/* SPECIAL CASE: Immediately mark node as IDLE on job kill reply */
if (is_kill_msg && (rc != ESLURMD_KILL_JOB_FAILED)) {
lock_slurmctld(node_write_lock);
if ((node_ptr = find_node_record(thread_ptr->node_name))) {
kill_job_msg_t *kill_job;
kill_job = (kill_job_msg_t *) task_ptr->msg_args_ptr;
debug3("Kill job_id %u on node %s ",
kill_job->job_id, thread_ptr->node_name);
make_node_idle(node_ptr,
find_job_record(kill_job->job_id));
/* scheduler(); Overhead too high,
* only do when last node registers */
}
unlock_slurmctld(node_write_lock);
}
/* SPECIAL CASE: Kill non-startable batch job */
if ((msg_type == REQUEST_BATCH_JOB_LAUNCH) && rc) {
batch_job_launch_msg_t *launch_msg_ptr = task_ptr->msg_args_ptr;
uint32_t job_id = launch_msg_ptr->job_id;
info("Killing non-startable batch job %u: %s",
job_id, slurm_strerror(rc));
thread_state = DSH_DONE;
lock_slurmctld(job_write_lock);
job_complete(job_id, 0, false, 1);
unlock_slurmctld(job_write_lock);
goto cleanup;
}
#endif
switch (rc) {
case SLURM_SUCCESS:
debug3("agent processed RPC to node %s",
thread_ptr->node_name);
thread_state = DSH_DONE;
break;
case ESLURMD_EPILOG_FAILED:
error("Epilog failure on host %s, setting DOWN",
thread_ptr->node_name);
thread_state = DSH_FAILED;
break;
case ESLURMD_PROLOG_FAILED:
error("Prolog failure on host %s, setting DOWN",
thread_ptr->node_name);
thread_state = DSH_FAILED;
break;
case ESLURM_INVALID_JOB_ID: /* Not indicative of a real error */
debug2("agent processed RPC to node %s, error %s",
thread_ptr->node_name, "Invalid Job Id");
thread_state = DSH_DONE;
break;
case ESLURMD_KILL_JOB_FAILED: /* non-killable process */
info("agent KILL_JOB RPC to node %s FAILED",
thread_ptr->node_name);
thread_state = DSH_JOB_HUNG;
break;
default:
error("agent error from host %s for msg type %d: %s",
thread_ptr->node_name, task_ptr->msg_type,
slurm_strerror(rc));
thread_state = DSH_DONE;
}
cleanup:
slurm_mutex_lock(task_ptr->thread_mutex_ptr);
thread_ptr->state = thread_state;
thread_ptr->end_time = (time_t) difftime(time(NULL),
thread_ptr->start_time);
/* Signal completion so another thread can replace us */
(*task_ptr->threads_active_ptr)--;
pthread_cond_signal(task_ptr->thread_cond_ptr);
slurm_mutex_unlock(task_ptr->thread_mutex_ptr);
xfree(args);
return (void *) NULL;
}
/*
* SIGALRM handler. This is just a stub because we are really interested
* in interrupting connect() in k4cmd/rcmd or select() in rsh() below and
* causing them to return EINTR.
*/
static void _alarm_handler(int dummy)
{
}
/*
* _queue_agent_retry - Queue any failed RPCs for later replay
* IN agent_info_ptr - pointer to info on completed agent requests
* IN count - number of agent requests which failed, count to requeue
*/
static void _queue_agent_retry(agent_info_t * agent_info_ptr, int count)
{
agent_arg_t *agent_arg_ptr;
thd_t *thread_ptr = agent_info_ptr->thread_struct;
int i, j;
if (count == 0)
return;
/* build agent argument with just the RPCs to retry */
agent_arg_ptr = xmalloc(sizeof(agent_arg_t));
agent_arg_ptr->node_count = count;
agent_arg_ptr->retry = 1;
agent_arg_ptr->slurm_addr = xmalloc(sizeof(struct sockaddr_in)
* count);
agent_arg_ptr->node_names = xmalloc(MAX_NAME_LEN * count);
agent_arg_ptr->msg_type = agent_info_ptr->msg_type;
agent_arg_ptr->msg_args = *(agent_info_ptr->msg_args_pptr);
*(agent_info_ptr->msg_args_pptr) = NULL;
j = 0;
for (i = 0; i < agent_info_ptr->thread_count; i++) {
if ( (thread_ptr[i].state != DSH_NO_RESP)
&& (thread_ptr[i].state != DSH_JOB_HUNG) )
continue;
agent_arg_ptr->slurm_addr[j] = thread_ptr[i].slurm_addr;
strncpy(&agent_arg_ptr->node_names[j * MAX_NAME_LEN],
thread_ptr[i].node_name, MAX_NAME_LEN);
if ((++j) == count)
break;
}
if (count != j) {
error("agent: Retry count (%d) != actual count (%d)",
count, j);
agent_arg_ptr->node_count = j;
}
debug2("Queue RPC msg_type=%u, count=%d for retry",
agent_arg_ptr->msg_type, j);
/* add the requeust to a list */
slurm_mutex_lock(&retry_mutex);
if (retry_list == NULL) {
retry_list = list_create(&_list_delete_retry);
if (retry_list == NULL)
fatal("list_create failed");
}
if (list_enqueue(retry_list, (void *) agent_arg_ptr) == 0)
fatal("list_append failed");
slurm_mutex_unlock(&retry_mutex);
}
/*
* _list_delete_retry - delete an entry from the retry list,
* see common/list.h for documentation
*/
static void _list_delete_retry(void *retry_entry)
{
agent_arg_t *agent_arg_ptr; /* pointer to part_record */
agent_arg_ptr = (agent_arg_t *) retry_entry;
xfree(agent_arg_ptr->slurm_addr);
xfree(agent_arg_ptr->node_names);
#if AGENT_IS_THREAD
xfree(agent_arg_ptr->msg_args);
#endif
xfree(agent_arg_ptr);
}
/*
* agent_retry - Agent for retrying pending RPCs (top one on the queue),
* IN args - unused
* RET always NULL (function format just for use as pthread)
*/
void *agent_retry(void *args)
{
agent_arg_t *agent_arg_ptr = NULL;
slurm_mutex_lock(&retry_mutex);
if (retry_list)
agent_arg_ptr = (agent_arg_t *) list_dequeue(retry_list);
slurm_mutex_unlock(&retry_mutex);
if (agent_arg_ptr)
_spawn_retry_agent(agent_arg_ptr);
return NULL;
}
/* retry_pending - retry all pending RPCs for the given node name
* IN node_name - name of a node to executing pending RPCs for */
void retry_pending(char *node_name)
{
int list_size = 0, i, j, found;
agent_arg_t *agent_arg_ptr = NULL;
slurm_mutex_lock(&retry_mutex);
if (retry_list) {
list_size = list_count(retry_list);
}
for (i = 0; i < list_size; i++) {
agent_arg_ptr = (agent_arg_t *) list_dequeue(retry_list);
found = 0;
for (j = 0; j < agent_arg_ptr->node_count; j++) {
if (strncmp
(&agent_arg_ptr->node_names[j * MAX_NAME_LEN],
node_name, MAX_NAME_LEN))
continue;
found = 1;
break;
}
if (found) /* issue this RPC */
_spawn_retry_agent(agent_arg_ptr);
else /* put the RPC back on the queue */
list_enqueue(retry_list, (void *) agent_arg_ptr);
}
slurm_mutex_unlock(&retry_mutex);
}
/* _spawn_retry_agent - pthread_crate an agent for the given task */
static void _spawn_retry_agent(agent_arg_t * agent_arg_ptr)
{
int retries = 0;
pthread_attr_t attr_agent;
pthread_t thread_agent;
if (agent_arg_ptr == NULL)
return;
debug2("Spawning RPC retry agent for msg_type %u",
agent_arg_ptr->msg_type);
if (pthread_attr_init(&attr_agent))
fatal("pthread_attr_init error %m");
if (pthread_attr_setdetachstate(&attr_agent,
PTHREAD_CREATE_DETACHED))
error("pthread_attr_setdetachstate error %m");
#ifdef PTHREAD_SCOPE_SYSTEM
if (pthread_attr_setscope(&attr_agent, PTHREAD_SCOPE_SYSTEM))
error("pthread_attr_setscope error %m");
#endif
while (pthread_create(&thread_agent, &attr_agent,
agent, (void *) agent_arg_ptr)) {
error("pthread_create error %m");
if (++retries > MAX_RETRIES)
fatal("Can't create pthread");
sleep(1); /* sleep and try again */
}
}
/* _slurmctld_free_job_launch_msg is a variant of slurm_free_job_launch_msg
* because all environment variables currently loaded in one xmalloc
* buffer (see get_job_env()), which is different from how slurmd
* assembles the data from a message
*/
static void _slurmctld_free_job_launch_msg(batch_job_launch_msg_t * msg)
{
if (msg) {
if (msg->environment) {
xfree(msg->environment[0]);
xfree(msg->environment);
}
slurm_free_job_launch_msg(msg);
}
}
/* agent_purge - purge all pending RPC requests */
void agent_purge(void)
{
#if AGENT_IS_THREAD
agent_arg_t *agent_arg_ptr = NULL;
if (retry_list == NULL)
return;
slurm_mutex_lock(&retry_mutex);
while ((agent_arg_ptr = (agent_arg_t *) list_dequeue(retry_list)))
_purge_agent_args(agent_arg_ptr);
slurm_mutex_unlock(&retry_mutex);
#endif
}
static void _purge_agent_args(agent_arg_t *agent_arg_ptr)
{
if (agent_arg_ptr == NULL)
return;
xfree(agent_arg_ptr->slurm_addr);
xfree(agent_arg_ptr->node_names);
if (agent_arg_ptr->msg_args) {
if (agent_arg_ptr->msg_type == REQUEST_BATCH_JOB_LAUNCH)
_slurmctld_free_job_launch_msg(agent_arg_ptr->msg_args);
else
xfree(agent_arg_ptr->msg_args);
}
xfree(agent_arg_ptr);
}