/**********************************************************************
 * 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: ossramboflush.cpp,v 1.8 2009/05/20 03:35:11 kchan Exp $
 *
 * Contributors:
 * IBM - Initial API and implementation
 **********************************************************************/

/*******************************************************************************

   Source File Name = ossramboflush.cpp

   Descriptive Name = OSSe RApid Memory Buffer Out (RAMBO) flushing routines

   Function:
      Defines:    The RAMBO flushing routines used by the RAC.  This file
                  is used on all platforms except OS/390 and is the same as
                  ossramboflush.c

   Dependencies:
      NOTE:       THIS FILE MUST BE KEPT IN SYNCH WITH ossramboflush.c !!!!

   Restrictions:
      None

   Change Activity:
   Defect Date        Who Description
   ====== =========== === ==============================================
   186134 Dec/17/2001 dns Created file - copied code from ossrambo.cpp
                          so it could be compiled as C code instead of C++
   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 ossRamboFlushToFD to read the data length from
                          the shared memory and use it when flushing
                          - added dataProcessor routine for flushing
   194483 Mar/20/2002 dns Changed ossRamboFlushToFunc and
                          ossRamboFlushFullChunksToFuncto to call
                          dataToFuncProcessor which calls a user routine
   205103 May/09/2002 dns for loop index i was being affected by inner
                          for loop - changed inner index variable to j
   205955 May/29/2002 dns added code to check if agent is still alive when
									flushing
   216026 Jul/23/2002 dns Added check for writers before detaching from shm buffer
   230611 Nov/19/2002 dns set numUsedSlots correctly when chunk is full
   232468 Nov/19/2002 dns added code to handle case when continueDataSize > dataLengh
   94473  Jun/10/2005 dns increased the wait before stopping flushing because
                          nobody else is attached to shm
   Last Changed =    05/06/10  12:15:11

*******************************************************************************/

#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
#ifdef SQLUNIX
#include <limits.h>
#include <memory.h>
#include <unistd.h>
#endif

#include <jni.h>

/* MACRO's to simplify the JNI code. */
#if defined __cplusplus
	#define ENV(e) e
	#define ENVPARM(e)
	#define ENVPARM1(e)
#else
	#define ENV(e) (*e)
	#define ENVPARM(e) e,
	#define ENVPARM1(e) e
#endif

#include "oss.h"
#include "ossdebug.h"
#include "osserror.h"
#include "ossipcmemory.h"
#include "RASharedMemory.h"
#include "ossramboinl.h"
#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)



/* 194483 begin */

/*******************************************************************************

   Function Name
      dataToFuncProcessor

   Function
      Processes data from a RAMBO buffer and passes it to a user function and
      a file descriptor if specified.

   Inputs
      pTraceData - addr of trace data to process
      dataLength - length of data to process
      continueDataSize - size of data at the start of the buffer to send without processing
      pcontinueFirstPartSize - ptr to size of first part of a record that was included in the previous buffer
      pcontinueBuffer - ptr to temporary buffer to hold data record that began in the previous buffer
      fileDescriptor - descriptor of file or socket to write data record to
      descriptorType - type of descriptor passed
      pUserDataProcessor - ptr to user data processing routine
      pArgs - addr of argument structure to pass to user data processing routine

   Normal Return
      returns size of last part of the data record that did not fit in this buffer

   Error Return
      -1
**********************************************************************************/

int dataToFuncProcessor(
      void *pTraceData,
      int dataLength,
      int continueDataSize,         /* size of end of msg from previous buffer to send */
      int* pcontinueFirstPartSize,  /* ptr to size of data in continue buffer */
	   void **pcontinueBuffer,       /* ptr to a continue buffer */
      int fileDescriptor,           /* file descriptor to flush data to */
	   int descriptorType,           /* type of descriptor */
      RA_Data_Processing_Routine pUserDataProcessor,  /* ptr to user data processing routine */
	   void *pArgs ) {               /* ptr to argument struct to pass to user data processing routine */

   char *pFlushData;
   Uint32 recordLength;
   int offset = 0;
   int endOffset = dataLength - OSS_RAMBO_DATA_HEADER_SIZE;
   int flushLen;     /* length of data to flush to file descriptor */
   int bFlushLen;    /* length of data to flush from buffer (includes rambo data header length) */
   ssize_t bytesWritten = 0;

#ifdef OSS_DEBUG
   ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  pTraceData=%p *pTraceData=%x/%x/%x/%x/%x  contSize=%d\n",
			                    pTraceData, (*(char *)pTraceData & 0x000000ff),(*((char *)pTraceData+1) & 0x000000ff),
								(*((char *)pTraceData+2) & 0x000000ff), (*((char *)pTraceData+3) & 0x000000ff),
                        (*((char *)pTraceData+4) & 0x000000ff), continueDataSize);
#endif

/* If there was a record that started in the previous buffer but the buffer was not big
   enough to contain the whole record, then the rest of the record is at the beginning
   of this buffer so flush the remainder of the record.
*/

   if (continueDataSize > 0) {
      /* The remainder of the record may be bigger than the size of this buffer */
      if (continueDataSize > dataLength) {
         /* Append the data in this buffer to the partial record from the previous buffer */
         memcpy((char *)*pcontinueBuffer + *pcontinueFirstPartSize, (char *)pTraceData, dataLength);

         /* Adjust the length values */
         *pcontinueFirstPartSize += dataLength;
         continueDataSize -= dataLength;
         offset += dataLength;
      }
      else {
         int totalRecordSize = *pcontinueFirstPartSize + continueDataSize;

         /* Append remainder of record to partial record from previous buffer */
         memcpy((char *)*pcontinueBuffer + *pcontinueFirstPartSize, (char *)pTraceData, continueDataSize);

         /* if a descriptor was passed then write data record to associated socket or file */
         if ( fileDescriptor > 0 ) {
            if ( descriptorType == OSS_RAMBO_DESCRIPTOR_SOCKET ) {
               /* Write data to socket */
#ifdef __OS400__
               bytesWritten = send( fileDescriptor, (char *)*pcontinueBuffer, totalRecordSize, 0 ) ;
#else
               bytesWritten = send( fileDescriptor, (const char *)*pcontinueBuffer, totalRecordSize, 0 ) ;
#endif
               if ( SOCKET_ERROR == bytesWritten )
               {
#ifdef OSS_DEBUG
#ifdef _WIN32
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. wsa error=%d\n",
                        fileDescriptor, WSAGetLastError() ) ;
#else
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
#endif
                  return -1 ;
               }
               /* If not all the data was sent then send the rest */
               else if (bytesWritten < totalRecordSize) {
                  char *dataptr = (char *)*pcontinueBuffer + bytesWritten;
                  int datalen = totalRecordSize - bytesWritten;
                  while (datalen > 0) {
                     bytesWritten = send( fileDescriptor, dataptr, datalen, 0 ) ;
                     if ( SOCKET_ERROR == bytesWritten )
                     {
#ifdef OSS_DEBUG
#ifdef _WIN32
                        ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. wsa error=%d\n",
					            dataptr, datalen, WSAGetLastError() ) ;
#else
                        ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. errno=%d\n",
					            dataptr, datalen, errno ) ;
#endif
#endif
                        return OSS_ERR_RAMBO_IO_ERROR ;
                     }
                     dataptr += bytesWritten;
                     datalen -= bytesWritten;
                  }
               }
            }
            /* Write data to file */
            else {
               bytesWritten = write( fileDescriptor, (char *)*pcontinueBuffer, totalRecordSize ) ;
               if ( (ssize_t)-1 == bytesWritten )
               {
#ifdef OSS_DEBUG
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to file %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
                  return -1 ;
               }
            }
         }

		/* if a user data processing routine was passed then pass the data record to it */
		if ( pUserDataProcessor != NULL ) {
			int err;
			err = pUserDataProcessor(*pcontinueBuffer, totalRecordSize, pArgs);

			if (err) {
				if(*pcontinueBuffer != NULL) {
					free(*pcontinueBuffer);
					*pcontinueBuffer = NULL;
				}
				return -1 ;
			}
		}

		if(*pcontinueBuffer != NULL) {
			free(*pcontinueBuffer);
			*pcontinueBuffer = NULL;
		}
         offset += OSS_RAMBO_ROUND_SLOT_SIZE( continueDataSize );
         continueDataSize = 0;
      }
   }

   /* Process the data until we have reached the end of the buffer */
   while (offset < endOffset) {

#ifdef OSS_DEBUG
      ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor: *pTraceData at offset %d =%x/%x/%x/%x %x/%x/%x/%x/%x\n",
			                    offset, (*((char *)pTraceData+offset) & 0x000000ff),(*((char *)pTraceData+offset+1) & 0x000000ff),
								(*((char *)pTraceData+offset+2) & 0x000000ff), (*((char *)pTraceData+offset+3) & 0x000000ff),
                        (*((char *)pTraceData+offset+4) & 0x000000ff), (*((char *)pTraceData+offset+5) & 0x000000ff),
                        (*((char *)pTraceData+offset+6) & 0x000000ff), (*((char *)pTraceData+offset+7) & 0x000000ff),
                        (*((char *)pTraceData+offset+8) & 0x000000ff));
#endif

      pFlushData = (char *)pTraceData + offset;
      flushLen = 0;
      bFlushLen = 0;
      /* read the data length from the buffer */
      recordLength = *(Uint32 *)pFlushData;

      /* If the whole record doesn't fit in this buffer then allocate a temporary
         buffer for the record and copy the part that is included in this buffer
         into the temp bufer*/
      if (offset + OSS_RAMBO_DATA_HEADER_SIZE + recordLength > dataLength) {
		
         continueDataSize = offset + OSS_RAMBO_DATA_HEADER_SIZE + recordLength - dataLength;
		   bFlushLen = dataLength - offset;
         *pcontinueFirstPartSize = bFlushLen - OSS_RAMBO_DATA_HEADER_SIZE;
         *pcontinueBuffer = (void *)malloc(recordLength);
         /* Return if malloc failed */
         if (*pcontinueBuffer == NULL) {
            return -1;
         }
         memcpy(*pcontinueBuffer, pFlushData + OSS_RAMBO_DATA_HEADER_SIZE, *pcontinueFirstPartSize);

#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  Reached end of buffer recordLength=%d bflushLen=%d  pFlushData=%p\n",
			                    recordLength, bFlushLen, pFlushData);
#endif
		}
      /* Else the whole record is contained in this buffer so flush it */
      else {
         bFlushLen = OSS_RAMBO_DATA_HEADER_SIZE + recordLength;

         flushLen = bFlushLen - OSS_RAMBO_DATA_HEADER_SIZE;
         pFlushData += OSS_RAMBO_DATA_HEADER_SIZE;

#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  recordLength=%d flushLen=%d bFlushLen=%d pFlushData=%p\n",
			                    recordLength, flushLen, bFlushLen, pFlushData);
#endif

         /* if a descriptor was passed then write data record to associated socket or file */
         if ( fileDescriptor > 0 ) {

            /* Write the data out to the file or socket */

            if ( descriptorType == OSS_RAMBO_DESCRIPTOR_SOCKET ) {
		         /* Write data to socket */
#ifdef __OS400__
               bytesWritten = send( fileDescriptor, (char *)pFlushData, flushLen, 0 ) ;
#else
               bytesWritten = send( fileDescriptor, (const char *)pFlushData, flushLen, 0 ) ;
#endif
               if ( SOCKET_ERROR == bytesWritten )
		         {
#ifdef OSS_DEBUG
#ifdef _WIN32
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. wsa error=%d\n",
                        fileDescriptor, WSAGetLastError() ) ;
#else
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
#endif
                  return -1 ;
               }
               /* If not all the data was sent then send the rest */
               else if (bytesWritten < flushLen) {
	               char *dataptr = pFlushData + bytesWritten;
	               int datalen = flushLen - bytesWritten;
	               while (datalen > 0) {
                     bytesWritten = send( fileDescriptor, dataptr, datalen, 0 ) ;
                     if ( SOCKET_ERROR == bytesWritten )
		   	         {
#ifdef OSS_DEBUG
#ifdef _WIN32
                         ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. wsa error=%d\n",
					                               dataptr, datalen, WSAGetLastError() ) ;
#else
                         ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. errno=%d\n",
					                               dataptr, datalen, errno ) ;
#endif
#endif
                         return -1 ;
			            }
		               dataptr += bytesWritten;
		               datalen -= bytesWritten;
		   	      }
               }
            }
		      /* Write data to file */
		      else {
               bytesWritten = write( fileDescriptor, pFlushData, flushLen ) ;
               if ( (ssize_t)-1 == bytesWritten )
		         {
#ifdef OSS_DEBUG
                   ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to file %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
                   return -1 ;
			      }
		      }

         }

          /* if a user data processing routine was passed then pass the data record to it */
         if ( pUserDataProcessor != NULL ) {
            int err;
            err = pUserDataProcessor(pFlushData, flushLen, pArgs);

			if (err) {
				if(*pcontinueBuffer != NULL) {
					free(*pcontinueBuffer);
					*pcontinueBuffer = NULL;
				}
				return -1 ;
			}
         }
      }
      /* increase the offset by the number of slots that were flushed */
      offset += OSS_RAMBO_ROUND_SLOT_SIZE( bFlushLen );
   }
   return continueDataSize;
}
/* 194483 end */
/* 172892   renamed this function because it is replaced by the following
            version of it */
/*******************************************************************************

   Function Name
      ossRamboFlushFullChunksToFunc

   Function
      Flushes the RAMBO buffer to a processing function.

   Inputs
      1. pRambo
         Address of a RAMBO buffer control block.
	  2. pDataProcessor
		   pointer to function that processes the flushed data from the RAMBO buffer
            - function takes two paramters a) the address of a chunk of the buffer
                 b) the address of a function specific argument structure
	  3. pArgs
	      pointer to arguments to pass to the data processing function

   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 ossRamboFlushFullChunksToFunc(
      OSSRambo *pRambo,
      RA_Data_Processing_Routine pDataProcessor,
	  void *pArgs )
{
   OSSErr osserr = OSS_OK ;
   ssize_t bytesWritten = 0 ;
   Uint32 chunkIndex = 0 ;
   Uint32 slotIndex = 0 ;
   OSSRamboCB * pRamboCB ;
   OSSRamboChunkCB * pChunkCB = NULL ;
   void * pChunk = NULL ;
   bool wrappedChunk = false ;
   bool flushUnfilledChunks = false ;
   bool stopping = false;
   Sint32 err = 0 ;
   Sint32 continueDataSize = 0;
   int firstPartSize;    /* 194483 */
   void *continueBuffer;         /* 194483 */
#ifdef OSS_DEBUG1
   FILE *dbgfd;
#endif

#ifdef OSS_DEBUG1
   dbgfd = fopen("flushdbg.txt", "ac+");
   if (dbgfd == NULL) {
	   osserr = errno;
	   goto exit;
   }
   if ( 0 > fprintf(dbgfd,"ossRamboFlushFD:  Entered.\n" ) ) {
	   osserr = OSS_ERR_RAMBO_IO_ERROR;
	   goto exit;
   }
   fflush(dbgfd);
#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 ) ) ) ;
#ifdef OSS_DEBUG1
   fprintf(dbgfd,"ossRamboFlushFD:  After asserts.\n" ) ;
   fflush(dbgfd);
#endif

   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) )
   {
      goto errorInvalid ;
   }

   pRamboCB = pRambo->pRamboCB ;

   OSS_ASSERT( ossRamboIsInitialized( pRamboCB ) ) ;

   if ( !ossRamboIsInitialized( pRamboCB ) )
   {
      goto errorNotInitialized ;
   }

   if ( NULL == pDataProcessor )
   {
      goto errorInvalid ;
   }

   /* If the auto flusher is on, don't allow manual flushing */
   if ( ossRamboIsAutoFlusherStarted( pRamboCB ) )
   {
      osserr = OSS_ERR_RAMBO_AUTOFLUSHER_ENABLED ;
      goto exit ;
   }

   /* If already flushing, wait. */
   /* 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 ;
   /* Pretend we are auto flushing instead of manual flushing so we don't hold up writing
      to the buffer */
   pRamboCB->l2.h2.status |= OSS_RAMBO_AUTOF_STARTED ;
   /* TODO */


   /* 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) ) ) ;


   /* TODO */
   /* Start flushing */
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
   fprintf(dbgfd,"ossRamboFlushFD:  Started flushing.\n" ) ;
   fflush(dbgfd);
#endif

   /* Try to find the least recently filled chunk if the buffer has
	  been wrapped and start flushing at it. */

   if ( ossAtomicPeek(&(pFirstChunkCB(pRamboCB)->fillCount)) > OSS_RAMBO_SLOTS_PER_CHUNK )
   {
      slotIndex   = ossAtomicPeek(&pRamboCB->l1.h1.nextSlotIndex) % pRamboCB->l2.h2.maxSlots ;
      chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK + 1;
   }
   else
	  chunkIndex = 0;

   for ( ; ; )
   {
      /* Get pointer to the chunk control block */
      pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
	  OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

      /* Wait until the chunk is full or we have been asked to stop */
      for ( ; ; )
      {
         if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
         {
            stopping = true ;
            break ;
         }

         if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
         {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushFD:  Data found in buffer.\n" ) ;
            fflush(dbgfd);
#endif
            break ;
         }

         ossSleep( 10 ) ;
      }

      /* If we should stop, break out of the flushing loop */
      if ( stopping )
      {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushFD:  We are stopping flushing so flush unfilled chunks.\n" ) ;
            fflush(dbgfd);
#endif
         flushUnfilledChunks = true ;
         break ;
      }

      /* If the chunk is full, flush it to disk */
      if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
      {
         unsigned int i;

         OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[0]) ) ;
         OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE-1]) ) ;

         if ( ossAtomicPeek(&pChunkCB->fillCount) > OSS_RAMBO_SLOTS_PER_CHUNK )
         {
            wrappedChunk = true ;
         }
/*
		 if (!flushStarted)
			 flushStarted = true;
*/
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

         err = dataToFuncProcessor( pChunk, OSS_RAMBO_CHUNK_SIZE, continueDataSize, &firstPartSize, &continueBuffer, 0, 0, pDataProcessor, pArgs ) ;
         if ( 0 > err ) {
//      perror("ossRamboFlushFD failed writing the storage header info");
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
		 else
			 continueDataSize = err;

         if ( wrappedChunk )
         {
            /*
             * The flusher has been lapped.  The chunk has been filled
             * more than once.  Log this so we know it has happened but
             * don't exit.
             */
#ifdef OSS_DEBUG1
//            printf("RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif
            wrappedChunk = false ;
         }
#ifdef OSS_DEBUG1
         else {
//            printf("Flushed chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
         }
#endif

         //pause() ;
         /* Reset the chunk control block */
/* 186134 begin
         memset( (void *)&pChunkCB->slotMap[0], 0x00, sizeof( pChunkCB->slotMap ) ) ;
*/
         for(i=0; i < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; i++) {
            ossAtomicPoke(&(pChunkCB->slotMap[i]), 0);
         }
/* 186134 end */
         ossAtomicPoke(&pChunkCB->fillCount, 0) ;
      }
#ifdef OSS_DEBUG1
      else if (ossAtomicPeek(&pChunkCB->fillCount) > 0) {
//         printf("Unfilled chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
         fprintf(dbgfd,"Unfilled chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
         fflush(dbgfd);
      }
#endif
/*
      else if (flushStarted)
         printf("Empty chunk %lu\n", chunkIndex) ;
*/
      /* Move to next chunk */
      chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

      /* Check if we have been asked to stop */
      if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
      {
         flushUnfilledChunks = true ;
         break ;
      }

   }

   /* Flush unfilled chunks */
	if ( flushUnfilledChunks )
	{
		int numUsedSlots;

		/* Start flushing at the current chunk index so the flushing continues in the correct order */
		unsigned int i;
		for ( i = 0; i < pRamboCB->l2.h2.maxChunks; i++ )
		{
			/* Get pointer to the chunk control block */
			pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
			OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

			/* If the chunk contains data, flush it to disk */
			if ( ossAtomicPeek(&pChunkCB->fillCount) > 0 )
			{
				unsigned int j;
				numUsedSlots = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;
				if (numUsedSlots == 0)
					numUsedSlots = OSS_RAMBO_SLOTS_PER_CHUNK;


            /* Get a pointer to the chunk of memory */
            pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                    + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
            OSS_ASSERT( pChunk != NULL ) ;
            OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

            /* Pass the chunk to the data processor, but only process the filled slots */

            err = dataToFuncProcessor( pChunk, numUsedSlots * OSS_RAMBO_SLOT_SIZE, continueDataSize, &firstPartSize, &continueBuffer, 0, 0, pDataProcessor, pArgs ) ;
            if ( 0 > err ) {
#ifdef OSS_DEBUG1
               fprintf(dbgfd,"ossRamboFlushFD:  failed writing the chunk %d of shared memory. error=%d\n", chunkIndex, err ) ;
               fflush(dbgfd);
#endif
               goto errorProcessingFailed ;
            }
			else
				continueDataSize = err;

#ifdef OSS_DEBUG1
//            printf("Flushed unfilled chunk index %lu has fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"Flushed unfilled chunk index %lu has fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif

            /* Reset the chunk control block */
/* 186134 begin
            memset( (void *)&pChunkCB->slotMap[0], 0x00, sizeof( pChunkCB->slotMap ) ) ;
*/
            for(j=0; j < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; j++) {
               ossAtomicPoke(&(pChunkCB->slotMap[j]), 0);
            }
/* 186134 end */
            ossAtomicPoke(&pChunkCB->fillCount, 0) ;
		 }
#ifdef OSS_DEBUG1
		 else {
//            printf("empty chunk index %lu\n", chunkIndex ) ;
            fprintf(dbgfd,"empty chunk index %lu\n", chunkIndex ) ;
            fflush(dbgfd);
         }
#endif
		 /* Move to next chunk */
		 chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

	  }
      /* Clear the auto-flusher status flags */
      pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
      pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
   }


// pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ;

   OSS_ASSERT( OSS_OK == osserr ) ;

exit :
   #ifdef OSS_DEBUG1
     fclose(dbgfd);
   #endif
   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 ;

errorProcessingFailed :
// pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
   osserr = OSS_ERR_RAMBO_IO_ERROR ;
   goto exit ;
}

/* 172892 begin */
/*******************************************************************************

   Function Name
      ossRamboFlushToFunc

   Function
      Flushes the RAMBO buffer to a processing function on a timed basis.  It
      does not wait for a chunk to be filled before flushing the data in it.

   Inputs
      1. pRambo
         Address of a RAMBO buffer control block.
	  2. pDataProcessor
		   pointer to function that processes the flushed data from the RAMBO buffer
            - function takes two paramters a) the address of a chunk of the buffer
                 b) the address of a function specific argument structure
	  3. pArgs
	      pointer to arguments to pass to the data processing function

   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 ossRamboFlushToFunc(
      OSSRambo *pRambo,
      RA_Data_Processing_Routine pDataProcessor,
	  void *pArgs )
{
   OSSErr osserr = OSS_OK ;
   ssize_t bytesWritten = 0 ;
   Uint32 chunkIndex = 0 ;
   Uint32 slotIndex = 0 ;
   Uint32 loopIndex = 0 ;
   Uint32 timedOut = 0 ;
   Uint32 attachCheckIndex = 0 ;  /* bugzilla 94473 */
   int flushDataLen;
   OSSRamboCB * pRamboCB ;
   OSSRamboChunkCB * pChunkCB = NULL ;
   void * pChunk = NULL ;
   bool wrappedChunk = false ;
   bool flushUnfilledChunks = false ;
   bool stopping = false;
   Sint32 err = 0 ;
   Sint32 continueDataSize = 0;
   int firstPartSize;    /* 194483 */
   void *continueBuffer = NULL;         /* 194483 */
#ifdef OSS_DEBUG1
   FILE *dbgfd;
#endif

#ifdef OSS_DEBUG1
   dbgfd = fopen("flushdbg.txt", "ac+");
   if (dbgfd == NULL) {
	   osserr = errno;
	   goto exit;
   }
   if ( 0 > fprintf(dbgfd,"ossRamboFlushFastToFunc:  Entered.\n" ) ) {
	   osserr = OSS_ERR_RAMBO_IO_ERROR;
	   goto exit;
   }
   fflush(dbgfd);
#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 ) ) ) ;
#ifdef OSS_DEBUG1
   fprintf(dbgfd,"ossRamboFlushFastToFunc:  After asserts.\n" ) ;
   fflush(dbgfd);
#endif

   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) )
   {
      goto errorInvalid ;
   }

   pRamboCB = pRambo->pRamboCB ;

   OSS_ASSERT( ossRamboIsInitialized( pRamboCB ) ) ;

   if ( !ossRamboIsInitialized( pRamboCB ) )
   {
      goto errorNotInitialized ;
   }

   if ( NULL == pDataProcessor )
   {
      goto errorInvalid ;
   }

   /* If the auto flusher is on, don't allow manual flushing */
   if ( ossRamboIsAutoFlusherStarted( pRamboCB ) )
   {
      osserr = OSS_ERR_RAMBO_AUTOFLUSHER_ENABLED ;
      goto exit ;
   }

   /* If already flushing, wait. */
   /* 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 ;
   /* Pretend we are auto flushing instead of manual flushing so we don't hold up writing
      to the buffer */
   pRamboCB->l2.h2.status |= OSS_RAMBO_AUTOF_STARTED ;
   /* TODO */


   /* 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) ) ) ;


   /* TODO */
   /* Start flushing */
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
   fprintf(dbgfd,"ossRamboFlushFastToFunc:  Started flushing.\n" ) ;
   fflush(dbgfd);
#endif

   /* Try to find the least recently filled chunk if the buffer has
	  been wrapped and start flushing at it. */

   if ( ossAtomicPeek(&(pFirstChunkCB(pRamboCB)->fillCount)) > OSS_RAMBO_SLOTS_PER_CHUNK )
   {
      slotIndex   = ossAtomicPeek(&pRamboCB->l1.h1.nextSlotIndex) % pRamboCB->l2.h2.maxSlots ;
      chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK + 1;
   }
   else
	  chunkIndex = 0;

   for ( ; ; )
   {
      /* Get pointer to the chunk control block */
      pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
	   OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

      loopIndex = 0;
      timedOut = 0;
      attachCheckIndex = 0;  /* 94473 */
      /* Wait until the chunk is full or we have been asked to stop or we have waited too long */
      for ( ; ; )
      {
         /* If flushing is to be stopped then set the stopping flag and quit the loop */
         if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
         {
            stopping = true ;
            break ;
         }

         /* If the chunk is full then quit the loop so it can be flushed. */
         if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
         {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushFastToFunc:  Data found in buffer.\n" ) ;
            fflush(dbgfd);
#endif
            break ;
         }
         /* Else If we have looped too long */
         else if (loopIndex == 100) {
            /* If there is any data in the chunk Then quit the loop to flush it */
            if ((ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) - ossAtomicPeek(&pChunkCB->flushedCount) > 0  ||
                (ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0)) {
               timedOut = 1;
#ifdef OSS_DEBUG1
               fprintf(dbgfd,"ossRamboFlushFastToFunc:  We've waited too long, flush data found in chunk.\n" ) ;
               fflush(dbgfd);
#endif
               break ;
            }
/* 205955 begin */
			/* Else get the attach count to see if anyone else is attached */
            else {
               int attcnt = 0;
               OSSErr osserr = OSS_OK ;
               osserr = ossIPCMemAttachCount( pRambo->ipcMemHandle, &attcnt ) ;
               attachCheckIndex++;  /* bugzilla 94473 */
               /* If nobody is attached to the shared memory except for us then stop flushing
                * bugzilla 94473 - only stop flushing if we have gone through the wait loop four
                * times to provide more time for the agent to attach to it (4 seconds instead of 1). */
               if ( osserr == OSS_OK  &&  attcnt == 1 && attachCheckIndex >= 4)
               {
                  /* set the user count to 1 so the shared memory gets deleted when it is destroyed */
                  ossIPCMemSetUserCount( pRambo->ipcMemHandle, attcnt ) ;
                  stopping = true ;
                  break ;
               }
               /* Else continue checking for writes to shared memory */
               loopIndex = 0;
            }
/* 205955 end */
         }

         ossSleep( 10 ) ;
         loopIndex++;
      }

      /* If we should stop, break out of the flushing loop */
      if ( stopping )
      {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushFastToFunc:  We are stopping flushing so flush unfilled chunks.\n" ) ;
            fflush(dbgfd);
#endif
         flushUnfilledChunks = true ;
         break ;
      }

      /* If the chunk is full, flush it to disk */
      if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
      {
         unsigned int i;
         if (ossAtomicPeek(&pChunkCB->flushedCount) == 0) {
            OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[0]) ) ;
         }
         OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE-1]) ) ;

         if ( ossAtomicPeek(&pChunkCB->fillCount) > OSS_RAMBO_SLOTS_PER_CHUNK )
         {
            wrappedChunk = true ;
         }
/*
		 if (!flushStarted)
			 flushStarted = true;
*/
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
         if (ossAtomicPeek(&pChunkCB->flushedCount) > 0) {
            pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            flushDataLen = OSS_RAMBO_CHUNK_SIZE - ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE;
         }
         else
            flushDataLen = OSS_RAMBO_CHUNK_SIZE;

         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

         err = dataToFuncProcessor( pChunk, flushDataLen, continueDataSize, &firstPartSize, &continueBuffer, 0, 0, pDataProcessor, pArgs ) ;
         if ( 0 > err ) {
//      perror("ossRamboFlushFD failed writing the storage header info");
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
         else
            continueDataSize = err;

         if ( wrappedChunk )
         {
            /*
             * The flusher has been lapped.  The chunk has been filled
             * more than once.  Log this so we know it has happened but
             * don't exit.
             */
#ifdef OSS_DEBUG1
//            printf("RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif
            wrappedChunk = false ;
         }
#ifdef OSS_DEBUG1
         else {
//            printf("Flushed chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
         }
#endif

         //pause() ;
         /* Reset the chunk control block */
/* 186134 begin
         memset( (void *)&pChunkCB->slotMap[0], 0x00, sizeof( pChunkCB->slotMap ) ) ;
*/
         for(i=0; i < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; i++) {
            ossAtomicPoke(&(pChunkCB->slotMap[i]), 0);
         }
/* 186134 end */
         ossAtomicPoke(&pChunkCB->fillCount, 0) ;
         ossAtomicPoke(&pChunkCB->flushedCount, 0) ;
      }
      else if ( timedOut ) {
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;

         slotIndex = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;
         if (ossAtomicPeek(&pChunkCB->flushedCount) > 0) {
            pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            flushDataLen = (slotIndex - ossAtomicPeek(&pChunkCB->flushedCount)) * OSS_RAMBO_SLOT_SIZE;
         }
         else
            flushDataLen = slotIndex * OSS_RAMBO_SLOT_SIZE;

         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

         err = dataToFuncProcessor( pChunk, flushDataLen, continueDataSize, &firstPartSize, &continueBuffer, 0, 0, pDataProcessor, pArgs ) ;
         if ( 0 > err ) {
//      perror("ossRamboFlushFD failed writing the storage header info");
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
         else
            continueDataSize = err;

         /* Reset the chunk control block */
         ossRamboMarkSlotsEmpty(pChunkCB, ossAtomicPeek(&pChunkCB->flushedCount), slotIndex-ossAtomicPeek(&pChunkCB->flushedCount));
      }
#ifdef OSS_DEBUG1
      else if (ossAtomicPeek(&pChunkCB->fillCount) > 0) {
//         printf("Unfilled chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
         fprintf(dbgfd,"Unfilled chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
         fflush(dbgfd);
      }
#endif
/*
      else if (flushStarted)
         printf("Empty chunk %lu\n", chunkIndex) ;
*/
      /* Move to next chunk if we didn't time out */
      if ( !timedOut )
         chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

      /* Check if we have been asked to stop */
      if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
      {
         flushUnfilledChunks = true ;
         break ;
      }

   }

   /* Flush unfilled chunks */
   if ( flushUnfilledChunks )
   {
      int numUsedSlots;

      /* Start flushing at the current chunk index so the flushing continues in the correct order */
      unsigned int i;
      for ( i = 0; i < pRamboCB->l2.h2.maxChunks; i++ )
      {
         /* Get pointer to the chunk control block */
         pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
         OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

         /* If the chunk contains data, flush it to disk */
         if ( ossAtomicPeek(&pChunkCB->fillCount) > 0 )
         {
            unsigned int j;
            numUsedSlots = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;

            /* Check if chunk is full */
            if (numUsedSlots == 0) {
               numUsedSlots = OSS_RAMBO_SLOTS_PER_CHUNK;
            }

            /* subtract the slots that have already been flushed */
            numUsedSlots -= ossAtomicPeek(&pChunkCB->flushedCount);

            /* if there are no used slots in this chunk */
            if (numUsedSlots == 0) {
               /* Move to next chunk */
               chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;
               continue;
            }
            else if (numUsedSlots < 0) {
               /* Note: flushedCount should never be bigger than fillCount but we'll
                  catch this case and stop flushing because we are in an inconsistent
                  state
               */
               goto errorProcessingFailed ;
            }

            /* Get a pointer to the chunk of memory */
            pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                    + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
            if (ossAtomicPeek(&pChunkCB->flushedCount) > 0)
               pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            OSS_ASSERT( pChunk != NULL ) ;
            OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

            /* Pass the chunk to the data processor, but only process the filled slots */

            err = dataToFuncProcessor( pChunk, numUsedSlots * OSS_RAMBO_SLOT_SIZE, continueDataSize, &firstPartSize, &continueBuffer, 0, 0, pDataProcessor, pArgs ) ;
            if ( 0 > err ) {
#ifdef OSS_DEBUG1
               fprintf(dbgfd,"ossRamboFlushFD:  failed writing the chunk %d of shared memory. error=%d\n", chunkIndex, err ) ;
               fflush(dbgfd);
#endif
               goto errorProcessingFailed ;
            }
			   else
				   continueDataSize = err;

#ifdef OSS_DEBUG1
//            printf("Flushed unfilled chunk index %lu has fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"Flushed unfilled chunk index %lu has fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif

            /* Reset the chunk control block */
/* 186134 begin
            memset( (void *)&pChunkCB->slotMap[0], 0x00, sizeof( pChunkCB->slotMap ) ) ;
*/
            for(j=0; j < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; j++) {
               ossAtomicPoke(&(pChunkCB->slotMap[j]), 0);
            }
/* 186134 end */
            ossAtomicPoke(&pChunkCB->fillCount, 0) ;
            ossAtomicPoke(&pChunkCB->flushedCount, 0) ;
         }
#ifdef OSS_DEBUG1
		   else {
//            printf("empty chunk index %lu\n", chunkIndex ) ;
            fprintf(dbgfd,"empty chunk index %lu\n", chunkIndex ) ;
            fflush(dbgfd);
         }
#endif
		 /* Move to next chunk */
         chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

      }
      /* Clear the auto-flusher status flags */
      pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
      pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
   }


// pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ;

   OSS_ASSERT( OSS_OK == osserr ) ;

exit :
   #ifdef OSS_DEBUG1
     fclose(dbgfd);
   #endif
   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 ;

errorProcessingFailed :
// pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
   osserr = OSS_ERR_RAMBO_IO_ERROR ;
   goto exit ;
}
/* 172892 end */

/* 198075 begin */

/*******************************************************************************

   Function Name
      dataProcessor

   Function
      Processes data from a RAMBO buffer and writes it out to the file descriptor

   Inputs
      pTraceData - addr of trace data to process
      dataLength - length of data to process
      continueDataSize - size of data at the start of the buffer to send without processing because it is the remainder of a record
      filepArgs - addr of I/O descriptor structure

   Normal Return
      returns size of last part of the data record that did not fit in this buffer

   Error Return
      -1
**********************************************************************************/

int dataProcessor(
      void *pTraceData,
      int dataLength,
      int continueDataSize,   /* size of end of msg from previous buffer to send */
      int fileDescriptor,     /* file descriptor to flush data to */
	   int descriptorType ) {  /* type of descriptor */

   char *pFlushData;
   Uint32 recordLength;
   int offset = 0;
   int endOffset = dataLength - OSS_RAMBO_DATA_HEADER_SIZE;
   int flushLen;     /* length of data to flush to file descriptor */
   int bFlushLen;    /* length of data to flush from buffer (includes rambo data header length) */
   ssize_t bytesWritten = 0;

#ifdef OSS_DEBUG
   ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  pTraceData=%p *pTraceData=%x/%x/%x/%x/%x  contSize=%d\n",
			                    pTraceData, (*(char *)pTraceData & 0x000000ff),(*((char *)pTraceData+1) & 0x000000ff),
								(*((char *)pTraceData+2) & 0x000000ff), (*((char *)pTraceData+3) & 0x000000ff),
                        (*((char *)pTraceData+4) & 0x000000ff), continueDataSize);
#endif

/* If there was a record that started in the previous buffer but the buffer was not big
   enough to contain the whole record, then the rest of the record is at the beginning
   of this buffer so flush the remainder of the record.
*/
   if (continueDataSize > 0) {
      int conFlushLen;

      /* The remainder of the record may be bigger than the size of this buffer */
      if (continueDataSize > dataLength) {
         /* Flush this whole buffer to the file or socket */
         conFlushLen = dataLength;
      }
      else {
         /* Write the remaninder of the previous record out to the file or socket */
         conFlushLen = continueDataSize;
      }

      if ( descriptorType == OSS_RAMBO_DESCRIPTOR_SOCKET ) {
		 /* Write data to socket */
#ifdef __OS400__
         bytesWritten = send( fileDescriptor, (char *)pTraceData, conFlushLen, 0 ) ;
#else
         bytesWritten = send( fileDescriptor, (const char *)pTraceData, conFlushLen, 0 ) ;
#endif
         if ( SOCKET_ERROR == bytesWritten )
         {
#ifdef OSS_DEBUG
#ifdef _WIN32
            ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. wsa error=%d\n",
                        fileDescriptor, WSAGetLastError() ) ;
#else
            ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
#endif
            return -1 ;
         }
         /* If not all the data was sent then send the rest */
         else if (bytesWritten < conFlushLen) {
	        char *dataptr = (char *)pTraceData + bytesWritten;
           int datalen = conFlushLen - bytesWritten;
           while (datalen > 0) {
               bytesWritten = send( fileDescriptor, dataptr, datalen, 0 ) ;
               if ( SOCKET_ERROR == bytesWritten )
               {
#ifdef OSS_DEBUG
#ifdef _WIN32
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. wsa error=%d",
					   dataptr, datalen, WSAGetLastError() ) ;
#else
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. errno=%d",
					   dataptr, datalen, errno ) ;
#endif
#endif
                  return -1 ;
               }
               dataptr += bytesWritten;
               datalen -= bytesWritten;
            }
         }
      }
      /* Write data to file */
      else {
         bytesWritten = write( fileDescriptor, (char *)pTraceData, conFlushLen ) ;
         if ( (ssize_t)-1 == bytesWritten )
         {
#ifdef OSS_DEBUG
            ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to file %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
             return -1 ;
         }
      }

      offset += OSS_RAMBO_ROUND_SLOT_SIZE( conFlushLen );
      continueDataSize -= conFlushLen;

   }

   /* Process the data until we have reached the end of the buffer */
   while (offset < endOffset) {

#ifdef OSS_DEBUG
      ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor: *pTraceData at offset %d =%x/%x/%x/%x %x/%x/%x/%x/%x\n",
			                    offset, (*((char *)pTraceData+offset) & 0x000000ff),(*((char *)pTraceData+offset+1) & 0x000000ff),
								(*((char *)pTraceData+offset+2) & 0x000000ff), (*((char *)pTraceData+offset+3) & 0x000000ff),
                        (*((char *)pTraceData+offset+4) & 0x000000ff), (*((char *)pTraceData+offset+5) & 0x000000ff),
                        (*((char *)pTraceData+offset+6) & 0x000000ff), (*((char *)pTraceData+offset+7) & 0x000000ff),
                        (*((char *)pTraceData+offset+8) & 0x000000ff));
#endif

      pFlushData = (char *)pTraceData + offset;
      flushLen = 0;
      bFlushLen = 0;
      /* read the data length from the buffer */
      recordLength = *(Uint32 *)pFlushData;

      /* If the whole record doesn't fit in this buffer then flush what is included
            in this buffer */
      if (offset + OSS_RAMBO_DATA_HEADER_SIZE + recordLength >= dataLength) {
		   continueDataSize = offset + OSS_RAMBO_DATA_HEADER_SIZE + recordLength - dataLength;
		   bFlushLen = dataLength - offset;
#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  Reached end of buffer recordLength=%d bflushLen=%d  pFlushData=%p\n",
			                    recordLength, bFlushLen, pFlushData);
#endif
		}
      /* Else the whole record is contained in this buffer so flush it */
      else {
         bFlushLen = OSS_RAMBO_DATA_HEADER_SIZE + recordLength;
      }

      flushLen = bFlushLen - OSS_RAMBO_DATA_HEADER_SIZE;
      pFlushData = (char *)pFlushData + OSS_RAMBO_DATA_HEADER_SIZE;

#ifdef OSS_DEBUG
      ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  recordLength=%d flushLen=%d bFlushLen=%d pFlushData=%p",
			                    recordLength, flushLen, bFlushLen, pFlushData);
#endif

      /* Write the data out to the file or socket */

      if ( descriptorType == OSS_RAMBO_DESCRIPTOR_SOCKET ) {
		 /* Write data to socket */
#ifdef __OS400__
         bytesWritten = send( fileDescriptor, (char *)pFlushData, flushLen, 0 ) ;
#else
         bytesWritten = send( fileDescriptor, (const char *)pFlushData, flushLen, 0 ) ;
#endif
         if ( SOCKET_ERROR == bytesWritten )
		   {
#ifdef OSS_DEBUG
#ifdef _WIN32
            ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. wsa error=%d",
                        fileDescriptor, WSAGetLastError() ) ;
#else
            ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to socket %d. errno=%d",
                        fileDescriptor, errno ) ;
#endif
#endif
            return -1 ;
         }
         /* If not all the data was sent then send the rest */
         else if (bytesWritten < flushLen) {
	        char *dataptr = pFlushData + bytesWritten;
	        int datalen = flushLen - bytesWritten;
	        while (datalen > 0) {
               bytesWritten = send( fileDescriptor, dataptr, datalen, 0 ) ;
               if ( SOCKET_ERROR == bytesWritten )
		   	   {
#ifdef OSS_DEBUG
#ifdef _WIN32
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. wsa error=%d",
					   dataptr, datalen, WSAGetLastError() ) ;
#else
                  ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of buffer starting at %p len %d. errno=%d",
					   dataptr, datalen, errno ) ;
#endif
#endif
                  return -1 ;
			      }
		         dataptr += bytesWritten;
		         datalen -= bytesWritten;
		   	}
         }
      }
		 /* Write data to file */
		else {
         bytesWritten = write( fileDescriptor, pFlushData, flushLen ) ;
         if ( (ssize_t)-1 == bytesWritten )
		   {
#ifdef OSS_DEBUG
            ra_logServiceMessage(__FILE__, __LINE__,RA_WARNING,"dataProcessor:  failed writing part of shared memory buffer to file %d. errno=%d\n",
                        fileDescriptor, errno ) ;
#endif
             return -1 ;
			}
		}

      /* increase the offset by the number of slots that were flushed */
      offset += OSS_RAMBO_ROUND_SLOT_SIZE( bFlushLen );
   }
   return continueDataSize;
}


/*******************************************************************************

   Function Name
      dataToStrProcessor

   Function
      Processes data from a RAMBO buffer and writes it out a java OutputStream

   Inputs
      pTraceData - addr of trace data to process
      dataLength - length of data to process
      continueDataSize - size of data at the start of the buffer to send without processing because it is the remainder of a record
      jenv - Java environment
	  jobj - Java OutputStream object
	  jmethod  - the OutputStream write method ID

   Normal Return
      returns size of last part of the data record that did not fit in this buffer

   Error Return
      -1
**********************************************************************************/

int dataToStrProcessor(
	void *pTraceData,
	int dataLength,
	int continueDataSize,   /* size of end of msg from previous buffer to send */
	JNIEnv *jenv,     /* Java environemnt */
	jobject jobj,     /* OutputStream object */
	jmethodID jmethod ) {  /* write method ID */

	char *pFlushData;
	Uint32 recordLength;
	int offset = 0;
	int endOffset = dataLength - OSS_RAMBO_DATA_HEADER_SIZE;
	int flushLen;     /* length of data to flush to file descriptor */
	int bFlushLen;    /* length of data to flush from buffer (includes rambo data header length) */

#ifdef OSS_DEBUG
	ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  pTraceData=%p *pTraceData=%x/%x/%x/%x/%x  contSize=%d\n",
		pTraceData, (*(char *)pTraceData & 0x000000ff),(*((char *)pTraceData+1) & 0x000000ff),
		(*((char *)pTraceData+2) & 0x000000ff), (*((char *)pTraceData+3) & 0x000000ff),
		(*((char *)pTraceData+4) & 0x000000ff), continueDataSize);
#endif

	/* If there was a record that started in the previous buffer but the buffer was not big
		enough to contain the whole record, then the rest of the record is at the beginning
		of this buffer so flush the remainder of the record.
	*/
	if (continueDataSize > 0) {
		int conFlushLen;

		/* The remainder of the record may be bigger than the size of this buffer */
		if (continueDataSize > dataLength) {
		/* Flush this whole buffer to the file or socket */
			conFlushLen = dataLength;
		}
		else {
			/* Write the remaninder of the previous record out to the file or socket */
			conFlushLen = continueDataSize;
		}

		/* Format the java input arguments */
		
		/* Possible memory leak here if we create a new byte array every time
		 *  - need to confirm this
		 */
		if(conFlushLen > 0) {
			jbyteArray jbarr;
			jthrowable jexc;
			jbyte *tmpBuffer;

			jbarr = ENV(jenv)->NewByteArray(ENVPARM(jenv) conFlushLen);
			if(jbarr == NULL) {
				return -1;
			}

			tmpBuffer = (signed char*)malloc(sizeof(signed char) * conFlushLen);
			memcpy(tmpBuffer, (signed char*)pTraceData, sizeof(signed char) * conFlushLen);

			ENV(jenv)->SetByteArrayRegion(ENVPARM(jenv) jbarr, 0, conFlushLen, tmpBuffer);

			/* Write the data out to the java OutputStream */
			ENV(jenv)->CallVoidMethod(ENVPARM(jenv) jobj, jmethod, jbarr, 0, conFlushLen);

			/* Check for a java exception */
			jexc = ENV(jenv)->ExceptionOccurred(ENVPARM1(jenv));
			if(jexc) {
				ENV(jenv)->ExceptionClear(ENVPARM1(jenv));
			}

//			ENV(jenv)->ReleaseByteArrayElements(ENVPARM(jenv) jbarr, tmpBuffer, 0);
			ENV(jenv)->DeleteLocalRef(ENVPARM(jenv) jbarr);
			free(tmpBuffer);

			if(jexc) {
				return -1;
			}
		}

		offset += OSS_RAMBO_ROUND_SLOT_SIZE( conFlushLen );
		continueDataSize -= conFlushLen;
   }

   /* Process the data until we have reached the end of the buffer */
   while (offset < endOffset) {

#ifdef OSS_DEBUG
      ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor: *pTraceData at offset %d =%x/%x/%x/%x %x/%x/%x/%x/%x\n",
			                    offset, (*((char *)pTraceData+offset) & 0x000000ff),(*((char *)pTraceData+offset+1) & 0x000000ff),
								(*((char *)pTraceData+offset+2) & 0x000000ff), (*((char *)pTraceData+offset+3) & 0x000000ff),
                        (*((char *)pTraceData+offset+4) & 0x000000ff), (*((char *)pTraceData+offset+5) & 0x000000ff),
                        (*((char *)pTraceData+offset+6) & 0x000000ff), (*((char *)pTraceData+offset+7) & 0x000000ff),
                        (*((char *)pTraceData+offset+8) & 0x000000ff));
#endif

      pFlushData = (char *)pTraceData + offset;
      flushLen = 0;
      bFlushLen = 0;
      /* read the data length from the buffer */
      recordLength = *(Uint32 *)pFlushData;

      /* If the whole record doesn't fit in this buffer then flush what is included
            in this buffer */
      if (offset + OSS_RAMBO_DATA_HEADER_SIZE + recordLength >= dataLength) {
		   continueDataSize = offset + OSS_RAMBO_DATA_HEADER_SIZE + recordLength - dataLength;
		   bFlushLen = dataLength - offset;
#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  Reached end of buffer recordLength=%d bflushLen=%d  pFlushData=%p\n",
			                    recordLength, bFlushLen, pFlushData);
#endif
		}
      /* Else the whole record is contained in this buffer so flush it */
      else {
         bFlushLen = OSS_RAMBO_DATA_HEADER_SIZE + recordLength;
      }

      flushLen = bFlushLen - OSS_RAMBO_DATA_HEADER_SIZE;
      pFlushData = (char *)pFlushData + OSS_RAMBO_DATA_HEADER_SIZE;

#ifdef OSS_DEBUG
		ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"dataProcessor:  recordLength=%d flushLen=%d bFlushLen=%d pFlushData=%p",
			recordLength, flushLen, bFlushLen, pFlushData);
#endif

		/* Format the java input arguments */
		
		/* Possible memory leak here if we create a new byte array every time
			*  - need to confirm this
		*/
		if(flushLen > 0) {
			jbyteArray jbarr;
			jthrowable jexc;
			jbyte *tmpBuffer;

			jbarr = ENV(jenv)->NewByteArray(ENVPARM(jenv) flushLen);
			if(jbarr == NULL) {
				return -1;
			}

			tmpBuffer = (signed char*)malloc(sizeof(signed char) * flushLen);
			memcpy(tmpBuffer, (signed char*)pFlushData, sizeof(signed char) * flushLen);

			ENV(jenv)->SetByteArrayRegion(ENVPARM(jenv) jbarr, 0, flushLen, tmpBuffer);

			/* Write the data out to the java OutputStream */
			ENV(jenv)->CallVoidMethod(ENVPARM(jenv) jobj, jmethod, jbarr, 0, flushLen);

			/* Check for a java exception */
			jexc = ENV(jenv)->ExceptionOccurred(ENVPARM1(jenv));
			if(jexc) {
				ENV(jenv)->ExceptionClear(ENVPARM1(jenv));
			}

//			ENV(jenv)->ReleaseByteArrayElements(ENVPARM(jenv) jbarr, tmpBuffer, 0);
			ENV(jenv)->DeleteLocalRef(ENVPARM(jenv) jbarr);
			free(tmpBuffer);

			if(jexc) {
				return -1;
			}
		}

		/* increase the offset by the number of slots that were flushed */
		offset += OSS_RAMBO_ROUND_SLOT_SIZE( bFlushLen );
   }
   return continueDataSize;
}



/*******************************************************************************

   Function Name
      ossRamboFlushToFD

   Function
      Flushes the RAMBO buffer to a file descriptor on a timed basis.  It
      does not wait for a chunk to be filled before flushing the data in it.

   Inputs
      1. pRambo
         Address of a RAMBO buffer control block.
	  2. fileDescriptor
		   file descriptor to flush buffer to
	  3. descriptorType
	      type of descriptor (1=file 2=socket)

   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_ERR_TIMEOUT

 ******************************************************************************/
OSS_EXTERNC OSSErr OSS_API ossRamboFlushToFD(
      OSSRambo *pRambo,
      int fileDescriptor,
	   int descriptorType )
{
   OSSErr osserr = OSS_OK ;
   ssize_t bytesWritten = 0 ;
   Uint32 chunkIndex = 0 ;
   Uint32 slotIndex = 0 ;
   Uint32 loopIndex = 0 ;
   Uint32 attachCheckIndex = 0 ;  /* bugzilla 94473 */
   Uint32 timedOut = 0 ;
   int flushDataLen;
   OSSRamboCB * pRamboCB ;
   OSSRamboChunkCB * pChunkCB = NULL ;
   void * pChunk = NULL ;
   bool wrappedChunk = false ;
   bool flushUnfilledChunks = false ;
   bool stopping = false;
   Sint32 err = 0 ;
   Sint32 continueDataSize = 0;
   bool dataFlashed = false;
#ifdef OSS_DEBUG1
   FILE *dbgfd;
#endif

#ifdef OSS_DEBUG1
   dbgfd = fopen("flushdbg.txt", "ac+");
   if (dbgfd == NULL) {
	   osserr = errno;
	   goto exit;
   }
   if ( 0 > fprintf(dbgfd,"ossRamboFlushToFD:  Entered.\n" ) ) {
	   osserr = OSS_ERR_RAMBO_IO_ERROR;
	   goto exit;
   }
   fflush(dbgfd);
#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 ) ) ) ;
   OSS_ASSERT( -1 != fileDescriptor ) ;
#ifdef OSS_DEBUG1
   fprintf(dbgfd,"ossRamboFlushToFD:  After asserts.\n" ) ;
   fflush(dbgfd);
#endif

   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) )
   {
      goto errorInvalid ;
   }

   pRamboCB = pRambo->pRamboCB ;

   OSS_ASSERT( ossRamboIsInitialized( pRamboCB ) ) ;

   if ( !ossRamboIsInitialized( pRamboCB ) )
   {
      goto errorNotInitialized ;
   }

   if ( -1 == fileDescriptor )
   {
      goto errorInvalid ;
   }

   /* If the auto flusher is on, don't allow manual flushing */
   if ( ossRamboIsAutoFlusherStarted( pRamboCB ) )
   {
      osserr = OSS_ERR_RAMBO_AUTOFLUSHER_ENABLED ;
      goto exit ;
   }

   /* If already flushing, wait. */
   /* 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 ;
   /* Pretend we are auto flushing instead of manual flushing so we don't hold up writing
      to the buffer */
   pRamboCB->l2.h2.status |= OSS_RAMBO_AUTOF_STARTED ;
   /* TODO */


   /* 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) ) ) ;


   /* TODO */
   /* Start flushing */
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
   fprintf(dbgfd,"ossRamboFlushToFD:  Started flushing.\n" ) ;
   fflush(dbgfd);
#endif

   /* Try to find the least recently filled chunk if the buffer has
	  been wrapped and start flushing at it. */

   if ( ossAtomicPeek(&(pFirstChunkCB(pRamboCB)->fillCount)) > OSS_RAMBO_SLOTS_PER_CHUNK )
   {
      slotIndex   = ossAtomicPeek(&pRamboCB->l1.h1.nextSlotIndex) % pRamboCB->l2.h2.maxSlots ;
      chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK + 1;
   }
   else
	  chunkIndex = 0;

   for ( ; ; )
   {
      /* Get pointer to the chunk control block */
      pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
	   OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

      loopIndex = 0;
      timedOut = 0;
      attachCheckIndex = 0;  /* 94473 */
      /* Wait until the chunk is full or we have been asked to stop or we have waited too long */
      for ( ; ; )
      {
         /* If flushing is to be stopped then set the stopping flag and quit the loop */
         if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
         {
            stopping = true ;
#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"flushToFD:  Stop flushing");
#endif
            break ;
         }

         /* If the chunk is full then quit the loop so it can be flushed. */
         if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
         {
#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"flushToFD:  Data found in buffer.");
#endif
            break ;
         }
         /* Else If we have looped too long */
         else if (loopIndex == 100) {
            /* If there is any data in the chunk Then quit the loop to flush it */
            if ((ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) - ossAtomicPeek(&pChunkCB->flushedCount) > 0  ||
                (ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0)) {
               timedOut = 1;
#ifdef OSS_DEBUG
				ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"flushToFD:  We've waited too long, flush data found in chunk.");
#endif
               break ;
            }
/* 205955 begin */
			/* Else get the attach count to see if anyone else is attached */
            else {
               int attcnt = 0;
               OSSErr osserr = OSS_OK ;
               osserr = ossIPCMemAttachCount( pRambo->ipcMemHandle, &attcnt ) ;
#ifdef OSS_DEBUG
					ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"flushToFD:  attach count = %d", attcnt);
#endif
               attachCheckIndex++;  /* bugzilla 94473 */
               /* If nobody is attached to the shared memory except for us then stop flushing
                * bugzilla 94473 - only stop flushing if we have gone through the wait loop four
                * times to provide more time for the agent to attach to it (4 seconds instead of 1). */
               if ( osserr == OSS_OK  &&  attcnt == 1 && attachCheckIndex >= 4)
               {
                  /* set the user count to 1 so the shared memory gets deleted when it is destroyed */
                  ossIPCMemSetUserCount( pRambo->ipcMemHandle, attcnt ) ;
                  stopping = true ;
#ifdef OSS_DEBUG
						ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"flushToFD:  only we are attached so stop flushing");
#endif
                  break ;
               }
               /* Else continue checking for writes to shared memory */
               loopIndex = 0;
            }
/* 205955 end */
         }

         ossSleep( 10 ) ;
         loopIndex++;
      }

      /* If we should stop, break out of the flushing loop */
      if ( stopping )
      {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushToFD:  We are stopping flushing so flush unfilled chunks.\n" ) ;
            fflush(dbgfd);
#endif
         flushUnfilledChunks = true ;
         break ;
      }

      /* If the chunk is full, flush it to disk */
#ifdef ENW
      if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
#else
      if ( pChunkCB->fillCount >= OSS_RAMBO_SLOTS_PER_CHUNK )
#endif
      {
         unsigned int i;
         if (ossAtomicPeek(&pChunkCB->flushedCount) == 0) {
            OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[0]) ) ;
         }
         OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE-1]) ) ;

         if ( ossAtomicPeek(&pChunkCB->fillCount) > OSS_RAMBO_SLOTS_PER_CHUNK )
         {
            wrappedChunk = true ;
         }
/*
		 if (!flushStarted)
			 flushStarted = true;
*/
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
         if (ossAtomicPeek(&pChunkCB->flushedCount) > 0) {
            pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            flushDataLen = OSS_RAMBO_CHUNK_SIZE - ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE;
         }
         else
            flushDataLen = OSS_RAMBO_CHUNK_SIZE;

         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

		 dataFlashed = true;
         err = dataProcessor( pChunk, flushDataLen, continueDataSize, fileDescriptor, descriptorType ) ;
         if ( 0 > err ) {
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushToFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
         else
            continueDataSize = err;

         if ( wrappedChunk )
         {
            /*
             * The flusher has been lapped.  The chunk has been filled
             * more than once.  Log this so we know it has happened but
             * don't exit.
             */
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif
            wrappedChunk = false ;
         }
#ifdef OSS_DEBUG1
         else {
            fprintf(dbgfd,"Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
         }
#endif

         /* Reset the chunk control block */
/* 186134 begin */
         for(i=0; i < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; i++) {
            ossAtomicPoke(&(pChunkCB->slotMap[i]), 0);
         }
/* 186134 end */
         ossAtomicPoke(&pChunkCB->fillCount, 0) ;
         ossAtomicPoke(&pChunkCB->flushedCount, 0) ;
      }
      else if ( timedOut ) {
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;

         slotIndex = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;
         if (ossAtomicPeek(&pChunkCB->flushedCount) > 0) {
            pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            flushDataLen = (slotIndex - ossAtomicPeek(&pChunkCB->flushedCount)) * OSS_RAMBO_SLOT_SIZE;
         }
         else
            flushDataLen = slotIndex * OSS_RAMBO_SLOT_SIZE;

         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

		 dataFlashed = true;
         err = dataProcessor( pChunk, flushDataLen, continueDataSize, fileDescriptor, descriptorType ) ;
         if ( 0 > err ) {
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushToFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
         else
            continueDataSize = err;

         /* Reset the chunk control block */
         ossRamboMarkSlotsEmpty(pChunkCB, ossAtomicPeek(&pChunkCB->flushedCount), slotIndex-ossAtomicPeek(&pChunkCB->flushedCount));
      }
#ifdef OSS_DEBUG1
      else if (ossAtomicPeek(&pChunkCB->fillCount) > 0) {
         fprintf(dbgfd,"Unfilled chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
         fflush(dbgfd);
      }
#endif
/*
      else if (flushStarted)
         printf("Empty chunk %lu\n", chunkIndex) ;
*/
      /* Move to next chunk if we didn't time out */
      if ( !timedOut )
         chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

      /* Check if we have been asked to stop */
      if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
      {
         flushUnfilledChunks = true ;
         break ;
      }

   }

	/* Flush unfilled chunks */
	if ( flushUnfilledChunks )
	{
		int numUsedSlots;

		/* Start flushing at the current chunk index so the flushing continues in the correct order */
		unsigned int i;
		for ( i = 0; i < pRamboCB->l2.h2.maxChunks; i++ )
		{
			/* Get pointer to the chunk control block */
			pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
			OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

			/* If the chunk contains data, flush it to disk */
			if ( ossAtomicPeek(&pChunkCB->fillCount) > 0 )
			{
				unsigned int j;
            numUsedSlots = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;

            /* Check if chunk is full */
            if (numUsedSlots == 0) {
               numUsedSlots = OSS_RAMBO_SLOTS_PER_CHUNK;
            }

            /* subtract the slots that have already been flushed */
            numUsedSlots -= ossAtomicPeek(&pChunkCB->flushedCount);

            /* if there are no used slots in this chunk */
            if (numUsedSlots == 0) {
               /* Move to next chunk */
               chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;
               continue;
            }
            else if (numUsedSlots < 0) {
               /* Note: flushedCount should never be bigger than fillCount but we'll
                  catch this case and stop flushing because we are in an inconsistent
                  state
               */
               goto errorProcessingFailed ;
            }

            /* Get a pointer to the chunk of memory */
            pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                    + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
            if (ossAtomicPeek(&pChunkCB->flushedCount) > 0)
               pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            OSS_ASSERT( pChunk != NULL ) ;
            OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

            /* Pass the chunk to the data processor, but only process the filled slots */

			dataFlashed = true;
            err = dataProcessor( pChunk, numUsedSlots * OSS_RAMBO_SLOT_SIZE, continueDataSize, fileDescriptor, descriptorType ) ;
            if ( 0 > err ) {
#ifdef OSS_DEBUG1
               fprintf(dbgfd,"ossRamboFlushToFD:  failed writing the chunk %d of shared memory. error=%d\n", chunkIndex, err ) ;
               fflush(dbgfd);
#endif
               goto errorProcessingFailed ;
            }
			   else
				   continueDataSize = err;

#ifdef OSS_DEBUG1
            fprintf(dbgfd,"Flushed unfilled chunk index %lu has fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif

				/* Reset the chunk control block */
/* 186134 begin */
				for(j=0; j < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; j++) {
					ossAtomicPoke(&(pChunkCB->slotMap[j]), 0);
				}
/* 186134 end */
				ossAtomicPoke(&pChunkCB->fillCount, 0) ;
				ossAtomicPoke(&pChunkCB->flushedCount, 0) ;
			}
#ifdef OSS_DEBUG1
			else {
				fprintf(dbgfd,"empty chunk index %lu\n", chunkIndex ) ;
				fflush(dbgfd);
			}
#endif
			/* Move to next chunk */
			chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

		}
		/* Clear the auto-flusher status flags */
		pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
		pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
	}

   OSS_ASSERT( OSS_OK == osserr ) ;

exit :
   #ifdef OSS_DEBUG1
     fclose(dbgfd);
   #endif
   OSS_ASSERT( !( pRamboCB->l2.h2.status & OSS_RAMBO_FLUSHING ) ) ;
   
   if (osserr == OSS_OK && timedOut && !dataFlashed)
   		osserr = OSS_ERR_TIMEOUT; 
   
   return osserr ;

errorInvalid :
   osserr = OSS_ERR_INVALID ;
   goto exit ;

errorNotInitialized :
   osserr = OSS_ERR_RAMBO_NOT_INIT ;
   goto exit ;

errorProcessingFailed :
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
   osserr = OSS_ERR_RAMBO_IO_ERROR ;
   goto exit ;
}
/* 198075 end */


/*******************************************************************************

   Function Name
      ossRamboFlushToStream

   Function
      Flushes the RAMBO buffer to a Java Stream on a timed basis.  It
      does not wait for a chunk to be filled before flushing the data in it.

   Inputs
      1. pRambo
         Address of a RAMBO buffer control block.
	  2. outStrObj
		   java OutputStream object

   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 ossRamboFlushToStream(
      OSSRambo *pRambo,
	  JNIEnv   *jenv,
      jobject   jOutStrObj )
{
   OSSErr osserr = OSS_OK ;
   ssize_t bytesWritten = 0 ;
   Uint32 chunkIndex = 0 ;
   Uint32 slotIndex = 0 ;
   Uint32 loopIndex = 0 ;
   Uint32 timedOut = 0 ;
   Uint32 attachCheckIndex = 0 ;  /* bugzilla 94473 */
   int flushDataLen;
   OSSRamboCB * pRamboCB ;
   OSSRamboChunkCB * pChunkCB = NULL ;
   void * pChunk = NULL ;
   bool wrappedChunk = false ;
   bool flushUnfilledChunks = false ;
   bool stopping = false;
   Sint32 err = 0 ;
   Sint32 continueDataSize = 0;
	jclass jcls = NULL;
	jmethodID jmethod = NULL;

#ifdef OSS_DEBUG1
   FILE *dbgfd;
#endif

#ifdef OSS_DEBUG1
   dbgfd = fopen("flushdbg.txt", "ac+");
   if (dbgfd == NULL) {
	   osserr = errno;
	   goto exit;
   }
   if ( 0 > fprintf(dbgfd,"ossRamboFlushToFD:  Entered.\n" ) ) {
	   osserr = OSS_ERR_RAMBO_IO_ERROR;
	   goto exit;
   }
   fflush(dbgfd);
#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 ) ) ) ;

#ifdef OSS_DEBUG1
   fprintf(dbgfd,"ossRamboFlushToFD:  After asserts.\n" ) ;
   fflush(dbgfd);
#endif

   if ( ( NULL == pRambo ) || ( NULL == pRambo->pRamboCB ) )
   {
      goto errorInvalid ;
   }

   pRamboCB = pRambo->pRamboCB ;

   OSS_ASSERT( ossRamboIsInitialized( pRamboCB ) ) ;

   if ( !ossRamboIsInitialized( pRamboCB ) )
   {
      goto errorNotInitialized ;
   }

   /* If the auto flusher is on, don't allow manual flushing */
   if ( ossRamboIsAutoFlusherStarted( pRamboCB ) )
   {
      osserr = OSS_ERR_RAMBO_AUTOFLUSHER_ENABLED ;
      goto exit ;
   }

	/* Get java method ID */
#ifdef OSS_DEBUG
	printf("ossRamboFlushToStream: Before trying to get java method ID\n");
#endif
	jcls = ENV(jenv)->GetObjectClass(ENVPARM(jenv) jOutStrObj);
	if(jcls != NULL) {
		jmethod = ENV(jenv)->GetMethodID(ENVPARM(jenv) jcls, "write", "([BII)V");
	}

	if (jmethod == NULL) {
		osserr = OSS_ERR_RAMBO_IO_ERROR ;
		goto exit ;
	}
#ifdef OSS_DEBUG
	printf("ossRamboFlushToStream: After getting java method ID\n");
#endif


   /* If already flushing, wait. */
   /* 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 ;
   /* Pretend we are auto flushing instead of manual flushing so we don't hold up writing
      to the buffer */
   pRamboCB->l2.h2.status |= OSS_RAMBO_AUTOF_STARTED ;
   /* TODO */


   /* 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) ) ) ;


   /* TODO */
   /* Start flushing */
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
   fprintf(dbgfd,"ossRamboFlushToFD:  Started flushing.\n" ) ;
   fflush(dbgfd);
#endif

   /* Try to find the least recently filled chunk if the buffer has
	  been wrapped and start flushing at it. */

   if ( ossAtomicPeek(&(pFirstChunkCB(pRamboCB)->fillCount)) > OSS_RAMBO_SLOTS_PER_CHUNK )
   {
      slotIndex   = ossAtomicPeek(&pRamboCB->l1.h1.nextSlotIndex) % pRamboCB->l2.h2.maxSlots ;
      chunkIndex  = slotIndex / OSS_RAMBO_SLOTS_PER_CHUNK + 1;
   }
   else
	  chunkIndex = 0;

#ifdef OSS_DEBUG
	printf("ossRamboFlushToStream: About to go into waiting loop\n");
#endif

   for ( ; ; )
   {
      /* Get pointer to the chunk control block */
      pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
	   OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

      loopIndex = 0;
      timedOut = 0;
      attachCheckIndex = 0;  /* 94473 */
      /* Wait until the chunk is full or we have been asked to stop or we have waited too long */
      for ( ; ; )
      {
         /* If flushing is to be stopped then set the stopping flag and quit the loop */
         if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
         {
            stopping = true ;
            break ;
         }

         /* If the chunk is full then quit the loop so it can be flushed. */
         if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
         {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushToFD:  Data found in buffer.\n" ) ;
            fflush(dbgfd);
#endif
            break ;
         }
         /* Else If we have looped too long */
         else if (loopIndex == 100) {
            /* If there is any data in the chunk Then quit the loop to flush it */
            if ((ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) - ossAtomicPeek(&pChunkCB->flushedCount) > 0  ||
                (ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0)) {
               timedOut = 1;
#ifdef OSS_DEBUG1
               fprintf(dbgfd,"ossRamboFlushToFD:  We've waited too long, flush data found in chunk.\n" ) ;
               fflush(dbgfd);
#endif
               break ;
            }
/* 205955 begin */
			/* Else get the attach count to see if anyone else is attached */
            else {
               int attcnt = 0;
               OSSErr osserr = OSS_OK ;
               osserr = ossIPCMemAttachCount( pRambo->ipcMemHandle, &attcnt ) ;
               attachCheckIndex++;  /* bugzilla 94473 */
               /* If nobody is attached to the shared memory except for us then stop flushing
                * bugzilla 94473 - only stop flushing if we have gone through the wait loop four
                * times to provide more time for the agent to attach to it (4 seconds instead of 1). */
               if ( osserr == OSS_OK  &&  attcnt == 1 && attachCheckIndex >= 4)
               {
#ifdef OSS_DEBUG
   				  printf("ossRamboFlushToStream: Nobody is attached to write to the buffer yet.\n");
#endif
                  /* set the user count to 1 so the shared memory gets deleted when it is destroyed */
/*
                  ossIPCMemSetUserCount( pRambo->ipcMemHandle, attcnt ) ;
                  stopping = true ;
                  break ;
*/
               }
               /* Else continue checking for writes to shared memory */
               loopIndex = 0;
            }
/* 205955 end */
         }

         ossSleep( 10 ) ;
         loopIndex++;
      }

      /* If we should stop, break out of the flushing loop */
      if ( stopping )
      {
#ifdef OSS_DEBUG1
// printf("ossRamboFlushFD:  Started flushing.\n" ) ;
            fprintf(dbgfd,"ossRamboFlushToFD:  We are stopping flushing so flush unfilled chunks.\n" ) ;
            fflush(dbgfd);
#endif
         flushUnfilledChunks = true ;
         break ;
      }

      /* If the chunk is full, flush it to disk */
#ifdef ENW
      if ( ossAtomicPeek(&pChunkCB->fillCount) > 0  &&  (ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK) == 0 )
#else
      if ( pChunkCB->fillCount >= OSS_RAMBO_SLOTS_PER_CHUNK )
#endif
      {
         unsigned int i;
         if (ossAtomicPeek(&pChunkCB->flushedCount) == 0) {
            OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[0]) ) ;
         }
         OSS_ASSERT( (Uint32)-1 == ossAtomicPeek(&pChunkCB->slotMap[OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE-1]) ) ;

         if ( ossAtomicPeek(&pChunkCB->fillCount) > OSS_RAMBO_SLOTS_PER_CHUNK )
         {
            wrappedChunk = true ;
         }
/*
		 if (!flushStarted)
			 flushStarted = true;
*/
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
         if (ossAtomicPeek(&pChunkCB->flushedCount) > 0) {
            pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            flushDataLen = OSS_RAMBO_CHUNK_SIZE - ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE;
         }
         else
            flushDataLen = OSS_RAMBO_CHUNK_SIZE;

         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

         err = dataToStrProcessor( pChunk, flushDataLen, continueDataSize, jenv, jOutStrObj, jmethod ) ;
         if ( 0 > err ) {
//      perror("ossRamboFlushFD failed writing the storage header info");
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushToFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
         else
            continueDataSize = err;

         if ( wrappedChunk )
         {
            /*
             * The flusher has been lapped.  The chunk has been filled
             * more than once.  Log this so we know it has happened but
             * don't exit.
             */
#ifdef OSS_DEBUG1
//            printf("RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"RC = ERR_CHUNK_OVERWRITTEN:  Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif
            wrappedChunk = false ;
         }
#ifdef OSS_DEBUG1
         else {
//            printf("Flushed chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"Flushed chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
         }
#endif

         //pause() ;
         /* Reset the chunk control block */
/* 186134 begin
         memset( (void *)&pChunkCB->slotMap[0], 0x00, sizeof( pChunkCB->slotMap ) ) ;
*/
         for(i=0; i < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; i++) {
            ossAtomicPoke(&(pChunkCB->slotMap[i]), 0);
         }
/* 186134 end */
         ossAtomicPoke(&pChunkCB->fillCount, 0) ;
         ossAtomicPoke(&pChunkCB->flushedCount, 0) ;
      }
      else if ( timedOut ) {
         /* Get a pointer to the chunk of memory */
         pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
                 + ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;

         slotIndex = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;
         if (ossAtomicPeek(&pChunkCB->flushedCount) > 0) {
            pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
            flushDataLen = (slotIndex - ossAtomicPeek(&pChunkCB->flushedCount)) * OSS_RAMBO_SLOT_SIZE;
         }
         else
            flushDataLen = slotIndex * OSS_RAMBO_SLOT_SIZE;

         OSS_ASSERT( pChunk != NULL ) ;
         OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

         /* Pass the chunk to the data processor */

         err = dataToStrProcessor( pChunk, flushDataLen, continueDataSize, jenv, jOutStrObj, jmethod ) ;
         if ( 0 > err ) {
//      perror("ossRamboFlushFD failed writing the storage header info");
#ifdef OSS_DEBUG1
            fprintf(dbgfd,"ossRamboFlushToFD:  failed writing the chunk %d of shared memory. wsa error=%d\n", chunkIndex, WSAGetLastError() ) ;
            fflush(dbgfd);
#endif
            goto errorProcessingFailed ;
         }
         else
            continueDataSize = err;

         /* Reset the chunk control block */
         ossRamboMarkSlotsEmpty(pChunkCB, ossAtomicPeek(&pChunkCB->flushedCount), slotIndex-ossAtomicPeek(&pChunkCB->flushedCount));
      }
#ifdef OSS_DEBUG1
      else if (ossAtomicPeek(&pChunkCB->fillCount) > 0) {
//         printf("Unfilled chunk %lu fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
         fprintf(dbgfd,"Unfilled chunk %lu fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
         fflush(dbgfd);
      }
#endif
/*
      else if (flushStarted)
         printf("Empty chunk %lu\n", chunkIndex) ;
*/
      /* Move to next chunk if we didn't time out */
      if ( !timedOut )
         chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

      /* Check if we have been asked to stop */
      if ( pRamboCB->l2.h2.status & OSS_RAMBO_AUTOF_STOP_NOW )
      {
         flushUnfilledChunks = true ;
         break ;
      }
#ifdef OSS_DEBUG
	   printf("ossRamboFlushToStream: About to do another iteration of the waiting loop\n");
#endif
   }

	/* Flush unfilled chunks */
	if ( flushUnfilledChunks )
	{
		int numUsedSlots;

		/* Start flushing at the current chunk index so the flushing continues in the correct order */
		unsigned int i;
#ifdef OSS_DEBUG
		printf("ossRamboFlushToStream: Flushing unfilled chunks\n");
#endif
		for ( i = 0; i < pRamboCB->l2.h2.maxChunks; i++ )
		{
			/* Get pointer to the chunk control block */
			pChunkCB = pFirstChunkCB(pRamboCB) + chunkIndex ;
			OSS_ASSERT( ossRamboIsValidChunkCB( pChunkCB, pRamboCB ) ) ;

			/* If the chunk contains data, flush it to disk */
			if ( ossAtomicPeek(&pChunkCB->fillCount) > 0 )
			{
				unsigned int j;
            numUsedSlots = ossAtomicPeek(&pChunkCB->fillCount) % OSS_RAMBO_SLOTS_PER_CHUNK;

            /* Check if chunk is full */
            if (numUsedSlots == 0) {
               numUsedSlots = OSS_RAMBO_SLOTS_PER_CHUNK;
            }

            /* subtract the slots that have already been flushed */
            numUsedSlots -= ossAtomicPeek(&pChunkCB->flushedCount);

            /* if there are no used slots in this chunk */
            if (numUsedSlots == 0) {
               /* Move to next chunk */
               chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;
               continue;
            }
            else if (numUsedSlots < 0) {
               /* Note: flushedCount should never be bigger than fillCount but we'll
                  catch this case and stop flushing because we are in an inconsistent
                  state
               */
               goto errorProcessingFailed ;
            }

				/* Get a pointer to the chunk of memory */
				pChunk  = (void *)( (uintptr_t)pMemBuffer(pRamboCB)
							+ ( chunkIndex * OSS_RAMBO_CHUNK_SIZE ) ) ;
				if (ossAtomicPeek(&pChunkCB->flushedCount) > 0)
					pChunk = (void *)((char *)pChunk + ossAtomicPeek(&pChunkCB->flushedCount) * OSS_RAMBO_SLOT_SIZE);
				OSS_ASSERT( pChunk != NULL ) ;
				OSS_ASSERT( (uintptr_t)pChunk >= (uintptr_t)pMemBuffer(pRamboCB) ) ;

				/* Pass the chunk to the data processor, but only process the filled slots */

				err = dataToStrProcessor( pChunk, numUsedSlots * OSS_RAMBO_SLOT_SIZE, continueDataSize, jenv, jOutStrObj, jmethod ) ;
				if ( 0 > err ) {
#ifdef OSS_DEBUG1
					fprintf(dbgfd,"ossRamboFlushToFD:  failed writing the chunk %d of shared memory. error=%d\n", chunkIndex, err ) ;
					fflush(dbgfd);
#endif
					goto errorProcessingFailed ;
				}
				else
					continueDataSize = err;

#ifdef OSS_DEBUG1
//            printf("Flushed unfilled chunk index %lu has fillCount %lu\n",
//                        chunkIndex, pChunkCB->fillCount ) ;
            fprintf(dbgfd,"Flushed unfilled chunk index %lu has fillCount %lu\n",
                        chunkIndex, ossAtomicPeek(&pChunkCB->fillCount) ) ;
            fflush(dbgfd);
#endif

				/* Reset the chunk control block */
/* 186134 begin
				memset( (void *)&pChunkCB->slotMap[0], 0x00, sizeof( pChunkCB->slotMap ) ) ;
*/
				for(j=0; j < OSS_RAMBO_SLOT_BITMAP_ARRAY_SIZE; j++) {
					ossAtomicPoke(&(pChunkCB->slotMap[j]), 0);
				}
/* 186134 end */
				ossAtomicPoke(&pChunkCB->fillCount, 0) ;
				ossAtomicPoke(&pChunkCB->flushedCount, 0) ;
			}
#ifdef OSS_DEBUG1
			else {
//            printf("empty chunk index %lu\n", chunkIndex ) ;
				fprintf(dbgfd,"empty chunk index %lu\n", chunkIndex ) ;
				fflush(dbgfd);
			}
#endif
			/* Move to next chunk */
			chunkIndex = (chunkIndex + 1) % pRamboCB->l2.h2.maxChunks ;

		}
		/* Clear the auto-flusher status flags */
		pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
		pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
	}
#ifdef OSS_DEBUG
	printf("ossRamboFlushToStream: Returning with osserr = %d\n", osserr);
#endif

// pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ;

   OSS_ASSERT( OSS_OK == osserr ) ;

exit :
   #ifdef OSS_DEBUG1
     fclose(dbgfd);
   #endif
   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 ;

errorProcessingFailed :
// pRamboCB->l2.h2.status &= ~OSS_RAMBO_FLUSHING ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STOP_NOW ;
   pRamboCB->l2.h2.status &= ~OSS_RAMBO_AUTOF_STARTED ;
   osserr = OSS_ERR_RAMBO_IO_ERROR ;
   goto exit ;
}


/*******************************************************************************

   Name
      ossRamboStopFlusherAndDetach

   Function
      Tells flusher routine to stop flushing buffer and detach from buffer.

   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 ossRamboStopFlusherAndDetach( 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

/* 216026 begin */
/* Wait until there are no more writers to the shared memory */

	while ( ossAtomicPeek(&pRambo->writers) > 0 ) {
#ifdef OSS_DEBUG
		printf("ossRamboStopFlusher: waiting because there are still %d writers\n", ossAtomicPeek(&pRambo->writers));
#endif
		ossYield() ;
	}
/* 216026 end */

/* 186134 - Add a call to detach from the shared memory so it can be destroyed
            by the server */
   ossIPCMemDetach( pRambo->ipcMemHandle ) ;

   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
      ossRamboStopFlusher

   Function
      Tells flusher routine to stop flushing buffer.

   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 ossRamboStopFlusher( 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

   OSS_ASSERT( OSS_OK == osserr ) ;

exit:
   return osserr ;

errorInvalid :
   osserr = OSS_ERR_RAMBO_PTR ;
   goto exit ;

errorNotInitialized :
   osserr = OSS_ERR_RAMBO_NOT_INIT ;
   goto exit ;
}


/*******************************************************************************

   Name
      ossRamboDetach

   Function
      Detachs from the Rambo buffer.

   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 ossRamboDetach( 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 ;
   }

	/* Wait until there are no more writers to the shared memory */

 	while ( ossAtomicPeek(&pRambo->writers) > 0 ) {
#ifdef OSS_DEBUG
		printf("ossRamboStopFlusher: waiting because there are still %d writers\n", ossAtomicPeek(&pRambo->writers));
#endif
		ossYield() ;
	}


/* Detach from the shared memory so it can be destroyed
            by the server */
   ossIPCMemDetach( pRambo->ipcMemHandle ) ;

   OSS_ASSERT( OSS_OK == osserr ) ;

exit:
   return osserr ;

errorInvalid :
   osserr = OSS_ERR_INVALID ;
   goto exit ;

errorNotInitialized :
   osserr = OSS_ERR_RAMBO_NOT_INIT ;
   goto exit ;
}


/* bugzilla 64229 begin */
int ra_getShmAttachCount(ra_shm_handle_t *handle) {
   int attcnt = 0;
   OSSErr osserr = OSS_OK ;

   if (handle != NULL) {
	   osserr = ossIPCMemAttachCount( ((OSSRambo *)handle)->ipcMemHandle, &attcnt ) ;
	   if (osserr != OSS_OK) {
		   attcnt = -1;

#ifdef OSS_DEBUG
         ra_logServiceMessage(__FILE__, __LINE__,RA_DEBUG,"ra_getShmAttachCount:  error occurred %d", osserr);
#endif

	   }
   }
   else {
      attcnt = -1;
   }

   return attcnt;
}

ra_shm_err_t ra_setShmAttachCount(ra_shm_handle_t *handle, int attcnt) {

   OSSErr osserr = OSS_OK ;

   if (handle != NULL) {
	   ossIPCMemSetUserCount( ((OSSRambo *)handle)->ipcMemHandle, attcnt ) ;
   }
   else {
      osserr = OSS_ERR_INVALID;
   }

   return osserr;
}
/* bugzilla 64229 end */


/* 175248 begin */
#include "RAShm.h"
ra_shm_err_t ra_attachToShm(char *name, ra_shm_handle_t **handle) {
   *(OSSRambo **)handle = (OSSRambo *)malloc(sizeof(OSSRambo));
   return ossRamboAttach(name, *(OSSRambo **)handle, NULL, NULL);
}

ra_shm_err_t ra_stopFlushingShm(ra_shm_handle_t **handle) {
   OSSErr shmerr;
/* 175881 begin */
   if (*handle != NULL) {
      shmerr = ossRamboStopFlusherAndDetach(*(OSSRambo **)handle);
      free(*handle);
      *handle = NULL;
   }
   else
      shmerr = OSS_OK;
/* 175881 end */
   return (ra_shm_err_t)shmerr;
}
/* 175248 end */
  
OSSErr ra_stopFlusher(OSSRambo *handle) {
   OSSErr shmerr;

   if (handle != NULL) 
      shmerr = ossRamboStopFlusher(handle);
   else
      shmerr = OSS_OK;

   return shmerr;
}

