shmem.cc

Source code for an Numeric Cmputer

    if (min_compatible_version < 3.44 && min_compatible_version > 0) {
	total_subdivisions = 1;
    }

    if (min_compatible_version > 2.57 || min_compatible_version <= 0) {
	if (!shm->created) {
	    char *cptr = (char *) shm->addr;
	    cptr[31] = 0;
	    if (strncmp(cptr, BufferName, 31)) {
		rcs_print_error
		    ("Shared memory buffers %s and %s may conflict. (key=%d(0x%X))\n",
		    BufferName, cptr, key, key);
		strncpy(cptr, BufferName, 32);
	    }
	}
	if (master) {
/*! \todo Another #if 0 */
#if 0				// PC Do we need to use autokey ?
	    if (use_autokey_for_connection_number) {
		void *autokey_table_end =
		    (void *) (((char *) shm->addr) + 32 + autokey_table_size);
		memset(autokey_table_end, 0, size - 32 - autokey_table_size);
	    }
#endif
	    strncpy((char *) shm->addr, BufferName, 32);
	}
/*! \todo Another #if 0 */
#if 0				// PC Do we need to use autokey ?
	if (use_autokey_for_connection_number) {
	    void *autokey_table = (void *) (((char *) shm->addr) + 32);
	    connection_number =
		autokey_getkey(autokey_table, total_connections, ProcessName);
	    shm_addr_offset =
		(void *) ((char *) (shm->addr) + 32 + autokey_table_size);
	    max_message_size -= (32 + autokey_table_size);	/* size of
								   cms buffer 
								   available
								   for user */
	} else {
#endif
	    shm_addr_offset = (void *) ((char *) (shm->addr) + 32);
	    max_message_size -= 32;	/* size of cms buffer available for
					   user */
/*! \todo Another #if 0 */
#if 0				// PC Do we need to use autokey ?
	}
#endif
	/* messages = size - CMS Header space */
	if (enc_max_size <= 0 || enc_max_size > size) {
	    if (neutral) {
		max_encoded_message_size -= 32;
	    } else {
		max_encoded_message_size -=
		    (cms_encoded_data_explosion_factor * 32);
	    }
	}
	/* Maximum size of message after being encoded. */
	guaranteed_message_space -= 32;	/* Largest size message before being
					   encoded that can be guaranteed to
					   fit after xdr. */
	size -= 32;
	size_without_diagnostics -= 32;
	subdiv_size =
	    (size_without_diagnostics -
	    total_connections) / total_subdivisions;
	subdiv_size -= (subdiv_size % 4);
    } else {
	if (master) {
	    memset(shm->addr, 0, size);
	}
	shm_addr_offset = shm->addr;
    }
    skip_area = 32 + total_connections + autokey_table_size;
    mao.data = shm_addr_offset;
    mao.timeout = timeout;
    mao.total_connections = total_connections;
    mao.sem_delay = sem_delay;
    mao.connection_number = connection_number;
    mao.split_buffer = split_buffer;
    mao.read_only = 0;
    mao.sem = sem;

    fast_mode = !queuing_enabled && !split_buffer && !neutral &&
	(mutex_type == NO_SWITCHING_MUTEX);
    handle_to_global_data = dummy_handle = new PHYSMEM_HANDLE;
    handle_to_global_data->set_to_ptr(shm_addr_offset, size);
    if ((connection_number < 0 || connection_number >= total_connections)
	&& (mutex_type == MAO_MUTEX || mutex_type == MAO_MUTEX_W_OS_SEM)) {
	rcs_print_error("Bad connection number %d\n", connection_number);
	status = CMS_MISC_ERROR;
	return -1;
    }
    return 0;
}

/* Closes the  shared memory and mutual exclusion semaphore  descriptors. */
int SHMEM::close()
{
    int nattch = 0;
    second_read = 0;

/*! \todo Another #if 0 */
#if 0				// PC Do we need to use autokey ?
    if (use_autokey_for_connection_number) {
	void *autokey_table = (void *) (((char *) shm->addr) + 32);
	autokey_releasekey(autokey_table, total_connections, ProcessName,
	    connection_number);
    }
#endif
    if (NULL != shm) {
	/* see if we're the last one */
	nattch = shm->nattch();
	shm->delete_totally = delete_totally;
	delete shm;
	shm = NULL;
    }
    if (NULL != sem) {
	/* if we're the last one, then make us the master so that the
	   semaphore will go away */
	if ((nattch <= 1 && nattch > -1) || delete_totally) {
	    sem->setflag(RCS_SEMAPHORE_CREATE);
	} else {
	    sem->setflag(RCS_SEMAPHORE_NOCREATE);
	}
	delete sem;
    }
    if (NULL != bsem) {
	/* if we're the last one, then make us the master so that the
	   semaphore will go away */
	if ((nattch <= 1 && nattch > -1) || delete_totally) {
	    bsem->setflag(RCS_SEMAPHORE_CREATE);
	} else {
	    bsem->setflag(RCS_SEMAPHORE_NOCREATE);
	}
	delete bsem;
    }
#ifdef DEBUG
    printf("SHMEM(%s): nattch = %d\n", BufferName, nattch);
#endif

    return 0;
}

/* Access the shared memory buffer. */
CMS_STATUS SHMEM::main_access(void *_local)
{

    /* Check pointers. */
    if (shm == NULL) {
	second_read = 0;
	return (status = CMS_MISC_ERROR);
    }

    if (bsem == NULL && not_zero(blocking_timeout)) {
	rcs_print_error
	    ("No blocking semaphore available. Can not call blocking_read(%f).\n",
	    blocking_timeout);
	second_read = 0;
	return (status = CMS_NO_BLOCKING_SEM_ERROR);
    }

    mao.read_only = ((internal_access_type == CMS_CHECK_IF_READ_ACCESS) ||
	(internal_access_type == CMS_PEEK_ACCESS) ||
	(internal_access_type == CMS_READ_ACCESS));

    switch (mutex_type) {
    case NO_MUTEX:
	break;

    case MAO_MUTEX:
    case MAO_MUTEX_W_OS_SEM:
	switch (mem_get_access(&mao)) {
	case -1:
	    rcs_print_error("SHMEM: Can't take semaphore\n");
	    second_read = 0;
	    return (status = CMS_MISC_ERROR);
	case -2:
	    if (timeout > 0) {
		rcs_print_error("SHMEM: Timed out waiting for semaphore.\n");
		rcs_print_error("buffer = %s, timeout = %lf sec.\n",
		    BufferName, timeout);
	    }
	    second_read = 0;
	    return (status = CMS_TIMED_OUT);
	default:
	    break;
	}
	toggle_bit = mao.toggle_bit;
	break;

    case OS_SEM_MUTEX:
	if (sem == NULL) {
	    second_read = 0;
	    return (status = CMS_MISC_ERROR);
	}
	switch (sem->wait()) {
	case -1:
	    rcs_print_error("SHMEM: Can't take semaphore\n");
	    second_read = 0;
	    return (status = CMS_MISC_ERROR);
	case -2:
	    if (timeout > 0) {
		rcs_print_error("SHMEM: Timed out waiting for semaphore.\n");
		rcs_print_error("buffer = %s, timeout = %lf sec.\n",
		    BufferName, timeout);
	    }
	    second_read = 0;
	    return (status = CMS_TIMED_OUT);
	default:
	    break;
	}
	break;

    case NO_INTERRUPTS_MUTEX:
	rcs_print_error("Interrupts can not be disabled.\n");
	second_read = 0;
	return (status = CMS_MISC_ERROR);
	break;

    case NO_SWITCHING_MUTEX:
	rcs_print_error("Interrupts can not be disabled.\n");
	return (status = CMS_MISC_ERROR);
	break;

    default:
	rcs_print_error("SHMEM: Invalid mutex type.(%d)\n", mutex_type);
	second_read = 0;
	return (status = CMS_MISC_ERROR);
	break;
    }

    if (second_read > 0 && enable_diagnostics) {
	disable_diag_store = 1;
    }

    /* Perform access function. */
    internal_access(shm->addr, size, _local);

    disable_diag_store = 0;

    if (NULL != bsem &&
	(internal_access_type == CMS_WRITE_ACCESS
	    || internal_access_type == CMS_WRITE_IF_READ_ACCESS)) {
	bsem->flush();
    }
    switch (mutex_type) {
    case NO_MUTEX:
	break;

    case MAO_MUTEX:
    case MAO_MUTEX_W_OS_SEM:
	mem_release_access(&mao);
	break;

    case OS_SEM_MUTEX:
	sem->post();
	break;
    case NO_INTERRUPTS_MUTEX:
	rcs_print_error("Can not restore interrupts.\n");
	break;

    case NO_SWITCHING_MUTEX:
	rcs_print_error("Can not restore interrupts.\n");
	break;
    }

    switch (internal_access_type) {

    case CMS_READ_ACCESS:
	if (NULL != bsem && status == CMS_READ_OLD &&
	    (blocking_timeout > 1e-6 || blocking_timeout < -1E-6)) {
	    if (second_read > 10 && total_subdivisions <= 1) {
		status = CMS_MISC_ERROR;
		rcs_print_error
		    ("CMS: Blocking semaphore error. The semaphore wait has returned %d times but there is still no new data.\n",
		    second_read);
		second_read = 0;
		return (status);
	    }
	    second_read++;
	    bsem->timeout = blocking_timeout;
	    int bsem_ret = bsem->wait();
	    if (bsem_ret == -2) {
		status = CMS_TIMED_OUT;
		second_read = 0;
		return (status);
	    }
	    if (bsem_ret == -1) {
		rcs_print_error("CMS: Blocking semaphore error.\n");
		status = CMS_MISC_ERROR;
		second_read = 0;
		return (status);
	    }
	    main_access(_local);
	}
	break;

    case CMS_WRITE_ACCESS:
    case CMS_WRITE_IF_READ_ACCESS:
	break;

    default:
	break;

    }

    second_read = 0;
    return (status);
}