contribs/perlapi/libslurm/perl/step_ctx.c - SchedMD/slurm - Git at Google

 /*
  * step_ctx.c - convert data between step context related messages and perl HVs
  */

 #include <EXTERN.h>
 #include <perl.h>
 #include <XSUB.h>
 #include "ppport.h"

 #include <slurm/slurm.h>
 #include "slurm-perl.h"

 /********** conversion functions for callback **********/
 static int
 launch_tasks_response_msg_to_hv(launch_tasks_response_msg_t *resp_msg, HV *hv)
 {
 	AV *av, *av2;
 	int i;

 	STORE_FIELD(hv, resp_msg, return_code, uint32_t);
 	if (resp_msg->node_name)
 		STORE_FIELD(hv, resp_msg, node_name, charp);
 	STORE_FIELD(hv, resp_msg, srun_node_id, uint32_t);
 	STORE_FIELD(hv, resp_msg, count_of_pids, uint32_t);
 	if (resp_msg->count_of_pids > 0) {
 		av = newAV();
 		av2 = newAV();
 		for (i = 0; i < resp_msg->count_of_pids; i ++) {
 			av_store_uint32_t(av, i, resp_msg->local_pids[i]);
 			av_store_uint32_t(av2, i, resp_msg->task_ids[i]);
 		}
 		hv_store_sv(hv, "local_pids", newRV_noinc((SV*)av));
 		hv_store_sv(hv, "task_ids", newRV_noinc((SV*)av2));
 	}
 	return 0;
 }

 static int
 task_exit_msg_to_hv(task_exit_msg_t *exit_msg, HV *hv)
 {
 	AV *av;
 	int i;
 	HV *step_id_hv = (HV*)sv_2mortal((SV*)newHV());

 	step_id_to_hv(&exit_msg->step_id, step_id_hv);
 	hv_store_sv(hv, "step_id", newRV((SV*)step_id_hv));

 	STORE_FIELD(hv, exit_msg, num_tasks, uint32_t);
 	if (exit_msg->num_tasks > 0) {
 		av = newAV();
 		for (i = 0; i < exit_msg->num_tasks; i ++) {
 			av_store_uint32_t(av, i, exit_msg->task_id_list[i]);
 		}
 		hv_store_sv(hv, "task_id_list", newRV_noinc((SV*)av));
 	}
 	STORE_FIELD(hv, exit_msg, return_code, uint32_t);

 	return 0;
 }


 /********** callback related functions **********/

 /*
  * In the C api, callbacks are associated with step_ctx->launch_state.
  * Since the callback functions have no parameter like "ctx" or "sls",
  * there is no simple way to map Perl callback to C callback.
  *
  * So, only one $step_ctx->launch() call is allowed in Perl, until
  * $step_ctx->launch_wait_finish().
  */

 static SV *task_start_cb_sv = NULL;
 static SV *task_finish_cb_sv = NULL;

 static PerlInterpreter *main_perl = NULL;
 static pthread_key_t cbs_key;

 typedef struct thread_callbacks {
 	SV *step_complete;
 	SV *step_signal;
 	SV *step_timeout;
 	SV *task_start;
 	SV *task_finish;
 } thread_callbacks_t;

 static void
 set_thread_perl(void)
 {
 	PerlInterpreter *thr_perl = PERL_GET_CONTEXT;

 	if (thr_perl == NULL) {
 		if (main_perl == NULL) { /* should never happen */
 			fprintf(stderr, "error: no main perl context\n");
 			exit(-1);
 		}
 		thr_perl = perl_clone(main_perl, CLONEf_COPY_STACKS | CLONEf_KEEP_PTR_TABLE);
 		/* seems no need to call PERL_SET_CONTEXT(thr_perl); */

 		/*
 		 * seems perl will destroy the interpreter associated with
 		 * a thread automatically.
 		 */
 	}
 }

 #define GET_THREAD_CALLBACKS ((thread_callbacks_t *)pthread_getspecific(cbs_key))
 #define SET_THREAD_CALLBACKS(cbs) (pthread_setspecific(cbs_key, (void *)cbs))

 static void
 clear_thread_callbacks(void *arg)
 {
 	thread_callbacks_t *cbs = (thread_callbacks_t *)arg;
 	if (cbs->task_start) {
 		/* segfault if called. dunno why */
 		/* SvREFCNT_dec(cbs->task_start); */
 	}
 	if (cbs->task_finish) {
 		/* SvREFCNT_dec(cbs->task_finish); */
 	}
 	xfree(cbs);
 }

 static void
 set_thread_callbacks(void)
 {
 	CLONE_PARAMS params;
 	thread_callbacks_t *cbs = GET_THREAD_CALLBACKS;

 	if (cbs != NULL)
 		return;

 	cbs = xmalloc(sizeof(thread_callbacks_t));
 	if (!cbs) {
 		fprintf(stderr, "set_thread_callbacks: memory exhausted\n");
 		exit(-1);
 	}

 	params.stashes = NULL;
 	params.flags = CLONEf_COPY_STACKS | CLONEf_KEEP_PTR_TABLE;
 	params.proto_perl = PERL_GET_CONTEXT;

 	if (task_start_cb_sv != NULL && task_start_cb_sv != &PL_sv_undef) {
 		cbs->task_start = sv_dup(task_start_cb_sv, &params);
 	}
 	if (task_finish_cb_sv != NULL && task_finish_cb_sv != &PL_sv_undef) {
 		cbs->task_finish = sv_dup(task_finish_cb_sv, &params);
 	}
 	if (SET_THREAD_CALLBACKS(cbs) != 0) {
 		fprintf(stderr, "set_thread_callbacks: failed to set thread specific value\n");
 		exit(-1);
 	}
 }

 void
 set_slcb(HV *callbacks)
 {
 	SV **svp, *cb;

 	svp = hv_fetch(callbacks, "task_start", 10, FALSE);
 	cb = svp ? *svp : &PL_sv_undef;
 	if (task_start_cb_sv == NULL) {
 		task_start_cb_sv = newSVsv(cb);
 	} else {
 		sv_setsv(task_start_cb_sv, cb);
 	}

 	svp = hv_fetch(callbacks, "task_finish", 11, FALSE);
 	cb = svp ? *svp : &PL_sv_undef;
 	if (task_finish_cb_sv == NULL) {
 		task_finish_cb_sv = newSVsv(cb);
 	} else {
 		sv_setsv(task_finish_cb_sv, cb);
 	}

 	if (main_perl == NULL) {
 		main_perl = PERL_GET_CONTEXT;
 		if ( pthread_key_create(&cbs_key, clear_thread_callbacks) != 0) {
 			fprintf(stderr, "set_slcb: failed to create cbs_key\n");
 			exit(-1);
 		}
 	}
 }

 static void
 step_complete_cb(srun_job_complete_msg_t *comp_msg)
 {
 	HV *hv;
 	thread_callbacks_t *cbs = NULL;

 	set_thread_perl();
 	set_thread_callbacks();

 	cbs = GET_THREAD_CALLBACKS;
 	if (cbs->step_complete == NULL)
 		return;

 	hv = newHV();
 	if (srun_job_complete_msg_to_hv(comp_msg, hv) < 0) {
 		Perl_warn( aTHX_ "failed to prepare parameter for step_complete callback");
 		SvREFCNT_dec(hv);
 		return;
 	}

 	dSP;

 	ENTER;
 	SAVETMPS;
 	PUSHMARK(SP);
 	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
 	PUTBACK;

 	call_sv(cbs->step_complete, G_SCALAR);

 	FREETMPS;
 	LEAVE;
 }

 static void
 step_signal_cb(int signo)
 {
 	thread_callbacks_t *cbs = NULL;

 	set_thread_perl();
 	set_thread_callbacks();

 	cbs = GET_THREAD_CALLBACKS;
 	if (cbs->step_signal == NULL)
 		return;

 	dSP;

 	ENTER;
 	SAVETMPS;
 	PUSHMARK(SP);
 	XPUSHs(sv_2mortal(newSViv(signo)));
 	PUTBACK;

 	call_sv(cbs->step_signal, G_SCALAR);

 	FREETMPS;
 	LEAVE;
 }

 static void
 step_timeout_cb(srun_timeout_msg_t *timeout_msg)
 {
 	HV *hv;
 	thread_callbacks_t *cbs = NULL;

 	set_thread_perl();
 	set_thread_callbacks();

 	cbs = GET_THREAD_CALLBACKS;
 	if (cbs->step_timeout == NULL)
 		return;

 	hv = newHV();
 	if (srun_timeout_msg_to_hv(timeout_msg, hv) < 0) {
 		Perl_warn( aTHX_ "failed to prepare parameter for step_timeout callback");
 		SvREFCNT_dec(hv);
 		return;
 	}

 	dSP;

 	ENTER;
 	SAVETMPS;
 	PUSHMARK(SP);
 	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
 	PUTBACK;

 	call_sv(cbs->step_timeout, G_SCALAR);

 	FREETMPS;
 	LEAVE;
 }

 static void
 task_start_cb(launch_tasks_response_msg_t *resp_msg)
 {
 	HV *hv;
 	thread_callbacks_t *cbs = NULL;

 	set_thread_perl();
 	set_thread_callbacks();

 	cbs = GET_THREAD_CALLBACKS;
 	if (cbs->task_start == NULL)
 		return;

 	hv = newHV();
 	if (launch_tasks_response_msg_to_hv(resp_msg, hv) < 0) {
 		Perl_warn( aTHX_ "failed to prepare parameter for task_start callback");
 		SvREFCNT_dec(hv);
 		return;
 	}

 	dSP;

 	ENTER;
 	SAVETMPS;
 	PUSHMARK(SP);
 	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
 	PUTBACK;

 	call_sv(cbs->task_start, G_SCALAR);

 	FREETMPS;
 	LEAVE;
 }

 static void
 task_finish_cb(task_exit_msg_t *exit_msg)
 {
 	HV *hv;
 	thread_callbacks_t *cbs = NULL;

 	set_thread_perl();
 	set_thread_callbacks();

 	cbs = GET_THREAD_CALLBACKS;
 	if (cbs->task_finish == NULL)
 		return;

 	hv = newHV();
 	if (task_exit_msg_to_hv(exit_msg, hv) < 0) {
 		Perl_warn( aTHX_ "failed to prepare parameter for task_exit callback");
 		SvREFCNT_dec(hv);
 		return;
 	}

 	dSP;

 	ENTER;
 	SAVETMPS;
 	PUSHMARK(SP);
 	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
 	PUTBACK;

 	call_sv(cbs->task_finish, G_VOID);

 	FREETMPS;
 	LEAVE;
 }

 slurm_step_launch_callbacks_t slcb = {
 	step_complete_cb,
 	step_signal_cb,
 	step_timeout_cb,
 	task_start_cb,
 	task_finish_cb
 };
	/*
	* step_ctx.c - convert data between step context related messages and perl HVs
	*/

	#include <EXTERN.h>
	#include <perl.h>
	#include <XSUB.h>
	#include "ppport.h"

	#include <slurm/slurm.h>
	#include "slurm-perl.h"

	/******** conversion functions for callback ********/
	static int
	launch_tasks_response_msg_to_hv(launch_tasks_response_msg_t resp_msg, HV hv)
	{
	AV av, av2;
	int i;

	STORE_FIELD(hv, resp_msg, return_code, uint32_t);
	if (resp_msg->node_name)
	STORE_FIELD(hv, resp_msg, node_name, charp);
	STORE_FIELD(hv, resp_msg, srun_node_id, uint32_t);
	STORE_FIELD(hv, resp_msg, count_of_pids, uint32_t);
	if (resp_msg->count_of_pids > 0) {
	av = newAV();
	av2 = newAV();
	for (i = 0; i < resp_msg->count_of_pids; i ++) {
	av_store_uint32_t(av, i, resp_msg->local_pids[i]);
	av_store_uint32_t(av2, i, resp_msg->task_ids[i]);
	}
	hv_store_sv(hv, "local_pids", newRV_noinc((SV*)av));
	hv_store_sv(hv, "task_ids", newRV_noinc((SV*)av2));
	}
	return 0;
	}

	static int
	task_exit_msg_to_hv(task_exit_msg_t exit_msg, HV hv)
	{
	AV *av;
	int i;
	HV step_id_hv = (HV)sv_2mortal((SV*)newHV());

	step_id_to_hv(&exit_msg->step_id, step_id_hv);
	hv_store_sv(hv, "step_id", newRV((SV*)step_id_hv));

	STORE_FIELD(hv, exit_msg, num_tasks, uint32_t);
	if (exit_msg->num_tasks > 0) {
	av = newAV();
	for (i = 0; i < exit_msg->num_tasks; i ++) {
	av_store_uint32_t(av, i, exit_msg->task_id_list[i]);
	}
	hv_store_sv(hv, "task_id_list", newRV_noinc((SV*)av));
	}
	STORE_FIELD(hv, exit_msg, return_code, uint32_t);

	return 0;
	}


	/******** callback related functions ********/

	/*
	* In the C api, callbacks are associated with step_ctx->launch_state.
	* Since the callback functions have no parameter like "ctx" or "sls",
	* there is no simple way to map Perl callback to C callback.
	*
	* So, only one $step_ctx->launch() call is allowed in Perl, until
	* $step_ctx->launch_wait_finish().
	*/

	static SV *task_start_cb_sv = NULL;
	static SV *task_finish_cb_sv = NULL;

	static PerlInterpreter *main_perl = NULL;
	static pthread_key_t cbs_key;

	typedef struct thread_callbacks {
	SV *step_complete;
	SV *step_signal;
	SV *step_timeout;
	SV *task_start;
	SV *task_finish;
	} thread_callbacks_t;

	static void
	set_thread_perl(void)
	{
	PerlInterpreter *thr_perl = PERL_GET_CONTEXT;

	if (thr_perl == NULL) {
	if (main_perl == NULL) { /* should never happen */
	fprintf(stderr, "error: no main perl context\n");
	exit(-1);
	}
	thr_perl = perl_clone(main_perl, CLONEf_COPY_STACKS \| CLONEf_KEEP_PTR_TABLE);
	/* seems no need to call PERL_SET_CONTEXT(thr_perl); */

	/*
	* seems perl will destroy the interpreter associated with
	* a thread automatically.
	*/
	}
	}

	#define GET_THREAD_CALLBACKS ((thread_callbacks_t *)pthread_getspecific(cbs_key))
	#define SET_THREAD_CALLBACKS(cbs) (pthread_setspecific(cbs_key, (void *)cbs))

	static void
	clear_thread_callbacks(void *arg)
	{
	thread_callbacks_t cbs = (thread_callbacks_t )arg;
	if (cbs->task_start) {
	/* segfault if called. dunno why */
	/* SvREFCNT_dec(cbs->task_start); */
	}
	if (cbs->task_finish) {
	/* SvREFCNT_dec(cbs->task_finish); */
	}
	xfree(cbs);
	}

	static void
	set_thread_callbacks(void)
	{
	CLONE_PARAMS params;
	thread_callbacks_t *cbs = GET_THREAD_CALLBACKS;

	if (cbs != NULL)
	return;

	cbs = xmalloc(sizeof(thread_callbacks_t));
	if (!cbs) {
	fprintf(stderr, "set_thread_callbacks: memory exhausted\n");
	exit(-1);
	}

	params.stashes = NULL;
	params.flags = CLONEf_COPY_STACKS \| CLONEf_KEEP_PTR_TABLE;
	params.proto_perl = PERL_GET_CONTEXT;

	if (task_start_cb_sv != NULL && task_start_cb_sv != &PL_sv_undef) {
	cbs->task_start = sv_dup(task_start_cb_sv, &params);
	}
	if (task_finish_cb_sv != NULL && task_finish_cb_sv != &PL_sv_undef) {
	cbs->task_finish = sv_dup(task_finish_cb_sv, &params);
	}
	if (SET_THREAD_CALLBACKS(cbs) != 0) {
	fprintf(stderr, "set_thread_callbacks: failed to set thread specific value\n");
	exit(-1);
	}
	}

	void
	set_slcb(HV *callbacks)
	{
	SV *svp, cb;

	svp = hv_fetch(callbacks, "task_start", 10, FALSE);
	cb = svp ? *svp : &PL_sv_undef;
	if (task_start_cb_sv == NULL) {
	task_start_cb_sv = newSVsv(cb);
	} else {
	sv_setsv(task_start_cb_sv, cb);
	}

	svp = hv_fetch(callbacks, "task_finish", 11, FALSE);
	cb = svp ? *svp : &PL_sv_undef;
	if (task_finish_cb_sv == NULL) {
	task_finish_cb_sv = newSVsv(cb);
	} else {
	sv_setsv(task_finish_cb_sv, cb);
	}

	if (main_perl == NULL) {
	main_perl = PERL_GET_CONTEXT;
	if ( pthread_key_create(&cbs_key, clear_thread_callbacks) != 0) {
	fprintf(stderr, "set_slcb: failed to create cbs_key\n");
	exit(-1);
	}
	}
	}

	static void
	step_complete_cb(srun_job_complete_msg_t *comp_msg)
	{
	HV *hv;
	thread_callbacks_t *cbs = NULL;

	set_thread_perl();
	set_thread_callbacks();

	cbs = GET_THREAD_CALLBACKS;
	if (cbs->step_complete == NULL)
	return;

	hv = newHV();
	if (srun_job_complete_msg_to_hv(comp_msg, hv) < 0) {
	Perl_warn( aTHX_ "failed to prepare parameter for step_complete callback");
	SvREFCNT_dec(hv);
	return;
	}

	dSP;

	ENTER;
	SAVETMPS;
	PUSHMARK(SP);
	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
	PUTBACK;

	call_sv(cbs->step_complete, G_SCALAR);

	FREETMPS;
	LEAVE;
	}

	static void
	step_signal_cb(int signo)
	{
	thread_callbacks_t *cbs = NULL;

	set_thread_perl();
	set_thread_callbacks();

	cbs = GET_THREAD_CALLBACKS;
	if (cbs->step_signal == NULL)
	return;

	dSP;

	ENTER;
	SAVETMPS;
	PUSHMARK(SP);
	XPUSHs(sv_2mortal(newSViv(signo)));
	PUTBACK;

	call_sv(cbs->step_signal, G_SCALAR);

	FREETMPS;
	LEAVE;
	}

	static void
	step_timeout_cb(srun_timeout_msg_t *timeout_msg)
	{
	HV *hv;
	thread_callbacks_t *cbs = NULL;

	set_thread_perl();
	set_thread_callbacks();

	cbs = GET_THREAD_CALLBACKS;
	if (cbs->step_timeout == NULL)
	return;

	hv = newHV();
	if (srun_timeout_msg_to_hv(timeout_msg, hv) < 0) {
	Perl_warn( aTHX_ "failed to prepare parameter for step_timeout callback");
	SvREFCNT_dec(hv);
	return;
	}

	dSP;

	ENTER;
	SAVETMPS;
	PUSHMARK(SP);
	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
	PUTBACK;

	call_sv(cbs->step_timeout, G_SCALAR);

	FREETMPS;
	LEAVE;
	}

	static void
	task_start_cb(launch_tasks_response_msg_t *resp_msg)
	{
	HV *hv;
	thread_callbacks_t *cbs = NULL;

	set_thread_perl();
	set_thread_callbacks();

	cbs = GET_THREAD_CALLBACKS;
	if (cbs->task_start == NULL)
	return;

	hv = newHV();
	if (launch_tasks_response_msg_to_hv(resp_msg, hv) < 0) {
	Perl_warn( aTHX_ "failed to prepare parameter for task_start callback");
	SvREFCNT_dec(hv);
	return;
	}

	dSP;

	ENTER;
	SAVETMPS;
	PUSHMARK(SP);
	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
	PUTBACK;

	call_sv(cbs->task_start, G_SCALAR);

	FREETMPS;
	LEAVE;
	}

	static void
	task_finish_cb(task_exit_msg_t *exit_msg)
	{
	HV *hv;
	thread_callbacks_t *cbs = NULL;

	set_thread_perl();
	set_thread_callbacks();

	cbs = GET_THREAD_CALLBACKS;
	if (cbs->task_finish == NULL)
	return;

	hv = newHV();
	if (task_exit_msg_to_hv(exit_msg, hv) < 0) {
	Perl_warn( aTHX_ "failed to prepare parameter for task_exit callback");
	SvREFCNT_dec(hv);
	return;
	}

	dSP;

	ENTER;
	SAVETMPS;
	PUSHMARK(SP);
	XPUSHs(sv_2mortal(newRV_noinc((SV*)hv)));
	PUTBACK;

	call_sv(cbs->task_finish, G_VOID);

	FREETMPS;
	LEAVE;
	}

	slurm_step_launch_callbacks_t slcb = {
	step_complete_cb,
	step_signal_cb,
	step_timeout_cb,
	task_start_cb,
	task_finish_cb
	};