libcm/cmThread.c

#include "cmPrefix.h"
#include "cmGlobal.h"

#include "cmRpt.h"
#include "cmErr.h"
#include "cmMem.h"
#include "cmMallocDebug.h"

#include "cmThread.h"

#include <pthread.h>
#include <unistd.h>  // usleep
//#include <atomic_ops.h>

cmThreadH_t cmThreadNullHandle = {NULL};

enum
{
  kDoExitThFl  = 0x01,
  kDoPauseThFl = 0x02,
  kDoRunThFl   = 0x04
};

typedef struct
{
  cmErr_t        err;
  cmThreadFunc_t funcPtr;
  pthread_t      pthreadH;
  cmThStateId_t  state;
  void*          funcParam;
  unsigned       doFlags;
  unsigned       pauseMicroSecs;
  unsigned       waitMicroSecs;
} cmThThread_t;

cmThRC_t _cmThError( cmErr_t* err, cmThRC_t rc, int sysErr, const char* fmt, ... )
{ 
  va_list vl;
  va_start(vl,fmt);
  cmErrVSysMsg(err,rc,sysErr,fmt,vl);
  va_end(vl);
  return rc;
}

void _cmThThreadCleanUpCallback(void* t)
{
  ((cmThThread_t*)t)->state = kExitedThId;
}

void* _cmThThreadCallback(void* param)
{
  cmThThread_t* t = (cmThThread_t*)param;

	// set a clean up handler - this will be called when the 
	// thread terminates unexpectedly or pthread_cleanup_pop() is called.
	pthread_cleanup_push(_cmThThreadCleanUpCallback,t);

  while( cmIsFlag(t->doFlags,kDoExitThFl) == false )
  {

    if( t->state == kPausedThId )
    {
      usleep( t->pauseMicroSecs );

      if( cmIsFlag(t->doFlags,kDoRunThFl) )
      {
        t->doFlags = cmClrFlag(t->doFlags,kDoRunThFl);
        t->state   = kRunningThId;
      }
    }
    else
    {

      if( t->funcPtr(t->funcParam)==false )
        break;

      if( cmIsFlag(t->doFlags,kDoPauseThFl) )
      {
        t->doFlags = cmClrFlag(t->doFlags,kDoPauseThFl);
        t->state   = kPausedThId;        
      }
    }
  }

	pthread_cleanup_pop(1);
	
  pthread_exit(NULL);

  return t;
}

cmThThread_t* _cmThThreadFromHandle( cmThreadH_t h )
{
  cmThThread_t* tp = (cmThThread_t*)h.h;

  assert(tp != NULL);
  
  return tp->state==kNotInitThId ? NULL : tp;
}

cmThRC_t _cmThWaitForState( cmThThread_t* t, unsigned stateId )
{
  unsigned waitTimeMicroSecs = 0;

  while( t->state != stateId && waitTimeMicroSecs < t->waitMicroSecs )
  {
    //usleep( t->waitMicroSecs );
    usleep( 15000 );
    waitTimeMicroSecs += 15000; //t->waitMicroSecs;
  }

  return t->state==stateId ? kOkThRC :  kTimeOutThRC;
}

cmThRC_t cmThreadCreate(  cmThreadH_t* hPtr, cmThreadFunc_t funcPtr, void* funcParam, cmRpt_t* rpt )
{
  //pthread_attr_t attr;
  cmThRC_t         rc     = kOkThRC;
  cmThThread_t*    tp     = cmMemAllocZ( cmThThread_t, 1 );
  int            sysErr;

  cmErrSetup(&tp->err,rpt,"Thread");

  tp->funcPtr        = funcPtr;
  tp->funcParam      = funcParam;
  tp->state          = kPausedThId;
  tp->doFlags        = 0;
  tp->pauseMicroSecs = 50000;
  tp->waitMicroSecs  = 1000000;

  if((sysErr = pthread_create(&tp->pthreadH,NULL,_cmThThreadCallback, (void*)tp )) != 0 )
  {
    tp->state = kNotInitThId;
    rc = _cmThError(&tp->err,kCreateFailThRC,sysErr,"Thread create failed.");
  }

  hPtr->h = tp;

  return rc;
}


cmThRC_t cmThreadDestroy( cmThreadH_t* hPtr )
{
  cmThRC_t      rc     = kOkThRC;

  if( hPtr==NULL || cmThreadIsValid(*hPtr)==false )
    return rc;

  cmThThread_t* t      = _cmThThreadFromHandle(*hPtr );

  if( t == NULL )
    return kInvalidHandleThRC;

  // tell the thread to exit
  t->doFlags = cmSetFlag(t->doFlags,kDoExitThFl);

  // wait for the thread to exit and then deallocate the thread object
  if((rc = _cmThWaitForState(t,kExitedThId)) == kOkThRC )
  {
    cmMemFree(t);
    hPtr->h = NULL;
  }
  else
  {
    rc = _cmThError(&t->err,rc,0,"Thread timed out waiting for destroy.");
  }

  return rc;
}

cmThRC_t cmThreadPause(   cmThreadH_t h, unsigned cmdFlags )
{
  cmThRC_t    rc         = kOkThRC;
  bool        pauseFl    = cmIsFlag(cmdFlags,kPauseThFl);
  bool        waitFl     = cmIsFlag(cmdFlags,kWaitThFl);
  
  cmThThread_t* t        = _cmThThreadFromHandle(h);
  unsigned      waitId;

  if( t == NULL )
    return kInvalidHandleThRC;

  bool        isPausedFl = t->state == kPausedThId;
 
  if( isPausedFl == pauseFl )
    return kOkThRC;

  if( pauseFl )
  {
    t->doFlags = cmSetFlag(t->doFlags,kDoPauseThFl);
    waitId = kPausedThId;
  }
  else
  {
    t->doFlags = cmSetFlag(t->doFlags,kDoRunThFl);
    waitId = kRunningThId;
  }

  if( waitFl )
    rc = _cmThWaitForState(t,waitId);

  if( rc != kOkThRC )
    _cmThError(&t->err,rc,0,"Thread timed out waiting for '%s'.", pauseFl ? "pause" : "un-pause");
  
  return rc;
}

cmThStateId_t cmThreadState( cmThreadH_t h )
{
  cmThThread_t* tp = _cmThThreadFromHandle(h);

  if( tp == NULL )
    return kNotInitThId;

  return tp->state;
}

bool cmThreadIsValid( cmThreadH_t h )
{ return h.h != NULL; }

unsigned      cmThreadPauseTimeOutMicros( cmThreadH_t h )
{ 
  cmThThread_t* tp = _cmThThreadFromHandle(h);
  return tp->pauseMicroSecs;
}

void          cmThreadSetPauseTimeOutMicros( cmThreadH_t h, unsigned usecs )
{ 
  cmThThread_t* tp = _cmThThreadFromHandle(h);
  tp->pauseMicroSecs = usecs;
}

unsigned      cmThreadWaitTimeOutMicros(  cmThreadH_t h )
{ 
  cmThThread_t* tp = _cmThThreadFromHandle(h);
  return tp->waitMicroSecs;
}

void          cmThreadSetWaitTimeOutMicros( cmThreadH_t h, unsigned usecs )
{ 
  cmThThread_t* tp = _cmThThreadFromHandle(h);
  tp->waitMicroSecs = usecs;
}


bool _cmThreadTestCb( void* p )
{
  unsigned* ip = (unsigned*)p;
  ip[0]++;
  return true;
}


void cmThreadTest(cmRpt_t* rpt)
{
  cmThreadH_t th0;
  unsigned    val = 0;

  if( cmThreadCreate(&th0,_cmThreadTestCb,&val,rpt) == kOkThRC )
  {
    if( cmThreadPause(th0,0) != kOkThRC )
    {
      cmRptPrintf(rpt,"Thread start failed.\n");
      return;
    }

    char c = 0;

    cmRptPrintf(rpt,"o=print p=pause s=state q=quit\n");

    while( c != 'q' )
    {

      c = (char)fgetc(stdin);
      fflush(stdin);

      switch(c)
      {
        case 'o':
          cmRptPrintf(rpt,"val: 0x%x\n",val);
          break;

        case 's':
          cmRptPrintf(rpt,"state=%i\n",cmThreadState(th0));
          break;

        case 'p':
        {
          cmRC_t rc;
          if( cmThreadState(th0) == kPausedThId )
            rc = cmThreadPause(th0,kWaitThFl);
          else
            rc = cmThreadPause(th0,kPauseThFl|kWaitThFl);

          if( rc == kOkThRC )
            cmRptPrintf(rpt,"new state:%i\n", cmThreadState(th0));
          else
            cmRptPrintf(rpt,"cmThreadPause() failed.");

        }
        break;
        
        case 'q':
          break;

          //default:
          //cmRptPrintf(rpt,"Unknown:%c\n",c);
          
      }
    }

    if( cmThreadDestroy(&th0) != kOkThRC )
      cmRptPrintf(rpt,"Thread destroy failed.\n");
  }

}

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

typedef struct 
{
  cmErr_t         err;
  pthread_mutex_t mutex;
  pthread_cond_t  cvar;
} cmThreadMutex_t;

cmThreadMutexH_t kThreadMutexNULL = {NULL};


cmThreadMutex_t* _cmThreadMutexFromHandle( cmThreadMutexH_t h )
{
  cmThreadMutex_t* p = (cmThreadMutex_t*)h.h;
  assert(p != NULL);
  return p;
}

cmThRC_t cmThreadMutexCreate(   cmThreadMutexH_t* hPtr, cmRpt_t* rpt )
{
  int sysErr;
  cmThreadMutex_t* p = cmMemAllocZ( cmThreadMutex_t, 1 );

  cmErrSetup(&p->err,rpt,"Thread Mutex");

  if((sysErr = pthread_mutex_init(&p->mutex,NULL)) != 0 )
    return _cmThError(&p->err,kCreateFailThRC,sysErr,"Thread mutex create failed.");

  if((sysErr = pthread_cond_init(&p->cvar,NULL)) != 0 )
    return _cmThError(&p->err,kCreateFailThRC,sysErr,"Thread Condition var. create failed.");

  hPtr->h = p;
  return kOkThRC;
}

cmThRC_t cmThreadMutexDestroy(  cmThreadMutexH_t* hPtr )
{
  int sysErr;
  cmThreadMutex_t* p = _cmThreadMutexFromHandle(*hPtr);

  if( p == NULL )
    return kInvalidHandleThRC;

  if((sysErr = pthread_cond_destroy(&p->cvar)) != 0)
    return _cmThError(&p->err,kDestroyFailThRC,sysErr,"Thread condition var. destroy failed.");

  if((sysErr = pthread_mutex_destroy(&p->mutex)) != 0)
    return _cmThError(&p->err,kDestroyFailThRC,sysErr,"Thread mutex destroy failed.");


  cmMemFree(p);
  hPtr->h = NULL;

  return kOkThRC;
}

cmThRC_t cmThreadMutexTryLock(  cmThreadMutexH_t  h, bool* lockFlPtr )
{
  cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);

  if( p == NULL )
    return kInvalidHandleThRC;

  int sysErr = pthread_mutex_trylock(&p->mutex);

  switch(sysErr)
  {
    case EBUSY:
      *lockFlPtr = false;
      break;

    case 0:
      *lockFlPtr = true;
      break;

    default:
      return _cmThError(&p->err,kLockFailThRC,sysErr,"Thread mutex try-lock failed.");;
  }

  return kOkThRC;
}

cmThRC_t cmThreadMutexLock(     cmThreadMutexH_t  h )
{
  cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);

  if( p == NULL )
    return kInvalidHandleThRC;

  int sysErr = pthread_mutex_lock(&p->mutex);

  if( sysErr == 0 )
    return kOkThRC;

  return _cmThError(&p->err,kLockFailThRC,sysErr,"Thread mutex lock failed.");
}

cmThRC_t cmThreadMutexUnlock(   cmThreadMutexH_t  h )
{
  cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);

  if( p == NULL )
    return kInvalidHandleThRC;

  int sysErr = pthread_mutex_unlock(&p->mutex);

  if( sysErr == 0 )
    return kOkThRC;

  return _cmThError(&p->err,kUnlockFailThRC,sysErr,"Thread mutex unlock failed.");
}

bool cmThreadMutexIsValid( cmThreadMutexH_t h )
{ return h.h != NULL; }

cmThRC_t cmThreadMutexWaitOnCondVar( cmThreadMutexH_t h, bool lockFl )
{
  cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);

  if( p == NULL )
    return kInvalidHandleThRC;

  int sysErr;
  
  if( lockFl )
    if( (sysErr=pthread_mutex_lock(&p->mutex)) != 0 )
      _cmThError(&p->err,kLockFailThRC,sysErr,"Thread lock failed on cond. var. wait.");

  if((sysErr = pthread_cond_wait(&p->cvar,&p->mutex)) != 0 )
    _cmThError(&p->err,kCVarWaitFailThRC,sysErr,"Thread cond. var. wait failed.");
        
  return kOkThRC;
}

cmThRC_t cmThreadMutexSignalCondVar( cmThreadMutexH_t h )
{
  int sysErr;
  cmThreadMutex_t* p = _cmThreadMutexFromHandle(h);

  if( p == NULL )
    return kInvalidHandleThRC;

  if((sysErr = pthread_cond_signal(&p->cvar)) != 0 )
    return _cmThError(&p->err,kCVarSignalFailThRC,sysErr,"Thread cond. var. signal failed.");

  return kOkThRC;
}


//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

cmTsQueueH_t cmTsQueueNullHandle = { NULL };

enum { cmTsQueueBufCnt = 2 };


typedef struct
{
  unsigned  allocCnt;  // count of bytes allocated for the buffer 
  unsigned  fullCnt;   // count of bytes used in the buffer
  char*     basePtr;   // base of buffer memory
  unsigned* msgPtr;    // pointer to first msg 
  unsigned  msgCnt;
} cmTsQueueBuf;

typedef struct
{
  cmThreadMutexH_t mutexH;
  cmTsQueueBuf     bufArray[cmTsQueueBufCnt];
  unsigned         inBufIdx;
  unsigned         outBufIdx;
  char*            memPtr;
  cmTsQueueCb_t    cbFunc;
  void*            userCbPtr;
} cmTsQueue_t;

cmTsQueue_t* _cmTsQueueFromHandle( cmTsQueueH_t h )
{
  cmTsQueue_t* p = h.h;
  assert(p != NULL);
  return p;
}

cmThRC_t _cmTsQueueDestroy( cmTsQueue_t* p )
{
  cmThRC_t rc;

  if( p == NULL )
    return kInvalidHandleThRC;

  if( p->mutexH.h != NULL )
    if((rc = cmThreadMutexDestroy(&p->mutexH)) != kOkThRC )
      return rc;

  if( p->memPtr != NULL )
    cmMemPtrFree(&p->memPtr);


  cmMemPtrFree(&p);


  return kOkThRC;
}


cmThRC_t   cmTsQueueCreate(     cmTsQueueH_t* hPtr, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* userCbPtr, cmRpt_t* rpt )
{
  cmTsQueue_t* p = cmMemAllocZ( cmTsQueue_t, 1 );
  unsigned   i;

  if( cmThreadMutexCreate(&p->mutexH,rpt) != kOkThRC )
    goto errLabel;  

  p->memPtr    = cmMemAllocZ( char, bufByteCnt*cmTsQueueBufCnt );
  p->outBufIdx = 0;
  p->inBufIdx  = 1;
  p->cbFunc    = cbFunc;
  p->userCbPtr = userCbPtr;

  for(i=0; i<cmTsQueueBufCnt; ++i)
  {
    p->bufArray[i].allocCnt = bufByteCnt;
    p->bufArray[i].fullCnt  = 0;
    p->bufArray[i].basePtr  = p->memPtr + (i*bufByteCnt);
    p->bufArray[i].msgPtr   = NULL;
    p->bufArray[i].msgCnt   = 0;
  }

  hPtr->h = p;

  return kOkThRC;
  
  errLabel:
  
  _cmTsQueueDestroy(p);

  return kCreateFailThRC;
}


cmThRC_t   cmTsQueueDestroy(    cmTsQueueH_t*  hPtr )
{
  cmThRC_t rc = kOkThRC;

  if( (hPtr != NULL) && cmTsQueueIsValid(*hPtr))
    if((rc = _cmTsQueueDestroy(_cmTsQueueFromHandle(*hPtr))) == kOkThRC )
      hPtr->h = NULL;

  return rc;
}

cmThRC_t   cmTsQueueSetCallback( cmTsQueueH_t h, cmTsQueueCb_t cbFunc, void* cbArg )
{
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  p->cbFunc    = cbFunc;
  p->userCbPtr = cbArg;
  return kOkThRC;
}

unsigned cmTsQueueAllocByteCount( cmTsQueueH_t h )
{
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  unsigned     n = 0;

  if( cmThreadMutexLock(p->mutexH) == kOkThRC )
  {
    n = p->bufArray[ p->inBufIdx ].allocCnt;
    cmThreadMutexUnlock(p->mutexH);
  }
  return n;
}

unsigned cmTsQueueAvailByteCount( cmTsQueueH_t h )
{
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  unsigned     n = 0;

  if(cmThreadMutexLock(p->mutexH) == kOkThRC )
  {
    n =  p->bufArray[ p->inBufIdx ].allocCnt - p->bufArray[ p->inBufIdx].fullCnt;
    cmThreadMutexUnlock(p->mutexH);
  }
  return n;
}

cmThRC_t   _cmTsQueueEnqueueMsg( cmTsQueueH_t  h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
{
  cmThRC_t     rc;
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);


  if( p == NULL )
    return kInvalidHandleThRC;

  // lock the mutex
  if((rc = cmThreadMutexLock(p->mutexH)) == kOkThRC )
  {
    
    cmTsQueueBuf*  b          = p->bufArray + p->inBufIdx;            // ptr to buf recd
    const char*    ep         = b->basePtr  + b->allocCnt;            // end of buf data space
    unsigned      *mp         = (unsigned*)(b->basePtr + b->fullCnt); // ptr to size of new msg space
    char*          dp         = (char*)(mp+1);                        // ptr to data area of new msg space
    unsigned       ttlByteCnt = 0;                                    // track size of msg data
    unsigned       i          = 0;   

    // get the total size of the msg
    for(i=0; i<arrayCnt; ++i)
      ttlByteCnt += msgByteCntArray[i];

    // if the msg is too big for the queue buf
    if( dp + ttlByteCnt > ep )
      rc = kBufFullThRC;
    else
    {
      // for each segment of the incoming msg
      for(i=0; i<arrayCnt; ++i)
      {
        // get the size of the segment
        unsigned n = msgByteCntArray[i]; 

        // copy in the segment
        memcpy(dp,msgPtrArray[i],n);
      
        dp   += n; //
      }

      assert(dp <= ep );

      // write the size ofthe msg into the buffer
      *mp = ttlByteCnt;

      // update the pointer to the first msg
      if( b->msgPtr == NULL )
        b->msgPtr = mp;

      // track the count of msgs in this buffer
      ++b->msgCnt;

      // update fullCnt last since dequeue uses fullCnt to
      // notice that a msg may be waiting
      b->fullCnt += sizeof(unsigned) + ttlByteCnt;    
    }

    cmThreadMutexUnlock(p->mutexH);
  }

  return rc;
}

cmThRC_t   cmTsQueueEnqueueSegMsg( cmTsQueueH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
{ return _cmTsQueueEnqueueMsg(h,msgPtrArray,msgByteCntArray,arrayCnt); }

cmThRC_t   cmTsQueueEnqueueMsg( cmTsQueueH_t h, const void* dataPtr, unsigned byteCnt )
{ 
  const void*  msgPtrArray[]     = { dataPtr };
  unsigned     msgByteCntArray[] = { byteCnt };

  return _cmTsQueueEnqueueMsg(h,msgPtrArray,msgByteCntArray,1);
}

cmThRC_t   cmTsQueueEnqueueIdMsg( cmTsQueueH_t h, unsigned id, const void* dataPtr, unsigned byteCnt )
{   
  const void* msgPtrArray[]     = { &id, dataPtr };
  unsigned    msgByteCntArray[] = { sizeof(id), byteCnt };

  return _cmTsQueueEnqueueMsg(h,msgPtrArray,msgByteCntArray,2);
}

cmThRC_t   _cmTsQueueDequeueMsg( cmTsQueue_t* p, void* retBuf, unsigned refBufByteCnt )
{
  cmTsQueueBuf* b = p->bufArray + p->outBufIdx;

  // if the output buffer is empty - there is nothing to do
  if( b->fullCnt == 0 )
    return kBufEmptyThRC;

  assert( b->msgPtr != NULL );

  // get the output msg size and data
  unsigned msgByteCnt = *b->msgPtr;
  char*    msgDataPtr = (char*)(b->msgPtr + 1);

  // transmit the msg via a callback
  if( retBuf == NULL && p->cbFunc != NULL )
    p->cbFunc(p->userCbPtr,msgByteCnt,msgDataPtr);
  else
  {
    // retBuf may be NULL if the func is being used by cmTsQueueDequeueByteCount()
    if( retBuf == NULL || msgByteCnt > refBufByteCnt )
      return kBufTooSmallThRC;

    // copy the msg to a buffer
    if( retBuf != NULL )
      memcpy(retBuf,msgDataPtr,msgByteCnt);

  }
  
  // update the buffer
  b->fullCnt -= sizeof(unsigned) + msgByteCnt;  
  b->msgPtr   = (unsigned*)(msgDataPtr + msgByteCnt);
  --(b->msgCnt);

  if( b->fullCnt == 0 )
  {
    assert(b->msgCnt == 0);
    b->msgPtr = NULL;
  }
  
  return kOkThRC;

}

cmThRC_t   cmTsQueueDequeueMsg( cmTsQueueH_t  h,  void* retBuf, unsigned refBufByteCnt )
{
  cmThRC_t rc;
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  
  if( p == NULL )
    return kInvalidHandleThRC;

  // dequeue the next msg from the current output buffer
  if((rc =_cmTsQueueDequeueMsg( p, retBuf, refBufByteCnt )) != kBufEmptyThRC )
    return rc;
  
  // the current output buffer was empty 

  cmTsQueueBuf* b = p->bufArray + p->inBufIdx;

  // if the input buffer has msg's ...
  if( b->fullCnt > 0 )
  {
    bool lockFl = false;

    //  ...attempt to lock the mutex ...
    if( (cmThreadMutexTryLock(p->mutexH,&lockFl) == kOkThRC) && lockFl )
    {
      // ... swap the input and the output buffers ...
      unsigned tmp = p->inBufIdx;
      p->inBufIdx  = p->outBufIdx;
      p->outBufIdx = tmp;

      // .. unlock the mutex 
      cmThreadMutexUnlock(p->mutexH);

      // ... and dequeue the first msg from the new output buffer
      rc = _cmTsQueueDequeueMsg( p, retBuf, refBufByteCnt );      
    }
    
  }

  return rc;
  
}

bool     cmTsQueueMsgWaiting( cmTsQueueH_t  h )
{
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  
  if( p == NULL )
    return false;

  if( p->bufArray[p->outBufIdx].fullCnt )
    return true;

  return p->bufArray[p->inBufIdx].fullCnt > 0;    
}

unsigned cmTsQueueDequeueMsgByteCount( cmTsQueueH_t h )
{
  cmTsQueue_t* p = _cmTsQueueFromHandle(h);
  
  if( p == NULL )
    return 0;

  // if output msgs are available then the msgPtr points to the size of the msg
  if( p->bufArray[p->outBufIdx].fullCnt )
    return *(p->bufArray[p->outBufIdx].msgPtr);

  // no msgs are waiting in the output buffer
  
  // force the buffers to swap - returns kBufEmptyThRC if there are
  // still no msgs waiting after the swap (the input buf was also empty)
  if( cmTsQueueDequeueMsg(h,NULL,0) == kBufTooSmallThRC )
  {
    // the buffers swapped so there must be msg waiting
    assert( p->bufArray[p->outBufIdx].fullCnt );

    return *(p->bufArray[p->outBufIdx].msgPtr);
  }

  return 0;
}

bool cmTsQueueIsValid( cmTsQueueH_t h )
{ return h.h != NULL; }



//--------------------------------------------------------------------------------------------------
//--------------------------------------------------------------------------------------------------
//--------------------------------------------------------------------------------------------------
#ifdef NOT_DEF
enum { kThBufCnt=2 };

typedef struct
{
  char*             buf;
  volatile unsigned ii;
  volatile unsigned oi;
} cmThBuf_t;

typedef struct
{
  cmErr_t           err;
  cmThBuf_t         a[kThBufCnt];
  volatile unsigned ibi;
  unsigned          bn;
  cmTsQueueCb_t     cbFunc;
  void*             cbArg;
} cmTs1p1c_t;

cmTs1p1c_t* _cmTs1p1cHandleToPtr( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p = (cmTs1p1c_t*)h.h;
  assert( p != NULL );
  return p;
}

cmThRC_t _cmTs1p1cDestroy( cmTs1p1c_t* p )
{
  unsigned i;
  for(i=0; i<kThBufCnt; ++i)
    cmMemFree(p->a[i].buf);

  cmMemFree(p);

  return kOkThRC;
}

cmThRC_t   cmTs1p1cCreate( cmTs1p1cH_t* hPtr, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* cbArg, cmRpt_t* rpt )
{
  cmThRC_t rc;
  if((rc = cmTs1p1cDestroy(hPtr)) != kOkThRC )
    return rc;
  
  unsigned    i;
  cmTs1p1c_t* p = cmMemAllocZ(cmTs1p1c_t,1);
  cmErrSetup(&p->err,rpt,"TS 1p1c Queue");
  for(i=0; i<kThBufCnt; ++i)
  {
    p->a[i].buf = cmMemAllocZ(char,bufByteCnt);
    p->a[i].ii  = 0;
    p->a[i].oi  = bufByteCnt;
  }

  p->ibi        = 0;
  p->bn         = bufByteCnt;
  p->cbFunc     = cbFunc;
  p->cbArg = cbArg;

  hPtr->h = p;

  return rc;
}

cmThRC_t cmTs1p1cDestroy( cmTs1p1cH_t* hp )
{
  cmThRC_t rc = kOkThRC;

  if( hp == NULL || cmTs1p1cIsValid(*hp)==false )
    return kOkThRC;

  cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(*hp);
  if(( rc = _cmTs1p1cDestroy(p)) != kOkThRC )
    return rc;
  
  hp->h = NULL;
  return rc;
}

cmThRC_t   cmTs1p1cEnqueueSegMsg( cmTs1p1cH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
{
  cmThRC_t    rc = kOkThRC;
  unsigned    mn = 0;
  unsigned    i;
  cmTs1p1c_t* p  = _cmTs1p1cHandleToPtr(h);
  cmThBuf_t*  ib = p->a + p->ibi;
  

  // get the total count of bytes for this msg
  for(i=0; i<arrayCnt; ++i)
    mn += msgByteCntArray[i];

  unsigned dn = mn + sizeof(unsigned);

  // if the message is too big for even an empty buffer
  if( dn > p->bn )
    return cmErrMsg(&p->err,kBufFullThRC,"A msg containing %i bytes will never be able to fit in a queue with an empty size of %i bytes.",dn,p->bn);

  // if the msg won't fit in the current input buffer then try swapping buffers.
  if( ib->ii + dn > p->bn )
  {
    // get the current output buffer
    cmThBuf_t* ob = p->a + (p->ibi==0 ? 1 : 0);

    // Empty buffers will be set such that:  oi==bn and ii==0.
    //
    // Note that setting ii to 0 in an output buffer is the last operation
    // performed on an empty output buffer. ii==0 is therefore the 
    // signal that an output buffer can be reused for input.

    // if the output buffer is not empty - then an overflow occurred
    if( ob->ii != 0 )
      return cmErrMsg(&p->err,kBufFullThRC,"The msq queue cannot accept a %i byte msg into %i bytes.",dn, p->bn - ib->ii);

    // setup the initial output location of the new output buffer
    ib->oi = 0;    
                    
    // swap buffers 
    p->ibi = (p->ibi + 1) % kThBufCnt; 

    // get the new input buffer 
    ib     = ob;                       
  }
  
  // get a pointer to the base of the write location
  char* dp = ib->buf + ib->ii;

  // write the length of the message
  *(unsigned*)dp = mn;
  dp += sizeof(unsigned);

  // write the body of the message
  for(i=0; i<arrayCnt; ++i)
  {
    memcpy(dp,msgPtrArray[i],msgByteCntArray[i]);
    dp += msgByteCntArray[i];
  }

  // this MUST be executed last - we'll use 'dp' in the calculation
  // (even though ib->ii += dn would be more straight forward way 
  // to accomplish the same thing) to prevent the optimizer from 
  // moving the assignment prior to the for loop.
  ib->ii += dp - (ib->buf + ib->ii);

  return rc;
}

cmThRC_t   cmTs1p1cEnqueueMsg( cmTsQueueH_t  h, const void* dataPtr, unsigned byteCnt )
{ return cmTs1p1cEnqueueSegMsg(h,&dataPtr,&byteCnt,1); }

unsigned   cmTs1p1cAllocByteCount( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  return p->bn;
}

unsigned   cmTs1p1cAvailByteCount( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  return p->bn - p->a[ p->ibi ].ii;
}
cmThRC_t   cmTs1p1cDequeueMsg( cmTs1p1cH_t  h,  void* dataPtr, unsigned byteCnt )
{
  cmThRC_t    rc = kOkThRC;
  cmTs1p1c_t* p  = _cmTs1p1cHandleToPtr(h);
  cmThBuf_t*  ob = p->a + (p->ibi == 0 ? 1 : 0);

  // empty buffers always are set to: oi==bn && ii==0
  if( ob->oi >= ob->ii )
    return kBufEmptyThRC;

  // get the size of the msg
  unsigned mn = *(unsigned*)(ob->buf + ob->oi);

  // increment the current output location to the msg body
  ob->oi += sizeof(unsigned);

  // copy or send the msg 
  if( dataPtr != NULL )
  {
    if( byteCnt < mn )
      return cmErrMsg(&p->err,kBufTooSmallThRC,"The return buffer constains too few bytes (%i) to contain %i bytes.",byteCnt,mn);

    memcpy(dataPtr, ob->buf + ob->oi, mn);    
  }
  else
  {
    p->cbFunc(p->cbArg, mn, ob->buf + ob->oi );
  }

  ob->oi += mn;
  
  // if we are reading correctly ob->oi should land 
  // exactly on ob->ii when the buffer is empty
  assert( ob->oi <= ob->ii );

  // if the buffer is empty
  if( ob->oi == ob->ii )
  {
    ob->oi = p->bn; // mark the buffer as empty
    ob->ii = 0;     //
  }

  return rc;
}


unsigned cmTs1p1cDequeueMsgByteCount( cmTsQueueH_t h )
{
  cmTs1p1c_t* p  = _cmTs1p1cHandleToPtr(h);
  cmThBuf_t*  ob = p->a + (p->ibi == 0 ? 1 : 0);

  // empty buffers always are set to: oi==bn && ii==0
  if( ob->oi >= ob->ii )
    return 0;

  // get the size of the msg
  return  *(unsigned*)(ob->buf + ob->oi);
}


bool     cmTs1p1cMsgWaiting( cmTsQueueH_t h )
{  return cmTs1p1cDequeueMsgByteCount(h) > 0;  }

bool cmTs1p1cIsValid( cmTs1p1cH_t h )
{ return h.h != NULL; }
#endif

//--------------------------------------------------------------------------------------------------
//--------------------------------------------------------------------------------------------------
//--------------------------------------------------------------------------------------------------

typedef struct
{
  volatile unsigned ii;
  cmErr_t           err;
  char*             buf;
  unsigned          bn;
  cmTsQueueCb_t     cbFunc;
  void*             cbArg;
  volatile unsigned oi;
} cmTs1p1c_t;

cmTs1p1c_t* _cmTs1p1cHandleToPtr( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p = (cmTs1p1c_t*)h.h;
  assert( p != NULL );
  return p;
}

cmThRC_t   cmTs1p1cCreate( cmTs1p1cH_t* hPtr, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* cbArg, cmRpt_t* rpt )
{
  cmThRC_t rc;
  if((rc = cmTs1p1cDestroy(hPtr)) != kOkThRC )
    return rc;
  
  cmTs1p1c_t* p = cmMemAllocZ(cmTs1p1c_t,1);
  cmErrSetup(&p->err,rpt,"1p1c Queue");
  p->buf        = cmMemAllocZ(char,bufByteCnt+sizeof(unsigned));
  p->ii         = 0;
  p->oi         = 0;
  p->bn         = bufByteCnt;
  p->cbFunc     = cbFunc;
  p->cbArg = cbArg;
  hPtr->h = p;

  return rc;
}

cmThRC_t cmTs1p1cDestroy( cmTs1p1cH_t* hp )
{
  cmThRC_t rc = kOkThRC;

  if( hp == NULL || cmTs1p1cIsValid(*hp)==false )
    return kOkThRC;

  cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(*hp);
  
  cmMemFree(p->buf);
  cmMemFree(p);
  
  hp->h = NULL;
  return rc;
}

cmThRC_t   cmTs1p1cEnqueueSegMsg( cmTs1p1cH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
{
  cmThRC_t    rc = kOkThRC;
  unsigned    mn = 0;
  unsigned    i;
  cmTs1p1c_t* p  = _cmTs1p1cHandleToPtr(h);

  // get the total count of bytes for this msg
  for(i=0; i<arrayCnt; ++i)
    mn += msgByteCntArray[i];

  int dn = mn + sizeof(unsigned);
  int oi = p->oi;
  int bi = p->ii;   // 'bi' is the idx of the leftmost cell which can be written 
  int en = p->bn;   // 'en' is the idx of the cell just to the right of the rightmost cell that can be written

  // note: If 'oi' marks the rightmost location then 'en' must be set
  // one cell to the left of 'oi', because 'ii' can never be allowed to
  // advance onto 'oi' - because 'oi'=='ii' marks an empty (NOT a full)
  // queue. 
  //
  // If 'bn' marks the rightmost location then 'ii' can advance onto 'bn'
  // beause the true queue length is bn+1.

  // if we need to wrap
  if( en-bi < dn && oi<=bi )
  {
    bi = 0;
    en = oi - 1;  // note if oi==0 then en is negative - see note above re: oi==ii
    assert( p->ii>=0 && p->ii <= p->bn );
    *(unsigned*)(p->buf + p->ii) = cmInvalidIdx; // mark the wrap location
  }

  // if oi is between ii and bn
  if( oi > bi )
    en = oi - 1;  // never allow ii to advance onto oi - see note above

  // if the msg won't fit
  if( en - bi < dn )
    return cmErrMsg(&p->err,kBufFullThRC,"%i consecutive bytes is not available in the queue.",dn); 

  // set the msg byte count - the msg byte cnt precedes the msg body
  char* dp = p->buf + bi;
  *(unsigned*)dp = dn - sizeof(unsigned);
  dp += sizeof(unsigned);
  
  // copy the msg into the buffer
  for(i=0,dn=0; i<arrayCnt; ++i)
  {
    memcpy(dp,msgPtrArray[i],msgByteCntArray[i]);
    dp += msgByteCntArray[i];
    dn += msgByteCntArray[i];
  }

  // incrementing p->ii must occur last - the unnecessary accumulation
  // of dn in the above loop is intended to prevent this line from
  // begin moved before the copy loop.
  p->ii = bi + dn + sizeof(unsigned);

  assert( p->ii >= 0 && p->ii <= p->bn);

  return rc;
}

cmThRC_t   cmTs1p1cEnqueueMsg( cmTs1p1cH_t  h, const void* dataPtr, unsigned byteCnt )
{ return cmTs1p1cEnqueueSegMsg(h,&dataPtr,&byteCnt,1); }

unsigned   cmTs1p1cAllocByteCount( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  return p->bn;
}

unsigned   cmTs1p1cAvailByteCount( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p = _cmTs1p1cHandleToPtr(h);
  unsigned oi = p->oi;
  unsigned ii = p->ii;
  return oi < ii ? p->bn - ii + oi : oi - ii;
}

unsigned _cmTs1p1cDequeueMsgByteCount( cmTs1p1c_t* p )
{
  // if the buffer is empty
  if( p->ii == p->oi )
    return 0;

  // get the length of the next msg
  unsigned mn = *(unsigned*)(p->buf + p->oi);

  // if the msg length is cmInvalidIdx ...
  if( mn == cmInvalidIdx )
  {
    p->oi = 0;  // ... wrap to buf begin and try again

    return _cmTs1p1cDequeueMsgByteCount(p);
  }
  
  return mn;
}

cmThRC_t   cmTs1p1cDequeueMsg( cmTs1p1cH_t  h,  void* dataPtr, unsigned byteCnt )
{
  cmThRC_t    rc = kOkThRC;
  cmTs1p1c_t* p  = _cmTs1p1cHandleToPtr(h);

  unsigned mn;

  if((mn = _cmTs1p1cDequeueMsgByteCount(p)) == 0 )
    return kBufEmptyThRC;

  void* mp = p->buf + p->oi + sizeof(unsigned);

  if( dataPtr != NULL )
  {
    if( byteCnt < mn )
      return cmErrMsg(&p->err,kBufTooSmallThRC,"The return buffer constains too few bytes (%i) to contain %i bytes.",byteCnt,mn);
      
    memcpy(dataPtr,mp,mn);
  }
  else
  {
    p->cbFunc(p->cbArg,mn,mp);
  }

  p->oi += mn + sizeof(unsigned);
  
  return rc;
}


unsigned cmTs1p1cDequeueMsgByteCount( cmTs1p1cH_t h )
{
  cmTs1p1c_t* p  = _cmTs1p1cHandleToPtr(h);
  return _cmTs1p1cDequeueMsgByteCount(p);
}

bool     cmTs1p1cMsgWaiting( cmTs1p1cH_t h )
{  return cmTs1p1cDequeueMsgByteCount(h) > 0;  }

bool cmTs1p1cIsValid( cmTs1p1cH_t h )
{ return h.h != NULL; }

//============================================================================================================================


bool     cmThIntCAS(   int*      addr, int      old, int      new )
{ return __sync_bool_compare_and_swap(addr,old,new); }

bool     cmThUIntCAS(  unsigned* addr, unsigned old, unsigned new )
{ return  __sync_bool_compare_and_swap(addr,old,new); }

bool     cmThFloatCAS( float*    addr, float    old, float    new )
{ return  __sync_bool_compare_and_swap((unsigned*)addr, *(unsigned*)(&old),*(unsigned*)(&new)); }

bool     cmThPtrCAS(   void*    addr, void*    old, void*    neww )
{
#ifdef OS_64
  return  __sync_bool_compare_and_swap((long long*)addr, (long long)old, (long long)neww); 
#else
  return  __sync_bool_compare_and_swap((int*)addr,(int)old,(int)neww); 
#endif
}



void     cmThIntIncr(  int*      addr, int      incr )
{
  // ... could also use __sync_add_and_fetch() ...
  __sync_fetch_and_add(addr,incr);
}

void     cmThUIntIncr( unsigned* addr, unsigned incr )
{
  __sync_fetch_and_add(addr,incr);
}

void     cmThFloatIncr(float*    addr, float    incr )
{

  float old,new;
  do
  {
    old = *addr;
    new = old + incr;
  }while( cmThFloatCAS(addr,old,new)==0 );
}


void     cmThIntDecr(  int*  addr, int      decr )
{
  __sync_fetch_and_sub(addr,decr);
}

void     cmThUIntDecr( unsigned* addr, unsigned decr )
{
  __sync_fetch_and_sub(addr,decr);
}

void     cmThFloatDecr(float*    addr, float    decr )
{

  float old,new;
  do
  {
    old = *addr;
    new = old - decr;
  }while( cmThFloatCAS(addr,old,new)==0 );

}



//============================================================================================================================
//
//
typedef  pthread_t cmThreadId_t;

typedef struct
{
  cmThreadId_t id;    // id of this thread as returned by pthread_self()
  char*        buf;   // buf[bn]
  int          ii;    // input index
  int          oi;    // output index  (oi==ii == empty buffer)
} cmTsBuf_t;

// msg header - which is actually written AFTER the msg it is associated with
typedef struct cmTsHdr_str
{
  int                 mn;     // length of the msg
  int                 ai;     // buffer index
  struct cmTsHdr_str* link;   // pointer to next msg
} cmTsHdr_t;

typedef struct
{
  cmErr_t       err;
  int           bn;        // bytes per thread buffer  
  cmTsBuf_t*    a;         // a[an] buffer array
  unsigned      an;        // length of a[] - one buffer per thread
  cmTsQueueCb_t cbFunc;
  void*         cbArg;
  cmTsHdr_t*    ilp;        // prev msg hdr record
  cmTsHdr_t*    olp;        // prev msg hdr record (wait for olp->link to be set to go to next record)
} cmTsMp1c_t;

cmTsMp1cH_t cmTsMp1cNullHandle = cmSTATIC_NULL_HANDLE;

void _cmTsMp1cPrint( cmTsMp1c_t* p )
{
  unsigned i;
  for(i=0; i<p->an; ++i)
    printf("%2i ii:%3i oi:%3i\n",i,p->a[i].ii,p->a[i].oi);
}

cmTsMp1c_t* _cmTsMp1cHandleToPtr( cmTsMp1cH_t h )
{
  cmTsMp1c_t* p = (cmTsMp1c_t*)h.h;
  assert(p != NULL);
  return p;
}

unsigned _cmTsMp1cBufIndex( cmTsMp1c_t* p, cmThreadId_t id )
{
  unsigned i;
  for(i=0; i<p->an; ++i)
    if( p->a[i].id == id )
      return i;

  p->an       = i+1;
  p->a        = cmMemResizePZ(cmTsBuf_t,p->a,p->an);
  p->a[i].buf = cmMemAllocZ(char,p->bn);
  p->a[i].id  = id;
 
  return i;        
}

cmThRC_t   cmTsMp1cDestroy(    cmTsMp1cH_t* hp )
{
  if( hp == NULL || cmTsMp1cIsValid(*hp) == false )
    return kOkThRC;

  cmTsMp1c_t* p = _cmTsMp1cHandleToPtr(*hp);
  unsigned i;
  for(i=0; i<p->an; ++i)
    cmMemFree(p->a[i].buf);

  cmMemPtrFree(&p->a);
  cmMemFree(p);

  hp->h = NULL;

  return kOkThRC;
}

cmThRC_t   cmTsMp1cCreate(  cmTsMp1cH_t* hp, unsigned bufByteCnt, cmTsQueueCb_t cbFunc, void* cbArg, cmRpt_t* rpt )
{
  cmThRC_t rc;
  if((rc = cmTsMp1cDestroy(hp)) != kOkThRC )
    return rc;

  cmTsMp1c_t* p = cmMemAllocZ(cmTsMp1c_t,1);

  cmErrSetup(&p->err,rpt,"TsMp1c Queue");

  p->a          = NULL;
  p->an         = 0;
  p->bn         = bufByteCnt;
  p->cbFunc     = cbFunc;
  p->cbArg = cbArg;
  p->ilp        = NULL;
  p->olp        = NULL;

  hp->h = p;
  return rc;
}

void cmTsMp1cSetCbFunc( cmTsMp1cH_t h,  cmTsQueueCb_t cbFunc, void* cbArg )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  p->cbFunc     = cbFunc;
  p->cbArg = cbArg;
}

cmTsQueueCb_t cmTsMp1cCbFunc( cmTsMp1cH_t h )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  return p->cbFunc;
}

void* cmTsMp1cCbArg( cmTsMp1cH_t h )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  return p->cbArg;
}

//#define CAS(addr,old,new) __sync_bool_compare_and_swap(addr,old,new)
//#define CAS(addr,old,neww) cmThPtrCAS(addr,old,neww)

cmThRC_t   cmTsMp1cEnqueueSegMsg( cmTsMp1cH_t h, const void* msgPtrArray[], unsigned msgByteCntArray[], unsigned arrayCnt )
{
  cmThRC_t    rc = kOkThRC;
  unsigned    mn = 0;
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  unsigned    ai = _cmTsMp1cBufIndex( p, pthread_self() );
  cmTsBuf_t*  b  = p->a + ai;
  int         i,bi,ei;
  cmTsHdr_t   hdr;

  // Use a stored oi for the duration of this function.
  // b->oi may be changed by the dequeue thread but storing it here
  // at least prevents it from changing during the course of the this function.
  // Note: b->oi is only used to check for buffer full.  Even if it changes
  // it would only mean that more bytes were available than calculated based
  // on the stored value.  A low estimate of the actual bytes available is 
  // never unsafe.
  volatile int    oi = b->oi; 

  // get the total count of bytes for this msg
  for(i=0; i<arrayCnt; ++i)
    mn += msgByteCntArray[i];

  // dn = count of msg bytes + count of header bytes
  int dn = mn + sizeof(hdr);

  // if oi is ahead of ii in the queue then we must write
  // in the area between ii and oi
  if( oi > b->ii )
  {
    ei = oi-1; // (never allow ii to equal oi (that's the empty condition))
    bi = b->ii;
  }
  else // otherwise oi is same or before ii in the queue and we have the option to wrap
  {
    // if the new msg will not fit at the end of the queue ....
    if( b->ii + dn > p->bn )
    {
      bi = 0;     // ... then wrap to the beginning
      ei = oi-1;  // (never allow ii to equal oi (that's the empty condition))
     }
    else 
    {
      ei = p->bn; // otherwise write at the current location
      bi = b->ii;
    }
  }
  
  if( bi + dn > ei )
    return cmErrMsg(&p->err,kBufFullThRC,"%i consecutive bytes is not available in the queue.",dn);

  char* dp = b->buf + bi;

  // write the msg
  for(i=0; i<arrayCnt; ++i)
  {
    memcpy(dp,msgPtrArray[i],msgByteCntArray[i]);
    dp += msgByteCntArray[i];
  }

  // setup the msg header
  hdr.ai   = ai;
  hdr.mn   = mn;
  hdr.link = NULL;

  // write the msg header (following the msg body in memory)
  cmTsHdr_t* hp = (cmTsHdr_t*)dp;
  memcpy(hp,&hdr,sizeof(hdr));

  // increment the buffers input index
  b->ii = bi + dn;

  // update the link list head to point to this msg hdr
  cmTsHdr_t* old_hp, *new_hp;
  do
  {
    old_hp = p->ilp;
    new_hp = hp;
  }while(!cmThPtrCAS(&p->ilp,old_hp,new_hp));

  // link the prev recd to this recd
  if( old_hp != NULL )
    old_hp->link = hp;

  // if this is the first record written by this queue then prime the output list 
  do
  {
    old_hp = p->olp;
    new_hp = hp;

    if( old_hp != NULL )
      break;
    
  }while(!cmThPtrCAS(&p->olp,old_hp,new_hp));

  return rc;
}

cmThRC_t   cmTsMp1cEnqueueMsg( cmTsMp1cH_t  h, const void* dataPtr, unsigned byteCnt )
{ return cmTsMp1cEnqueueSegMsg(h,&dataPtr,&byteCnt,1); }

unsigned   cmTsMp1cAllocByteCount( cmTsMp1cH_t h )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  return p->bn;
}

unsigned   cmTsMp1cAvailByteCount( cmTsMp1cH_t h )
{
  cmTsMp1c_t*      p  = _cmTsMp1cHandleToPtr(h);
  unsigned         ai = _cmTsMp1cBufIndex(p,pthread_self());
  const cmTsBuf_t* b  = p->a + ai;

  if( b->oi > b->ii )
    return b->oi - b->ii - 1;

  return (p->bn - b->ii) + b->oi - 1;
}

unsigned _cmTsMp1cNextMsgByteCnt( cmTsMp1c_t* p )
{ 
  if( p->olp == NULL )
    return 0;

  // if the current msg has not yet been read
  if( p->olp->mn != 0 )
    return p->olp->mn;

  // if the current msg has been read but a new next msg has been linked
  if( p->olp->mn == 0 && p->olp->link != NULL )
  {
    // advance the buffer output msg past the prev msg header
    char* hp = (char*)(p->olp + 1);
    p->a[p->olp->ai].oi =  hp - p->a[p->olp->ai].buf;

    // advance msg pointer to point to the new msg header
    p->olp = p->olp->link;

    // return the size of the new msg
    return p->olp->mn;
  }

  return 0;
}

cmThRC_t   cmTsMp1cDequeueMsg( cmTsMp1cH_t  h,  void* dataPtr, unsigned byteCnt )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);

  // if there are no messages waiting
  if( _cmTsMp1cNextMsgByteCnt(p) == 0 )
    return kBufEmptyThRC;

  char*    hp = (char*)p->olp;
  char*    dp = hp - p->olp->mn;  // the msg body is before the msg hdr

  if( dataPtr == NULL )
  {
    p->cbFunc(p->cbArg,p->olp->mn,dp);
  }
  else
  {
    if( p->olp->mn > byteCnt )
      return cmErrMsg(&p->err,kBufTooSmallThRC,"The return buffer constains too few bytes (%i) to contain %i bytes.",byteCnt,p->olp->mn);

    memcpy(dataPtr,dp,p->olp->mn);
  }

  // advance the buffers output index past the msg body
  p->a[p->olp->ai].oi = hp - p->a[p->olp->ai].buf;

  // mark the msg as read
  p->olp->mn = 0;

  return kOkThRC;
}


bool       cmTsMp1cMsgWaiting( cmTsMp1cH_t h )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  return _cmTsMp1cNextMsgByteCnt(p) != 0;
}

unsigned   cmTsMp1cDequeueMsgByteCount( cmTsMp1cH_t h )
{
  cmTsMp1c_t* p  = _cmTsMp1cHandleToPtr(h);
  return _cmTsMp1cNextMsgByteCnt(p);
}
  
bool       cmTsMp1cIsValid( cmTsMp1cH_t h )
{ return h.h != NULL; }
  


//============================================================================================================================/
//
// cmTsQueueTest()
//

// param recd for use by _cmTsQueueCb0() and
// the msg record passed between the sender
// threads and the receiver thread
typedef struct
{
  unsigned     id;
  cmTsQueueH_t qH;
  int          val;
} _cmTsQCbParam_t;


// Generate a random number and put it in a TS queue
bool _cmTsQueueCb0(void* param)
{
  _cmTsQCbParam_t* p = (_cmTsQCbParam_t*)param;

  p->val = rand(); // generate a random number

  // send the msg
  if( cmTsQueueEnqueueMsg( p->qH, p, sizeof(_cmTsQCbParam_t)) == kOkThRC )
    printf("in:%i %i\n",p->id,p->val);
  else
    printf("in error %i\n",p->id);
  

  usleep(100*1000);

  return true;
}

// Monitor a TS queue for incoming messages from _cmTsQueueCb1()
bool _cmTsQueueCb1(void* param)
{
  // the thread param is a ptr to the TS queue to monitor.
  cmTsQueueH_t*   qp = (cmTsQueueH_t*)param;
  cmThRC_t        rc;
  _cmTsQCbParam_t msg;

  // dequeue any waiting messages
  if((rc = cmTsQueueDequeueMsg( *qp, &msg, sizeof(msg))) == kOkThRC )
    printf("out:%i %i\n",msg.id,msg.val);
  else
  {
    if( rc != kBufEmptyThRC )
      printf("out error:%i\n", rc);
  }

  return true;
}

// Test the TS queue by starting sender threads (threads 0 & 1)
// and a receiver thread (thread 2) and sending messages
// from the sender to the receiver.
void cmTsQueueTest( cmRpt_t* rpt )
{
  cmThreadH_t th0=cmThreadNullHandle,th1=cmThreadNullHandle,th2=cmThreadNullHandle;
  cmTsQueueH_t q=cmTsQueueNullHandle;
  _cmTsQCbParam_t param0, param1;

  // create a TS Queue
  if( cmTsQueueCreate(&q,100,NULL,NULL,rpt) != kOkThRC )
    goto errLabel;

  // create thread 0
  param0.id = 0;
  param0.qH = q;
  if( cmThreadCreate(&th0,_cmTsQueueCb0,&param0,rpt) != kOkThRC )
    goto errLabel;

  // create thread 1
  param1.id = 1;
  param1.qH = q;
  if( cmThreadCreate(&th1,_cmTsQueueCb0,&param1,rpt) != kOkThRC )
    goto errLabel;

  // create thread 2
  if( cmThreadCreate(&th2,_cmTsQueueCb1,&q,rpt) != kOkThRC )
    goto errLabel;

  // start thread 0
  if( cmThreadPause(th0,0) != kOkThRC )
    goto errLabel;

  // start thread 1
  if( cmThreadPause(th1,0) != kOkThRC )
    goto errLabel;

  // start thread 2
  if( cmThreadPause(th2,0) != kOkThRC )
    goto errLabel;

  printf("any key to quit.");
  getchar();

 
 errLabel:

  if( cmThreadIsValid(th0) )
    if( cmThreadDestroy(&th0) != kOkThRC )
      printf("Error destroying thread 0\n");

  if( cmThreadIsValid(th1) )
    if( cmThreadDestroy(&th1) != kOkThRC )
      printf("Error destroying thread 1\n");

  if( cmThreadIsValid(th2) )
    if( cmThreadDestroy(&th2) != kOkThRC )
      printf("Error destroying thread 1\n");

  if( cmTsQueueIsValid(q) )
    if( cmTsQueueDestroy(&q) != kOkThRC )
      printf("Error destroying queue\n");
  
}


//============================================================================================================================/
//
// cmTs1p1cTest()
//

// param recd for use by _cmTsQueueCb0() and
// the msg record passed between the sender
// threads and the receiver thread
typedef struct
{
  unsigned     id;
  cmTs1p1cH_t  qH;
  int          val;
} _cmTs1p1cCbParam_t;

cmTs1p1cH_t cmTs1p1cNullHandle = cmSTATIC_NULL_HANDLE;

// Generate a random number and put it in a TS queue
bool _cmTs1p1cCb0(void* param)
{
  _cmTs1p1cCbParam_t* p = (_cmTs1p1cCbParam_t*)param;

  p->val = rand(); // generate a random number

  // send the msg
  if( cmTs1p1cEnqueueMsg( p->qH, p, sizeof(_cmTs1p1cCbParam_t)) == kOkThRC )
    printf("in:%i %i\n",p->id,p->val);
  else
    printf("in error %i\n",p->id);
  
  ++p->id;

  usleep(100*1000);

  return true;
}

// Monitor a TS queue for incoming messages from _cmTs1p1cCb1()
bool _cmTs1p1cCb1(void* param)
{
  // the thread param is a ptr to the TS queue to monitor.
  cmTs1p1cH_t*   qp = (cmTs1p1cH_t*)param;
  cmThRC_t        rc;
  _cmTs1p1cCbParam_t msg;

  // dequeue any waiting messages
  if((rc = cmTs1p1cDequeueMsg( *qp, &msg, sizeof(msg))) == kOkThRC )
    printf("out:%i %i\n",msg.id,msg.val);
  else
  {
    if( rc != kBufEmptyThRC )
      printf("out error:%i\n", rc);
  }

  return true;
}

// Test the TS queue by starting sender threads (threads 0 & 1)
// and a receiver thread (thread 2) and sending messages
// from the sender to the receiver.
void cmTs1p1cTest( cmRpt_t* rpt )
{
  cmThreadH_t th0=cmThreadNullHandle,th1=cmThreadNullHandle,th2=cmThreadNullHandle;
  cmTs1p1cH_t q=cmTs1p1cNullHandle;
  _cmTs1p1cCbParam_t param1;

  // create a TS Queue
  if( cmTs1p1cCreate(&q,28*2,NULL,NULL,rpt) != kOkThRC )
    goto errLabel;

  // create thread 1
  param1.id = 0;
  param1.qH = q;
  if( cmThreadCreate(&th1,_cmTs1p1cCb0,&param1,rpt) != kOkThRC )
    goto errLabel;

  // create thread 2
  if( cmThreadCreate(&th2,_cmTs1p1cCb1,&q,rpt) != kOkThRC )
    goto errLabel;

  // start thread 1
  if( cmThreadPause(th1,0) != kOkThRC )
    goto errLabel;

  // start thread 2
  if( cmThreadPause(th2,0) != kOkThRC )
    goto errLabel;

  printf("any key to quit.");
  getchar();

 
 errLabel:

  if( cmThreadIsValid(th0) )
    if( cmThreadDestroy(&th0) != kOkThRC )
      printf("Error destroying thread 0\n");

  if( cmThreadIsValid(th1) )
    if( cmThreadDestroy(&th1) != kOkThRC )
      printf("Error destroying thread 1\n");

  if( cmThreadIsValid(th2) )
    if( cmThreadDestroy(&th2) != kOkThRC )
      printf("Error destroying thread 1\n");

  if( cmTs1p1cIsValid(q) )
    if( cmTs1p1cDestroy(&q) != kOkThRC )
      printf("Error destroying queue\n");
  
}


//============================================================================================================================/
//
// cmTsMp1cTest()
//
// param recd for use by _cmTsQueueCb0() and
// the msg record passed between the sender
// threads and the receiver thread
typedef struct
{
  unsigned     id;
  cmTsMp1cH_t  qH;
  int          val;
} _cmTsMp1cCbParam_t;


unsigned _cmTsMp1cVal = 0;

// Incr the global value _cmTsMp1cVal and put it in a TS queue
bool _cmTsMp1cCb0(void* param)
{
  _cmTsMp1cCbParam_t* p = (_cmTsMp1cCbParam_t*)param;

  p->val = __sync_fetch_and_add(&_cmTsMp1cVal,1);

  // send the msg
  if( cmTsMp1cEnqueueMsg( p->qH, p, sizeof(_cmTsMp1cCbParam_t)) == kOkThRC )
    printf("in:%i %i\n",p->id,p->val);
  else
    printf("in error %i\n",p->id);

  usleep(100*1000);

  return true;
}

// Monitor a TS queue for incoming messages from _cmTsMp1cCb1()
bool _cmTsMp1cCb1(void* param)
{
  // the thread param is a ptr to the TS queue to monitor.
  cmTsMp1cH_t*   qp = (cmTsMp1cH_t*)param;
  cmThRC_t        rc;
  _cmTsMp1cCbParam_t msg;

  // dequeue any waiting messages
  if((rc = cmTsMp1cDequeueMsg( *qp, &msg, sizeof(msg))) == kOkThRC )
    printf("out - cons id:%i val:%i\n",msg.id,msg.val);
  else
  {
    if( rc != kBufEmptyThRC )
      printf("out error:%i\n", rc);
  }

  return true;
}

// Test the TS queue by starting sender threads (threads 0 & 1)
// and a receiver thread (thread 2) and sending messages
// from the sender to the receiver.
void cmTsMp1cTest( cmRpt_t* rpt )
{
  cmThreadH_t th0=cmThreadNullHandle,th1=cmThreadNullHandle,th2=cmThreadNullHandle;
  cmTsMp1cH_t q=cmTsMp1cNullHandle;
  _cmTsMp1cCbParam_t param0, param1;

  // create a TS Queue
  if( cmTsMp1cCreate(&q,1000,NULL,NULL,rpt) != kOkThRC )
    goto errLabel;

  // create thread 0 - producer 0
  param0.id = 0;
  param0.qH = q;
  if( cmThreadCreate(&th0,_cmTsMp1cCb0,&param0,rpt) != kOkThRC )
    goto errLabel;

  // create thread 1 - producer 1
  param1.id = 1;
  param1.qH = q;
  if( cmThreadCreate(&th1,_cmTsMp1cCb0,&param1,rpt) != kOkThRC )
    goto errLabel;

  // create thread 2 - consumer 0
  if( cmThreadCreate(&th2,_cmTsMp1cCb1,&q,rpt) != kOkThRC )
    goto errLabel;

  // start thread 0
  if( cmThreadPause(th0,0) != kOkThRC )
    goto errLabel;

  // start thread 1
  if( cmThreadPause(th1,0) != kOkThRC )
    goto errLabel;

  // start thread 2
  if( cmThreadPause(th2,0) != kOkThRC )
    goto errLabel;

  printf("any key to quit.");
  getchar();

 
 errLabel:

  if( cmThreadIsValid(th0) )
    if( cmThreadDestroy(&th0) != kOkThRC )
      printf("Error destroying thread 0\n");

  if( cmThreadIsValid(th1) )
    if( cmThreadDestroy(&th1) != kOkThRC )
      printf("Error destroying thread 1\n");

  if( cmThreadIsValid(th2) )
    if( cmThreadDestroy(&th2) != kOkThRC )
      printf("Error destroying thread 1\n");

  if( cmTsMp1cIsValid(q) )
    if( cmTsMp1cDestroy(&q) != kOkThRC )
      printf("Error destroying queue\n");
  
}