libcm/cmTaskMgr.c

#include "cmGlobal.h"
#include "cmRpt.h"
#include "cmErr.h"
#include "cmCtx.h"
#include "cmMem.h"
#include "cmMallocDebug.h"
#include "cmThread.h"
#include "cmTime.h"
#include "cmTaskMgr.h"
#include "cmLinkedHeap.h"
#include "cmText.h"

cmTaskMgrH_t cmTaskMgrNullHandle = cmSTATIC_NULL_HANDLE;

struct cmTmInst_str;

typedef struct cmTmTask_str
{
  unsigned             taskId;
  cmChar_t*            label; 
  cmTaskMgrFunc_t      func;
  cmTaskMgrRecv_t      recv;
  struct cmTmTask_str* link;
} cmTmTask_t;

typedef struct cmTmInst_str
{
  unsigned             instId;             // Task instance id.
  struct cmTmTask_str* task;               // Pointer to task record for this task instance.
  void*                funcArg;            // Client supplied pointer to cmTaskMgrFuncArg_t.arg;
  unsigned             progCnt;            // Maximum expected progress (cmTaskMgrStatusArg_t.prog) to be used by this task instance.
  cmChar_t*            label;              // Optional instance label.
  cmStatusTmId_t       status;             // Current instance status (See cmStatusTmId_t)
  void*                result;             // Task instance result pointer. 
  unsigned             resultByteCnt;      // Size of the task instance result pointer in bytes.
  cmTaskMgrCtlId_t     ctlId;              // ctlId must only be written from the client thread
  cmTs1p1cH_t          msgQueH;            // client->inst 'msg' communication queue
  bool                 deleteOnCompleteFl; // delete this instance when its status indicates that it is killed or complete
  struct cmTmInst_str* link;
} cmTmInst_t;

struct cmTm_str* p;

typedef struct cmTmThread_str
{
  struct cmTm_str*       p;            // Pointer to task mgr.
  cmThreadH_t            thH;          // Thread handle. 
  cmTmInst_t*            inst;         // Ptr to the task instance this thread is executing.
  double                 durSecs;      // Duration of the instance currently assigned to this thread in seconds.
  cmTimeSpec_t           t0;           // Task start time.  
  cmTimeSpec_t           t1;           // Time of last review by the master thread.
  bool                   deactivateFl; // True if this instance has been deactivated by the system.
  cmTaskMgrFuncArg_t     procArg;      // 
  cmChar_t*              text;         // Temporary text buffer
  struct cmTmThread_str* link;         // p->threads link.
} cmTmThread_t;

typedef struct cmTm_str
{
  cmErr_t             err;              // Task manager error object.
  cmThreadH_t         mstrThH;          // Master thread handle.
  cmTmThread_t*       threads;          // Thread record list.
  unsigned            threadRecdCnt;    // Current count of records in 'threads' list.
  unsigned            maxActiveTaskCnt; // Max. number of active tasks.
  cmTaskMgrStatusCb_t statusCb;         // Client task status callback.
  void*               statusCbArg;      // Client task status callback argument.
  unsigned            pauseSleepMs;     // 
  cmTs1p1cH_t         callQueH;         // client->mgr 'inst' communication queue
  cmTsMp1cH_t         outQueH;          // mgr->client communication queue
  cmTmTask_t*         tasks;            // Task list.
  cmTmInst_t*         insts;            // Task instance list.
  unsigned            nextInstId;       // Next available task instance id.
  unsigned            activeTaskCnt;    // Current active task count.
} cmTm_t;


void  _cmTaskMgrStatusArgSetup(
  cmTaskMgrStatusArg_t* s,
  void*                 arg,
  unsigned              instId,
  cmSelTmId_t           selId,
  cmStatusTmId_t        statusId,
  unsigned              prog,
  const cmChar_t*       text,
  const void*           msg,
  unsigned              msgByteCnt )
{
  s->arg           = arg;
  s->instId        = instId;
  s->selId         = selId;
  s->statusId      = statusId;
  s->prog          = prog;
  s->text          = text;
  s->msg           = msg;
  s->msgByteCnt    = msgByteCnt;
}


// Called by MASTER and WORKER.
cmTmRC_t _cmTmEnqueueStatusMsg0( cmTm_t* p, const cmTaskMgrStatusArg_t* s )
{
  enum { arrayCnt = 3 };
  const void*    msgPtrArray[arrayCnt];
  unsigned       msgSizeArray[arrayCnt];

  msgPtrArray[0] = s;
  msgPtrArray[1] = s->text==NULL ? "" : s->text;
  msgPtrArray[2] = s->msg;

  msgSizeArray[0] = sizeof(cmTaskMgrStatusArg_t);
  msgSizeArray[1] = s->text==NULL ? 1 : strlen(s->text)+1;
  msgSizeArray[2] = s->msgByteCnt;


  if( cmTsMp1cEnqueueSegMsg(p->outQueH, msgPtrArray, msgSizeArray, arrayCnt ) != kOkThRC )
    return kQueueFailTmRC;
  
  return kOkTmRC;
}


// Called by MASTER and WORKER.
// This function is called by the worker thread wrapper _cmTmWorkerStatusCb()
// function to enqueue messages being sent back to the client.
cmTmRC_t _cmTmEnqueueStatusMsg1( 
  cmTm_t*         p,
  unsigned        instId,
  cmSelTmId_t     selId,
  cmStatusTmId_t  statusId,
  unsigned        prog,
  const cmChar_t* text,
  const void*     msg,
  unsigned        msgByteCnt )
{
  cmTaskMgrStatusArg_t s;
  _cmTaskMgrStatusArgSetup(&s,p->statusCbArg,instId,selId,statusId,prog,text,msg,msgByteCnt);
  return _cmTmEnqueueStatusMsg0(p,&s);
}

// Called by WORKER.
// Worker threads call this function to enqueue status messages
// for delivery to the task mgr client.
void _cmTmWorkerStatusCb( const cmTaskMgrStatusArg_t* status )
{
  cmTmThread_t* trp = (cmTmThread_t*)status->arg;

  if( _cmTmEnqueueStatusMsg0( trp->p, status ) != kOkTmRC )
  {
    /// ??????? HOW DO WE HANDLE ERRORS IN THE WORKER THREAD
    /// (set an error code in trp and let the master thread notice it.)
    assert(0);
  }

}

// Called by WORKER.
// This function is called in the worker thread by
// cmTs1p1cDequeueMsg() from within  cmTaskMgrWorkerHandleCommand()
// to transfer msg's waiting in the worker's incoming msg queue
// (cmTmInst_t.msgQueH) to the instance recv function (cmTmInst_t.recv).
cmRC_t _cmTmWorkerRecvCb( void* arg, unsigned msgByteCnt, const void* msg )
{
  cmTmThread_t* trp = (cmTmThread_t*)arg;
  assert(trp->inst->task->recv);
  trp->inst->task->recv(&trp->procArg,msg,msgByteCnt);
  return cmOkRC;
}

// Called by WORKER.
// This is the wrapper for all worker threads.
bool _cmTmWorkerThreadFunc(void* arg)
{
  cmTmThread_t*      trp = (cmTmThread_t*)arg;
  
  trp->procArg.reserved    = trp;
  trp->procArg.arg         = trp->inst->funcArg;
  trp->procArg.instId      = trp->inst->instId;
  trp->procArg.statusCb    = _cmTmWorkerStatusCb;
  trp->procArg.statusCbArg = trp;
  trp->procArg.progCnt     = trp->inst->progCnt;
  trp->procArg.pauseSleepMs= trp->p->pauseSleepMs;
  
  // if the task was paused or killed while it was queued then
  // cmTaskMgrHandleCommand() will do the right thing
  if( cmTaskMgrWorkerHandleCommand(&trp->procArg) != kStopTmwRC )
  { 
    trp->inst->status = kStartedTmId;

    // Notify the client that the instance has started.
    _cmTmEnqueueStatusMsg1(trp->p,trp->inst->instId,kStatusTmId,trp->inst->status,0,NULL,NULL,0);

    // Execute the client provided task function.
    trp->inst->task->func(&trp->procArg);
  }

  // Notify the client if the instance was killed
  if( trp->inst->ctlId == kKillTmId )
  {
    trp->inst->status = kKilledTmId;
    _cmTmEnqueueStatusMsg1(trp->p,trp->inst->instId,kStatusTmId,trp->inst->status,0,NULL,NULL,0);
  }

  // Notify the client that the instance is completed 
  // (but don't actually set the status yet)
  _cmTmEnqueueStatusMsg1(trp->p,trp->inst->instId,kStatusTmId,kCompletedTmId,0,NULL,NULL,0);

  trp->inst->status = kCompletedTmId;

  // Force the thread to go into the 'pause' state when it 
  // returns to it's internal loop. The master thread recognizes paused
  // threads as available for reuse.
  cmThreadPause(trp->thH,kPauseThFl);

  return true;
}

void _cmTmMasterRptError( cmTm_t* p, unsigned rc, const cmChar_t* msg )
{
  _cmTmEnqueueStatusMsg1(p,cmInvalidId,kErrorTmId,kInvalidTmId,rc,msg,NULL,0);
}

int _cmTmSortThreadByDur( const void* t0, const void* t1 )
{  
  double d = ((cmTmThread_t*)t0)->durSecs - ((cmTmThread_t*)t1)->durSecs; 

  return d== 0 ? 0 : (d<0 ? -1 : 1);
}


// This is the master thread function.
bool _cmTmMasterThreadFunc(void* arg)
{
  cmTm_t*       p         = (cmTm_t*)arg;
  unsigned      activeThreadCnt = 0;
  unsigned      activeTaskCnt   = 0;
  cmTmThread_t* trp       = p->threads;
 
  if( p->threadRecdCnt > 0 )
  {
    cmTmThread_t* thArray[p->threadRecdCnt];
    unsigned      deactivatedCnt = 0;

    //
    // Determine the number of active threads and tasks
    //

    // for each thread record
    for(trp=p->threads; trp!=NULL; trp=trp->link)
    {
      cmThStateId_t thState;

      // if this thread is active ...
      if( (thState = cmThreadState(trp->thH)) != kPausedThId )
      {        
        assert(trp->inst!=NULL);

        thArray[activeThreadCnt] = trp;
        ++activeThreadCnt;

        // if the task assigned to this thread is started then the task is active
        if( trp->inst->status == kStartedTmId )  
          ++activeTaskCnt;

        // if the deactivatedFl is set then this thread has been deactivated by the system
        if( trp->deactivateFl )
          ++deactivatedCnt;

        // update the task lifetime duration
        if( trp->inst->status != kCompletedTmId )
        {
          cmTimeSpec_t t2;
          cmTimeGet(&t2);
          trp->durSecs += (double)cmTimeElapsedMicros(&trp->t1,&t2) / 1000000.0;
          trp->t1 = t2;
        }

      }
    }

    //
    // thArray[activeThreadCnt] now holds pointers to the
    // cmTmThread_t records of the active threads
    //


    // if more tasks are active than should be - then deactive the youngest
    if( activeTaskCnt > p->maxActiveTaskCnt )
    {
      // sort the active tasks in increasing order of lifetime
      qsort(&thArray[0],activeThreadCnt,sizeof(thArray[0]),_cmTmSortThreadByDur);

      // determine the number of threads that need to be deactivated
      int n = activeTaskCnt - p->maxActiveTaskCnt;
      int i,j;

      // pause the active threads with the lowest lifetime
      for(i=0,j=0; i<activeThreadCnt && j<n; ++i)
        if( thArray[i]->deactivateFl == false )
        {
          thArray[i]->deactivateFl = true;
          ++deactivatedCnt;
          ++j;
        }
    }
    
    //
    // if there are deactivated tasks and the max thread count has not been reached
    // then re-activate some of the deactivated tasks.
    //
    
    if( activeTaskCnt < p->maxActiveTaskCnt && deactivatedCnt > 0 )
    {
      // sort the active tasks in increasing order of lifetime
      qsort(&thArray[0],activeThreadCnt,sizeof(thArray[0]),_cmTmSortThreadByDur);

      int n = cmMin(p->maxActiveTaskCnt - activeTaskCnt, deactivatedCnt );
      int i;

      // re-activate the oldest deactivated tasks first
      for(i=activeThreadCnt-1; i>=0 && n>0; --i)
        if( thArray[i]->deactivateFl )
        {
          thArray[i]->deactivateFl = false;
          --n;
          ++activeTaskCnt;
        }

    }
  }
  
  
  // If the number of activeTaskCnt is less than the limit and a queued task exists
  while( activeTaskCnt < p->maxActiveTaskCnt && cmTs1p1cMsgWaiting(p->callQueH) )
  {
    cmTmInst_t*   ip        = NULL;
    cmTmThread_t* atrp      = NULL;


    // Find a worker thread that is in the 'paused' state.
    // This is the definitive indication that the thread
    // does not have an assigned instance and the thread recd can be reused.
    for(trp=p->threads; trp!=NULL; trp=trp->link)
      if( cmThreadState(trp->thH) == kPausedThId )
      {
        atrp = trp;
        break;
      }

    // If all the existing worker threads are in use ...
    if( atrp==NULL )
    {
      // ... then create a new worker thread recd
      atrp = cmMemAllocZ(cmTmThread_t,1);

      // ... create the new worker thread
      if( cmThreadCreate(&atrp->thH,_cmTmWorkerThreadFunc,atrp,p->err.rpt) != kOkThRC )
      {
        cmMemFree(atrp);
        atrp = NULL;
        _cmTmMasterRptError(p,kThreadFailTmRC,"Worker thread create failed.");
        break;
      }
      else
      {
        // ... setup the new thread record
        atrp->p           = p;
        atrp->link        = p->threads;
        p->threads        = atrp;
        p->threadRecdCnt += 1;
      }
    }

    // if the thread creation failed
    if( atrp == NULL )
      break;

    // dequeue a pending task instance pointer from the input queue
    if(cmTs1p1cDequeueMsg(p->callQueH,&ip,sizeof(ip)) != kOkThRC )
    {
      _cmTmMasterRptError(p,kQueueFailTmRC,"Dequeue failed on incoming task instance queue.");
      break;
    }

    // if the task has a msg recv callback then assign it here
    if( ip->task->recv != NULL )
      if( cmTs1p1cSetCallback( ip->msgQueH, _cmTmWorkerRecvCb, atrp ) != kOkThRC )
      {
        _cmTmMasterRptError(p,kQueueFailTmRC,"Worker thread msg queue callback assignment failed.");
        break;
      }

    // setup the thread record associated with the new task
    atrp->inst         = ip;
    atrp->durSecs      = 0;
    atrp->deactivateFl = false;
    cmTimeGet(&atrp->t0);
    atrp->t1 = atrp->t0;


    // start the worker thread 
    if( cmThreadPause(atrp->thH,kWaitThFl) != kOkThRC )
      _cmTmMasterRptError(p,kThreadFailTmRC,"Worker thread start failed.");

    ++activeTaskCnt;
  }


  cmSleepMs(p->pauseSleepMs);

  p->activeTaskCnt = activeTaskCnt;

  return true;
}


cmTm_t* _cmTmHandleToPtr( cmTaskMgrH_t h )
{
  cmTm_t* p = (cmTm_t*)h.h;
  assert( p != NULL );
  return p;
}

cmTmTask_t* _cmTmTaskFromId( cmTm_t* p, unsigned taskId )
{
  cmTmTask_t* tp;
  for(tp=p->tasks; tp!=NULL; tp=tp->link)
    if( tp->taskId == taskId )
      return tp;
  return NULL;
}

cmTmInst_t* _cmTmInstFromId( cmTm_t* p, unsigned instId )
{
  cmTmInst_t* ip;
  for(ip=p->insts; ip!=NULL; ip=ip->link)
    if( ip->instId == instId )
      return ip;
  return NULL;
}


cmTmRC_t _cmTmInstFree( cmTm_t* p, unsigned instId )
{
  cmTmInst_t* ip = p->insts;  
  cmTmInst_t* pp = NULL;

  for(; ip!=NULL; ip=ip->link)
  {
    if( ip->instId == instId )
    {
      if( pp == NULL )
        p->insts = ip->link;
      else
        pp->link = ip->link;

      if( cmTs1p1cDestroy(&ip->msgQueH) != kOkThRC )
        return cmErrMsg(&p->err,kQueueFailTmRC,"The 'msg' input queue destroy failed.");

      cmMemFree(ip->label);
      cmMemFree(ip->result);
      cmMemFree(ip);
      return kOkTmRC;
    }
    pp = ip;
  }

  return cmErrMsg(&p->err,kAssertFailTmRC,"The instance %i could not be found to be deleted.",instId);
}


cmTmRC_t _cmTmDestroy( cmTm_t* p )
{
  cmTmRC_t rc = kOkTmRC;
  unsigned i;

  // stop and destroy the master thread
  if( cmThreadDestroy(&p->mstrThH) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kThreadFailTmRC,"Master thread destroy failed.");
    goto errLabel;
  }

  // stop and destroy all the worker threads
  for(i=0; p->threads != NULL; ++i )
  {
    if( cmThreadDestroy(&p->threads->thH) != kOkThRC )
    {
      rc = cmErrMsg(&p->err,kThreadFailTmRC,"Thread destruction failed for the worker thread at index %i.",i);
      goto errLabel;      
    }

    cmTmThread_t* trp = p->threads;
    p->threads = p->threads->link;
    cmMemFree(trp->text); 
    cmMemFree(trp);
  }

  // release the call input queue
  if( cmTs1p1cDestroy(&p->callQueH) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"The 'call' input queue destroy failed.");
    goto errLabel;
  }

  // draining the output queue
  while( cmTsMp1cMsgWaiting(p->outQueH) )
    if(cmTsMp1cDequeueMsg(p->outQueH,NULL,0) != kOkThRC )
      cmErrMsg(&p->err,kQueueFailTmRC,"The output queue failed while draingin.");

  // release the output queue
  if( cmTsMp1cDestroy(&p->outQueH) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"The input queue destroy failed.");
    goto errLabel;
  }


  // release instance list
  while( p->insts != NULL )
    _cmTmInstFree(p,p->insts->instId);

  // release the task list
  cmTmTask_t* tp = p->tasks;
  while( tp != NULL )
  {
    cmTmTask_t* np = tp->link;
    cmMemFree(tp->label);
    cmMemFree(tp);
    tp = np;
  }

  cmMemFree(p);
 errLabel:
  return rc;
}

cmRC_t _cmTmMasterOutQueueCb(void* arg, unsigned msgByteCnt, const void* msgDataPtr );

cmTmRC_t cmTaskMgrCreate(  
  cmCtx_t*             ctx, 
  cmTaskMgrH_t*        hp, 
  cmTaskMgrStatusCb_t  statusCb, 
  void*                statusCbArg, 
  unsigned             maxActiveTaskCnt,
  unsigned             queueByteCnt,
  unsigned             pauseSleepMs)
{
  cmTmRC_t rc = kOkTmRC;

  if((rc = cmTaskMgrDestroy(hp)) != kOkTmRC )
    return rc;

  cmTm_t* p = cmMemAllocZ(cmTm_t,1);
 
  cmErrSetup(&p->err,&ctx->rpt,"Task Mgr.");

  p->maxActiveTaskCnt = maxActiveTaskCnt;
  p->statusCb     = statusCb;
  p->statusCbArg  = statusCbArg;
  p->pauseSleepMs = pauseSleepMs;

  // create the master thread
  if( cmThreadCreate(&p->mstrThH, _cmTmMasterThreadFunc,p,&ctx->rpt) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kThreadFailTmRC,"Thread index %i create failed.");
    goto errLabel;
  }

  // create the call input queue
  if(cmTs1p1cCreate( &p->callQueH, queueByteCnt, NULL, NULL, p->err.rpt ) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"The call input queue creation failed.");
    goto errLabel;
  }

  // create the output queue
  if( cmTsMp1cCreate( &p->outQueH, queueByteCnt, _cmTmMasterOutQueueCb, p, p->err.rpt ) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"The output queue creation failed.");
    goto errLabel;
  }

  hp->h = p;

 errLabel:
  return rc;  
}

cmTmRC_t cmTaskMgrDestroy( cmTaskMgrH_t* hp )
{
  cmTmRC_t rc = kOkTmRC;
  if( hp==NULL || cmTaskMgrIsValid(*hp)==false )
    return rc;

  cmTm_t* p = _cmTmHandleToPtr(*hp);

  if((rc = _cmTmDestroy(p)) != kOkTmRC )
    return rc;

  hp->h = NULL;

  return rc;
}

void _cmTmWaitForCompletion( cmTm_t* p, unsigned timeOutMs )
{
  unsigned durMs = 0;
  cmTimeSpec_t  t0,t1;

  cmTimeGet(&t0);

  // Go into timeout loop - waiting for all instances to finish
  while( timeOutMs==0 || durMs < timeOutMs )
  {
    cmTimeGet(&t1);
    durMs += cmTimeElapsedMicros(&t0,&t1) / 1000;
    t0 = t1;

    cmSleepMs(p->pauseSleepMs);

    if( p->activeTaskCnt == 0 )
      break;
  }

}


cmTmRC_t cmTaskMgrClose( cmTaskMgrH_t h, unsigned flags, unsigned timeOutMs )
{
  cmTmRC_t rc = kOkTmRC;
  cmTm_t*  p  = _cmTmHandleToPtr(h);
  bool     fl = false;

  // if requested kill any queued tasks
  if( cmIsFlag(flags,kKillQueuedTmFl) )
  {
    cmTmInst_t* ip = p->insts;
    for(; ip!=NULL; ip=ip->link)
      if( ip->status == kQueuedTmId )
        ip->ctlId = kKillTmId;
  }

  // wait for any existing or queued tasks to complete
  _cmTmWaitForCompletion(p,timeOutMs);

  // force any queued msgs for the client to be sent
  cmTaskMgrOnIdle(h);

  // if the 'kill on timeout' flag is set then kill any remaining active tasks
  if( cmIsFlag(flags,kTimeOutKillTmFl) )
  {
    cmTmInst_t* ip = p->insts;
    for(; ip!=NULL; ip=ip->link)
      if( ip->status != kCompletedTmId )
      {
        ip->ctlId = kKillTmId;    
        fl = true;
      }
  }
  
  // wait for the remaining tasks to complete
  if( fl )
    _cmTmWaitForCompletion(p,timeOutMs);

  // force any queued msgs for the client to be sent
  cmTaskMgrOnIdle(h);

  return rc;
}

unsigned cmTaskMgrActiveTaskCount( cmTaskMgrH_t h )
{
  cmTm_t*  p  = _cmTmHandleToPtr(h);
  return p->activeTaskCnt;
}


bool     cmTaskMgrIsValid(   cmTaskMgrH_t h )
{ return h.h != NULL; }

const cmChar_t* cmTaskMgrStatusIdToLabel( cmStatusTmId_t statusId )
{
  typedef struct map_str
  {
    cmStatusTmId_t  id;
    const cmChar_t* label;
  } map_t;

  map_t a[] = 
  {
    { kQueuedTmId,      "Queued" },
    { kStartedTmId,     "Started" },
    { kPausedTmId,      "Paused" },
    { kDeactivatedTmId, "Deactivated" },
    { kCompletedTmId,   "Completed" },
    { kKilledTmId,      "Killed" },
    { kInvalidTmId,     "<Invalid>" },
  };

  int i;
  for(i=0; a[i].id!=kInvalidTmId; ++i)
    if( a[i].id == statusId )
      return a[i].label;

  return "<Unknown>";
}

// This function is called by cmTaskMgrIdle() to dispatch
// status updates to the client.
cmRC_t _cmTmMasterOutQueueCb(void* arg, unsigned msgByteCnt, const void* msgDataPtr )
{
  cmTm_t* p = (cmTm_t*)arg;
  cmTaskMgrStatusArg_t s;

  // This is probably not nesessary since changing the memory  
  // pointed to by msgDataPtr should be safe even though it is marked as const.
  memcpy(&s,msgDataPtr,sizeof(s));

  // The 'text' and 'msg' data have been serialized after the status record.
  // The 'text' is guaranteed to at least contain a terminating zero.
  s.text = ((char*)msgDataPtr) + sizeof(s);

  // if the 'resultByteCnt' > 0 then there is a result record
  if( s.msgByteCnt > 0 )
    s.msg = ((char*)msgDataPtr) + sizeof(s) + strlen(s.text) + 1;
  else
    s.msg = NULL;

  s.arg = p->statusCbArg;

  p->statusCb( &s );
  return cmOkRC;
}


cmTmRC_t cmTaskMgrOnIdle( cmTaskMgrH_t h )
{
  cmTmRC_t rc = kOkTmRC;
  cmTm_t*  p  = _cmTmHandleToPtr(h);

  // Transmit any msgs waiting to be sent to the client.
  while( cmTsMp1cMsgWaiting(p->outQueH) )
  {
    // calling this function calls: _cmTmMasterOutQueueCb()
    if(cmTsMp1cDequeueMsg(p->outQueH,NULL,0) != kOkThRC )
    {
      rc = cmErrMsg(&p->err,kQueueFailTmRC,"The output queue failed during a dequeue.");
      goto errLabel;
    }
  }

  // Step through the instance list and delete instances that are
  // completed and also marked for deletion.
  cmTmInst_t* ip = p->insts;
  while( ip != NULL )
  {
    cmTmInst_t* np = ip->link;

    if( ip->status==kCompletedTmId && ip->deleteOnCompleteFl )
      _cmTmInstFree(p,ip->instId);
    
    ip = np;
  }

 errLabel:
  return rc;
}

bool     cmTaskMgrIsEnabled( cmTaskMgrH_t h )
{
  cmTm_t*  p = _cmTmHandleToPtr(h);

  return cmThreadState(p->mstrThH) != kPausedThId;
}

cmTmRC_t cmTaskMgrEnable(    cmTaskMgrH_t h, bool enableFl )
{
  cmTmRC_t rc    = kOkTmRC;
  cmTm_t*  p     = _cmTmHandleToPtr(h);
  unsigned flags = (enableFl ? 0 : kPauseThFl) | kWaitThFl;

  if( cmThreadPause(p->mstrThH, flags ) != kOkThRC )
    rc = cmErrMsg(&p->err,kThreadFailTmRC,"The master thread failed to %s.",enableFl ? "enable" : "disable" );

  return rc;
}

cmTmRC_t cmTaskMgrInstall( 
  cmTaskMgrH_t    h, 
  unsigned        taskId, 
  const cmChar_t* label, 
  cmTaskMgrFunc_t func,
  cmTaskMgrRecv_t recv)
{
  cmTmRC_t    rc = kOkTmRC;
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmTask_t* tp = cmMemAllocZ(cmTmTask_t,1);

  if( _cmTmTaskFromId(p,taskId) != NULL )
  {
    rc = cmErrMsg(&p->err,kInvalidArgTmRC,"The task id %i is already in use.",taskId);
    goto errLabel;
  }

  tp->taskId = taskId;
  tp->func   = func;
  tp->recv   = recv;
  tp->label  = cmMemAllocStr(label);
  
  tp->link   = p->tasks;
  p->tasks   = tp;
  
 errLabel:
  return rc;
}


cmTmRC_t cmTaskMgrCall( 
  cmTaskMgrH_t    h, 
  unsigned        taskId, 
  void*           funcArg, 
  unsigned        progCnt, 
  unsigned        queueByteCnt,
  const cmChar_t* label,
  unsigned*       retInstIdPtr )
{
  cmTmRC_t    rc = kOkTmRC;
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmTask_t* tp = NULL;
  cmTmInst_t* ip = NULL;

  if( retInstIdPtr != NULL )
    *retInstIdPtr = cmInvalidId;

  // locate the task for this instance
  if((tp = _cmTmTaskFromId(p,taskId)) == NULL )
  {
    rc = cmErrMsg(&p->err,kInvalidArgTmRC,"Task not found for task id=%i.",taskId);
    goto errLabel;
  }

  // allocate a new instance record
  ip = cmMemAllocZ(cmTmInst_t,1);

  // setup the instance record
  ip->instId         = p->nextInstId++;
  ip->task           = tp;
  ip->funcArg        = funcArg;
  ip->progCnt        = progCnt;
  ip->label          = label==NULL ? NULL : cmMemAllocStr(label);
  ip->status         = kQueuedTmId;
  ip->ctlId          = kStartTmId;
 
  // create the msg input queue
  if(cmTs1p1cCreate( &ip->msgQueH, queueByteCnt, NULL, NULL, p->err.rpt ) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"The msg input queue creation failed.");
    goto errLabel;
  }

  // insert the new instance at the end of the instance list
  if( p->insts == NULL )
    p->insts = ip;
  else
  {
    cmTmInst_t* pp = p->insts;
    for(; pp != NULL; pp=pp->link )
      if( pp->link == NULL )
      {
        pp->link = ip;
        break;
      }
  }

  // enqueue the instance ptr in the input queue 
  if( cmTs1p1cEnqueueMsg(p->callQueH,&ip,sizeof(ip)) != kOkThRC )
  {
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"New task instance command enqueue failed.");    
    goto errLabel;
  }

  // set the returned instance id
  if( retInstIdPtr != NULL )
    *retInstIdPtr = ip->instId;

  // notify the client that the instance was enqueued
  cmTaskMgrStatusArg_t s;
  _cmTaskMgrStatusArgSetup(&s,p->statusCbArg,ip->instId,kStatusTmId,kQueuedTmId,progCnt,NULL,NULL,0);
  
  p->statusCb( &s );
  
 errLabel:
  return rc;
}

cmTmRC_t cmTaskMgrCtl( cmTaskMgrH_t h, unsigned instId, cmTaskMgrCtlId_t ctlId )
{
  cmTmRC_t    rc = kOkTmRC;
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmInst_t* ip = NULL;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
  {
    cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);
    goto errLabel;
  }

  // Once an instance ctlId is set to kKillTmId don't allow it to change.
  if( ip->ctlId == kKillTmId )
    return rc;

  switch(ctlId )
  {
    case kStartTmId:
      // Acting on a 'start' cmd only makes sense if the previous command was 'pause'
      if( ip->ctlId == kPauseTmId )
        ip->ctlId = kStartTmId;
      break;

    case kPauseTmId:
      // Acting on a 'pause' command only makes sense if  the previous command was a 'start'
      if( ip->ctlId == kStartTmId )
        ip->ctlId = kPauseTmId;
      break;

    case kKillTmId:
      ip->ctlId = kKillTmId;
      break;
  }


 errLabel:
  return rc;
}

cmStatusTmId_t cmTaskMgrStatus( cmTaskMgrH_t h, unsigned instId )
{
  cmTm_t*        p      = _cmTmHandleToPtr(h);
  cmTmInst_t*    ip     = NULL;
  cmStatusTmId_t status = kInvalidTmId;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
  {
    cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);
    goto errLabel;
  }
  
  status = ip->status;

 errLabel:
  return status;
}

cmTmRC_t  cmTaskMgrSendMsg( cmTaskMgrH_t h, unsigned instId, const void* msg, unsigned msgByteCnt )
{
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmRC_t    rc = kOkTmRC;  
  cmTmInst_t* ip = NULL;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
    return cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);

  if( cmTs1p1cEnqueueMsg(ip->msgQueH, msg, msgByteCnt ) != kOkThRC )
    rc = cmErrMsg(&p->err,kQueueFailTmRC,"Task msg enqueue failed.");    

  return rc;
}

const cmChar_t* cmTaskMgrTaskIdToLabel( cmTaskMgrH_t h, unsigned taskId )
{
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmTask_t* tp;
  if((tp = _cmTmTaskFromId(p,taskId)) == NULL )
    return NULL;
  return tp->label;
}

const cmChar_t* cmTaskMgrInstIdToLabel( cmTaskMgrH_t h, unsigned instId )
{
  cmTm_t* p = _cmTmHandleToPtr(h);
  cmTmInst_t* ip;
  if((ip = _cmTmInstFromId(p,instId)) == NULL )
    return NULL;
  return ip->label;
}
  
const void* cmTaskMgrResult( cmTaskMgrH_t h, unsigned instId )
{
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmInst_t* ip = NULL;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
  {
    cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);
    goto errLabel;
  }

  if( ip->status != kCompletedTmId )
  {
    cmErrMsg(&p->err,kOpFailTmRC,"The result of a running task (id:%i) may not be accessed.",instId);
    goto errLabel;
  }

  return ip->result;

 errLabel:
  return NULL;
}

unsigned    cmTaskMgrResultByteCount( cmTaskMgrH_t h, unsigned instId )
{
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmInst_t* ip = NULL;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
  {
    cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);
    goto errLabel;
  }

  if( ip->status != kCompletedTmId )
  {
    cmErrMsg(&p->err,kOpFailTmRC,"The result byte count of a running task (id:%i) may not be accessed.",instId);
    goto errLabel;
  }

  return ip->resultByteCnt;

 errLabel:
  return 0;
}

void* cmTaskMgrFuncArg( cmTaskMgrH_t h, unsigned instId )
{
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmInst_t* ip = NULL;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
  {
    cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);
    goto errLabel;
  }

  if( ip->status != kCompletedTmId )
  {
    cmErrMsg(&p->err,kOpFailTmRC,"The function argument of a running task (id:%i) may not be accessed.",instId);
    goto errLabel;
  }

  return ip->funcArg;

 errLabel:
  return NULL;
}


cmTmRC_t    cmTaskMgrInstDelete(    cmTaskMgrH_t h, unsigned instId )
{
  cmTmRC_t    rc = kOkTmRC;
  cmTm_t*     p  = _cmTmHandleToPtr(h);
  cmTmInst_t* ip = NULL;

  if((ip = _cmTmInstFromId(p,instId)) == NULL )
  {
    cmErrMsg(&p->err,kInvalidArgTmRC,"The task instance associated with id %i could not be found.",instId);
    return 0;
  }

  ip->deleteOnCompleteFl = true;

  return rc;
}

void _cmTaskMgrWorkerHelper( cmTaskMgrFuncArg_t* a, cmSelTmId_t selId, cmStatusTmId_t statusId, unsigned prog, const cmChar_t* text )
{
  cmTaskMgrStatusArg_t s;
  
  _cmTaskMgrStatusArgSetup(
    &s,
    a->statusCbArg,
    a->instId,
    statusId == kInvalidTmId ? kProgTmId    : kStatusTmId,
    statusId == kInvalidTmId ? kStartedTmId : statusId,
    statusId == kInvalidTmId ? prog         : 0,
    text,NULL,0);

  a->statusCb(&s);
}

cmTmWorkerRC_t cmTaskMgrWorkerHandleCommand( cmTaskMgrFuncArg_t* a )
{
  cmTmThread_t*  trp       = a->reserved;

  // Check if we should go into the paused or deactivated state.
  if( trp->inst->ctlId == kPauseTmId || trp->deactivateFl == true )
  {
    cmStatusTmId_t prvStatus = kInvalidTmId;

    do
    {
      // Note that it is possible that the state of the task switch from
      // paused <-> deactivated during the course of this loop.  
      // In either case we continue looping but should report the change
      // to the client via a status callback.

      // change the instance status to reflect the true status
      trp->inst->status = trp->deactivateFl ? kDeactivatedTmId : kPausedTmId;

      // if the status actually changed then notify the client
      if( trp->inst->status != prvStatus )
      {
        _cmTaskMgrWorkerHelper(a,kStatusTmId,trp->inst->status,0,NULL);
        prvStatus = trp->inst->status;
      }

      // sleep the thread for pauseSleepMs milliseconds
      cmSleepMs(a->pauseSleepMs);

      // if the task was unpaused while we slept
    }while( trp->inst->ctlId == kPauseTmId || trp->deactivateFl==true );
      
    // we are leaving the paused state because we were restarted or killed.
    switch( trp->inst->ctlId )
    {
      case kStartTmId:
        // change the instance status to 'started'.
        trp->inst->status = kStartedTmId;

        // notify the client of the change in state
        _cmTaskMgrWorkerHelper(a,kStatusTmId,kStartedTmId,0,NULL);
        break;

      case kKillTmId:
        // if killed the client will be notified in the worker thread wrapper
        // function: _cmTmWorkerThreadFunc()
        break;

      default:
        { assert(0); }
    }
  }
  else  // There was no command to handle so check for incoming msg's.
  {
    
    if( cmTs1p1cMsgWaiting(trp->inst->msgQueH) )
    {
      // if the task registered a msg receive callback
      if( trp->inst->task->recv != NULL )
      {
        
        if( cmTs1p1cDequeueMsg(trp->inst->msgQueH,  NULL, 0 ) != kOkThRC )
        {
          
          // ?????
          // ????? how do we send error messages back to the client
          // ??????
          return kOkTmwRC;
        }
        

      }
      else
      {
      }
      
      return kRecvTmwRC;
    }
  }

  // if ctlId==kKillTmId then the status update will be handled 
  // when the task custom function returns in  _cmTmWorkerThreadFunc()

  return trp->inst->ctlId == kKillTmId  ? kStopTmwRC : kOkTmwRC;
}



cmTmRC_t cmTaskMgrWorkerSendStatus( cmTaskMgrFuncArg_t* a, cmStatusTmId_t statusId )
{  
  _cmTaskMgrWorkerHelper(a,kStatusTmId,statusId,0,NULL); 
  return kOkTmRC;
}

cmTmRC_t cmTaskMgrWorkerSendProgress( cmTaskMgrFuncArg_t* a, unsigned prog, const cmChar_t* text )
{ 
  _cmTaskMgrWorkerHelper(a,kProgTmId,kInvalidTmId,prog,text);  
  return kOkTmRC;
}

cmTmRC_t       cmTaskMgrWorkerSendProgressV( cmTaskMgrFuncArg_t* a, unsigned prog, const cmChar_t* fmt, va_list vl )
{
  cmTmThread_t*  trp  = a->reserved;
  cmTsVPrintfP(trp->text,fmt,vl);
  return  cmTaskMgrWorkerSendProgress(a,prog,trp->text);
}

cmTmRC_t       cmTaskMgrWorkerSendProgressF( cmTaskMgrFuncArg_t* a, unsigned prog, const cmChar_t* fmt, ... )
{
  va_list vl;
  va_start(vl,fmt);
  cmTmRC_t rc = cmTaskMgrWorkerSendProgressV(a,prog,fmt,vl);
  va_end(vl);
  return rc;
}

cmTmRC_t  cmTaskMgrWorkerError(  cmTaskMgrFuncArg_t* a, unsigned rc, const cmChar_t* text )
{
  _cmTaskMgrWorkerHelper(a, kErrorTmId, kInvalidTmId, rc, text);
  return rc;
}

cmTmRC_t  cmTaskMgrWorkerErrorV( cmTaskMgrFuncArg_t* a, unsigned rc, const cmChar_t* fmt, va_list vl )
{
  cmTmThread_t*  trp  = a->reserved;
  cmTsVPrintfP(trp->text,fmt,vl);  
  return cmTaskMgrWorkerError(a,rc,trp->text);
}
 
cmTmRC_t  cmTaskMgrWorkerErrorF( cmTaskMgrFuncArg_t* a, unsigned rc, const cmChar_t* fmt, ... )
{
  va_list vl;
  va_start(vl,fmt);
  cmTmRC_t rc0 = cmTaskMgrWorkerErrorV(a,rc,fmt,vl);
  va_end(vl);
  return rc0;
}

cmTmRC_t cmTaskMgrWorkerSetResult(    cmTaskMgrFuncArg_t* a, void* result, unsigned resultByteCnt )
{
  cmTmThread_t*  trp = a->reserved;
  trp->inst->result        = result;
  trp->inst->resultByteCnt = resultByteCnt;
  return kOkTmRC;
}

unsigned cmTaskMgrWorkerMsgByteCount( cmTaskMgrFuncArg_t* a )
{
  cmTmThread_t*  trp       = a->reserved;
  return cmTs1p1cDequeueMsgByteCount(trp->inst->msgQueH);
}

unsigned cmTaskMgrWorkerMsgRecv( cmTaskMgrFuncArg_t* a, void* buf, unsigned bufByteCnt )
{
  cmTmThread_t*  trp    = a->reserved;
  unsigned       retVal = bufByteCnt;

  switch( cmTs1p1cDequeueMsg(trp->inst->msgQueH,  buf, bufByteCnt ) )
  {
    case kOkThRC:
      break;

    case kBufEmptyThRC:
      retVal = 0;
      break;

    case kBufTooSmallThRC:
      retVal = cmInvalidCnt;
      break;

    default:
      { assert(0); }
  }

  return retVal;
}

cmTmRC_t cmTaskMgrWorkerMsgSend( cmTaskMgrFuncArg_t* a, const void* buf, unsigned bufByteCnt )
{
  cmTmThread_t*  trp    = a->reserved;
  return _cmTmEnqueueStatusMsg1(trp->p,trp->inst->instId,kMsgTmId,trp->inst->status,0,NULL,buf,bufByteCnt);
}


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

enum { kMaxTestInstCnt = 3 };

typedef struct cmTmTestInst_str
{
  unsigned instId;
} cmTmTestInst_t;

typedef struct cmTmTestApp_str
{
  cmErr_t*       err;
  cmTmTestInst_t insts[kMaxTestInstCnt];
} cmTmTestApp_t;

void _cmTmTestReportStatus( cmRpt_t* rpt, const cmTaskMgrStatusArg_t* s )
{
  cmRptPrintf(rpt,"inst:%i ",s->instId );

  switch( s->selId )
  {
    case kStatusTmId:
      {
        const cmChar_t* label = cmTaskMgrStatusIdToLabel(s->statusId); 
        cmRptPrintf(rpt,"status '%s'",label);
      }
      break;

    case kProgTmId:
      cmRptPrintf(rpt,"prog %i",s->prog);
      break;

    case kErrorTmId:
      cmRptPrintf(rpt,"error %s",cmStringNullGuard(s->msg));
      break;

    default:
      { assert(0); }
  }

  cmRptPrintf(rpt,"\n");

}

// Test client status callback function.
void _cmTmTestStatusCb( const cmTaskMgrStatusArg_t* s  )
{
   // s.arg set from cmTaskMgrCreate( ..., statusCbArg, ...);
  cmTmTestApp_t* app = (cmTmTestApp_t*)s->arg;
  unsigned i;

  // locate the instance record assoc'd with this callback
  for(i=0; i<kMaxTestInstCnt; ++i)
    if( app->insts[i].instId ==  s->instId )
      break;

  if( i==kMaxTestInstCnt )
    cmRptPrintf(app->err->rpt,"instId %i not found.\n",s->instId);

  _cmTmTestReportStatus(app->err->rpt,s);
}


// Test worker function.
void _cmTmTestFunc(cmTaskMgrFuncArg_t* arg )
{
  if( cmTaskMgrWorkerHandleCommand(arg) == kStopTmwRC )
    return;

  unsigned prog = 0;

  for(; prog<arg->progCnt; ++prog)
  {
    if( cmTaskMgrWorkerHandleCommand(arg) == kStopTmwRC )
      break;

    cmSleepMs(1000); 

    if( cmTaskMgrWorkerSendProgress(arg,prog,NULL) == kStopTmwRC )
      break;
  }
  
}


cmTmRC_t cmTaskMgrTest(cmCtx_t* ctx)
{
  cmTmRC_t        rc           = kOkTmRC;
  cmTaskMgrH_t    tmH          = cmTaskMgrNullHandle;
  unsigned        threadCnt    = 2;
  unsigned        queueByteCnt = 1024;
  unsigned        pauseSleepMs = 50;
  unsigned        nextInstId   = 0;
  unsigned        taskId       = 0;
  const cmChar_t* taskLabel    = "Task Label";
  cmTmTestApp_t   app;
  char            c;

  memset(&app,0,sizeof(app));
  app.err = &ctx->err;

  // create the task mgr
   if( cmTaskMgrCreate( ctx,&tmH,_cmTmTestStatusCb,&app,threadCnt,queueByteCnt,pauseSleepMs) != kOkTmRC )
  {
    rc = cmErrMsg(&ctx->err,kTestFailTmRC,"Task mgr create failed.");
    goto errLabel;
  }

  // install a task
   if( cmTaskMgrInstall(tmH, taskId, taskLabel, _cmTmTestFunc, NULL ) != kOkTmRC )
  {
    rc = cmErrMsg(&ctx->err,kTestFailTmRC,"Task mgr task install failed.");
    goto errLabel;    
  }


  // go into interactive mode
  printf("q=quit e=enable c=call i=idle\n");
  while((c = getchar()) != 'q')
  {
    switch(c)
    {
      case 'i':
        cmTaskMgrOnIdle(tmH);
        cmRptPrintf(&ctx->rpt,"idled\n");
        break;

      case 'e':
        {
          // toggle the enable state of the task mgr.
          bool fl = !cmTaskMgrIsEnabled(tmH);
          if( cmTaskMgrEnable(tmH,fl) != kOkTmRC )
            rc = cmErrMsg(&ctx->err,kTestFailTmRC,"Test enable failed.");
          else
            cmRptPrintf(&ctx->rpt,"%s\n", fl ? "enabled" : "disabled" );
        }
        break;

      case 'c':
          if( nextInstId < kMaxTestInstCnt )
          {
            void* funcArg = app.insts + nextInstId;
            unsigned progCnt = 5;
            if( cmTaskMgrCall( tmH, taskId, funcArg, progCnt, queueByteCnt, "My Inst", &app.insts[nextInstId].instId ) != kOkTmRC )
              rc = cmErrMsg(&ctx->err,kTestFailTmRC,"Test call failed.");            
            else
            {
              ++nextInstId;
              cmRptPrintf(&ctx->rpt,"called\n");
            }
          }
    }
  }

  

 errLabel:
  // destroy the task mgr
  if( cmTaskMgrDestroy(&tmH) != kOkTmRC )
    rc = cmErrMsg(&ctx->err,kTestFailTmRC,"Task mgr destroy failed.");

  return rc;
}