slurm_protocol_pack.c

/****************************************************************************\
 *  slurm_protocol_pack.c - functions to pack and unpack structures for RPCs
 *****************************************************************************
 *  Copyright (C) 2002 The Regents of the University of California.
 *  Produced at Lawrence Livermore National Laboratory (cf, DISCLAIMER).
 *  Written by Kevin Tew <tew1@llnl.gov>, Moe Jette <jette1@llnl.gov>, et. al.
 *  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.
\*****************************************************************************/

#if HAVE_CONFIG_H
#  include "config.h"
#endif

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>

#include "src/common/bitstring.h"
#include "src/common/log.h"
#include "src/common/pack.h"
#include "src/common/slurm_auth.h"
#include "src/common/slurm_protocol_api.h"
#include "src/common/slurm_protocol_defs.h"
#include "src/common/slurm_protocol_pack.h"
#include "src/common/xmalloc.h"

#define _pack_job_info_msg(msg,buf)		_pack_buffer_msg(msg,buf)
#define _pack_job_step_info_msg(msg,buf)	_pack_buffer_msg(msg,buf)

static void _pack_update_node_msg(update_node_msg_t * msg, Buf buffer);
static int _unpack_update_node_msg(update_node_msg_t ** msg, Buf buffer);

static void
 _pack_node_registration_status_msg(slurm_node_registration_status_msg_t *
				    msg, Buf buffer);
static int
 _unpack_node_registration_status_msg(slurm_node_registration_status_msg_t
				      ** msg, Buf buffer);

static void
 _pack_resource_allocation_response_msg(resource_allocation_response_msg_t *
					msg, Buf buffer);
static int
 _unpack_resource_allocation_response_msg(resource_allocation_response_msg_t
					  ** msg, Buf buffer);

static void
 _pack_resource_allocation_and_run_response_msg
    (resource_allocation_and_run_response_msg_t * msg, Buf buffer);
static int
 _unpack_resource_allocation_and_run_response_msg
    (resource_allocation_and_run_response_msg_t ** msg, Buf buffer);

static void _pack_submit_response_msg(submit_response_msg_t * msg,
				      Buf buffer);
static int _unpack_submit_response_msg(submit_response_msg_t ** msg,
				       Buf buffer);

static void _pack_node_info_msg(slurm_msg_t * msg, Buf buffer);
static int _unpack_node_info_msg(node_info_msg_t ** msg, Buf buffer);
static int _unpack_node_info_members(node_info_t * node, Buf buffer);

static void _pack_update_partition_msg(update_part_msg_t * msg, Buf buffer);
static int _unpack_update_partition_msg(update_part_msg_t ** msg, Buf buffer);

static void _pack_job_step_create_request_msg(job_step_create_request_msg_t
					      * msg, Buf buffer);
static int
 _unpack_job_step_create_request_msg(job_step_create_request_msg_t ** msg,
				     Buf buffer);

static void _pack_revoke_credential_msg(revoke_credential_msg_t * msg,
					Buf buffer);
static int _unpack_revoke_credential_msg(revoke_credential_msg_t ** msg,
					 Buf buffer);

static void _pack_update_job_time_msg(job_time_msg_t * msg, Buf buffer);
static int _unpack_update_job_time_msg(job_time_msg_t ** msg, Buf buffer);

static void
 _pack_job_step_create_response_msg(job_step_create_response_msg_t * msg,
				    Buf buffer);
static int
 _unpack_job_step_create_response_msg(job_step_create_response_msg_t ** msg,
				      Buf buffer);

static void _pack_partition_info_msg(slurm_msg_t * msg, Buf buffer);
static int _unpack_partition_info_msg(partition_info_msg_t ** msg,
				      Buf buffer);
static int _unpack_partition_info_members(partition_info_t * part,
					  Buf buffer);

static void _pack_launch_tasks_request_msg(launch_tasks_request_msg_t *
					   msg, Buf buffer);
static int _unpack_launch_tasks_request_msg(launch_tasks_request_msg_t **
					    msg_ptr, Buf buffer);

static void _pack_cancel_tasks_msg(kill_tasks_msg_t * msg, Buf buffer);
static int _unpack_cancel_tasks_msg(kill_tasks_msg_t ** msg_ptr, Buf buffer);

static void _pack_launch_tasks_response_msg(launch_tasks_response_msg_t *
					    msg, Buf buffer);
static int _unpack_launch_tasks_response_msg(launch_tasks_response_msg_t **
					     msg_ptr, Buf buffer);

static void _pack_shutdown_msg(shutdown_msg_t * msg, Buf buffer);
static int _unpack_shutdown_msg(shutdown_msg_t ** msg_ptr, Buf buffer);

static void _pack_reattach_tasks_request_msg(reattach_tasks_request_msg_t *,
					     Buf);
static int _unpack_reattach_tasks_request_msg(reattach_tasks_request_msg_t **,
					      Buf);

static void
 _pack_reattach_tasks_response_msg(reattach_tasks_response_msg_t *, Buf);
static int
 _unpack_reattach_tasks_response_msg(reattach_tasks_response_msg_t **, Buf);

static void _pack_task_exit_msg(task_exit_msg_t * msg, Buf buffer);
static int _unpack_task_exit_msg(task_exit_msg_t ** msg_ptr, Buf buffer);

static void _pack_old_job_desc_msg(old_job_alloc_msg_t * job_desc_ptr,
				   Buf buffer);
static int _unpack_old_job_desc_msg(old_job_alloc_msg_t **
				    job_desc_buffer_ptr, Buf buffer);

static void _pack_return_code_msg(return_code_msg_t * msg, Buf buffer);
static int _unpack_return_code_msg(return_code_msg_t ** msg, Buf buffer);

static void _pack_slurm_ctl_conf_msg(slurm_ctl_conf_info_msg_t * build_ptr,
				     Buf buffer);
static int _unpack_slurm_ctl_conf_msg(slurm_ctl_conf_info_msg_t **
				      build_buffer_ptr, Buf buffer);

static void _pack_job_step_id_msg(job_step_id_t * msg, Buf buffer);
static int _unpack_job_step_id_msg(job_step_id_t ** msg_ptr, Buf buffer);

static void _pack_get_job_step_info_msg(job_step_info_request_msg_t * msg,
					Buf buffer);
static int _unpack_get_job_step_info_msg(job_step_info_request_msg_t **
					 msg, Buf buffer);
static int _unpack_job_step_info_response_msg(job_step_info_response_msg_t
					      ** msg, Buf buffer);
static int _unpack_job_step_info_members(job_step_info_t * step, Buf buffer);

static void _pack_complete_job_step_msg(complete_job_step_msg_t * msg,
					Buf buffer);
static int _unpack_complete_job_step_msg(complete_job_step_msg_t **
					 msg_ptr, Buf buffer);
static int _unpack_job_info_members(job_info_t * job, Buf buffer);

static void _pack_batch_job_launch_msg(batch_job_launch_msg_t * msg,
				       Buf buffer);
static int _unpack_batch_job_launch_msg(batch_job_launch_msg_t ** msg,
					Buf buffer);

static void _pack_job_desc_msg(job_desc_msg_t * job_desc_ptr, Buf buffer);
static int _unpack_job_desc_msg(job_desc_msg_t ** job_desc_buffer_ptr,
				Buf buffer);
static int _unpack_job_info_msg(job_info_msg_t ** msg, Buf buffer);

static void _pack_last_update_msg(last_update_msg_t * msg, Buf buffer);
static int _unpack_last_update_msg(last_update_msg_t ** msg, Buf buffer);

static void _pack_slurm_addr_array(slurm_addr * slurm_address,
				   uint16_t size_val, Buf buffer);
static int _unpack_slurm_addr_array(slurm_addr ** slurm_address,
				    uint16_t * size_val, Buf buffer);

static void _pack_job_id_request_msg(job_id_request_msg_t * msg, Buf buffer);
static int  _unpack_job_id_request_msg(job_id_request_msg_t ** msg, Buf buffer);

static void _pack_job_id_response_msg(job_id_response_msg_t * msg, Buf buffer);
static int  _unpack_job_id_response_msg(job_id_response_msg_t ** msg, 
					Buf buffer);

static void _pack_job_step_kill_msg(job_step_kill_msg_t * msg, Buf buffer);
static int  _unpack_job_step_kill_msg(job_step_kill_msg_t ** msg_ptr, 
				      Buf buffer);

static void _pack_buffer_msg(slurm_msg_t * msg, Buf buffer);

/* pack_header
 * packs a slurm protocol header that proceeds every slurm message
 * IN header - the header structure to pack
 * IN/OUT buffer - destination of the pack, contains pointers that are
 *			automatically updated
 */
void
pack_header(header_t * header, Buf buffer)
{
	pack16(header->version, buffer);
	pack16(header->flags, buffer);
	pack16((uint16_t) header->msg_type, buffer);
	pack32(header->body_length, buffer);
}

/* unpack_header
 * unpacks a slurm protocol header that proceeds every slurm message
 * OUT header - the header structure to unpack
 * IN/OUT buffer - source of the unpack data, contains pointers that are
 *			automatically updated
 * RET 0 or error code
 */
int
unpack_header(header_t * header, Buf buffer)
{
	uint16_t tmp = 0;

	safe_unpack16(&header->version, buffer);
	safe_unpack16(&header->flags, buffer);
	safe_unpack16(&tmp, buffer);
	header->msg_type = (slurm_msg_type_t) tmp;
	safe_unpack32(&header->body_length, buffer);
	return SLURM_SUCCESS;

      unpack_error:
	return SLURM_ERROR;
}

/* size_io_stream_header - get the size of an I/O stream header
 * RET number of bytes in an I/O steam header
 */
int
size_io_stream_header(void)
{
	/* must match un/pack_io_stream_header and size_io_stream_header */
	return (SLURM_SSL_SIGNATURE_LENGTH +
		2 * sizeof(uint16_t) + sizeof(uint32_t));
}

/* pack_io_stream_header
 * packs an i/o stream protocol header used for stdin/out/err
 * IN header - the header structure to pack
 * IN/OUT buffer - destination of the pack, contains pointers that are
 *			automatically updated
 */
void
pack_io_stream_header(slurm_io_stream_header_t * header, Buf buffer)
{				/* must match un/pack_io_stream_header and size_io_stream_header */

	assert(header != NULL);

	pack16(header->version, buffer);
	packmem_array(header->key, (uint32_t) SLURM_SSL_SIGNATURE_LENGTH,
		      buffer);
	pack32(header->task_id, buffer);
	pack16(header->type, buffer);
}

/* unpack_io_stream_header
 * unpacks an i/o stream protocol header used for stdin/out/err
 * OUT header - the header structure to unpack
 * IN/OUT buffer - source of the unpack data, contains pointers that are
 *			automatically updated
 * RET 0 or error code
 */
int
unpack_io_stream_header(slurm_io_stream_header_t * header, Buf buffer)
{
	/* must match un/pack_io_stream_header and size_io_stream_header */
	safe_unpack16(&header->version, buffer);
	safe_unpackmem_array(header->key,
			     (uint32_t) SLURM_SSL_SIGNATURE_LENGTH, buffer);
	safe_unpack32(&header->task_id, buffer);
	safe_unpack16(&header->type, buffer);
	return SLURM_SUCCESS;

      unpack_error:
	return SLURM_ERROR;
}


/* pack_msg
 * packs a generic slurm protocol message body
 * IN msg - the body structure to pack (note: includes message type)
 * IN/OUT buffer - destination of the pack, contains pointers that are 
 *			automatically updated
 * RET 0 or error code
 */
int
pack_msg(slurm_msg_t const *msg, Buf buffer)
{
	switch (msg->msg_type) {
	 case REQUEST_BUILD_INFO:
	 case REQUEST_NODE_INFO:
	 case REQUEST_PARTITION_INFO:
	 case REQUEST_ACCTING_INFO:
		 _pack_last_update_msg((last_update_msg_t *)
				       msg->data, buffer);
		 break;
	 case RESPONSE_BUILD_INFO:
		 _pack_slurm_ctl_conf_msg((slurm_ctl_conf_info_msg_t *)
					  msg->data, buffer);
		 break;
	 case RESPONSE_JOB_INFO:
		 _pack_job_info_msg((slurm_msg_t *) msg, buffer);
		 break;
	 case RESPONSE_PARTITION_INFO:
		 _pack_partition_info_msg((slurm_msg_t *) msg, buffer);
		 break;
	 case RESPONSE_NODE_INFO:
		 _pack_node_info_msg((slurm_msg_t *) msg, buffer);
		 break;
	 case MESSAGE_NODE_REGISTRATION_STATUS:
		 _pack_node_registration_status_msg(
			(slurm_node_registration_status_msg_t *) msg->data, 
			buffer);
		 break;
	 case REQUEST_RESOURCE_ALLOCATION:
	 case REQUEST_SUBMIT_BATCH_JOB:
	 case REQUEST_JOB_WILL_RUN:
	 case REQUEST_ALLOCATION_AND_RUN_JOB_STEP:
	 case REQUEST_UPDATE_JOB:
		 _pack_job_desc_msg((job_desc_msg_t *)
				    msg->data, buffer);
		 break;
	 case REQUEST_OLD_JOB_RESOURCE_ALLOCATION:
		 _pack_old_job_desc_msg((old_job_alloc_msg_t *) msg->data,
					buffer);
		 break;
	 case REQUEST_NODE_REGISTRATION_STATUS:
	 case REQUEST_RECONFIGURE:
	 case REQUEST_SHUTDOWN_IMMEDIATE:
	 case REQUEST_PING:
	 case REQUEST_CONTROL:
		 /* Message contains no body/information */
		 break;
	 case REQUEST_SHUTDOWN:
		 _pack_shutdown_msg((shutdown_msg_t *) msg->data, buffer);
		 break;
	 case RESPONSE_SUBMIT_BATCH_JOB:
		 _pack_submit_response_msg((submit_response_msg_t *) 
					   msg->data, buffer);
		 break;
	 case RESPONSE_RESOURCE_ALLOCATION:
	 case RESPONSE_JOB_WILL_RUN:
		 _pack_resource_allocation_response_msg
		     ((resource_allocation_response_msg_t *) msg->data,
		      buffer);
		 break;
	 case RESPONSE_ALLOCATION_AND_RUN_JOB_STEP:
		 _pack_resource_allocation_and_run_response_msg(
			(resource_allocation_and_run_response_msg_t *) 
			msg->data, buffer);
		 break;
	 case REQUEST_UPDATE_NODE:
		 _pack_update_node_msg((update_node_msg_t *) msg->data,
				       buffer);
		 break;
	 case REQUEST_UPDATE_PARTITION:
		 _pack_update_partition_msg((update_part_msg_t *) msg->
					    data, buffer);
		 break;
	 case REQUEST_REATTACH_TASKS:
		 _pack_reattach_tasks_request_msg(
			(reattach_tasks_request_msg_t *) msg->data, buffer);
		 break;
	 case RESPONSE_REATTACH_TASKS:
		 _pack_reattach_tasks_response_msg(
			(reattach_tasks_response_msg_t *) msg->data, buffer);
		 break;
	 case REQUEST_LAUNCH_TASKS:
		 _pack_launch_tasks_request_msg(
				(launch_tasks_request_msg_t *) msg->data, 
				buffer);
		 break;
	 case RESPONSE_LAUNCH_TASKS:
		 _pack_launch_tasks_response_msg((launch_tasks_response_msg_t
						  *) msg->data, buffer);
		 break;
	 case REQUEST_KILL_TASKS:
		 _pack_cancel_tasks_msg((kill_tasks_msg_t *) msg->data,
					buffer);
		 break;
	 case REQUEST_JOB_STEP_INFO:
		 _pack_get_job_step_info_msg((job_step_info_request_msg_t
					      *) msg->data, buffer);
		 break;
		/********  job_step_id_t Messages  ********/
	 case REQUEST_JOB_INFO:
		 _pack_job_step_id_msg((job_step_id_t *) msg->data, buffer);
		 break;
	 case REQUEST_CANCEL_JOB_STEP:
		 _pack_job_step_kill_msg((job_step_kill_msg_t *) 
					 msg->data, buffer);
		 break;
	 case REQUEST_COMPLETE_JOB_STEP:
		 _pack_complete_job_step_msg((complete_job_step_msg_t *)
					     msg->data, buffer);
		 break;

	 case REQUEST_KILL_TIMELIMIT:
	 case REQUEST_REVOKE_JOB_CREDENTIAL:
		 _pack_revoke_credential_msg((revoke_credential_msg_t *)
					     msg->data, buffer);
		 break;
	 case REQUEST_UPDATE_JOB_TIME:
		 _pack_update_job_time_msg((job_time_msg_t *)
					     msg->data, buffer);
		 break;
	 case REQUEST_SIGNAL_JOB:
		 break;
	 case REQUEST_SIGNAL_JOB_STEP:
		 break;
	 case RESPONSE_RECONFIGURE:
	 case RESPONSE_SHUTDOWN:
	 case RESPONSE_CANCEL_JOB_STEP:
	 case RESPONSE_COMPLETE_JOB_STEP:
	 case RESPONSE_SIGNAL_JOB:
	 case RESPONSE_SIGNAL_JOB_STEP:
		 break;
	 case REQUEST_JOB_ATTACH:
		 break;
	 case RESPONSE_JOB_ATTACH:
		 break;
	 case RESPONSE_JOB_STEP_INFO:
		 _pack_job_step_info_msg((slurm_msg_t *) msg, buffer);
		 break;
	 case REQUEST_JOB_RESOURCE:
		 break;
	 case RESPONSE_JOB_RESOURCE:
		 break;
	 case REQUEST_RUN_JOB_STEP:
		 break;
	 case RESPONSE_RUN_JOB_STEP:
		 break;
	 case MESSAGE_TASK_EXIT:
		 _pack_task_exit_msg((task_exit_msg_t *) msg->data, buffer);
		 break;
	 case REQUEST_BATCH_JOB_LAUNCH:
		 _pack_batch_job_launch_msg((batch_job_launch_msg_t *)
					    msg->data, buffer);
		 break;
	 case MESSAGE_UPLOAD_ACCOUNTING_INFO:
		 break;
	 case RESPONSE_SLURM_RC:
		 _pack_return_code_msg((return_code_msg_t *) msg->data,
				       buffer);
		 break;
	 case RESPONSE_JOB_STEP_CREATE:
		 _pack_job_step_create_response_msg(
				(job_step_create_response_msg_t *)
				msg->data, buffer);
		 break;
	 case REQUEST_JOB_STEP_CREATE:
		 _pack_job_step_create_request_msg(
				(job_step_create_request_msg_t *)
				msg->data, buffer);
		 break;
	 case REQUEST_JOB_ID:
		 _pack_job_id_request_msg(
				(job_id_request_msg_t *)msg->data,
				buffer);
		 break;
	 case RESPONSE_JOB_ID:
		 _pack_job_id_response_msg(
				(job_id_response_msg_t *)msg->data,
				buffer);
		 break;
	 default:
		 debug("No pack method for msg type %i", msg->msg_type);
		 return EINVAL;
		 break;

	}
	return SLURM_SUCCESS;
}

/* unpack_msg
 * unpacks a generic slurm protocol message body
 * OUT msg - the body structure to unpack (note: includes message type)
 * IN/OUT buffer - source of the unpack, contains pointers that are 
 *			automatically updated
 * RET 0 or error code
 */
int
unpack_msg(slurm_msg_t * msg, Buf buffer)
{
	int rc = SLURM_SUCCESS;
	msg->data = NULL;	/* Initialize to no data for now */

	switch (msg->msg_type) {
	 case REQUEST_BUILD_INFO:
	 case REQUEST_NODE_INFO:
	 case REQUEST_PARTITION_INFO:
	 case REQUEST_ACCTING_INFO:
		 rc = _unpack_last_update_msg((last_update_msg_t **) &
					      (msg->data), buffer);
		 break;
	 case RESPONSE_BUILD_INFO:
		 rc = _unpack_slurm_ctl_conf_msg((slurm_ctl_conf_info_msg_t
						  **)
						 & (msg->data), buffer);
		 break;
	 case RESPONSE_JOB_INFO:
		 rc = _unpack_job_info_msg((job_info_msg_t **) & (msg->data),
					   buffer);
		 break;
	 case RESPONSE_PARTITION_INFO:
		 rc = _unpack_partition_info_msg((partition_info_msg_t **) &
						 (msg->data), buffer);
		 break;
	 case RESPONSE_NODE_INFO:
		 rc = _unpack_node_info_msg((node_info_msg_t **) &
					    (msg->data), buffer);
		 break;
	 case MESSAGE_NODE_REGISTRATION_STATUS:
		 rc = _unpack_node_registration_status_msg(
				(slurm_node_registration_status_msg_t **)
				& (msg->data), buffer);
		 break;
	 case REQUEST_RESOURCE_ALLOCATION:
	 case REQUEST_SUBMIT_BATCH_JOB:
	 case REQUEST_JOB_WILL_RUN:
	 case REQUEST_ALLOCATION_AND_RUN_JOB_STEP:
	 case REQUEST_UPDATE_JOB:
		 rc = _unpack_job_desc_msg((job_desc_msg_t **) & (msg->data),
					   buffer);
		 break;
	 case REQUEST_OLD_JOB_RESOURCE_ALLOCATION:
		 rc = _unpack_old_job_desc_msg((old_job_alloc_msg_t **) &
					       (msg->data), buffer);
		 break;
	 case REQUEST_NODE_REGISTRATION_STATUS:
	 case REQUEST_RECONFIGURE:
	 case REQUEST_SHUTDOWN_IMMEDIATE:
	 case REQUEST_PING:
	 case REQUEST_CONTROL:
		 /* Message contains no body/information */
		 break;
	 case REQUEST_SHUTDOWN:
		 rc = _unpack_shutdown_msg((shutdown_msg_t **) & (msg->data),
					   buffer);
		 break;
	 case RESPONSE_SUBMIT_BATCH_JOB:
		 rc = _unpack_submit_response_msg((submit_response_msg_t **)
						  & (msg->data), buffer);
		 break;
	 case RESPONSE_RESOURCE_ALLOCATION:
	 case RESPONSE_JOB_WILL_RUN:
		 rc = _unpack_resource_allocation_response_msg(
				(resource_allocation_response_msg_t **)
				& (msg->data), buffer);
		 break;
	 case RESPONSE_ALLOCATION_AND_RUN_JOB_STEP:
		 rc = _unpack_resource_allocation_and_run_response_msg(
				(resource_allocation_and_run_response_msg_t **)
				&(msg->data), buffer);
		 break;
	 case REQUEST_UPDATE_NODE:
		 rc = _unpack_update_node_msg((update_node_msg_t **) &
					      (msg->data), buffer);
		 break;
	 case REQUEST_UPDATE_PARTITION:
		 rc = _unpack_update_partition_msg((update_part_msg_t **) &
						   (msg->data), buffer);
		 break;
	 case REQUEST_LAUNCH_TASKS:
		 rc = _unpack_launch_tasks_request_msg(
					(launch_tasks_request_msg_t **)
					& (msg->data), buffer);
		 break;
	 case RESPONSE_LAUNCH_TASKS:
		 rc = _unpack_launch_tasks_response_msg(
					(launch_tasks_response_msg_t **)
					& (msg->data), buffer);
		 break;
	 case REQUEST_REATTACH_TASKS:
		 rc = _unpack_reattach_tasks_request_msg(
			(reattach_tasks_request_msg_t **) & msg->data, 
			buffer);
		 break;
	 case RESPONSE_REATTACH_TASKS:
		 rc = _unpack_reattach_tasks_response_msg(
					(reattach_tasks_response_msg_t **) 
					& msg->data, buffer);
		 break;
	 case REQUEST_KILL_TASKS:
		 rc = _unpack_cancel_tasks_msg((kill_tasks_msg_t **) &
					       (msg->data), buffer);
		 break;
	 case REQUEST_JOB_STEP_INFO:
		 rc = _unpack_get_job_step_info_msg(
				(job_step_info_request_msg_t **)
				& (msg->data), buffer);
		 break;
		/********  job_step_id_t Messages  ********/
	 case REQUEST_JOB_INFO:
		 rc = _unpack_job_step_id_msg((job_step_id_t **)
					      & (msg->data), buffer);
		 break;
	 case REQUEST_CANCEL_JOB_STEP:
		 rc = _unpack_job_step_kill_msg((job_step_kill_msg_t **)
					        & (msg->data), buffer);
		 break;
	 case REQUEST_COMPLETE_JOB_STEP:
		 rc = _unpack_complete_job_step_msg((complete_job_step_msg_t
						     **) & (msg->data),
						    buffer);
		 break;
	 case REQUEST_KILL_TIMELIMIT:
	 case REQUEST_REVOKE_JOB_CREDENTIAL:
		 rc = _unpack_revoke_credential_msg(
					(revoke_credential_msg_t **)
					& (msg->data), buffer);
		 break;
	 case REQUEST_UPDATE_JOB_TIME:
		 rc = _unpack_update_job_time_msg(
					(job_time_msg_t **)
					& (msg->data), buffer);
		 break;
	 case REQUEST_SIGNAL_JOB:
		 break;
	 case REQUEST_SIGNAL_JOB_STEP:
		 break;
	 case RESPONSE_RECONFIGURE:
	 case RESPONSE_SHUTDOWN:
	 case RESPONSE_CANCEL_JOB_STEP:
	 case RESPONSE_COMPLETE_JOB_STEP:
	 case RESPONSE_SIGNAL_JOB:
	 case RESPONSE_SIGNAL_JOB_STEP:
		 break;
	 case REQUEST_JOB_ATTACH:
		 break;
	 case RESPONSE_JOB_ATTACH:
		 break;
	 case RESPONSE_JOB_STEP_INFO:
		 rc = _unpack_job_step_info_response_msg(
					(job_step_info_response_msg_t **)
					& (msg->data), buffer);
		 break;
	 case REQUEST_JOB_RESOURCE:
		 break;
	 case RESPONSE_JOB_RESOURCE:
		 break;
	 case REQUEST_RUN_JOB_STEP:
		 break;
	 case RESPONSE_RUN_JOB_STEP:
		 break;
	 case MESSAGE_TASK_EXIT:
		 rc = _unpack_task_exit_msg((task_exit_msg_t **)
					    & (msg->data), buffer);
		 break;
	 case REQUEST_BATCH_JOB_LAUNCH:
		 rc = _unpack_batch_job_launch_msg((batch_job_launch_msg_t **)
						   & (msg->data), buffer);
		 break;
	 case MESSAGE_UPLOAD_ACCOUNTING_INFO:
		 break;
	 case RESPONSE_SLURM_RC:
		 rc = _unpack_return_code_msg((return_code_msg_t **)
					      & (msg->data), buffer);
		 break;
	 case RESPONSE_JOB_STEP_CREATE:
		 rc = _unpack_job_step_create_response_msg(
				(job_step_create_response_msg_t **)
				& msg->data, buffer);
		 break;
	 case REQUEST_JOB_STEP_CREATE:
		 rc = _unpack_job_step_create_request_msg(
					(job_step_create_request_msg_t **)
					& msg->data, buffer);
		 break;
	 case REQUEST_JOB_ID:
		 rc = _unpack_job_id_request_msg(
				(job_id_request_msg_t **) & msg->data,
				buffer);
		 break;
	 case RESPONSE_JOB_ID:
		 rc = _unpack_job_id_response_msg(
				(job_id_response_msg_t **) & msg->data,
				buffer);
		 break;
	 default:
		 debug("No unpack method for msg type %i", msg->msg_type);
		 return EINVAL;
		 break;
	}

	if (rc)
		error("Malformed RPC of type %u received", msg->msg_type);
	return rc;
}

static void
_pack_update_node_msg(update_node_msg_t * msg, Buf buffer)
{
	assert(msg != NULL);

	packstr(msg->node_names, buffer);
	pack16(msg->node_state, buffer);
}

static int
_unpack_update_node_msg(update_node_msg_t ** msg, Buf buffer)
{
	uint16_t uint16_tmp;
	update_node_msg_t *tmp_ptr;

	/* alloc memory for structure */
	assert(msg != NULL);
	tmp_ptr = xmalloc(sizeof(update_node_msg_t));
	*msg = tmp_ptr;

	safe_unpackstr_xmalloc(&tmp_ptr->node_names, &uint16_tmp, buffer);
	safe_unpack16(&tmp_ptr->node_state, buffer);
	return SLURM_SUCCESS;

      unpack_error:
	xfree(tmp_ptr->node_names);
	xfree(tmp_ptr);
	*msg = NULL;
	return SLURM_ERROR;
}

static void
_pack_node_registration_status_msg(slurm_node_registration_status_msg_t *
				   msg, Buf buffer)
{
	int i;
	assert(msg != NULL);

	pack_time(msg->timestamp, buffer);
	pack32(msg->status, buffer);
	packstr(msg->node_name, buffer);
	pack32(msg->cpus, buffer);
	pack32(msg->real_memory_size, buffer);
	pack32(msg->temporary_disk_space, buffer);
	pack32(msg->job_count, buffer);
	for (i = 0; i < msg->job_count; i++) {
		pack32(msg->job_id[i], buffer);
	}
	for (i = 0; i < msg->job_count; i++) {
		pack16(msg->step_id[i], buffer);
	}
}

static int
_unpack_node_registration_status_msg(slurm_node_registration_status_msg_t
				     ** msg, Buf buffer)
{
	uint16_t uint16_tmp;
	int i;
	slurm_node_registration_status_msg_t *node_reg_ptr;

	/* alloc memory for structure */
	assert(msg != NULL);
	node_reg_ptr = xmalloc(sizeof(slurm_node_registration_status_msg_t));
	*msg = node_reg_ptr;

	/* unpack timestamp of snapshot */
	safe_unpack_time(&node_reg_ptr->timestamp, buffer);
	/* load the data values */
	safe_unpack32(&node_reg_ptr->status, buffer);
	safe_unpackstr_xmalloc(&node_reg_ptr->node_name, &uint16_tmp, buffer);
	safe_unpack32(&node_reg_ptr->cpus, buffer);
	safe_unpack32(&node_reg_ptr->real_memory_size, buffer);
	safe_unpack32(&node_reg_ptr->temporary_disk_space, buffer);
	safe_unpack32(&node_reg_ptr->job_count, buffer);
	node_reg_ptr->job_id =
	    xmalloc(sizeof(uint32_t) * node_reg_ptr->job_count);
	for (i = 0; i < node_reg_ptr->job_count; i++) {
		safe_unpack32(&node_reg_ptr->job_id[i], buffer);
	}
	node_reg_ptr->step_id =
	    xmalloc(sizeof(uint16_t) * node_reg_ptr->job_count);
	for (i = 0; i < node_reg_ptr->job_count; i++) {
		safe_unpack16(&node_reg_ptr->step_id[i], buffer);
	}
	return SLURM_SUCCESS;

      unpack_error:
	xfree(node_reg_ptr->node_name);
	xfree(node_reg_ptr->job_id);
	xfree(node_reg_ptr->step_id);
	xfree(node_reg_ptr);
	*msg = NULL;
	return SLURM_ERROR;
}

static void
_pack_resource_allocation_response_msg(resource_allocation_response_msg_t *
				       msg, Buf buffer)
{
	assert(msg != NULL);

	pack32(msg->error_code, buffer);
	pack32(msg->job_id, buffer);
	packstr(msg->node_list, buffer);

	pack16(msg->num_cpu_groups, buffer);
	pack32_array(msg->cpus_per_node, msg->num_cpu_groups, buffer);
	pack32_array(msg->cpu_count_reps, msg->num_cpu_groups, buffer);

	pack16(msg->node_cnt, buffer);
	_pack_slurm_addr_array(msg->node_addr, msg->node_cnt, buffer);
}

static int
_unpack_resource_allocation_response_msg(resource_allocation_response_msg_t
					 ** msg, Buf buffer)
{
	uint16_t uint16_tmp;
	uint32_t uint32_tmp;
	resource_allocation_response_msg_t *tmp_ptr;

	/* alloc memory for structure */
	assert(msg != NULL);
	tmp_ptr = xmalloc(sizeof(resource_allocation_response_msg_t));
	*msg = tmp_ptr;

	/* load the data values */
	safe_unpack32(&tmp_ptr->error_code, buffer);
	safe_unpack32(&tmp_ptr->job_id, buffer);
	safe_unpackstr_xmalloc(&tmp_ptr->node_list, &uint16_tmp, buffer);

	safe_unpack16(&tmp_ptr->num_cpu_groups, buffer);
	if (tmp_ptr->num_cpu_groups > 0) {
		tmp_ptr->cpus_per_node = (uint32_t *)
		    xmalloc(sizeof(uint32_t) * tmp_ptr->num_cpu_groups);
		tmp_ptr->cpu_count_reps = (uint32_t *)
		    xmalloc(sizeof(uint32_t) * tmp_ptr->num_cpu_groups);
		safe_unpack32_array((uint32_t **) &
				    (tmp_ptr->cpus_per_node), &uint32_tmp,
				    buffer);
		if (tmp_ptr->num_cpu_groups != uint32_tmp)
			goto unpack_error;
		safe_unpack32_array((uint32_t **) &
				    (tmp_ptr->cpu_count_reps), &uint32_tmp,
				    buffer);
		if (tmp_ptr->num_cpu_groups != uint32_tmp)
			goto unpack_error;
	} else {
		tmp_ptr->cpus_per_node = NULL;
		tmp_ptr->cpu_count_reps = NULL;
	}

	safe_unpack16(&tmp_ptr->node_cnt, buffer);
	if (tmp_ptr->node_cnt > 0) {
		if (_unpack_slurm_addr_array(&(tmp_ptr->node_addr),
					     &(tmp_ptr->node_cnt), buffer))
			goto unpack_error;
	} else
		tmp_ptr->node_addr = NULL;

	debug("job id is %u", tmp_ptr->job_id);
	return SLURM_SUCCESS;

      unpack_error:
	xfree(tmp_ptr->node_list);
	xfree(tmp_ptr->cpus_per_node);
	xfree(tmp_ptr->cpu_count_reps);
	xfree(tmp_ptr);
	*msg = NULL;
	return SLURM_ERROR;
}

static void
 _pack_resource_allocation_and_run_response_msg
    (resource_allocation_and_run_response_msg_t * msg, Buf buffer) {
	assert(msg != NULL);

	pack32(msg->job_id, buffer);
	packstr(msg->node_list, buffer);
	pack16(msg->num_cpu_groups, buffer);
	pack32_array(msg->cpus_per_node, msg->num_cpu_groups, buffer);
	pack32_array(msg->cpu_count_reps, msg->num_cpu_groups, buffer);
	pack32(msg->job_step_id, buffer);

	pack16(msg->node_cnt, buffer);
	_pack_slurm_addr_array(msg->node_addr, msg->node_cnt, buffer);

	pack_job_credential(msg->credentials, buffer);
#ifdef HAVE_LIBELAN3
	qsw_pack_jobinfo(msg->qsw_job, buffer);
#endif
}

static int
 _unpack_resource_allocation_and_run_response_msg
    (resource_allocation_and_run_response_msg_t ** msg, Buf buffer) {
	uint16_t uint16_tmp;
	uint32_t uint32_tmp;
	resource_allocation_and_run_response_msg_t *tmp_ptr;

	/* alloc memory for structure */
	assert(msg != NULL);
	tmp_ptr = xmalloc(sizeof(resource_allocation_and_run_response_msg_t));
	*msg = tmp_ptr;

	/* load the data values */
	safe_unpack32(&tmp_ptr->job_id, buffer);
	safe_unpackstr_xmalloc(&tmp_ptr->node_list, &uint16_tmp, buffer);
	safe_unpack16(&tmp_ptr->num_cpu_groups, buffer);

	if (tmp_ptr->num_cpu_groups > 0) {
		tmp_ptr->cpus_per_node = (uint32_t *)
		    xmalloc(sizeof(uint32_t) * tmp_ptr->num_cpu_groups);
		tmp_ptr->cpu_count_reps = (uint32_t *)
		    xmalloc(sizeof(uint32_t) * tmp_ptr->num_cpu_groups);
		safe_unpack32_array((uint32_t **) &
				    (tmp_ptr->cpus_per_node), 
				    &uint32_tmp, buffer);
		if (tmp_ptr->num_cpu_groups != uint32_tmp)
			goto unpack_error;
		safe_unpack32_array((uint32_t **) &
				    (tmp_ptr->cpu_count_reps), 
				    &uint32_tmp, buffer);
		if (tmp_ptr->num_cpu_groups != uint32_tmp)
			goto unpack_error;
	}

	safe_unpack32(&tmp_ptr->job_step_id, buffer);
	safe_unpack16(&tmp_ptr->node_cnt, buffer);
	if (tmp_ptr->node_cnt > 0) {
		if (_unpack_slurm_addr_array(&(tmp_ptr->node_addr),
					     &(tmp_ptr->node_cnt), buffer))
			goto unpack_error;
	} else
		tmp_ptr->node_addr = NULL;

	if (unpack_job_credential(&tmp_ptr->credentials, buffer))
		goto unpack_error;
#ifdef HAVE_LIBELAN3
	qsw_alloc_jobinfo(&tmp_ptr->qsw_job);
	if (qsw_unpack_jobinfo(tmp_ptr->qsw_job, buffer) < 0) {
		error("qsw_unpack_jobinfo: %m");
		qsw_free_jobinfo(tmp_ptr->qsw_job);
		goto unpack_error;
	}
#endif
	return SLURM_SUCCESS;

      unpack_error:
	xfree(tmp_ptr->node_list);
	xfree(tmp_ptr->cpus_per_node);
	xfree(tmp_ptr->cpu_count_reps);
	xfree(tmp_ptr);
	*msg = NULL;
	return SLURM_ERROR;
}

static void
_pack_submit_response_msg(submit_response_msg_t * msg, Buf buffer)
{
	assert(msg != NULL);

	pack32(msg->job_id, buffer);
	pack32(msg->error_code, buffer);
}

static int
_unpack_submit_response_msg(submit_response_msg_t ** msg, Buf buffer)
{
	submit_response_msg_t *tmp_ptr;

	/* alloc memory for structure */
	assert(msg != NULL);
	tmp_ptr = xmalloc(sizeof(submit_response_msg_t));
	*msg = tmp_ptr;

	/* load the data values */
	safe_unpack32(&tmp_ptr->job_id, buffer);
	safe_unpack32(&tmp_ptr->error_code, buffer);
	return SLURM_SUCCESS;

      unpack_error:
	xfree(tmp_ptr);
	*msg = NULL;
	return SLURM_ERROR;
}
static void
_pack_node_info_msg(slurm_msg_t * msg, Buf buffer)
{
	packmem_array(msg->data, msg->data_size, buffer);
}

static int
_unpack_node_info_msg(node_info_msg_t ** msg, Buf buffer)
{
	int i;