chan.c

#include <stdio.h>   // for printf used in reporting errors, etc...
#include <string.h>
#include "config.h"
#include "thread.h"
#include "chan.h"
#include "fifo.h"

#define SUCCESS (0) 
#define FAILURE (1)

#define DEBUG_CHAN

//////////////////////////////////////////////////////////////////////
//  Logging    ///////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////// 
void chan_breakme() {}
#define chan_warning(...) printf(__VA_ARGS__);
#define chan_error(...)   do{fprintf(stderr,__VA_ARGS__);chan_breakme(); exit(-1);}while(0)

#ifdef DEBUG_CHAN
#define chan_debug(...)   printf(__VA_ARGS__)
#else
#define chan_debug(...)
#endif

// debug
#ifdef DEBUG_CHAN
#define Chan_Assert(expression) \
  if(!(expression))\
    chan_error("Assertion failed: %s\n\tIn %s (line: %u)\n", #expression, __FILE__ , __LINE__ )
#else
#define Chan_Assert(expression) (expression)  
#define DEBUG_REQUEST_STORAGE
#endif

#if 0
#define DEBUG_SHOW_REFS chan_debug("%s(%d) - Refs: %d"ENDL,__FILE__,__LINE__,c->q->ref_count)
#else
#define DEBUG_SHOW_REFS
#endif

#define CHAN_ERR__INVALID_MODE chan_error("Error: At %s(%d)"ENDL \
                                          "\tChannel has invalid read/write mode."ENDL, \
                                          __FILE__,__LINE__)

#define CHAN_WRN__NULL_ARG(e)  chan_warning("Warning: At %s(%d)"ENDL \
                                            "\tArgument <%s> was NULL."ENDL, \
                                            __FILE__,__LINE__,#e)

#define HERE printf("HERE: Line % 5d File: %s\n",__LINE__,__FILE__)

//////////////////////////////////////////////////////////////////////
//  Utilities  ///////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////
#define return_if_fail(cond)          { if(!(cond)) return; }
#define return_val_if(cond,val)       { if( (cond)) return (val); }
#define goto_if_not(e,lbl)            { if(!(e)) goto lbl; }
#define goto_if(e,lbl)                { if(e) goto lbl; }

//////////////////////////////////////////////////////////////////////
//  Chan       ///////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////

typedef uint32_t u32;

typedef struct
{ Fifo *fifo;  

  u32 ref_count;
  u32 nreaders;
  u32 nwriters;
  u32 expand_on_full;
  u32 flush;
  
  Mutex              lock;
  Condition          notfull;  //predicate: not full  || expand_on_full
  Condition          notempty; //predicate: not empty || no writers 
  Condition          changedRefCount; //predicate: refcount != n
  Condition          haveWriter; //predicate: nwriters>0
  Condition          haveReader; //predicate: nreaders>0

  void              *workspace;  // Token buffer used for copy operations.
} __chan_t;

typedef struct _chan
{ 
  __chan_t *q;
  ChanMode  mode;
} chan_t;

__chan_t* chan_alloc(size_t buffer_count, size_t buffer_size_bytes)
{ __chan_t *c=0;
  Fifo *fifo;
  if(fifo=Fifo_Alloc(buffer_count,buffer_size_bytes))
  { Chan_Assert(c=(__chan_t*)calloc(1,sizeof(__chan_t)));
    c->fifo = fifo;
    c->lock = MUTEX_INITIALIZER_INSTANCE;
    Condition_Initialize(&c->notfull);
    Condition_Initialize(&c->notempty);
    Condition_Initialize(&c->changedRefCount);
    Condition_Initialize(&c->haveWriter);
    Condition_Initialize(&c->haveReader);
    c->workspace = Fifo_Alloc_Token_Buffer(c->fifo);
    c->ref_count=1;
  }
  return c;
}

void chan_destroy(__chan_t* c)
{ //precondition - called when the last reference is released
  //             - nobody should be waiting
  Fifo_Free_Token_Buffer(c->workspace);
  Fifo_Free(c->fifo);
  free(c);
}

Chan* Chan_Alloc( size_t buffer_count, size_t buffer_size_bytes)
{ chan_t   *c=0;
  __chan_t *q=0;
  if(q=chan_alloc(buffer_count,buffer_size_bytes))
  { Chan_Assert(c=(chan_t*)malloc(sizeof(chan_t)));
    c->q = q;
    c->mode = CHAN_NONE;
  }
  return (Chan*)c;
}

inline
Chan *Chan_Alloc_Copy( Chan *chan)
{ return Chan_Alloc(Chan_Buffer_Count(chan),Chan_Buffer_Size_Bytes(chan));
}

// must be called from inside a lock
chan_t* incref(chan_t *c)
{ chan_t *n;
  ++(c->q->ref_count);
  goto_if_not(n=malloc(sizeof(chan_t)),ErrorAlloc);
  memcpy(n,c,sizeof(chan_t));
  DEBUG_SHOW_REFS;
  Condition_Notify_All(&c->q->changedRefCount);
  return n;
ErrorAlloc:
  --(c->q->ref_count);
  //Mutex_Unlock(&c->q->lock);
  chan_error("malloc() call failed."ENDL);
  return 0;
}

void decref(chan_t **pc)
{ u32 remaining;
  chan_t *c;
  Chan_Assert(pc);
  c=*pc; *pc=0;
  Chan_Assert(c);
  Mutex_Lock(&c->q->lock);
  remaining = --(c->q->ref_count);
  DEBUG_SHOW_REFS;
  Mutex_Unlock(&c->q->lock);
  Chan_Assert(remaining>=0);
  if(remaining==0)
    chan_destroy(c->q);
  else
    Condition_Notify_All(&c->q->changedRefCount);
    
  free(c);
}

/** \returns NULL on error, otherwise a valid Chan* 
    
    Errors usually cause a panic.
*/
Chan* Chan_Open( Chan *self, ChanMode mode)
{ chan_t *n,*c;
  c = (chan_t*)self;
  Chan_Assert( Chan_Get_Ref_Count(self)>0 );
  Mutex_Lock(&c->q->lock);
  goto_if_not(n = incref(c),ErrorIncref);
  n->mode = mode;
  switch(mode)
  { case CHAN_READ:      
      ++(n->q->nreaders);
      if(Fifo_Is_Empty(n->q->fifo))
        n->q->flush=0;
      Condition_Notify_All(&n->q->haveReader);
      break;
    case CHAN_WRITE: 
      ++(n->q->nwriters);
      n->q->flush=0;
      Condition_Notify_All(&n->q->haveWriter);
      break;
    case CHAN_NONE:
      break;
    default:
      CHAN_ERR__INVALID_MODE;
      break;
  }
ErrorIncref: 
  //n will be null if there's an error.
  //The error should already be reported.
  //This is here just in case the error doesn't cause a panic.
  Mutex_Unlock(&c->q->lock);
  return (Chan*)n;
}

int Chan_Close( Chan *self_ )
{ chan_t *self = (chan_t*)self_;
  int notify=0;
  if(!self)
  {
    //CHAN_WRN__NULL_ARG(self);
    return SUCCESS;
  }
  Mutex_Lock(&self->q->lock);
  { __chan_t *q = self->q;
    switch(self->mode)
    { case CHAN_READ:  
        Chan_Assert( (--(q->nreaders))>=0 );
        if(q->nreaders==0)
          q->flush=0;
        break;
      case CHAN_WRITE:
        Chan_Assert( (--(q->nwriters))>=0 );
        Condition_Notify_All(&q->haveWriter);
        notify = (q->nwriters==0);
        if(notify)
          q->flush=1;
        break;
    }
  }
  if(notify)
    Condition_Notify_All(&self->q->notempty);
  Mutex_Unlock(&self->q->lock);
  decref(&self);
  return SUCCESS;
}

unsigned Chan_Get_Ref_Count(Chan* self_)
{ chan_t *self = (chan_t*)self_; 
  return self->q->ref_count;
}

void Chan_Wait_For_Ref_Count(Chan* self_,size_t n)
{ chan_t *self = (chan_t*)self_; 
  __chan_t *q = self->q;
  Mutex_Lock(&q->lock);
  while(q->ref_count!=n)
    Condition_Wait(&q->changedRefCount,&q->lock);
  chan_debug("WaitForRefCount: target: %zu        now: %u"ENDL,n,q->ref_count);
  Mutex_Unlock(&q->lock);
}

void Chan_Wait_For_Writer_Count(Chan* self_,size_t n)
{ chan_t *self = (chan_t*)self_; 
  __chan_t *q = self->q;
  Mutex_Lock(&q->lock);
  while(q->nwriters!=n)
    Condition_Wait(&q->haveWriter,&q->lock);
  chan_debug("WaitForWriter - writer count: %d"ENDL,q->nwriters);
  Mutex_Unlock(&q->lock);
}

void Chan_Wait_For_Have_Reader(Chan* self_)
{ chan_t *self = (chan_t*)self_; 
  __chan_t *q = self->q;
  Mutex_Lock(&q->lock);
  while(q->nreaders==0)
    Condition_Wait(&q->haveReader,&q->lock);
  chan_debug("WaitForHaveReader - reader count: %d"ENDL,q->nreaders);
  Mutex_Unlock(&q->lock);
}

void Chan_Set_Expand_On_Full( Chan* self_, int expand_on_full)
{ chan_t *self = (chan_t*)self_;  
  self->q->expand_on_full=expand_on_full;
  if(expand_on_full)
    Condition_Notify_All(&self->q->notfull);
}

// ----
// Next
// ----

unsigned int chan_push__locked(__chan_t *q, void **pbuf, size_t sz, unsigned timeout_ms)
{ 
  while(Fifo_Is_Full(q->fifo) && q->expand_on_full==0)
    if(!Condition_Timed_Wait(&q->notfull,&q->lock,timeout_ms))
      return FAILURE; // timeout
  if(FIFO_SUCCESS(Fifo_Push(q->fifo,pbuf,sz,q->expand_on_full)))
    return SUCCESS;
  return FAILURE;
}

inline int _pop_bypass_wait(__chan_t *q)
{ 
  return q->nwriters==0 && q->flush;  
}

unsigned int chan_pop__locked(__chan_t *q, void **pbuf, size_t sz, unsigned timeout_ms)
{ //int starved;
  while(Fifo_Is_Empty(q->fifo) && !_pop_bypass_wait(q))
    if(!Condition_Timed_Wait(&q->notempty,&q->lock,timeout_ms))
      return FAILURE; //timeout
  //starved = Fifo_Is_Empty(q->fifo) && q->nwriters==0;
  if(FIFO_SUCCESS(Fifo_Pop(q->fifo,pbuf,sz)))
    return SUCCESS;
  return FAILURE;
}

inline int _peek_bypass_wait(__chan_t *q)
{ 
  return q->nwriters==0 && q->flush;  
}

unsigned int chan_peek__locked(__chan_t *q, void **pbuf, size_t sz, unsigned timeout_ms)
{ //int starved;
  while(Fifo_Is_Empty(q->fifo) && !_peek_bypass_wait(q))
    Condition_Wait(&q->notempty,&q->lock); //ingore timeout on peek
  //starved = Fifo_Is_Empty(q->fifo) && q->nwriters==0;
  if(FIFO_SUCCESS(Fifo_Peek(q->fifo,pbuf,sz)))
    return SUCCESS;
  return FAILURE;
}

unsigned int chan_push(chan_t *self, void **pbuf, size_t sz, int copy, unsigned timeout_ms)
{ // TO SELF: use timeout=0 for try 
  // precondition: this should be a "Write" mode channel
  Mutex_Lock(&self->q->lock);
  { __chan_t *q = self->q;
    if(timeout_ms==0)
      goto_if(Fifo_Is_Full(q->fifo),NoPush);
    if(copy)
    { Fifo_Resize(q->fifo,sz);
      Fifo_Resize_Token_Buffer(q->fifo,&q->workspace);
      memcpy(q->workspace,*pbuf,sz);
      goto_if(CHAN_FAILURE(chan_push__locked(q,&q->workspace,sz,timeout_ms)),NoPush);
    } else
    {
      goto_if(CHAN_FAILURE(chan_push__locked(q,pbuf,sz,timeout_ms)),NoPush);
    }
  }
  Mutex_Unlock(&self->q->lock);
  Condition_Notify(&self->q->notempty);
  return SUCCESS;
NoPush:
  Mutex_Unlock(&self->q->lock);
  return FAILURE;
}

unsigned int chan_pop(chan_t *self, void **pbuf, size_t sz, int copy, unsigned timeout_ms)
{ 
  Mutex_Lock(&self->q->lock);
  { __chan_t *q = self->q;
    if(timeout_ms==0)
      goto_if(Fifo_Is_Empty(q->fifo),NoPop);
    if(copy)
    { Fifo_Resize(q->fifo,sz);
      Fifo_Resize_Token_Buffer(q->fifo,&q->workspace);
      goto_if(CHAN_FAILURE(chan_pop__locked(q,&q->workspace,sz,timeout_ms)),NoPop);
      memcpy(*pbuf,q->workspace,sz);
    } else
      goto_if(CHAN_FAILURE(chan_pop__locked(q,pbuf,sz,timeout_ms)),NoPop);
  }            
  Condition_Notify(&self->q->notfull);
  Mutex_Unlock(&self->q->lock);
  return SUCCESS;
NoPop:
  Mutex_Unlock(&self->q->lock);
  return FAILURE;
}

unsigned int chan_peek(chan_t *self, void **pbuf, size_t sz, unsigned timeout_ms)
{ 
  Mutex_Lock(&self->q->lock);
  { __chan_t *q = self->q;
    if(timeout_ms==0)
      goto_if(Fifo_Is_Empty(q->fifo),NoPeek);
    goto_if(Fifo_Is_Empty(q->fifo) && q->nwriters==0,NoPeek); // possibly avoid the resize/copy
    goto_if(CHAN_FAILURE(chan_peek__locked(q,pbuf,sz,timeout_ms)),NoPeek);
  }
  Mutex_Unlock(&self->q->lock);
  // no size change so no notify
  return SUCCESS;
NoPeek:
  Mutex_Unlock(&self->q->lock);
  return FAILURE;
}

unsigned int Chan_Next( Chan *self_, void **pbuf, size_t sz)
{ chan_t *self = (chan_t*)self_;
  switch(self->mode)
  { case CHAN_READ:  return chan_pop (self,pbuf,sz,0,(unsigned)-1); break;
    case CHAN_WRITE: return chan_push(self,pbuf,sz,0,(unsigned)-1); break;
    default:
      CHAN_ERR__INVALID_MODE;
      break;
  }
  return FAILURE;
}

unsigned int Chan_Next_Copy( Chan *self_, void  *buf,  size_t sz) 
{ chan_t *self = (chan_t*)self_;
  switch(self->mode)
  { case CHAN_READ:  return chan_pop (self,&buf,sz,1,(unsigned)-1); break;
    case CHAN_WRITE: return chan_push(self,&buf,sz,1,(unsigned)-1); break;
    default:
      CHAN_ERR__INVALID_MODE;
      break;
  }
  return FAILURE;
}

unsigned int Chan_Next_Try( Chan *self_, void **pbuf, size_t sz)                     
{ chan_t *self = (chan_t*)self_;
  switch(self->mode)
  { case CHAN_READ:  return chan_pop (self,pbuf,sz,0,0); break;
    case CHAN_WRITE: return chan_push(self,pbuf,sz,0,0); break;
    default:
      CHAN_ERR__INVALID_MODE;
      break;
  }
  return FAILURE;
}

unsigned int Chan_Next_Copy_Try( Chan *self_, void  *buf,  size_t sz) 
{ chan_t *self = (chan_t*)self_;
  switch(self->mode)
  { case CHAN_READ:  return chan_pop (self,&buf,sz,1,0); break;
    case CHAN_WRITE: return chan_push(self,&buf,sz,1,0); break;
    default:
      CHAN_ERR__INVALID_MODE;
      break;
  }
  return FAILURE;
}

unsigned int Chan_Next_Timed( Chan *self_, void **pbuf, size_t sz, unsigned timeout_ms )
{ chan_t *self = (chan_t*)self_;
  switch(self->mode)
  { case CHAN_READ:  return chan_pop (self,pbuf,sz,0,timeout_ms); break;
    case CHAN_WRITE: return chan_push(self,pbuf,sz,0,timeout_ms); break;
    default:
      CHAN_ERR__INVALID_MODE;
      break;
  }
  return FAILURE;
}


// ----
// Peek
// ----
//
// Requires a read mode channel.
//
unsigned int Chan_Peek( Chan *self_, void **pbuf, size_t sz )
{ chan_t *self = (chan_t*)self_; 
  return chan_peek(self,pbuf,sz,(unsigned)-1);
}

unsigned int Chan_Peek_Try( Chan *self_, void **pbuf, size_t sz )
{ chan_t *self = (chan_t*)self_; 
  return chan_peek(self,pbuf,sz,0);
}

unsigned int Chan_Peek_Timed ( Chan *self_, void **pbuf, size_t sz, unsigned timeout_ms )
{ chan_t *self = (chan_t*)self_;   
  return chan_peek(self,pbuf,sz,timeout_ms);
}


// -----------------
// Memory management
// -----------------

#define FIFO(e) (((chan_t*)(e))->q->fifo)
int Chan_Is_Full( Chan *self )
{ return Fifo_Is_Full( FIFO(self) );
}

int Chan_Is_Empty( Chan *self )
{ return Fifo_Is_Empty( FIFO(self) );
}

inline void Chan_Resize( Chan* self, size_t nbytes)
{ Fifo_Resize( FIFO(self),nbytes );
}


void* Chan_Token_Buffer_Alloc( Chan *self )
{ return Fifo_Alloc_Token_Buffer(FIFO(self));
}

void* Chan_Token_Buffer_Alloc_And_Copy( Chan *self, void *src )
{ Fifo *fifo = FIFO(self);
  size_t sz  = Fifo_Buffer_Size_Bytes(fifo);
  void *buf  = Fifo_Alloc_Token_Buffer(fifo);
  memcpy(buf,src,sz);
  return buf;
}

void Chan_Token_Buffer_Free( void *buf )
{ Fifo_Free_Token_Buffer(buf);
}

inline
size_t Chan_Buffer_Size_Bytes( Chan *self )
{ return Fifo_Buffer_Size_Bytes(FIFO(self));
} 

inline
size_t Chan_Buffer_Count( Chan *self )
{ return Fifo_Buffer_Count(FIFO(self));
} 

inline Chan* Chan_Id( Chan *self )
{ return ((chan_t*)self)->q;
}