nml_srv.cc

Source code for an Numeric Cmputer

/********************************************************************
* Description: nml_srv.cc
*
*   Derived from a work by Fred Proctor & Will Shackleford
*
* Author:
* License: LGPL Version 2
* System: Linux
*    
* Copyright (c) 2004 All rights reserved.
*
* Last change: 
* $Revision: 1.5 $
* $Author: paul_c $
* $Date: 2005/05/23 16:34:15 $
********************************************************************/

#ifdef __cplusplus
extern "C" {
#endif

#include <string.h>		/* memcpy() */

#include <signal.h>		/* kill() */
#include <sys/types.h>
#include <unistd.h>		/* getpid() */
#include <sys/wait.h>		/* waitpid() */
#include <stdlib.h>		/* atexit() */

#ifdef __cplusplus
}
#endif
#include "nml.hh"
#include "nmlmsg.hh"
#include "cms.hh"
#include "nml_srv.hh"
#include "rem_msg.hh"		/* struct REMOTE_READ_REQUEST, */
#include "rcs_print.hh"		/* rcs_print_error() */
#include "timer.hh"		// esleep()
#include "rcs_exit.hh"		// rcs_exit
#include "cmsdiag.hh"
NML_SERVER::NML_SERVER(NML * _nml, int _set_to_master):CMS_SERVER()
{
    NML_SERVER_LOCAL_PORT *new_local_port = NULL;
    being_deleted = 0;
    if (NULL != _nml) {
	if (NULL != _nml->cms) {
	    if (CMS_REMOTE_TYPE != _nml->cms->ProcessType) {
		NML *new_nml;
		if (_nml->cms->isserver &&
		    (0 == _set_to_master ||
			(_nml->cms->is_local_master == 1
			    && _set_to_master == 1)
			|| (_nml->cms->is_local_master == 0
			    && _set_to_master == -1))) {
		    new_nml = _nml;
		    if (NULL != new_nml) {
			new_local_port = new NML_SERVER_LOCAL_PORT(new_nml);
			add_local_port(new_local_port);
		    }
		    new_local_port->local_channel_reused = 1;
		} else {
		    new_nml = new NML(_nml, 1, -1);
		    if (NULL != new_nml) {
			new_local_port = new NML_SERVER_LOCAL_PORT(new_nml);
			add_local_port(new_local_port);
		    }
		    new_local_port->local_channel_reused = 0;
		}
	    } else {
		rcs_print_error
		    ("NML_SERVER:(ERROR) ProcessType was REMOTE.\n");
		_nml = (NML *) NULL;
	    }
	} else {
	    rcs_print_error("NML_SERVER:(ERROR) cms was NULL.\n");
	}
    } else {
	rcs_print_error("NML_SERVER:(ERROR) nml_ptr was NULL.\n");
    }
    add_to_nml_server_list();
}

void NML_SERVER::add_to_nml_server_list()
{
    if (NULL == NML_Default_Super_Server) {
	NML_Default_Super_Server = new NML_SUPER_SERVER;
    }
    if (NULL != NML_Default_Super_Server) {
	NML_Default_Super_Server->add_to_list(this);
    }
}

NML_SERVER::~NML_SERVER()
{
    being_deleted = 1;
    delete_from_list();
}

void NML_SERVER::delete_from_list()
{
    CMS_SERVER::delete_from_list();
    if (NULL != NML_Default_Super_Server) {
	if (NULL != NML_Default_Super_Server->servers) {
	    NML_Default_Super_Server->servers->
		delete_node(super_server_list_id);
	}
    }
}

NML_SERVER_LOCAL_PORT::NML_SERVER_LOCAL_PORT(NML * _nml):CMS_SERVER_LOCAL_PORT((CMS
	*)
    NULL)
{
    local_channel_reused = 1;
    nml = _nml;
    if (NULL != nml) {
	cms = nml->cms;
	if (NULL != cms) {
	    buffer_number = cms->buffer_number;
	}
    }
}

NML_SERVER_LOCAL_PORT::~NML_SERVER_LOCAL_PORT()
{
    if (NULL != nml && !local_channel_reused) {
	delete nml;
    }
    nml = (NML *) NULL;
    cms = (CMS *) NULL;
}

REMOTE_READ_REPLY *NML_SERVER_LOCAL_PORT::reader(REMOTE_READ_REQUEST * _req)
{
    if ((NULL == cms) || (NULL == nml)) {
	rcs_print_error("NMLserver:reader: CMS object is NULL.\n");
	return ((REMOTE_READ_REPLY *) NULL);
    }

    /* Setup CMS channel from request arguments. */
    cms->in_buffer_id = _req->last_id_read;

    /* Read and encode the buffer. */
    switch (_req->access_type) {
    case CMS_READ_ACCESS:
	nml->read();
	break;
    case CMS_PEEK_ACCESS:
	nml->peek();
	break;
    default:
	rcs_print_error("NML_SERVER: Invalid access type.(%d)\n",
	    _req->access_type);
	break;
    }

    /* Setup reply structure to be returned to remote process. */
    read_reply.status = (int) cms->status;
    if (cms->status == CMS_READ_OLD) {
	read_reply.size = 0;
	read_reply.data = NULL;
	read_reply.write_id = _req->last_id_read;
	read_reply.was_read = 1;
    } else {
	read_reply.size = cms->header.in_buffer_size;
	read_reply.data = (unsigned char *) cms->encoded_data;
	read_reply.write_id = cms->in_buffer_id;
	read_reply.was_read = cms->header.was_read;
    }

    /* Reply structure contains the latest shared memory info-- now return it 
       to cms_dispatch for return to caller */
    return (&read_reply);
}

REMOTE_READ_REPLY *NML_SERVER_LOCAL_PORT::blocking_read(REMOTE_READ_REQUEST *
    _req)
{
    if ((NULL == cms) || (NULL == nml)) {
	rcs_print_error("NMLserver:blocking_read: CMS object is NULL.\n");
	return ((REMOTE_READ_REPLY *) NULL);
    }
    nml->cms->first_diag_store = 0;
    if (_req->type != REMOTE_CMS_BLOCKING_READ_REQUEST_TYPE) {
	rcs_print_error
	    ("NMLserver::blocking_read: Invalid request type(%d)\n",
	    _req->type);
	return NULL;
    }
    double orig_bytes_moved = 0.0;

    REMOTE_BLOCKING_READ_REQUEST *breq =
	(REMOTE_BLOCKING_READ_REQUEST *) _req;
    breq->_nml = new NML(nml, 1, -1);
    NML *nmlcopy = (NML *) breq->_nml;
    CMS *cmscopy = nmlcopy->cms;
    double blocking_timeout = (double) (breq->timeout_millis / 1000.0);
    REMOTE_READ_REPLY *temp_read_reply = new REMOTE_READ_REPLY();
    breq->_reply = temp_read_reply;
    long data_size = (long) cmscopy->max_encoded_message_size;
    temp_read_reply->data = malloc(data_size);
    breq->_data = temp_read_reply->data;
    if ((NULL == cmscopy) || (NULL == nmlcopy)) {
	rcs_print_error("NMLserver:blocking_read: CMS object is NULL.\n");
	return ((REMOTE_READ_REPLY *) NULL);
    }
    if (NULL != cmscopy->handle_to_global_data) {
	orig_bytes_moved = cmscopy->handle_to_global_data->total_bytes_moved;
    }
    if (NULL == temp_read_reply->data) {
	rcs_print_error
	    ("NMLserver:blocking_read: temp_read_reply->data object is NULL.\n");
	return ((REMOTE_READ_REPLY *) NULL);
    }
    nmlcopy->cms->set_encoded_data(temp_read_reply->data, data_size);

    /* Setup CMS channel from request arguments. */
    cmscopy->in_buffer_id = _req->last_id_read;

    /* Read and encode the buffer. */
    nmlcopy->blocking_read(blocking_timeout);

    /* Setup reply structure to be returned to remote process. */
    temp_read_reply->status = (int) cmscopy->status;
    if (cmscopy->status == CMS_READ_OLD) {
	temp_read_reply->size = 0;
	if (NULL != temp_read_reply->data) {
	    breq->_data = NULL;
	    free(temp_read_reply->data);
	    temp_read_reply->data = NULL;
	}
	temp_read_reply->write_id = _req->last_id_read;
	temp_read_reply->was_read = 1;
    } else {
	temp_read_reply->size = cmscopy->header.in_buffer_size;
	temp_read_reply->write_id = cmscopy->in_buffer_id;
	temp_read_reply->was_read = cmscopy->header.was_read;
    }
    if (NULL != nml->cms->handle_to_global_data &&
	NULL != cmscopy->handle_to_global_data) {
	nml->cms->handle_to_global_data->total_bytes_moved
	    +=
	    (cmscopy->handle_to_global_data->total_bytes_moved -
	    orig_bytes_moved);
	nml->cms->first_diag_store = cmscopy->first_diag_store;
    }
    breq->_nml = NULL;
    delete nmlcopy;

    /* Reply structure contains the latest shared memory info-- now return it 
       to cms_dispatch for return to caller */
    return (temp_read_reply);
}

/* REMOTE local_port function for writes */
REMOTE_WRITE_REPLY *NML_SERVER_LOCAL_PORT::writer(REMOTE_WRITE_REQUEST * _req)
{
    NMLmsg *temp;		/* Temporary Pointer */

    if ((NULL == cms) || (NULL == nml)) {
	rcs_print_error("NMLserver:writer: CMS object is NULL.\n");
	return ((REMOTE_WRITE_REPLY *) NULL);
    }

    temp = (NMLmsg *) cms->data;
    /* Check to see if remote process writing too much into local buffer. */
    if (_req->size > cms_encoded_data_explosion_factor * cms->size) {
	rcs_print_error
	    ("CMSserver:cms_writer: CMS buffer size is too small.\n");
	return ((REMOTE_WRITE_REPLY *) NULL);
    }

    /* Copy the encoded data to the location set up in CMS. */
    // memcpy(cms->encoded_data, _req->data, _req->size);
    cms->header.in_buffer_size = _req->size;
    temp->size = _req->size;

    switch (_req->access_type) {
    case CMS_WRITE_ACCESS:
	nml->write(*temp);
	break;
    case CMS_WRITE_IF_READ_ACCESS:
	nml->write_if_read(*temp);
	break;
    default:
	rcs_print_error("NML_SERVER: Invalid Access type. (%d)\n",
	    _req->access_type);
	break;
    }

    write_reply.status = (int) cms->status;
    write_reply.was_read = cms->header.was_read;