/**********************************************************************
 * Copyright (c) 2005, 2009 IBM Corporation and others.
 * All rights reserved.   This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 * $Id: ossrambo.cpp,v 1.8 2009/07/10 00:11:33 jwest Exp $
 * 
 * Contributors: 
 * IBM - Initial API and implementation
 **********************************************************************/

/******************************************************************************* 
 
   Source File Name = ossrambo.cpp
 
   Descriptive Name = OSSe RApid Memory Buffer Out (RAMBO) 
 
   Function: 
      Defines: 
 
   Dependencies: 
      None 
 
   Restrictions: 
      None 
 
   Change Activity: 
   Defect Date        Who Description 
   ====== =========== === ============================================== 
   164204 Jan/15/2001 jpk Initial drop 
   171596 Apr/05/2001 dns added ossRamboStopFlusher which is asynchronous 
   170752 Apr/25/2001 dns Porting changes for MVS 
   172892 May/01/2001 dns created new version of ossRamboFlushToFunc and 
                      saved old version in ossRamboFlushFullChunksToFunc 
   175248 Jun/01/2001 dns added defintions for ra_attachToShm and 
                          ra_stopFlushingShm 
   175881 Sep/17/2001 dns added code to ra_stopFlushingShm to check if it 
                          was called already 
   194539 Jan/24/2002 dns changed slot copy routines to return OSS_ERR_INVALID 
                          if the input parameters are invalid 
   186134 Mar/01/2002 dns Use new atomic types and operations 
   186132 Mar/01/2002 dns Add a call to detach from the shared memory in the 
                           ossRamboStopFlusher function so shared memory will 
                           be deleted 
   198075 Mar/05/2002 dns Changed ossRamboSlotCopy, ossRamboMultiSlotCopy, 
                          ossRamboFlushToFD to write the data length to the 
                          shared memory and read it an use it when flushing 
                          - added dataProcessor routine for flushing 
   194483 Mar/22/2002 dns Moved the flushing functions to ossramboflush.c/cpp 
   216026 Jul/23/2002 dns Added writers field to OSSRambo structure 
                          - increment and decrement it in ossRamboMultiSlotCopy 
   Last Changed =    02/08/27  12:31:30 
 
*******************************************************************************/ 
 
#ifdef ENW 
#include <stdio.h> 
#include <stdlib.h> 
#include <fcntl.h> 
#include <sys/types.h> 
#include <sys/stat.h> 
#include <string.h> 
#ifdef SQLWINT 
#include <limits.h> 
#include <io.h> 
#include <winsock2.h> /* 134924 */
#else 
#ifdef OSS_DEBUG 
#include <errno.h>      /* 198075 */ 
#endif 
#include <sys/socket.h> 
#define SOCKET_ERROR -1 
#endif 
#else 
#include <fcntl.h> 
#endif 
#ifdef SQLUNIX 
#include <limits.h> 
#include <memory.h> 
#include <unistd.h> 
#endif 
 
#include "oss.h" 
#include "ossdebug.h" 
#include "osserror.h" 
#include "ossipcmemory.h" 
#ifdef ENW 
#include "RASharedMemory.h" 
#include "RAShm.h"            /* 175248 */ 
#include "ossramboinl.h" 
#else 
#include "ossrambo.h" 
#endif 
#include "osstime.h" 
#ifdef OSS_DEBUG 
#include "RAServerConfigBinding.h" 
#endif 


/* 198075 - size of header information for data stored in rambo buffer */ 
#define OSS_RAMBO_DATA_HEADER_SIZE  sizeof(Uint32) 
 
/******************************************************************************* 
 
   Name 
      round2 
 
   Function 
      Returns the value of x rounded down to the next power of 2.  If x is 
      already an exact power of 2, the returned value is the same as x.  If 
      x is 0, the return value is 0. 
 
 ******************************************************************************/ 
Uint32 round2( Uint32 x ) 
{ 
   Uint32 temp ; 
   Sint32 bitShift ; 
 
   if ( 0 == x ) 
      return 0 ; 
 
   /* Figure out the number of bits in x */ 
   temp = x ; 
   bitShift = -1 ; 
   do 
   { 
      temp >>= 1 ; 
      bitShift++ ; 
   } 
   while ( temp > 0 ) ; 
 
   return ( 1 << bitShift )  ; 
} 

#if defined(__linux__) && defined(USE_PTHREAD_ATOMICS)

pthr_atom_critsec_t * globalLatchLock = 0;
volatile int globalLatchLockInit = 0;

void initGlobalLatchLock() {
	if(!globalLatchLockInit || globalLatchLock == 0) {
			globalLatchLock = (pthr_atom_critsec_t *)malloc(sizeof(pthr_atom_critsec_t));
			pthr_atom_mutexCreate(globalLatchLock);
			globalLatchLockInit = 1;
	} 
} 

#endif
	    
/******************************************************************************* 
 
   Name 
      ossRamboCreate 
 
   Function 
      Creates a RAMBO buffer in shared memory. 
 
   Inputs 
      1. pszBufferName 
         Address of a null-terminated string containing a name for the 
         RAMBO buffer.  This name is used to identify the buffer. 
      2. bufferSize 
         The size of the RAMBO buffer in bytes.  The size must be: 
            a) >= OSS_RAMBO_BUFFER_SIZE_MIN 
            b) and <= OSS_RAMBO_BUFFER_SIZE_MAX 
         If not, the function returns OSS_ERR_INVALID.  If bufferSize is not 
         an exact power of 2 it will be rounded down to the next power of 2. 
      3. appInMemDataSize 
         The size of application-specific in memory data buffer that will 
         be associated with the RAMBO buffer.  If the size is 0, no 
         application-specific in memory data buffer will be allocated and the 
         ppAppInMemData output parameter will not be changed. 
      4. appOnDiskDataSize 
         The size of application-specific on disk data buffer that will 
         be associated with the RAMBO buffer.  If the size is 0, no 
         application-specific in memory data buffer will be allocated and the 
         ppAppOnDiskData output parameter will not be changed. 
 
   Outputs 
      5. pRambo 
         If the function is successful, this contains the address of a 
         RAMBO handle. 
      6. ppAppInMemData 
         If the function is successful and appInMemDataSize > 0, this contains 
         the address of a pointer to the application-specific in memory data 
         area associated with the RAMBO buffer.  If this data area does not 
         exist, the pointer is NULL. 
      7. ppAppOnDiskData 
         If the function is successful and appOnDiskSize > 0, this contains 
         the address of a pointer to the application-specific on disk data 
         area associated with the RAMBO buffer.  If this data area does not 
         exist, the pointer is NULL. 
 
   Normal Returns 
      OSS_OK 
 
   Error Returns 
      OSS_ERR_RAMBO_BUFFER_SIZE
	  OSS_ERR_RAMBO_BUFFER_NAME
	  OSS_ERR_RAMBO_BUFFER_PTR
      OSS_ERR_IPC_CREATE_MAP_FAILED
      OSS_ERR_IPC_MEM_MAP_FAILED
      OSS_ERR_IPC_GET_KEY_FAILED
	  OSS_ERR_IPC_SHMEM_GET_FAILED
      OSS_ERR_IPC_SHMEM_ATTACH_FAILED
      OSS_ERR_IPC_NOMEMORY
      OSS_ERR_IPC_DUPLICATE
      OSS_ERR_FAILED 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboCreate( 
      char        * pszBufferName, 
      Uint32        bufferSize, 
      Uint32        appInMemDataSize, 
      Uint32        appOnDiskDataSize, 
      OSSRambo    * pRambo, 
      void       ** ppAppInMemData, 
      void       ** ppAppOnDiskData ) 
{ 
   /* 
    * The RAMBO buffer. 
    * 
    *     *** IN MEMORY LAYOUT *** 
    * 
    * +--------------------------------+ 
    * | OSSRamboCB                     | 
    * +--------------------------------+ 
    * | OSSRamboChunkCB #1             | 
    * +--------------------------------+ 
    * | OSSRamboChunkCB #2             | 
    * +--------------------------------+ 
    * | ...                            | 
    * +--------------------------------+ 
    * | OSSRamboChunkCB #n             | 
    * +--------------------------------+         *** ON DISK LAYOUT *** 
    * | App-specific in memory buffer* | 
    * +--------------------------------+   +-------------------------------+ 
    * | OSSRamboOnDisk                 |   | OSSRamboOnDisk                | 
    * +--------------------------------+   +-------------------------------+ 
    * | App-specific on disk buffer*   |   | App-specific on disk buffer*  | 
    * +--------------------------------+   +-------------------------------+ 
    * | Memory output buffer           |   | Memory output buffer          | 
    * +--------------------------------+   +-------------------------------+ 
    * 
    * The areas indicated with the "*" may not exist if they were not 
    * requested. 
    */ 
 
   OSSErr osserr = OSS_OK ; 
   Uint32 adjBufferSize = 0 ; 
   Uint32 numChunks = 0 ; 
   Uint32 chunkCBSize = 0 ; 
   Uint32 totalSize = 0 ; 
   OSSRamboCB * pRamboCB = NULL ; 
   char szBufferName[OSS_RAMBO_BUFFER_NAME_SIZE+1] = { 0, } ; 
   void * ipcMemHandle = NULL ; 
   OSSRamboChunkCB * pChunkCB = NULL ; 
   OSSRamboOnDiskCB * pOnDskCB = NULL ; 
   Nuint i ; 
   
	#if defined(__linux__) && defined(USE_PTHREAD_ATOMICS)
		initGlobalLatchLock();
	#endif
   
	// printf("size of ra_shm_handle_t inside ossrambo.cpp %d\n", sizeof(OSSRambo)); // jgw-shmem
 
   /* Validate the buffer name */ 
   OSS_ASSERT( NULL != pszBufferName ) ; 
   OSS_ASSERT( strlen( pszBufferName ) > 0 ) ; 
   if ( ( NULL == pszBufferName ) && ( strlen( pszBufferName ) == 0 ) ) 
   {
      osserr = OSS_ERR_RAMBO_BUFFER_NAME ; 
      goto exit ; 
   } 
 
   /* Validate the buffer size */ 
   OSS_ASSERT( bufferSize >= OSS_RAMBO_BUFFER_SIZE_MIN ) ; 
   OSS_ASSERT( bufferSize <= OSS_RAMBO_BUFFER_SIZE_MAX ) ; 
   if ( ( bufferSize < OSS_RAMBO_BUFFER_SIZE_MIN ) 
         || ( bufferSize > OSS_RAMBO_BUFFER_SIZE_MAX ) ) 
   {
      osserr = OSS_ERR_RAMBO_BUFFER_SIZE ; 
      goto exit ; 
   } 
 
   /* Validate the RAMBO handle pointer */ 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo, sizeof( *pRambo ) ) ) ; 
   if ( NULL == pRambo ) 
   {
      osserr = OSS_ERR_RAMBO_PTR ; 
      goto exit ; 
   } 
 
#ifdef OSS_DEBUG 
   /* Validate the application-specific data areas */ 
   if ( appInMemDataSize > 0 ) 
   { 
      OSS_ASSERT(!ossIsBadWritePtr(ppAppInMemData,sizeof(*ppAppInMemData))); 
   } 
 
   if ( appOnDiskDataSize > 0 ) 
   { 
      OSS_ASSERT(!ossIsBadWritePtr(ppAppOnDiskData,sizeof(*ppAppOnDiskData))); 
   } 
#endif 
 
   /* Round down the buffer size to a power of 2 */ 
   adjBufferSize = round2( bufferSize  ) ; 
   OSS_ASSERT( adjBufferSize <= bufferSize ) ; 
   OSS_ASSERT( adjBufferSize >= OSS_RAMBO_BUFFER_SIZE_MIN ) ; 
   OSS_ASSERT( adjBufferSize <= OSS_RAMBO_BUFFER_SIZE_MAX ) ; 
 
   /* Determine the number of chunks required */ 
   numChunks = adjBufferSize / OSS_RAMBO_CHUNK_SIZE ; 
   OSS_ASSERT( numChunks > 0 ) ; 
   chunkCBSize = numChunks * sizeof( OSSRamboChunkCB ) ; 
 
   /* Allocate the RAMBO buffer */ 
   totalSize = sizeof( *pRamboCB ) 
             + ossAlignP( chunkCBSize ) 
             + ossAlignP( appInMemDataSize ) 
             + sizeof( OSSRamboOnDiskCB ) 
             + ossAlignP( appOnDiskDataSize ) 
             + adjBufferSize ; 
 
   /* Truncate the buffer name */ 
   strncpy( szBufferName, pszBufferName, OSS_RAMBO_BUFFER_NAME_SIZE ) ; 
   szBufferName[OSS_RAMBO_BUFFER_NAME_SIZE] = '\0' ; 
 
#ifdef OSS_DEBUG 
   printf("ossRamboCreate:  About to create shared memory with name %s new name %s totalsize=%d  adjbuffersize=%d\n", 
	   pszBufferName, szBufferName, totalSize, adjBufferSize); 
#endif 
 
   ossIPCMemInitResSize();     /* 186134 */ 
 
   ossIPCMemSetResSize( szBufferName, totalSize ) ; 
 
   osserr = ossIPCMemCreate( &ipcMemHandle, 
                             szBufferName, 
                             OSS_RAMBO_IPC_MEM_ID, 
                             totalSize, 
                             0, 
                             NULL ) ; 
   if ( OSS_OK != osserr ) 
   { 
      OSS_ASSERT( !("ossIPCMemCreated failed") ) ; 
      goto exit ; 
   } 
 
   OSS_ASSERT( NULL != ipcMemHandle ) ; 
   pRamboCB = (OSSRamboCB *)ossIPCMemGetFixed( ipcMemHandle ) ; 
   OSS_ASSERT( !ossIsBadReadPtr ( pRamboCB, totalSize ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRamboCB, totalSize ) ) ; 
 
   /* Initialize the RAMBO handle */ 
   memcpy( pRamboCB->l1.h1.eye, OSS_RAMBO_RAMBOCB_EYE, 
           OSS_RAMBO_EYE_CATCHER_SIZE ) ; 
   memcpy( pRamboCB->l1.h1.szBufferName, szBufferName, 
           sizeof( szBufferName ) ) ; 
/* DNS 
   pRamboCB->l2.h2.pFirstChunkCB 
         = (OSSRamboChunkCB *)( (uintptr_t)pRamboCB 
         + (uintptr_t)ossAlignP( sizeof( *pRamboCB ) ) ) ; 
*/ 
   pRamboCB->l2.h2.oFirstChunkCB 
         = (Sint32)ossAlignP( sizeof( *pRamboCB ) ) ; 
 
   pRamboCB->l2.h2.chunkCBSize = chunkCBSize ; 
 
   /* 
    * Set the pointer to the application-specific in memory buffer or to NULL 
    * if there is no such area.  Then setup the pointer to the RAMBO 
    * on disk structure. 
    */ 
   if ( appInMemDataSize > 0 ) 
   { 
/* DNS 
      pRamboCB->l1.h1.pAppInMemData 
         = (void *)( (uintptr_t)pRamboCB->l2.h2.pFirstChunkCB 
         + (uintptr_t)ossAlignP( pRamboCB->l2.h2.chunkCBSize ) )  ; 
 
      pRamboCB->l2.h2.pOnDiskCB 
         = (OSSRamboOnDiskCB *)( (uintptr_t)pRamboCB->l1.h1.pAppInMemData 
         + (uintptr_t)ossAlignP( appInMemDataSize ) )  ; 
*/ 
      pRamboCB->l1.h1.oAppInMemData 
         = pRamboCB->l2.h2.oFirstChunkCB 
         + (Sint32)ossAlignP( pRamboCB->l2.h2.chunkCBSize )  ; 
 
      pRamboCB->l2.h2.oOnDiskCB 
         = pRamboCB->l1.h1.oAppInMemData 
         + (Sint32)ossAlignP( appInMemDataSize )  ; 
   } 
   else 
   { 
/* DNS 
      pRamboCB->l1.h1.pAppInMemData = NULL ; 
      pRamboCB->l2.h2.pOnDiskCB 
         = (OSSRamboOnDiskCB *)( (uintptr_t)pRamboCB->l2.h2.pFirstChunkCB 
         + (uintptr_t)ossAlignP( pRamboCB->l2.h2.chunkCBSize ) )  ; 
*/ 
      pRamboCB->l1.h1.oAppInMemData = 0 ; 
      pRamboCB->l2.h2.oOnDiskCB 
         = pRamboCB->l2.h2.oFirstChunkCB 
         + (Sint32)ossAlignP( pRamboCB->l2.h2.chunkCBSize )  ; 
   } 
 
   pOnDskCB = pOnDiskCB(pRamboCB); 
   /* 
    * Set the pointer to the application-specific data buffer or to NULL 
    * if there is no such area.  Then setup the pointer to the memory 
    * buffer. 
    */ 
   if ( appOnDiskDataSize > 0 ) 
   { 
/* DNS 
      pRamboCB->l1.h1.pAppOnDiskData 
         = (void *)( (uintptr_t)pRamboCB->l2.h2.pOnDiskCB 
         + (uintptr_t)ossAlignP( sizeof( *pRamboCB->l2.h2.pOnDiskCB ) ) ) ; 
 
      pRamboCB->l2.h2.pMemBuffer 
         = (void *)( (uintptr_t)pRamboCB->l1.h1.pAppOnDiskData 
         + (uintptr_t)ossAlignP( appOnDiskDataSize ) ) ; 
*/ 
      pRamboCB->l1.h1.oAppOnDiskData 
         = pRamboCB->l2.h2.oOnDiskCB 
         + (Sint32)ossAlignP( sizeof( OSSRamboOnDiskCB ) ) ; 
 
      pRamboCB->l2.h2.oMemBuffer 
         = pRamboCB->l1.h1.oAppOnDiskData 
         + (Sint32)ossAlignP( appOnDiskDataSize ) ; 
   } 
   else 
   { 
/* DNS 
      pRamboCB->l1.h1.pAppOnDiskData = NULL ; 
      pRamboCB->l2.h2.pMemBuffer 
         = (void *)( (uintptr_t)pRamboCB->l2.h2.pOnDiskCB 
         + (uintptr_t)ossAlignP( sizeof( *pRamboCB->l2.h2.pOnDiskCB ) ) ) ; 
*/ 
      pRamboCB->l1.h1.oAppOnDiskData = 0 ; 
      pRamboCB->l2.h2.oMemBuffer 
         = pRamboCB->l2.h2.oOnDiskCB 
         + (Sint32)ossAlignP( sizeof( OSSRamboOnDiskCB ) ) ; 
   } 
 
#ifdef OSS_DEBUG 
   printf("ossRamboCreate:\n   pRamboCB=%p\n   mem buffer=%p\n   OnDiskCB=%p\n   FirstChunkCB=%p\n   chunkCBsize=%d\n", 
	   pRamboCB, pMemBuffer(pRamboCB), pOnDiskCB(pRamboCB), pFirstChunkCB(pRamboCB), chunkCBSize ); 
   /* Fill with a special character to make it easier to recognize */ 
   memset( pMemBuffer(pRamboCB), 0xCC, adjBufferSize ) ; 
#else 
   memset( pMemBuffer(pRamboCB), 0x00, adjBufferSize ) ; 
#endif 
 
   /* Initialize the atomic slot index */ 
   ossAtomicInit(&pRamboCB->l1.h1.nextSlotIndex, 0) ; 
 
   /* Initialize the storage*/ 
/* DNS 
   memcpy( pRamboCB->l2.h2.pOnDiskCB->eye, OSS_RAMBO_ONDISKCB_EYE, 
           OSS_RAMBO_EYE_CATCHER_SIZE ) ; 
   pRamboCB->l2.h2.pOnDiskCB->appOnDiskDataSize = appOnDiskDataSize ; 
   pRamboCB->l2.h2.pOnDiskCB->bufferSize = adjBufferSize ; 
   pRamboCB->l2.h2.pOnDiskCB->bufferWrapped = 0 ;      False 
#ifndef SQLZ_BIG_ENDIAN 
   pRamboCB->l2.h2.pOnDiskCB->smallEndian = 1 ;        True 
#else 
   pRamboCB->l2.h2.pOnDiskCB->smallEndian = 0 ;        False 
#endif 
*/ 
   memcpy( pOnDskCB->eye, OSS_RAMBO_ONDISKCB_EYE, 
           OSS_RAMBO_EYE_CATCHER_SIZE ) ; 
   pOnDskCB->appOnDiskDataSize = appOnDiskDataSize ; 
   pOnDskCB->bufferSize = adjBufferSize ; 
   pOnDskCB->bufferWrapped = 0 ;     /* False */ 
#ifndef SQLZ_BIG_ENDIAN 
   pOnDskCB->smallEndian = 1 ;       /* True */ 
#else 
   pOnDskCB->smallEndian = 0 ;       /* False */ 
#endif 
 
   /* Calculate the max number of chunks and slots */ 
   pRamboCB->l2.h2.maxChunks = numChunks ; 
   pRamboCB->l2.h2.maxSlots 
      = pRamboCB->l2.h2.maxChunks * OSS_RAMBO_SLOTS_PER_CHUNK ; 
 
   OSS_ASSERT( numChunks 
                  == ( pOnDskCB->bufferSize 
                        / OSS_RAMBO_CHUNK_SIZE ) ) ; 
 
   /* Initialize each chunk control block */ 
   pChunkCB = pFirstChunkCB(pRamboCB) ; 
   for ( i = 0; i < pRamboCB->l2.h2.maxChunks; i++, pChunkCB++ ) 
   { 
      int j; 
      OSS_ASSERT( (uintptr_t)pChunkCB 
                     < (uintptr_t)pOnDskCB ) ; 
/* 186134 begin 
      Have to properly initialize the atomic elements of the chunk CBs 
      instead of using memset 
      memset( pChunkCB, 0x00, sizeof( *pChunkCB ) ) ; 
*/ 
      ossAtomicInit(&pChunkCB->fillCount, 0) ; 
      ossAtomicInit(&pChunkCB->flushedCount, 0) ; 
      for(j=0; j < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; j++) { 
         ossAtomicInit(&(pChunkCB->slotMap[j]), 0); 
      } 
/* 186134 end */ 
 
      memcpy( pChunkCB->eye, OSS_RAMBO_CHUNKCB_EYE, 
              OSS_RAMBO_EYE_CATCHER_SIZE ) ; 
   } 
 
   /* Setup the application-specific data area pointers */ 
   if ( NULL != ppAppInMemData ) 
      *ppAppInMemData = (appInMemDataSize > 0) ? pAppInMemData(pRamboCB) : NULL ; 
 
   if ( NULL != ppAppOnDiskData ) 
      *ppAppOnDiskData = (appOnDiskDataSize > 0) ? pAppOnDiskData(pRamboCB) : NULL; 
 
   /* Give the all clear sign */ 
   pRamboCB->l2.h2.status = OSS_RAMBO_CREATED ; 
 
   /* Setup the RAMBO handle */ 
   pRambo->ipcMemHandle = ipcMemHandle ; 
   pRambo->pRamboCB = pRamboCB ; 
   ossAtomicInit(&pRambo->writers, 0) ;   /* 216026 */ 
 
   OSS_ASSERT( NULL != pRambo->ipcMemHandle ) ; 
   OSS_ASSERT( NULL != pRambo->pRamboCB ) ; 
   OSS_ASSERT( OSS_OK == osserr ) ; 
 
exit: 
   return osserr ; 
 
} 
 
 
/******************************************************************************* 
 
   Name 
      ossRamboAttach 
 
   Function 
      Attaches to an existing, initialized RAMBO buffer. 
 
   Inputs 
      1. pszBufferName 
         The address of a null-terminated string containing the name of the 
         RAMBO buffer to attach to. 
 
   Outputs 
      2. pRambo 
         If the function is successful, this contains the address of a 
         RAMBO handle. 
      3. ppAppInMemData 
         If the function is successful and appInMemDataSize > 0, this contains 
         the address of a pointer to the application-specific in memory data 
         area associated with the RAMBO buffer.  If this data area does not 
         exist, the pointer is NULL. 
      4. ppAppOnDiskData 
         If the function is successful and appOnDiskSize > 0, this contains 
         the address of a pointer to the application-specific on disk data 
         area associated with the RAMBO buffer.  If this data area does not 
         exist, the pointer is NULL. 
 
   Normal Returns 
      OSS_OK 
 
   Error Returns
      OSS_ERR_RAMBO_BUFFER_NAME
      OSS_ERR_RAMBO_PTR
      OSS_ERR_IPC_NOTFOUND
      OSS_ERR_IPC_OPEN_MAP_FAILED
      OSS_ERR_IPC_MEM_MAP_FAILED
      OSS_ERR_IPC_GET_KEY_FAILED
	  OSS_ERR_IPC_SHMEM_GET_FAILED
      OSS_ERR_IPC_SHMEM_ATTACH_FAILED
      OSS_ERR_FAILED 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboAttach( 
      char     *  pszBufferName, 
      OSSRambo *  pRambo, 
      void     ** ppAppInMemData, 
      void     ** ppAppOnDiskData ) 
{ 
   OSSErr osserr = OSS_OK ; 
   void * ipcMemHandle = NULL ; 
   OSSRamboCB * pRamboCB = NULL ; 
   char szBufferName[OSS_RAMBO_BUFFER_NAME_SIZE+1] = { 0, } ; 
 
   /* Validate the parameters */ 
   OSS_ASSERT( NULL != pszBufferName ) ; 
   OSS_ASSERT( strlen( pszBufferName ) > 0 ) ; 
   if ( ( NULL == pszBufferName ) && ( 0 == strlen( pszBufferName ) ) ) 
   {
      osserr = OSS_ERR_RAMBO_BUFFER_NAME ; 
      goto exit ; 
   } 
 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo, sizeof( *pRambo ) ) ) ; 
   if ( NULL == pRambo ) 
   {
      osserr = OSS_ERR_RAMBO_PTR ; 
      goto exit ; 
   } 
 
 	#if defined(__linux__) && defined(USE_PTHREAD_ATOMICS)
		initGlobalLatchLock();
	#endif
 
#ifdef OSS_DEBUG 
   printf("ossRamboAttach:  About to attach to shared memory with name %s\n", pszBufferName); 
   if ( NULL != ppAppInMemData ) 
      OSS_ASSERT( !ossIsBadWritePtr( ppAppInMemData, 
                                     sizeof( *ppAppInMemData ) ) ) ; 
 
   if ( NULL != ppAppOnDiskData ) 
      OSS_ASSERT( !ossIsBadWritePtr( ppAppOnDiskData, 
                                     sizeof( *ppAppOnDiskData ) ) ) ; 
#endif 
 
   /* Truncate the buffer name */ 
   strncpy( szBufferName, pszBufferName, OSS_RAMBO_BUFFER_NAME_SIZE ) ; 
   szBufferName[OSS_RAMBO_BUFFER_NAME_SIZE] = '\0' ; 
#ifdef OSS_DEBUG 
   printf("ossRamboAttach:  About to attach to shared memory with new name %s\n", szBufferName); 
#endif 
   /* Attach to the shared memory */ 
   osserr = ossIPCMemAttach( &ipcMemHandle, 
                             szBufferName, 
                             OSS_RAMBO_IPC_MEM_ID, 
                             NULL ) ; 
   if ( osserr != OSS_OK ) 
   { 
      goto exit ; 
   } 
 
   OSS_ASSERT( NULL != ipcMemHandle ) ; 
   pRamboCB = (OSSRamboCB *)ossIPCMemGetFixed( ipcMemHandle ) ; 
#ifdef OSS_DEBUG 
   printf("ossRamboAttach:  Successfully attached to shared memory %p\n", pRamboCB); 
#endif 
 
   OSS_ASSERT( !ossIsBadReadPtr ( pRamboCB, sizeof( *pRamboCB ) ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRamboCB, sizeof( *pRamboCB ) ) ) ; 
   OSS_ASSERT( ossRamboIsInitialized( pRamboCB ) ) ; 
 
   /* Setup the application data area pointers */ 
   if ( NULL != ppAppInMemData ) 
      *ppAppInMemData = pRamboCB->l1.h1.oAppInMemData ? pAppInMemData(pRamboCB) : NULL ; 
/* DNS 
      *ppAppInMemData = pRamboCB->l1.h1.pAppInMemData ; 
*/ 
 
   if ( NULL != ppAppOnDiskData ) 
      *ppAppOnDiskData = pRamboCB->l1.h1.oAppOnDiskData ? pAppOnDiskData(pRamboCB) : NULL ; 
/* DNS 
      *ppAppOnDiskData = pRamboCB->l1.h1.pAppOnDiskData ; 
*/ 
   /* Setup the RAMBO handle */ 
   pRambo->ipcMemHandle = ipcMemHandle ; 
   pRambo->pRamboCB = pRamboCB ; 
   ossAtomicInit(&pRambo->writers, 0) ;   /* 216026 */ 
 
   OSS_ASSERT( NULL != pRambo->ipcMemHandle ) ; 
   OSS_ASSERT( NULL != pRambo->pRamboCB ) ; 
   OSS_ASSERT( OSS_OK == osserr ) ; 
 
exit : 
   return osserr ; 
 
} 
 
 
/******************************************************************************* 
 
   Name 
      ossRamboDestroy 
 
   Function 
      Destroys an existing RAMBO buffer and frees allocated resources. 
 
   Inputs 
      1. pRambo 
         The address of a RAMBO handle. 
 
   Normal Return 
      OSS_OK 
 
   Error Return
      OSS_ERR_RAMBO_PTR
      OSS_ERR_RAMBO_NOT_INIT 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboDestroy( OSSRambo * pRambo ) 
{ 
   OSSErr osserr = OSS_OK ; 
 
   OSS_ASSERT( !ossIsBadReadPtr( pRambo, sizeof( *pRambo ) ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) ) 
   {
	  osserr = OSS_ERR_RAMBO_PTR ;
      goto exit ; 
   } 
 
   OSS_ASSERT( ossRamboIsInitialized( pRambo->pRamboCB ) ) ; 
   if ( !ossRamboIsInitialized( pRambo->pRamboCB ) ) 
   {
      osserr = OSS_ERR_RAMBO_NOT_INIT ; 
      goto exit ; 
   } 
 
   /* Stop the auto-flusher */ 
   if ( ossRamboIsAutoFlusherStarted( pRambo->pRamboCB ) ) 
   { 
      OSS_VERIFY( OSS_OK == ossRamboAutoFlusherStop( pRambo ) ) ; 
   } 
 
   /* Free the IPC memory */ 
   OSS_ASSERT( NULL != pRambo->ipcMemHandle ) ; 
   ossIPCMemDestroy( pRambo->ipcMemHandle ) ; 
 
   OSS_ASSERT( OSS_OK == osserr ) ; 
 
exit : 
   return osserr ; 
  
errorNotInitialized : 
   osserr = OSS_ERR_RAMBO_NOT_INIT ; 
   goto exit ; 
} 
 
OSS_EXTERNC OSSErr OSS_API ossRamboClose( OSSRambo * pRambo ) 
{ 
   return ossIPCMemClose( pRambo->ipcMemHandle ) ; 
} 
 
/******************************************************************************* 
 
   Name 
      ossRamboAutoFlusherStart 
 
   Function 
      Creates a auto-flusher process associated with the specified 
      RAMBO buffer. 
 
   Inputs 
      1. pRambo 
         Address of a RAMBO handle. 
      2. pszAutoFlusherPath 
         Address of a null-terminated string containing the 
         fully qualified path to the auto-flusher executable. 
         This cannot include the filename or trailing separator. 
      2. pszAutoFlusherExe 
         Address of a null-terminuated string containing the filename 
         of the RAMBO auto-flusher executable.  This cannot include the 
         complete path. 
      3. pszOutputFilename 
         Address of a null-terminated string containing the complete 
         path and filename of the output file that the auto-flusher 
         will use. 
 
   Normal Return 
      OSS_OK 
 
   Error Return 
      OSS_ERR_INVALID 
      OSS_ERR_FAILED 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboAutoFlusherStart( 
      OSSRambo   * pRambo, 
      const char * pszAutoFlusherPath, 
      const char * pszAutoFlusherExe, 
      const char * pszOutputFilename ) 
{ 
   OSSErr osserr = OSS_OK ; 
   int rc = 0 ; 
#ifndef ENW 
   pid_t pid ; 
#endif 
   /* Validate the parameters */ 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo, sizeof( *pRambo ) ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) ) 
   { 
      goto errorInvalid ; 
   } 
 
   OSS_ASSERT( ossRamboIsInitialized( pRambo->pRamboCB ) ) ; 
   if ( !ossRamboIsInitialized( pRambo->pRamboCB ) ) 
   { 
      goto errorNotInitialized ; 
   } 
 
   OSS_ASSERT( NULL != pszAutoFlusherExe ) ; 
   OSS_ASSERT( strlen( pszAutoFlusherExe ) > 0 ) ; 
   if ( ( NULL == pszAutoFlusherExe ) || ( 0 == strlen( pszAutoFlusherExe ) ) ) 
   { 
      goto errorInvalid ; 
   } 
 
   OSS_ASSERT( NULL != pszOutputFilename ) ; 
   OSS_ASSERT( strlen( pszOutputFilename ) > 0 ) ; 
   if ( ( NULL == pszOutputFilename ) || ( 0 == strlen( pszOutputFilename ) ) ) 
   { 
      goto errorInvalid ; 
   } 
 
   /* Check if an auto-flusher is already started */ 
   if ( ossRamboIsAutoFlusherStarted( pRambo->pRamboCB ) ) 
   { 
      osserr = OSS_OK ; 
      goto exit ; 
   } 
 
#ifndef ENW 
   pid = fork() ; 
   if ( pid < 0 ) 
   { 
      OSS_ASSERT( !"fork failed" ) ; 
      osserr = OSS_ERR_FAILED ; 
      goto exit ; 
   } 
   else if ( 0 == pid )    /* Child */ 
   { 
      rc = execl( pszAutoFlusherPath, 
                  pszAutoFlusherExe, 
                  pRambo->pRamboCB->l1.h1.szBufferName, 
                  pszOutputFilename, 
                  (char *)0 ) ; 
      if ( rc < 0 ) 
      { 
         OSS_ASSERT( !"execl failed" ) ; 
         osserr = OSS_ERR_FAILED ; 
         goto exit ; 
      } 
   } 
   else                    /* Parent */ 
   { 
      /* Wait until the autoflusher has started */ 
      for ( ; ; ) 
      { 
         if ( pRambo->pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STARTED ) 
            break ; 
 
#ifdef SQLUNIX 
         /* 
          * Sending signal 0 to a process is a way to detect if the 
          * process is still living. 
          */ 
         if ( 0 != kill( pid, 0 ) ) 
         { 
            OSS_ASSERT( !"kill 0 failed" ) ; 
            osserr = OSS_ERR_FAILED ; 
            goto exit ; 
         } 
#endif 
      } 
   } 
#endif /* ENW */ 
 
   OSS_ASSERT( ossRamboIsAutoFlusherStarted( pRambo->pRamboCB ) ) ; 
   OSS_ASSERT( OSS_OK == osserr ) ; 
 
exit : 
   return osserr ; 
 
errorInvalid : 
   osserr = OSS_ERR_INVALID ; 
   goto exit ; 
 
errorNotInitialized : 
   osserr = OSS_ERR_RAMBO_NOT_INIT ; 
   goto exit ; 
} 
 
 
/******************************************************************************* 
 
   Name 
      ossRamboAutoFlusherStop 
 
   Function 
      Terminates the auto-flusher process. 
 
   Inputs 
      1. pRambo 
         Address of a RAMBO handle. 
 
   Normal Return 
      OSS_OK 
 
   Error Return 
      OSS_ERR_INVALID 
      OSS_ERR_RAMBO_NOT_INIT 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboAutoFlusherStop( OSSRambo * pRambo ) 
{ 
   OSSErr osserr = OSS_OK ; 
 
   /* Validate parameters */ 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo, sizeof( *pRambo ) ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   if ( NULL == pRambo ) 
   { 
      goto errorInvalid ; 
   } 
 
   OSS_ASSERT( ossRamboIsInitialized( pRambo->pRamboCB ) ) ; 
   if ( !ossRamboIsInitialized( pRambo->pRamboCB ) ) 
   { 
      goto errorNotInitialized ; 
   } 
 
   /* Check if the auto flusher is started */ 
   if ( !ossRamboIsAutoFlusherStarted( pRambo->pRamboCB ) ) 
   { 
      osserr = OSS_OK ; 
      goto exit ; 
   } 
 
   /* Notify the auto-flusher to stop */ 
   pRambo->pRamboCB->l2.h2.status |= OSS_RAMBO_AUTOF_STOP_NOW ; 
 
#ifdef OSS_DEBUG 
   printf("After setting the stop flag\n"); 
#endif 
   /* Wait until the autoflusher has stopped */ 
   for ( ; ; ) 
   { 
      if ( !( pRambo->pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW ) 
            && !( pRambo->pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STARTED ) ) 
      { 
         break ; 
      } 
#ifdef ENW 
	  /* DNS - added call to yield because of a timing problem when destroy is called by the 
	     flushing process. */ 
      ossYield() ; 
#endif 
#ifdef OSS_DEBUG 
      printf("Waiting for flags to be reset\n"); 
#endif 
    } 
 
   OSS_ASSERT( !( pRambo->pRamboCB->l2.h2.status 
                     & OSS_RAMBO_AUTOF_STOP_NOW ) ) ; 
   OSS_ASSERT( !( pRambo->pRamboCB->l2.h2.status 
                     & OSS_RAMBO_AUTOF_STARTED ) ) ; 
   OSS_ASSERT( OSS_OK == osserr ) ; 
 
exit: 
   return osserr ; 
 
errorInvalid : 
   osserr = OSS_ERR_INVALID ; 
   goto exit ; 
 
errorNotInitialized : 
   osserr = OSS_ERR_RAMBO_NOT_INIT ; 
   goto exit ; 
} 
 
 
/******************************************************************************* 
 
   Name 
      ossRamboFlush 
 
   Function 
 
   Inputs 
      1. pRambo 
         Address of a RAMBO handle. 
      2. pszFilename 
         Address of a null-terminated string containing a fully qualified 
         filename. 
 
   Normal Returns 
      OSS_OK 
 
   Error Returns 
      OSS_ERR_INVALID 
      OSS_ERR_RAMBO_NOT_INIT 
      OSS_ERR_RAMBO_AUTOFLUSHER_ENABLED 
      OSS_ERR_RAMBO_IO_ERROR 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboFlush( 
      OSSRambo   * pRambo, 
      const char * pszFilename ) 
{ 
   OSSErr osserr = OSS_OK ; 
   int fd = - 1 ; 
   ssize_t bytesWritten = 0 ; 
   Nuint i = 0 ; 
   Nuint slotIndex = 0 ; 
   Nuint chunkIndex = 0 ; 
   Nuint slotInChunk = 0 ; 
   OSSRamboChunkCB * pChunkCB = NULL ; 
   OSSRamboCB * pRamboCB = NULL ; 
   void * slotAddress = NULL ; 
 
   /* Validate the parameters */ 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo, sizeof( *pRambo ) ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) ) 
   { 
      goto errorInvalid ; 
   } 
 
   pRamboCB = pRambo->pRamboCB ; 
 
   OSS_ASSERT( ossRamboIsInitialized( pRamboCB ) ) ; 
   if ( !ossRamboIsInitialized( pRamboCB ) ) 
   { 
      goto errorNotInitialized ; 
   } 
 
   OSS_ASSERT( NULL != pszFilename ) ; 
   OSS_ASSERT( strlen( pszFilename ) > 0 ) ; 
   if ( ( NULL == pszFilename ) || ( 0 == strlen( pszFilename ) ) ) 
   { 
      goto errorInvalid ; 
   } 
 
   /* If the auto flusher is on, don't allow manual flushing */ 
   if ( ossRamboIsAutoFlusherStarted( pRamboCB ) ) 
   { 
      osserr = OSS_ERR_RAMBO_AUTOFLUSHER_ENABLED ; 
      goto exit ; 
   } 
 
   /* TODO */ 
   /* 
    * There is a timing hole here.  The flushing status may have been 
    * cleared and then set by another process before we get to set it. 
    */ 
   while ( ossRamboIsFlushing( pRamboCB ) ) 
   { 
      ossYield() ; 
   } 
   pRamboCB->l2.h2.status |= OSS_RAMBO_FLUSHING ; 
   /* TODO */ 
 
   /* Open the file */ 
#ifdef ENW 

	#ifdef _WIN32
	   fd = open( pszFilename, 
	                O_WRONLY | O_CREAT | O_TRUNC ) ; 

	#else
	   fd = open( pszFilename, 
	                O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU ) ; 
	#endif

 
#else 
   fd = open64( pszFilename, 
                O_WRONLY | O_CREAT | O_TRUNC, 
                S_IRUSR | S_IWUSR ) ; 
#endif 
   if ( -1 == fd ) 
   { 
      goto errorOpenFailed ; 
   } 
 
   /* Examine the first chunk control block to see if we wrapped. */ 
   if ( ossAtomicPeek(&(pFirstChunkCB(pRamboCB)->fillCount)) > OSS_RAMBO_SLOTS_PER_CHUNK ) 
   { 
      pOnDiskCB(pRamboCB)->bufferWrapped = 1 ;   /* True */ 
   } 
 
   /* Write RAMBO storage header information */ 
   OSS_ASSERT( ossRamboIsValidOnDiskCB( pOnDiskCB(pRamboCB) ) ) ; 
   bytesWritten = write( fd, pOnDiskCB(pRamboCB), 
                         sizeof( OSSRamboOnDiskCB) ) ; 
   if ( (ssize_t)-1 == bytesWritten ) 
   { 
      goto errorWriteFailed ; 
   } 
 
   /* Write the application-specific data area if necessary */ 
   if ( pOnDiskCB(pRamboCB)->appOnDiskDataSize > 0 ) 
   { 
      bytesWritten = write( fd, pAppOnDiskData(pRamboCB), 
                            pOnDiskCB(pRamboCB)->appOnDiskDataSize ) ; 
      if ( (ssize_t)-1 == bytesWritten ) 
      { 
         goto errorWriteFailed ; 
      } 
   } 
 
   /* 
    * Start flushing at the slot identified by nextSlotIndex.  Why? 
    * If the buffer has wrapped, nextSlotIndex will be a slot in the 
    * least recently used chunk.  If the buffer has not wrapped, we will 
    * in the chunk that contains the nextSlotIndex and advance one chunk 
    * at a time until we find the first filled slot. 
    */ 
   slotIndex   = ossAtomicPeek(&pRamboCB->l1.h1.nextSlotIndex) % pRamboCB->l2.h2.maxSlots ; 
   chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK ; 
   slotInChunk = slotIndex % OSS_RAMBO_SLOTS_PER_CHUNK ; 
   OSS_ASSERT( slotIndex < pRamboCB->l2.h2.maxSlots ) ; 
   OSS_ASSERT( chunkIndex < pRamboCB->l2.h2.maxChunks ) ; 
   OSS_ASSERT( slotInChunk < OSS_RAMBO_SLOTS_PER_CHUNK ) ; 
 
   /* Address of the current chunk control block. */ 
   pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ; 
   OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ; 
 
   /* Loop over all the slots only once */ 
   for ( i = 0; i < pRamboCB->l2.h2.maxSlots; i++ ) 
   { 
      OSS_ASSERT( slotInChunk < OSS_RAMBO_SLOTS_PER_CHUNK ) ; 
      OSS_ASSERT( chunkIndex < pRamboCB->l2.h2.maxChunks ) ; 
 
      if ( ossRamboIsSlotFull( pChunkCB, slotInChunk ) ) 
      { 
         /* If a slot is full, write the slot to disk */ 
         slotAddress = (void *)( (uintptr_t)pMemBuffer(pRamboCB) 
                     + ( chunkIndex  * OSS_RAMBO_CHUNK_SIZE ) 
                     + ( slotInChunk * OSS_RAMBO_SLOT_SIZE ) ) ; 
         OSS_ASSERT( ossRamboIsValidSlot( slotAddress, pRamboCB ) ) ; 
 
         bytesWritten = write( fd, slotAddress, OSS_RAMBO_SLOT_SIZE ) ; 
         if ( (ssize_t)-1 == bytesWritten ) 
         { 
            goto errorWriteFailed ; 
         } 
      } 
 
      /* Move to the next slot in this chunk */ 
      slotInChunk = (slotInChunk + 1) % OSS_RAMBO_SLOTS_PER_CHUNK ; 
      if ( 0 == slotInChunk  ) 
      { 
         /* Move to next chunk */ 
         chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ; 
         pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ; 
         OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ; 
      } 
   } 
 
   /* Flushing is complete */ 
   OSS_VERIFY( 0 == close( fd ) ) ; 
 
   /* Clear the flushing status */ 
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ; 
 
   OSS_ASSERT( OSS_OK == osserr ) ; 
 
exit : 
   OSS_ASSERT( !( pRamboCB->l2.h2.status & OSS_RAMBO_FLUSHING ) ) ; 
   return osserr ; 
 
errorInvalid : 
   osserr = OSS_ERR_INVALID ; 
   goto exit ; 
 
errorNotInitialized : 
   osserr = OSS_ERR_RAMBO_NOT_INIT ; 
   goto exit ; 
 
errorOpenFailed : 
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ; 
   osserr = OSS_ERR_RAMBO_IO_ERROR ; 
   goto exit ; 
 
errorWriteFailed : 
   OSS_VERIFY( 0 == close( fd ) ) ; 
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ; 
   osserr = OSS_ERR_RAMBO_IO_ERROR ; 
   goto exit ; 
} 
 
/******************************************************************************* 
 
   Name 
      ossRamboSlotCopy 
 
 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboSlotCopy(     /* 194539 - changed to return OSSErr */ 
      OSSRambo   * pRambo, 
      const void * pBuffer, 
      Uint32       size ) 
{ 
   OSSRegister slotIndex ;       /* 186134 */ 
   Nuint chunkIndex ; 
   Nuint slotInChunk ; 
   void * slotAddress ; 
   OSSRamboCB * pRamboCB ; 
   OSSRamboChunkCB * pChunkCB ; 
   Uint32 newSize = size + OSS_RAMBO_DATA_HEADER_SIZE;   /* 198075 - new size of data to write to buffer */ 
 
/* 194539 begin */ 
   if (pRambo == NULL || pBuffer == NULL) 
      return OSS_ERR_INVALID; 
/* 194539 end */ 
 
/* 198075 begin */ 
/* Don't do anything if the size is zero */ 
   if (size == 0) 
      return OSS_OK; 
 
/* If the buffer size plus the size of the data header  is larger than a slot 
   Then call the mult-slot copy function 
*/ 
   if (newSize > OSS_RAMBO_SLOT_SIZE) 
      return(ossRamboMultiSlotCopy(pRambo, pBuffer, size)); 
/* 198075 end */ 
 
   /* 216026  Increment writer count */ 
   ossAtomicIncAndRet( &pRambo->writers ); 
 
   /* Validate the parameters */ 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo, sizeof( *pRambo ) ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( ossRamboIsInitialized( pRambo->pRamboCB ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr( pBuffer, newSize ) ) ; 
   OSS_ASSERT( newSize <= OSS_RAMBO_SLOT_SIZE ) ; 
 
   /* Atomically reserve our slot if we are not flushing */ 
   pRamboCB = pRambo->pRamboCB ; 
   while ( ossRamboIsFlushing( pRamboCB ) ) 
   { 
      ossYield() ; 
   } 
   slotIndex = ossAtomicIncAndRet( &pRamboCB->l1.h1.nextSlotIndex ) ; 
 
   /* 
    * Determine our chunk index (range: 0 to maxChunks-1) and the 
    * chunk relative slot index (range: 0 to OSS_RAMBO_SLOTS_PER_CHUNK-1). 
    */ 
   slotIndex   = slotIndex % pRamboCB->l2.h2.maxSlots ; 
   chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK ; 
   slotInChunk = slotIndex % OSS_RAMBO_SLOTS_PER_CHUNK ; 
   OSS_ASSERT( slotIndex < pRamboCB->l2.h2.maxSlots ) ; 
   OSS_ASSERT( chunkIndex < pRamboCB->l2.h2.maxChunks ) ; 
   OSS_ASSERT( slotInChunk < OSS_RAMBO_SLOTS_PER_CHUNK ) ; 
 
   /* Determine the memory address of the slot and chunk control block */ 
   pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ; 
   OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ; 
 
   slotAddress = (void *)( (uintptr_t)pMemBuffer(pRamboCB) 
               + ( slotIndex  * OSS_RAMBO_SLOT_SIZE ) ) ; 
   OSS_ASSERT( ossRamboIsValidSlot( slotAddress, pRamboCB ) ) ; 
 
   /* 
    * Before copying the buffer into the memory slot make sure that the 
    * auto-flusher has flushed the chunk that this slot belongs to. 
    * Otherwise, we will clobber the previous contents of this slot. 
    * If the auto-flusher isn't enabled, then ignore this step. 
    */ 
   while ( ossRamboIsAutoFlusherStarted( pRamboCB ) 
            && ( ossAtomicPeek(&pChunkCB->fillCount) == OSS_RAMBO_SLOTS_PER_CHUNK ) ) 
   { 
      ossYield() ; 
/* 216026 begin */ 
      /* Check whether the flusher is stopping and if so then don't bother 
         to write the data to the shared memory buffer */ 
      if (ossRamboIsAutoFlusherStopping( pRamboCB )) { 
         /* Decrement writer count */ 
         ossAtomicDecAndRet( &pRambo->writers ); 
         return OSS_OK; 
      } 
/* 216026 end */ 
   } 
 
   /* Check that we don't copy beyond our buffer */ 
   OSS_ASSERT( ((uintptr_t)slotAddress + newSize ) 
                  <= ((uintptr_t)pMemBuffer(pRamboCB) 
                        + pOnDiskCB(pRamboCB)->bufferSize ) ) ; 
 
/* 198075 begin */ 
   /* Write the data length to the buffer */ 
   *(Uint32 *)slotAddress = size; 
   /* increment slotAddress past the length */ 
   slotAddress = (void *)((Uint32 *)slotAddress + 1); 
/* 198075 end */ 
 
   /* Copy the buffer into the memory slot and mark the slot filled */ 
   memcpy( slotAddress, pBuffer, size ) ; 
   ossRamboMarkSlotsFilled( pChunkCB, slotInChunk, 1 ) ; 
#ifdef OSS_DEBUG 
   printf("wrote to chunk %d slot %d\n", chunkIndex, slotIndex); 
#endif 
   /* 216026  Decrement writer count */ 
   ossAtomicDecAndRet( &pRambo->writers ); 
 
   return OSS_OK;    /* 194539 */ 
} 
 
 
/******************************************************************************* 
 
   Name 
      ossRamboMultiSlotCopy 

   Function 
 
   Inputs 
      1. pRambo 
         Address of a RAMBO handle. 
      2. pBuffer 
         Address of data bytes to copy into the RAMBO buffer 
      3. size 
	     Number of data bytes to copy into the RAMBO buffer
 
   Normal Returns 
      OSS_OK 
 
   Error Returns 
      OSS_ERR_INVALID  

 ******************************************************************************/ 
OSS_EXTERNC OSSErr OSS_API ossRamboMultiSlotCopy(   /* 194539 - changed to return OSSErr */ 
      OSSRambo   * pRambo, 
      const void * pBuffer, 
      Uint32       size ) 
{ 
   Nuint numSlotsToFill ; 
   OSSRegister slotIndex ;       /* 186134 */ 
   Nuint chunkIndex ; 
   Nuint slotInChunk ; 
   Nuint numEmptySlots ; 
   Nuint numSlotsFilled ; 
   Nuint bytesCopied ; 
   Nuint bytesOfBufferCopied ;  /* 198075 - bytes of the actual data buffer copied to shm */ 
   void * slotAddress ; 
   char * pb ; 
   OSSRamboCB * pRamboCB ; 
   OSSRamboChunkCB * pChunkCB ; 
   Uint32 newSize = size + OSS_RAMBO_DATA_HEADER_SIZE;   /* 198075 - new size of data to write to buffer */ 
   int loopIndex = 1;   /* 198075 - number of times through loop writing to buffer */ 
 
/* 194539 begin */ 
   if (pRambo == NULL || pBuffer == NULL) 
      return OSS_ERR_INVALID; 
/* 194539 end */ 
 
/* 198075 begin */ 
/* Don't do anything if the size is zero */ 
   if (size == 0) 
      return OSS_OK; 
/* 198075 end */ 
 
   /* 216026  Increment writer count */ 
   ossAtomicIncAndRet( &pRambo->writers ); 
 
   /* Validate the parameters */ 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo, sizeof( *pRambo ) ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr ( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( !ossIsBadWritePtr( pRambo->pRamboCB, sizeof( *pRambo->pRamboCB ) ) ) ; 
   OSS_ASSERT( ossRamboIsInitialized( pRambo->pRamboCB ) ) ; 
   OSS_ASSERT( !ossIsBadReadPtr( pBuffer, newSize ) ) ; 
   OSS_ASSERT( size > 0 ) ; 
   OSS_ASSERT( newSize <= OSS_RAMBO_BUFFER_SIZE_MAX ) ; 
 
   pRamboCB = pRambo->pRamboCB ; 
 
   /* Determine the number of slots and chunks that we will require. */ 
   OSS_ASSERT( OSS_RAMBO_ROUND_SLOT_SIZE( newSize ) >= newSize ) ; 
 
   numSlotsToFill = OSS_RAMBO_ROUND_SLOT_SIZE( newSize ) / OSS_RAMBO_SLOT_SIZE ; 
 
   OSS_ASSERT( numSlotsToFill > 0 ) ; 
   OSS_ASSERT( numSlotsToFill <= pRamboCB->l2.h2.maxSlots ) ; 
 
   /* Atomically reserve our slot(s) if we are not flushing */ 
   while ( ossRamboIsFlushing( pRamboCB ) ) 
   { 
      ossYield() ; 
   } 
#ifdef OSS_DEBUG 
   printf("ossRamboMultiSlotCopy: number of slots to write %d buffer CB %p\n", numSlotsToFill, pRamboCB); 
#endif 
 
   slotIndex = ossAtomicIncByValAndRet( &pRamboCB->l1.h1.nextSlotIndex, 
                                          numSlotsToFill ) ; 
 
   /* 
    * Determine our chunk index (range: 0 to maxChunks-1) and the 
    * chunk relative slot index (range: 0 to OSS_RAMBO_SLOTS_PER_CHUNK-1). 
    */ 
   slotIndex   = slotIndex % pRamboCB->l2.h2.maxSlots ; 
   chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK ; 
   slotInChunk = slotIndex % OSS_RAMBO_SLOTS_PER_CHUNK ; 
   OSS_ASSERT( slotIndex < pRamboCB->l2.h2.maxSlots ) ; 
   OSS_ASSERT( chunkIndex < pRamboCB->l2.h2.maxChunks ) ; 
   OSS_ASSERT( slotInChunk < OSS_RAMBO_SLOTS_PER_CHUNK ) ; 
 
   /* How many empty slots are within the current chunk. */ 
   numEmptySlots = OSS_RAMBO_SLOTS_PER_CHUNK - slotInChunk ; 
   OSS_ASSERT( numEmptySlots > 0 ) ; 
   OSS_ASSERT( numEmptySlots <= OSS_RAMBO_SLOTS_PER_CHUNK ) ; 
#ifdef OSS_DEBUG 
   printf("ossRamboMultiSlotCopy: number of empty slots %d  slotIndex=%d  chunckIndex=%d  slotInChunk=%d\n", 
	       numEmptySlots, slotIndex, chunkIndex, slotInChunk); 
   printf("ossRamboMultiSlotCopy: mem buffer=%p\n", 
	       pMemBuffer(pRamboCB)); 
   printf("ossRamboMultiSlotCopy: OnDiskCB=%p\n", 
	       pOnDiskCB(pRamboCB)); 
   printf("ossRamboMultiSlotCopy: FirstChunkCB=%p\n", 
	       pFirstChunkCB(pRamboCB)); 
   printf("ossRamboMultiSlotCopy: chunkCBsize=%d\n", 
	       pRamboCB->l2.h2.chunkCBSize); 
#endif 
 
   /* Address of the first empty slot in this chunk to copy into. */ 
   slotAddress = (void *)( (uintptr_t)pMemBuffer(pRamboCB) 
               + ( slotIndex  * OSS_RAMBO_SLOT_SIZE ) ) ; 
   OSS_ASSERT( ossRamboIsValidSlot( slotAddress, pRamboCB ) ) ; 
 
   /* Copy the buffer into the RAMBO buffer on a per-chunk basis. */ 
   pb = (char *)pBuffer ; 
   for ( ; ; ) 
   { 
#ifdef OSS_DEBUG 
   printf("ossRamboMultiSlotCopy: number of empty slots %d  slotAddress=%p\n", numEmptySlots, slotAddress); 
#endif 
      /* 
       * Determine how many bytes to copy and how many slots to mark 
       * filled within the current chunk. 
       */ 
      if ( numEmptySlots < numSlotsToFill ) 
      { 
         numSlotsFilled = numEmptySlots ; 
         bytesCopied    = numSlotsFilled * OSS_RAMBO_SLOT_SIZE ; 
      } 
      else 
      { 
         numSlotsFilled = numSlotsToFill ; 
         bytesCopied    = newSize ; 
      } 
      OSS_ASSERT( bytesCopied > 0 ) ; 
      OSS_ASSERT( bytesCopied <= newSize ) ; 
 
      /* Address of the current chunk control block. */ 
      pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ; 
      OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ; 
 
      /* 
       * Before copying the buffer into the memory slot make sure that the 
       * auto-flusher has flushed the chunk that this slot belongs to. 
       * Otherwise, we will clobber the previous contents of this slot. 
       * If the auto-flusher isn't enabled, then ignore this step. 
       */ 
      while ( ossRamboIsAutoFlusherStarted( pRamboCB ) && 
              ossAtomicPeek(&pChunkCB->fillCount) > 0  && 
              (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0  ) 
/* DNS 194539 && ( pChunkCB->fillCount == OSS_RAMBO_SLOTS_PER_CHUNK ) )  replaced with above two lines 
              because the buffer could have wrapped before flushing is started so 
              it could be true that pChunkCB->fillCount > OSS_RAMBO_SLOTS_PER_CHUNK */ 
      { 
#ifdef OSS_DEBUG 
   printf("ossRamboMultiSlotCopy: waiting because fillCount=%d\n", ossAtomicPeek(&pChunkCB->fillCount)); 
#endif 
 
         ossYield() ; 
 
/* 216026 begin */ 
         /* Check whether the flusher is stopping and if so then don't bother 
            to write the data to the shared memory buffer */ 
         if (ossRamboIsAutoFlusherStopping( pRamboCB )) { 
            /* Decrement writer count */ 
            ossAtomicDecAndRet( &pRambo->writers ); 
            return OSS_OK; 
         } 
/* 216026 end */ 
      } 
 
      /* Check that we don't copy beyond our buffer */ 
      OSS_ASSERT( ((uintptr_t)slotAddress + bytesCopied ) 
                     <= ((uintptr_t)pMemBuffer(pRamboCB) 
                           + pOnDiskCB(pRamboCB)->bufferSize ) ) ; 
/* 198075 begin */ 
      /* If this is the first time through the loop Then */ 
      if (loopIndex == 1) { 
         /* Write the data length to the buffer */ 
         *(Uint32 *)slotAddress = size; 
         /* increment slotAddress past the length */ 
         slotAddress = (void *)((Uint32 *)slotAddress + 1); 
         bytesOfBufferCopied = bytesCopied - OSS_RAMBO_DATA_HEADER_SIZE;  /* 198075 */ 
      } 
      else 
         bytesOfBufferCopied = bytesCopied;  /* 198075 */ 
/* 198075 end */ 
 
      /* Copy the buffer into the memory slots and mark the slots filled */ 
      memcpy( slotAddress, pb, bytesOfBufferCopied ) ; 
#ifdef OSS_DEBUG 
   printf("ossRamboMultiSlotCopy: copied data - now mark %d slots filled\n", numSlotsFilled); 
#endif 
 
      ossRamboMarkSlotsFilled( pChunkCB, slotInChunk, numSlotsFilled ) ; 
#ifdef OSS_DEBUG 
      printf("wrotem chunk %d slot %d\n", chunkIndex, slotIndex); 
#endif 
 
      /* Check if we are done */ 
      if ( bytesCopied == newSize ) { 
#if 0 
/* 198075 - this is not necessary now because we don't search byte by byte throught the data */ 
	      Nuint rem = newSize % OSS_RAMBO_SLOT_SIZE; 
      	if (rem > 0) { 
/*            printf("wrotem chunk %d slot %d - zeroing %d bytes at %p \n", chunkIndex, slotIndex, 
				           OSS_RAMBO_SLOT_SIZE - rem, (void *)((uintptr_t)slotAddress + (uintptr_t)bytesCopied)); 
*/ 
            memset((void *)((uintptr_t)slotAddress + bytesCopied), 0x00, OSS_RAMBO_SLOT_SIZE - rem); 
         } 
#endif 
         /* we have written all the data so quit */ 
         break ; 
      } 
 
      newSize -= bytesCopied ; 
      numSlotsToFill -= numSlotsFilled ; 
      pb += bytesOfBufferCopied ; 
 
      /* Move to the next chunk */ 
      chunkIndex = ( chunkIndex + 1 ) % pRamboCB->l2.h2.maxChunks ; 
      numEmptySlots = OSS_RAMBO_SLOTS_PER_CHUNK ; 
 
      /* Address of the first slot to in this chunk to copy into. */ 
      slotInChunk = 0 ; 
      slotAddress = (void *)( (uintptr_t)pMemBuffer(pRamboCB) 
                  + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) 
                  + ( slotInChunk * OSS_RAMBO_SLOT_SIZE ) ) ; 
      OSS_ASSERT( ossRamboIsValidSlot( slotAddress, pRamboCB ) ) ; 
      loopIndex++;  /* 198075 */ 
   } 
 
   /* 216026  Decrement writer count */ 
   ossAtomicDecAndRet( &pRambo->writers ); 
 
   return OSS_OK;    /* 194539 */ 
} 
 
