互斥和条件变量实现线程间通信

这部分代码是从某项目中看到的，写的挺好，值得我学习。

头文件：

#ifndef _XXX_MBX_H_
#define _XXX_MBX_H_

#include <XXX_msgq.h>

#define XXX_MBX_WAIT_ACK      0x0002 ///< Message Flag : Wait for ACK
#define XXX_MBX_FREE_PRM      0x0004 ///< Message Flag : Message parameters are malloc'ed and need to be free'ed

#define XXX_MBX_BROADCAST_MAX     10 ///< Max message queues/PRC's that can be broadcast to


#define XXX_memAlloc(size) (void*)malloc((size))


/**
  \brief Mailbox handle
*/
typedef struct {

  XXX_MsgqHndl rcvMbx; ///< Receive mailbox
  XXX_MsgqHndl ackMbx; ///< ACK mailbox

} XXX_MbxHndl;


int XXX_mbxCreate(XXX_MbxHndl *pHndl);
int XXX_mbxDelete(XXX_MbxHndl *pHndl);
int XXX_mbxSendMsg(XXX_MbxHndl *pTo, XXX_MbxHndl *pFrom, Uint16 cmd, void *pPrm, Uint16 flags);
int XXX_mbxBroadcastMsg(XXX_MbxHndl *pToList[], XXX_MbxHndl *pFrom, Uint16 cmd, void *pPrm, Uint16 flags);
int XXX_mbxAckOrFreeMsg(XXX_MsgHndl *pMsg, int ackRetVal);
int XXX_mbxWaitMsg(XXX_MbxHndl *pHndl, XXX_MsgHndl **pMsg);
int XXX_mbxCheckMsg(XXX_MbxHndl *pHndl, XXX_MsgHndl **pMsg);
int XXX_mbxWaitCmd(XXX_MbxHndl *pHndl, XXX_MsgHndl **pMsg, Uint16 waitCmd);
int XXX_mbxFlush(XXX_MbxHndl *pHndl);


#endif /* _XXX_MBX_H_ */

C文件：

#include <XXX_msgq.h>

int XXX_msgqCreate(XXX_MsgqHndl *hndl)
{
  pthread_mutexattr_t mutex_attr;
  pthread_condattr_t cond_attr;
  int status=XXX_SOK;

  status |= pthread_mutexattr_init(&mutex_attr);
  status |= pthread_condattr_init(&cond_attr);

  status |= pthread_mutex_init(&hndl->lock, &mutex_attr);
  status |= pthread_cond_init(&hndl->condRd, &cond_attr);
  status |= pthread_cond_init(&hndl->condWr, &cond_attr);

  hndl->curRd = hndl->curWr = 0;
  hndl->count = 0;
  hndl->len = XXX_MSGQ_LEN_MAX;

  if(status!=XXX_SOK)
    XXX_ERROR("XXX_msgqCreate() = %d \r\n", status);

  pthread_condattr_destroy(&cond_attr);
  pthread_mutexattr_destroy(&mutex_attr);

  return status;
}

int XXX_msgqDelete(XXX_MsgqHndl *hndl)
{
  pthread_cond_destroy(&hndl->condRd);
  pthread_cond_destroy(&hndl->condWr);
  pthread_mutex_destroy(&hndl->lock);

  return XXX_SOK;
}

XXX_MsgHndl *XXX_msgqAllocMsgHndl(XXX_MsgqHndl *to, XXX_MsgqHndl *from, Uint16 cmd, void *prm, Uint16 msgFlags)
{
  XXX_MsgHndl *msg;

  msg = XXX_memAlloc( sizeof(XXX_MsgHndl) );

  if(msg!=NULL) {
    msg->pTo = to;
    msg->pFrom = from;
    msg->pPrm = prm;
    msg->status = XXX_SOK;
    msg->cmd = cmd;
    msg->flags = msgFlags;
  }

  return msg;
}

int XXX_msgqSend(XXX_MsgqHndl *hndl, XXX_MsgHndl *msg, Uint32 timeout)
{
  int status = XXX_EFAIL;

  pthread_mutex_lock(&hndl->lock);

  while(1) {
    if( hndl->count < hndl->len ) {
      hndl->queue[hndl->curWr] = msg;
      hndl->curWr = (hndl->curWr+1)%hndl->len; /* 周而复始 */
      hndl->count++;
      status = XXX_SOK;
      pthread_cond_signal(&hndl->condRd);
      break;
    } else {
      if(timeout == XXX_TIMEOUT_NONE)
        break;

      status = pthread_cond_wait(&hndl->condWr, &hndl->lock);
    }
  }

  pthread_mutex_unlock(&hndl->lock);

  return status;
}


int XXX_msgqSendMsg(XXX_MsgqHndl *to, XXX_MsgqHndl *from, Uint16 cmd, void *prm, Uint16 msgFlags, XXX_MsgHndl **msg)
{
  int status;
  XXX_MsgHndl *data;

  data = XXX_msgqAllocMsgHndl(to, from, cmd, prm, msgFlags);
  if(data==NULL)
    return XXX_EFAIL;

  status = XXX_msgqSend(to, data, XXX_TIMEOUT_FOREVER);

  if(status==XXX_SOK) {
    if(msg!=NULL)
      *msg = data;
  }

  return status;
}

int XXX_msgqRecvMsg(XXX_MsgqHndl *hndl, XXX_MsgHndl **msg, Uint32 timeout)
{
  int status = XXX_EFAIL;

  pthread_mutex_lock(&hndl->lock);

  while(1) {
    if(hndl->count > 0 ) {

      if(msg!=NULL) {
        *msg = hndl->queue[hndl->curRd];
      }

      hndl->curRd = (hndl->curRd+1)%hndl->len;
      hndl->count--;
      status = XXX_SOK;
      pthread_cond_signal(&hndl->condWr);
      break;
    } else {
      if(timeout == XXX_TIMEOUT_NONE)
        break;

      status = pthread_cond_wait(&hndl->condRd, &hndl->lock);
    }
  }

  pthread_mutex_unlock(&hndl->lock);

  return status;
}

int XXX_msgqSendAck(XXX_MsgHndl *msg, int ackRetVal)
{
  int status;

  msg->status = ackRetVal;

  status = XXX_msgqSend(msg->pFrom, msg, XXX_TIMEOUT_FOREVER);

  return status;
}

int XXX_msgqFreeMsgHndl(XXX_MsgHndl *msg)
{
  XXX_memFree(msg);
  return XXX_SOK;
}