/*
 * Start the thread to extend cookie leases.
 */
extern int start_lease_extender(void)
{
	pthread_attr_t attr_agent;
	pthread_t thread_agent;
	int retries = 0;

	// Start lease extender in the slurmctld
	if (!_in_slurmctld())
		return SLURM_SUCCESS;

	/* spawn an agent */
	slurm_attr_init(&attr_agent);
	if (pthread_attr_setdetachstate(&attr_agent,
					PTHREAD_CREATE_DETACHED)) {
		CRAY_ERR("pthread_attr_setdetachstate error %m");
	}

	retries = 0;
	while (pthread_create(&thread_agent, &attr_agent,
			      &_lease_extender, NULL)) {
		error("pthread_create error %m");
		if (++retries > 1) {
			CRAY_ERR("Can't create pthread");
			slurm_attr_destroy(&attr_agent);
			return SLURM_ERROR;
		}

		usleep(1000);	/* sleep and retry */
	}
	slurm_attr_destroy(&attr_agent);
	return SLURM_SUCCESS;
}

/*
 * start the PMI2 agent thread
 */
extern int
pmi2_start_agent(void)
{
	int retries = 0;
	pthread_attr_t attr;
	pthread_t pmi2_agent_tid = 0;

	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	while ((errno = pthread_create(&pmi2_agent_tid, &attr,
				       &_agent, NULL))) {
		if (++retries > MAX_RETRIES) {
			error ("mpi/pmi2: pthread_create error %m");
			slurm_attr_destroy(&attr);
			return SLURM_ERROR;
		}
		sleep(1);
	}
	slurm_attr_destroy(&attr);
	debug("mpi/pmi2: started agent thread (%lu)",
	      (unsigned long) pmi2_agent_tid);

	/* wait for the agent to start */
	while (!_agent_running_test()) {
		sched_yield();
	}

	return SLURM_SUCCESS;
}

extern int bridge_status_init(void)
{
	pthread_attr_t attr;

	if (bridge_status_inited)
		return SLURM_ERROR;

	bridge_status_inited = true;
	if (!kill_job_list)
		kill_job_list = bg_status_create_kill_job_list();

	pthread_mutex_lock(&thread_flag_mutex);
	if (block_thread) {
		debug2("Bluegene threads already running, not starting "
		       "another");
		pthread_mutex_unlock(&thread_flag_mutex);
		return SLURM_ERROR;
	}

	slurm_attr_init(&attr);
	/* since we do a join on this later we don't make it detached */
	if (pthread_create(&block_thread, &attr, _block_state_agent, NULL))
		error("Failed to create block_agent thread");
	slurm_attr_init(&attr);
	/* since we do a join on this later we don't make it detached */
	if (pthread_create(&state_thread, &attr, _mp_state_agent, NULL))
		error("Failed to create state_agent thread");
	pthread_mutex_unlock(&thread_flag_mutex);
	slurm_attr_destroy(&attr);

	return SLURM_SUCCESS;
}

static void
_spawn_registration_engine(void)
{
	int            rc;
	pthread_attr_t attr;
	pthread_t      id;
	int            retries = 0;

	slurm_attr_init(&attr);
	rc = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	if (rc != 0) {
		errno = rc;
		fatal("Unable to set detachstate on attr: %m");
		slurm_attr_destroy(&attr);
		return;
	}

	while (pthread_create(&id, &attr, &_registration_engine, NULL)) {
		error("msg_engine: pthread_create: %m");
		if (++retries > 3)
			fatal("msg_engine: pthread_create: %m");
		usleep(10);	/* sleep and again */
	}

	return;
}

/*
 * init() is called when the plugin is loaded, before any other functions
 * are called.  Put global initialization here.
 */
extern int init (void)
{
    pthread_attr_t attr;

    verbose("jobcomp/script plugin loaded init");

    slurm_mutex_lock(&thread_flag_mutex);

    if (comp_list)
        error("Creating duplicate comp_list, possible memory leak");
    if (!(comp_list = list_create((ListDelF) _jobcomp_info_destroy))) {
        slurm_mutex_unlock(&thread_flag_mutex);
        return SLURM_ERROR;
    }

    if (script_thread) {
        debug2( "Script thread already running, not starting another");
        slurm_mutex_unlock(&thread_flag_mutex);
        return SLURM_ERROR;
    }

    slurm_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    pthread_create(&script_thread, &attr, _script_agent, NULL);

    slurm_mutex_unlock(&thread_flag_mutex);
    slurm_attr_destroy(&attr);

    return SLURM_SUCCESS;
}

void step_terminate_monitor_start(uint32_t jobid, uint32_t stepid)
{
	slurm_ctl_conf_t *conf;
	pthread_attr_t attr;

	pthread_mutex_lock(&lock);

	if (running_flag) {
		pthread_mutex_unlock(&lock);
		return;
	}

	conf = slurm_conf_lock();
	if (conf->unkillable_program == NULL) {
		/* do nothing */
		slurm_conf_unlock();
		pthread_mutex_unlock(&lock);
		return;
	}
	timeout = conf->unkillable_timeout;
	program_name = xstrdup(conf->unkillable_program);
	slurm_conf_unlock();

	slurm_attr_init(&attr);
	pthread_create(&tid, &attr, monitor, NULL);
	slurm_attr_destroy(&attr);
	running_flag = 1;
	recorded_jobid = jobid;
	recorded_stepid = stepid;

	pthread_mutex_unlock(&lock);

	return;
}

static void _spawn_timeslicer_thread(void)
{
	pthread_attr_t thread_attr_msg;
	int retries = 0;

	slurm_mutex_lock( &thread_flag_mutex );
	if (thread_running) {
		error("timeslicer thread already running, not starting "
		      "another");
		slurm_mutex_unlock(&thread_flag_mutex);
		return;
	}

	slurm_attr_init(&thread_attr_msg);
	while (pthread_create(&timeslicer_thread_id, &thread_attr_msg,
			      _timeslicer_thread, NULL)) {
		error("pthread_create error %m");
		if (++retries > 3)
			fatal("Can't create pthread");
		usleep(10000);	/* sleep and retry */
	}
	slurm_attr_destroy(&thread_attr_msg);
	thread_running = true;
	slurm_mutex_unlock(&thread_flag_mutex);
}

int init( void )
{
	pthread_attr_t attr;

	if (slurmctld_config.scheduling_disabled)
		return SLURM_SUCCESS;


	verbose( "sched: Backfill scheduler plugin loaded" );

	slurm_mutex_lock( &thread_flag_mutex );
	if ( backfill_thread ) {
		debug2( "Backfill thread already running, not starting "
			"another" );
		slurm_mutex_unlock( &thread_flag_mutex );
		return SLURM_ERROR;
	}

	slurm_attr_init( &attr );
	/* since we do a join on this later we don't make it detached */
	if (pthread_create( &backfill_thread, &attr, backfill_agent, NULL))
		error("Unable to start backfill thread: %m");
	slurm_mutex_unlock( &thread_flag_mutex );
	slurm_attr_destroy( &attr );

	return SLURM_SUCCESS;
}

extern void jobacct_gather_p_change_poll(uint16_t frequency)
{
#ifdef HAVE_AIX
	if(freq == 0 && frequency != 0) {
		pthread_attr_t attr;
		pthread_t _watch_tasks_thread_id;
		/* create polling thread */
		slurm_attr_init(&attr);
		if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
			error("pthread_attr_setdetachstate error %m");

		if  (pthread_create(&_watch_tasks_thread_id, &attr,
				    &_watch_tasks, NULL)) {
			debug("jobacct-gather failed to create _watch_tasks "
			      "thread: %m");
			frequency = 0;
		} else
			debug3("jobacct-gather AIX dynamic logging enabled");
		slurm_attr_destroy(&attr);
		jobacct_shutdown = false;
	}

	freq = frequency;
	debug("jobacct-gather: frequency changed = %d", frequency);
	if (freq == 0)
		jobacct_shutdown = true;
#endif
	return;
}

extern int acct_gather_infiniband_startpoll(uint32_t frequency)
{
	int retval = SLURM_SUCCESS;
	pthread_attr_t attr;

	if (acct_gather_infiniband_init() < 0)
		return SLURM_ERROR;

	if (!acct_shutdown) {
		error("acct_gather_infiniband_startpoll: "
		      "poll already started!");
		return retval;
	}

	acct_shutdown = false;

	freq = frequency;

	if (frequency == 0) {   /* don't want dynamic monitoring? */
		debug2("acct_gather_infiniband dynamic logging disabled");
		return retval;
	}

	/* create polling thread */
	slurm_attr_init(&attr);
	if (pthread_create(&watch_node_thread_id, &attr, &_watch_node, NULL)) {
		debug("acct_gather_infiniband failed to create _watch_node "
		      "thread: %m");
	} else
		debug3("acct_gather_infiniband dynamic logging enabled");
	slurm_attr_destroy(&attr);

	return retval;
}

void step_terminate_monitor_start(stepd_step_rec_t *job)
{
	slurm_ctl_conf_t *conf;
	pthread_attr_t attr;

	slurm_mutex_lock(&lock);

	if (running_flag) {
		slurm_mutex_unlock(&lock);
		return;
	}

	conf = slurm_conf_lock();
	timeout = conf->unkillable_timeout;
	program_name = xstrdup(conf->unkillable_program);
	slurm_conf_unlock();

	slurm_attr_init(&attr);
	pthread_create(&tid, &attr, _monitor, job);
	slurm_attr_destroy(&attr);
	running_flag = 1;
	recorded_jobid = job->jobid;
	recorded_stepid = job->stepid;

	slurm_mutex_unlock(&lock);

	return;
}

int main(int argc, char *argv[])
{
	log_options_t log_opts = LOG_OPTS_INITIALIZER;
	hostlist_t hl = NULL;
	char *node_name;
	pthread_attr_t attr_work;
	pthread_t thread_work = 0;

	xstrfmtcat(prog_name, "%s[%u]", argv[0], (uint32_t) getpid());
	_read_config();
	log_opts.stderr_level = LOG_LEVEL_QUIET;
	log_opts.syslog_level = LOG_LEVEL_QUIET;
	if (slurm_get_debug_flags() && DEBUG_FLAG_NODE_FEATURES)
		log_opts.logfile_level += 3;
	(void) log_init(argv[0], log_opts, LOG_DAEMON, log_file);

	if ((hl = hostlist_create(argv[1])) == NULL) {
		error("%s: Invalid hostlist (%s)", prog_name, argv[1]);
		exit(2);
	}
	while ((node_name = hostlist_pop(hl))) {
		slurm_mutex_lock(&thread_cnt_mutex);
		while (1) {
			if (thread_cnt <= MAX_THREADS) {
				thread_cnt++;
				break;
			} else {	/* wait for state change and retry */
				pthread_cond_wait(&thread_cnt_cond,
						  &thread_cnt_mutex);
			}
		}
		slurm_mutex_unlock(&thread_cnt_mutex);

		slurm_attr_init(&attr_work);
		(void) pthread_attr_setdetachstate
			(&attr_work, PTHREAD_CREATE_DETACHED);
		if (pthread_create(&thread_work, &attr_work, _node_update,
				   (void *) node_name)) {
			_node_update((void *) node_name);
		}
		slurm_attr_destroy(&attr_work);
	}

	/* Wait for work threads to complete */
	slurm_mutex_lock(&thread_cnt_mutex);
	while (1) {
		if (thread_cnt == 0)
			break;
		else	/* wait for state change and retry */
			pthread_cond_wait(&thread_cnt_cond, &thread_cnt_mutex);
	}
	slurm_mutex_unlock(&thread_cnt_mutex);
	hostlist_destroy(hl);

	exit(0);
}

static void _spawn_state_save_thread(char *dir)
{
	pthread_attr_t attr;
	pthread_t id;

	slurm_attr_init(&attr);

	if (pthread_create(&id, &attr, &_state_save_thread, (void *)dir) != 0)
		error("Could not start switch/nrt state saving pthread");

	slurm_attr_destroy(&attr);
}

static int _signal_job_by_str(void)
{
	job_cancel_info_t *cancel_info;
	int err, i, rc = 0;
	pthread_t dummy;

	slurm_attr_init(&attr);
	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
		error("pthread_attr_setdetachstate error %m");
	slurm_mutex_init(&num_active_threads_lock);
	slurm_cond_init(&num_active_threads_cond, NULL);

	for (i = 0; opt.job_list[i]; i++) {
		cancel_info = (job_cancel_info_t *)
			xmalloc(sizeof(job_cancel_info_t));
		cancel_info->job_id_str = xstrdup(opt.job_list[i]);
		cancel_info->rc      = &rc;
		cancel_info->sig     = opt.signal;
		cancel_info->num_active_threads = &num_active_threads;
		cancel_info->num_active_threads_lock =
			&num_active_threads_lock;
		cancel_info->num_active_threads_cond =
			&num_active_threads_cond;

		slurm_mutex_lock(&num_active_threads_lock);
		num_active_threads++;
		while (num_active_threads > MAX_THREADS) {
			slurm_cond_wait(&num_active_threads_cond,
					&num_active_threads_lock);
		}
		slurm_mutex_unlock(&num_active_threads_lock);

		err = pthread_create(&dummy, &attr, _cancel_job_id,cancel_info);
		if (err)	/* Run in-line if thread create fails */
			_cancel_job_id(cancel_info);
	}

	/* Wait all spawned threads to finish */
	slurm_mutex_lock( &num_active_threads_lock );
	while (num_active_threads > 0) {
		slurm_cond_wait(&num_active_threads_cond,
				&num_active_threads_lock);
	}
	slurm_mutex_unlock(&num_active_threads_lock);

	slurm_attr_destroy(&attr);

	return rc;
}

static int
_setup_srun_task_launch_detection(void)
{
	int retries = 0;
	pthread_t tid;
	pthread_attr_t attr;

	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	while ((errno = pthread_create(&tid, &attr,
				       &_task_launch_detection, NULL))) {
		if (++retries > 5) {
			error ("mpi/pmi2: pthread_create error %m");
			slurm_attr_destroy(&attr);
			return SLURM_ERROR;
		}
		sleep(1);
	}
	slurm_attr_destroy(&attr);
	debug("mpi/pmi2: task launch detection thread (%lu) started",
	      (unsigned long) tid);

	return SLURM_SUCCESS;
}

/*
 * init() is called when the plugin is loaded, before any other functions
 * are called.  Put global initialization here.
 */
extern int init ( void )
{
	pthread_attr_t attr;

	slurm_attr_init(&attr);
	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
		error("pthread_attr_setdetachstate: %m");
	if (pthread_create(&ckpt_agent_tid, &attr, _ckpt_agent_thr, NULL)) {
		error("pthread_create: %m");
		return SLURM_ERROR;
	}
	slurm_attr_destroy(&attr);

	return SLURM_SUCCESS;
}

static void _spawn_signal_thread(uint64_t cont_id, int signal)
{
	agent_arg_t *agent_arg_ptr;
	pthread_attr_t attr_agent;
	pthread_t thread_agent;

	slurm_attr_init(&attr_agent);
	if (pthread_attr_setdetachstate(&attr_agent, PTHREAD_CREATE_DETACHED))
		error("pthread_attr_setdetachstate error %m");
	agent_arg_ptr = xmalloc(sizeof(agent_arg_t));
	agent_arg_ptr->cont_id = cont_id;
	agent_arg_ptr->signal  = signal;
	(void) pthread_create(&thread_agent, &attr_agent,
			     _sig_agent, (void *) agent_arg_ptr);
	slurm_attr_destroy(&attr_agent);
}

static pthread_t _spawn_msg_handler(srun_job_t *job)
{
	pthread_attr_t attr;
	pthread_t msg_thread;
	static int slurmctld_fd;

	slurmctld_fd = job->step_ctx->launch_state->slurmctld_socket_fd;
	if (slurmctld_fd < 0)
		return (pthread_t) 0;
	job->step_ctx->launch_state->slurmctld_socket_fd = -1;

	slurm_attr_init(&attr);
	if (pthread_create(&msg_thread, &attr, _msg_thr_internal,
			   (void *) &slurmctld_fd))
		error("pthread_create of message thread: %m");
	slurm_attr_destroy(&attr);
	return msg_thread;
}

/* Transmit the KVS keypairs to all tasks, waiting at a barrier
 * This will take some time, so we work with a copy of the KVS keypairs.
 * We also work with a private copy of the barrier data and clear the
 * global data pointers so any new barrier requests get treated as
 * completely independent of this one. */
static void _kvs_xmit_tasks(void)
{
	struct agent_arg *args;
	pthread_attr_t attr;
	pthread_t agent_id;

#if _DEBUG
	info("All tasks at barrier, transmit KVS keypairs now");
#endif

	/* Target KVS_TIME should be about ave processing time */
	debug("kvs_put processing time min=%d, max=%d ave=%d (usec)",
		min_time_kvs_put, max_time_kvs_put,
		(tot_time_kvs_put / barrier_cnt));
	min_time_kvs_put = 1000000;
	max_time_kvs_put = 0;
	tot_time_kvs_put = 0;

	/* reset barrier info */
	args = xmalloc(sizeof(struct agent_arg));
	args->barrier_xmit_ptr = barrier_ptr;
	args->barrier_xmit_cnt = barrier_cnt;
	barrier_ptr = NULL;
	barrier_resp_cnt = 0;
	barrier_cnt = 0;

	/* copy the new kvs data */
	if (kvs_updated) {
		args->kvs_xmit_ptr = _kvs_comm_dup();
		args->kvs_xmit_cnt = kvs_comm_cnt;
		kvs_updated = 0;
	} else {	/* No new data to transmit */
		args->kvs_xmit_ptr = xmalloc(0);
		args->kvs_xmit_cnt = 0;
	}

	/* Spawn a pthread to transmit it */
	slurm_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	if (pthread_create(&agent_id, &attr, _agent, (void *) args))
		fatal("pthread_create");
	slurm_attr_destroy(&attr);
}

extern int jobacct_gather_startpoll(uint16_t frequency)
{
	int retval = SLURM_SUCCESS;
	pthread_attr_t attr;
	pthread_t _watch_tasks_thread_id;

	if (!plugin_polling)
		return SLURM_SUCCESS;

	if (jobacct_gather_init() < 0)
		return SLURM_ERROR;

	if (!jobacct_shutdown) {
		error("jobacct_gather_startpoll: poll already started!");
		return retval;
	}

	jobacct_shutdown = false;

	freq = frequency;

	task_list = list_create(jobacctinfo_destroy);
	if (frequency == 0) {	/* don't want dynamic monitoring? */
		debug2("jobacct_gather dynamic logging disabled");
		return retval;
	}

	/* create polling thread */
	slurm_attr_init(&attr);
	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
		error("pthread_attr_setdetachstate error %m");

	if  (pthread_create(&_watch_tasks_thread_id, &attr,
			    &_watch_tasks, NULL)) {
		debug("jobacct_gather failed to create _watch_tasks "
		      "thread: %m");
		frequency = 0;
	} else
		debug3("jobacct_gather dynamic logging enabled");
	slurm_attr_destroy(&attr);

	return retval;
}