cms_in.cc

Source code for an Numeric Cmputer

/********************************************************************
* Description: cms_in.cc
*   Provides the internal interface of CMS to the buffer.
*   The following applies to the member functions in this file:
*   They work through a handle to a shared physical memory object.
*   They should only be called when this process has sole access to
*   this shared physical memory object. This is normally achieved by
*   taking a mutual-exclusion semaphore before calling the
*   internal_access function above from the main_access function
*   of a derived class.
*   If they begin with "queue" then they are for buffers where messages
*   are to be queued, other wise they are for buffers with will have
*   only one message at a time.
*   Queuing buffers store a CMS_QUEUING_HEADER at the beginning and a
*   CMS_HEADER before each message. Non-queuing buffers have only a
*   CMS_HEADER before the only message.
*   If they end in "encoded" they are for buffers that will neutrally
*   encoded in some processor architecture independant data format such
*   as XDR (eXternal Data Representation), otherwise the buffer must be
*   in the format used by the same compiler as this is compiled in and
*   for the same processor architecture and the function name will end
*   in "raw".
*
*   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.6 $
* $Author: paul_c $
* $Date: 2005/05/23 16:34:11 $
********************************************************************/

#include "cms.hh"		/* class CMS */
#include "cmsdiag.hh"		// class CMS_DIAG_PROC_INFO, CMS_DIAG_HEADER
#include "rcs_print.hh"		/* rcs_print_error() */
#include "physmem.hh"		/* class PHYSMEM_HANDLE */

/* CMS Member Functions. */

int cms_print_queue_free_space = 0;
int cms_print_queue_full_messages = 1;

/*************************************************************************
* This function should be called by functions overloading main_access()
* It uses a dummy physmem handle so that reads and writes work on
* the memory at _global.
* Parameters:
* _local - Address of local buffer where user has stored messages in or will
* read messages from whithin this process.
* _global - Address of shared or global memory buffer used to communicate with
* other  process.
************************************************************************/
CMS_STATUS CMS::internal_access(void *_global, long _global_size,
    void *_local)
{
    /* Don't bother trying to create a physmem handle for a NULL pointer. */
    if (NULL == _global) {
	return (status = CMS_INTERNAL_ACCESS_ERROR);
    }

    /* Create a dummy physmem handle if I don't already have one. */
    if (NULL == dummy_handle) {
	dummy_handle = new PHYSMEM_HANDLE;
    }

    /* Check for problem allocating memory. */
    if (NULL == dummy_handle) {
	rcs_print_error("CMS: Couldn't create PHYSMEM_HANDLE.\n");
	return (status = CMS_INTERNAL_ACCESS_ERROR);
    }

    dummy_handle->set_to_ptr(_global, _global_size);
    internal_access(dummy_handle, _local);
    return (status);
}

 /* This function should be called by classes which overload the main_access
    function. */
CMS_STATUS CMS::internal_access(PHYSMEM_HANDLE * _global, void *_local)
{
    status = CMS_STATUS_NOT_SET;
    if (NULL == _global) {
	rcs_print_error("CMS: Handle to global memory is NULL.\n");
	return (status = CMS_INTERNAL_ACCESS_ERROR);
    }
    handle_to_global_data = _global;

    if (CMS_CLEAR_ACCESS == internal_access_type) {
	internal_clear();
	return (status);
    }

    if (min_compatible_version > 2.58 || min_compatible_version < 1E-6) {
	handle_to_global_data->offset += skip_area;
    }

    if (CMS_GET_DIAG_INFO_ACCESS == internal_access_type) {
	internal_retrieve_diag_info(handle_to_global_data, _local);
	return (status);
    }

    long orig_offset = handle_to_global_data->offset;
    if (enable_diagnostics) {
	handle_to_global_data->offset +=
	    sizeof(CMS_DIAG_HEADER) +
	    total_connections * sizeof(CMS_DIAG_PROC_INFO);
	handle_to_global_data->enable_byte_counting = 1;
	pre_op_total_bytes_moved = handle_to_global_data->total_bytes_moved;
    }

    char was_read_byte;
    write_just_completed = 0;
    int read_only = ((internal_access_type == CMS_CHECK_IF_READ_ACCESS) ||
	(internal_access_type == CMS_READ_ACCESS) ||
	(internal_access_type == CMS_PEEK_ACCESS));

    long offset_before_split = handle_to_global_data->offset;

    if (total_subdivisions >= 1 && current_subdivision > 0
	&& current_subdivision < total_subdivisions) {
	handle_to_global_data->offset += (current_subdivision * subdiv_size);
	handle_to_global_data->size =
	    ((current_subdivision + 1) * subdiv_size);
	if (handle_to_global_data->size > size) {
	    handle_to_global_data->size = size;
	}
    }

    if (split_buffer) {
	if (internal_access_type == CMS_WRITE_IF_READ_ACCESS) {
	    handle_to_global_data->offset++;
	    handle_to_global_data->read(&was_read_byte, 1);
	    handle_to_global_data->offset--;
	    header.was_read = (was_read_byte == toggle_bit + 1);
	    if (!header.was_read) {
		status = CMS_WRITE_WAS_BLOCKED;
		return (status);
	    }
	    internal_access_type = CMS_WRITE_ACCESS;
	}
	if (read_only == toggle_bit) {
	    handle_to_global_data->offset += 2;
	    handle_to_global_data->size = half_size;
	} else {
	    handle_to_global_data->offset += half_offset;
	    handle_to_global_data->size = size;
	}
    }

    if (!queuing_enabled) {
	if (neutral) {
	    switch (internal_access_type) {
	    case CMS_CHECK_IF_READ_ACCESS:
		check_if_read_encoded();
		break;
	    case CMS_READ_ACCESS:
		read_encoded();
		break;
	    case CMS_PEEK_ACCESS:
		peek_encoded();
		break;
	    case CMS_WRITE_ACCESS:
		write_encoded();
		break;
	    case CMS_WRITE_IF_READ_ACCESS:
		write_if_read_encoded();
		break;
	    case CMS_GET_MSG_COUNT_ACCESS:
		get_msg_count_encoded();
		break;
	    default:
		return (status = CMS_INTERNAL_ACCESS_ERROR);
	    }
	} else {
	    switch (internal_access_type) {
	    case CMS_CHECK_IF_READ_ACCESS:
		check_if_read_raw();
		break;
	    case CMS_READ_ACCESS:
		read_raw();
		break;
	    case CMS_PEEK_ACCESS:
		peek_raw();
		break;
	    case CMS_WRITE_ACCESS:
		write_raw(_local);
		break;
	    case CMS_WRITE_IF_READ_ACCESS:
		write_if_read_raw(_local);
		break;
	    case CMS_GET_MSG_COUNT_ACCESS:
		get_msg_count_raw();
		break;
	    default:
		return (status = CMS_INTERNAL_ACCESS_ERROR);
	    }
	}
    } else {
	if (neutral) {
	    switch (internal_access_type) {
	    case CMS_CHECK_IF_READ_ACCESS:
		queue_check_if_read_encoded();
		break;
	    case CMS_READ_ACCESS:
		queue_read_encoded();
		break;
	    case CMS_PEEK_ACCESS:
		queue_peek_encoded();
		break;
	    case CMS_WRITE_ACCESS:
		queue_write_encoded();
		break;
	    case CMS_WRITE_IF_READ_ACCESS:
		queue_write_if_read_encoded();
		break;
	    case CMS_GET_QUEUE_LENGTH_ACCESS:
		queue_get_queue_length_encoded();
		break;
	    case CMS_GET_SPACE_AVAILABLE_ACCESS:
		queue_get_space_available_encoded();
		break;
	    case CMS_GET_MSG_COUNT_ACCESS:
		queue_get_msg_count_encoded();
		break;
	    default:
		return (status = CMS_INTERNAL_ACCESS_ERROR);
	    }
	} else {
	    switch (internal_access_type) {
	    case CMS_CHECK_IF_READ_ACCESS:
		queue_check_if_read_raw();
		break;
	    case CMS_READ_ACCESS:
		queue_read_raw();
		break;
	    case CMS_PEEK_ACCESS:
		queue_peek_raw();
		break;
	    case CMS_WRITE_ACCESS:
		queue_write_raw(_local);
		break;
	    case CMS_WRITE_IF_READ_ACCESS:
		queue_write_if_read_raw(_local);
		break;
	    case CMS_GET_QUEUE_LENGTH_ACCESS:
		queue_get_queue_length_raw();
		break;
	    case CMS_GET_SPACE_AVAILABLE_ACCESS:
		queue_get_space_available_raw();
		break;
	    case CMS_GET_MSG_COUNT_ACCESS:
		queue_get_msg_count_raw();
		break;
	    default:
		return (status = CMS_INTERNAL_ACCESS_ERROR);
	    }
	}
    }

    if (split_buffer) {
	handle_to_global_data->offset = offset_before_split + 1;
	if (internal_access_type == CMS_READ_ACCESS) {
	    was_read_byte = 1;
	} else if (!read_only) {
	    was_read_byte = 0;
	}
	if (-1 == handle_to_global_data->write(&was_read_byte, 1)) {
	    rcs_print_error("CMS: can not set was read flag.\n");
	}
    }

    if (enable_diagnostics) {
	handle_to_global_data->offset = orig_offset;
	calculate_and_store_diag_info(handle_to_global_data, _local);
    }
    return (status);
}

/* Clear the shared or global memory. */
CMS_STATUS CMS::internal_clear()
{
    in_buffer_id = 0;

    /* Check that the handle to global memory exists. */
    if (NULL == handle_to_global_data) {
	rcs_print_error("CMS: handle_to_global_data is NULL.\n");
	return (status = CMS_INTERNAL_ACCESS_ERROR);
    }

    if (-1 == handle_to_global_data->clear_memory()) {
	rcs_print_error("CMS: Can't clear global_memory.\n");
	return (status = CMS_INTERNAL_ACCESS_ERROR);
    }
    int temp_offset = handle_to_global_data->offset;
    handle_to_global_data->offset = 0;
    if (-1 == handle_to_global_data->write(BufferName, 32)) {
	rcs_print_error("CMS: Can't clear reset name in global memory.\n");
	return (status = CMS_INTERNAL_ACCESS_ERROR);
    }
    handle_to_global_data->offset = temp_offset;

    return (status = CMS_CLEAR_OK);
}

/* Determine whether the message in the buffer has been read at least once. */
int CMS::check_if_read_raw()
{
    /* Check that the handle to global memory exists. */
    if (NULL == handle_to_global_data) {
	rcs_print_error("CMS: handle_to_global_data is NULL.\n");
	status = CMS_INTERNAL_ACCESS_ERROR;
	return 0;
    }

    /* Read the header for the message. */
    if (-1 == handle_to_global_data->read(&header, sizeof(CMS_HEADER))) {
	rcs_print_error
	    ("CMS: Error reading from global memory for %s at %s:%d",
	    BufferName, __FILE__, __LINE__);
	status = CMS_INTERNAL_ACCESS_ERROR;
	return 0;
    }
    return ((int) header.was_read);
}

 /* Determine if all of the messages in the buffer have been read. This means 
    the queue is empty. */
int CMS::queue_check_if_read_raw()
{
    /* Check that the handle to global memory exists. */
    if (NULL == handle_to_global_data) {
	rcs_print_error("CMS: handle_to_global_data is NULL.\n");
	status = CMS_INTERNAL_ACCESS_ERROR;
	return 0;
    }

    /* Read the header for the buffer. */
    if (-1 == handle_to_global_data->read(&queuing_header,
	    sizeof(CMS_QUEUING_HEADER))) {
	rcs_print_error
	    ("CMS: Error reading from global memory for %s at %s:%d\n",
	    BufferName, __FILE__, __LINE__);
	status = CMS_INTERNAL_ACCESS_ERROR;
	return 0;
    }

    return ((int) (queuing_header.queue_length == 0));
}

/* Determine whether the message in the buffer has been read at least once. */
int CMS::check_if_read_encoded()
{
    /* Check that the handle to global memory exists. */
    if (NULL == handle_to_global_data) {
	rcs_print_error("CMS: handle_to_global_data is NULL.\n");
	status = CMS_INTERNAL_ACCESS_ERROR;
	return 0;
    }

    /* Read the header for the buffer. */
    if (-1 == handle_to_global_data->read(encoded_header,
	    encoded_header_size)) {
	rcs_print_error
	    ("CMS: Error reading from global memory for %s at %s:%d\n",
	    BufferName, __FILE__, __LINE__);
	status = CMS_INTERNAL_ACCESS_ERROR;
	return 0;
    }

    /* Decode the header and store in header structure. */
    decode_header();

    return ((int) header.was_read);
}

 /* Determine if all of the messages in the buffer have been read. This means 
    the queue is empty. */
int CMS::queue_check_if_read_encoded()
{
    /* Check that the handle to global memory exists. */
    if (NULL == handle_to_global_data) {
	rcs_print_error("CMS: handle_to_global_data is NULL.\n");