text: /*	$Header: /usr/sctpCVS/rserpool/lib/distributor.c,v 1.1 2006-05-30 18:02:01 randall Exp $ */
  
  /*
   * Copyright (C) 2002 Cisco Systems Inc,
   * All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions
   * are met:
   * 1. Redistributions of source code must retain the above copyright
   *    notice, this list of conditions and the following disclaimer.
   * 2. Redistributions in binary form must reproduce the above copyright
   *    notice, this list of conditions and the following disclaimer in the
   *    documentation and/or other materials provided with the distribution.
   * 3. Neither the name of the project nor the names of its contributors
   *    may be used to endorse or promote products derived from this software
   *    without specific prior written permission.
   *
   * THIS SOFTWARE IS PROVIDED BY THE PROJECT AND CONTRIBUTORS ``AS IS'' AND
   * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
   * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   * ARE DISCLAIMED.  IN NO EVENT SHALL THE PROJECT OR CONTRIBUTORS BE LIABLE
   * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
   * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
   * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
   * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
   * SUCH DAMAGE.
   */
  
  #include <distributor.h>
  
  hashableHash *
  create_hashableHash(int pid,char *string,int sz)
  {
    hashableHash *o;
    o = calloc(1,sizeof(hashableHash));
    if(o != NULL){
      o->hash = HashedTbl_create(string,sz);
      if(o->hash != NULL){
        o->protocolId = pid;
      }else{
        free(o);
        o = NULL;
      }
    }
    return(o);
  }
  
  void
  destroy_hashableHash(hashableHash *o)
  {
    HashedTbl_destroy(o->hash);
    free(o);
  }
  
  
  
  hashableDlist *
  create_hashdlist(int strm)
  {
    hashableDlist *o;
    o = calloc(1,sizeof(hashableDlist));
    if(o != NULL){
      o->list = dlist_create();
      if(o->list != NULL){
        o->streamNo = strm;
      }else{
        free(o);
        o = NULL;
      }
    }
    return(o);
  }
  
  void
  destroy_hashDlist(hashableDlist *o)
  {
    dlist_destroy(o->list);
    free(o);
  }
  
  
  distributor *
  createDistributor()
  {
    int ret,i;
    distributor *o;
    o = (distributor *)calloc(1,sizeof(distributor));
    if(o == NULL)
      return(NULL);
    o->fdSubscribersList = dlist_create();
    if(o->fdSubscribersList == NULL){
      /* failed to create */
      free(o);	
      return(NULL);
    }
    o->lazyClockTickList = dlist_create();
    if(o->lazyClockTickList == NULL){
      dlist_destroy(o->fdSubscribersList);
      free(o);	
      return(NULL);
    }
    o->startStopList = dlist_create();
    if(o->startStopList == NULL){
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      free(o);	
      return(NULL);
    }
  
  
    /* we specify 11 subscribers */
    o->fdSubscribers = HashedTbl_create("File Descriptor Subscribers", 11);
    if(o->fdSubscribers == NULL){
      /* failed to create */
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);	
      return(NULL);
    }
    o->msgSubscriberList = dlist_create();
    if(o->msgSubscriberList == NULL){
      /* failed to create */
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);	
      return(NULL);
    }
    o->msgSubscribers = HashedTbl_create("Master Hash of Hashes", 11);
    if(o->msgSubscribers == NULL){
      /* failed to create */
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);	
      return(NULL);
    }
    o->msgSubZero = create_hashableHash(DIST_SCTP_PROTO_ID_DEFAULT,
  				      "Master Hash of Zero PID",
  				      11);
    if(o->msgSubZero == NULL){
      /* failed to create */
      HashedTbl_destroy(o->msgSubscribers);
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);	
      return(NULL);
    }
    o->msgSubZeroNoStrm = create_hashdlist(DIST_STREAM_DEFAULT);
    if(o->msgSubZeroNoStrm == NULL){
      /* failed to create */
      destroy_hashableHash(o->msgSubZero);
      HashedTbl_destroy(o->msgSubscribers);
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);	
      return(NULL);
    }
    o->numfds = DEFAULT_POLL_LIST_SZ;
    o->numused = 0;
    o->fdlist = (struct pollfd *)calloc(DEFAULT_POLL_LIST_SZ,sizeof(struct pollfd));
    if(o->fdlist == NULL){
      /* failed to create */
      destroy_hashDlist(o->msgSubZeroNoStrm);
      destroy_hashableHash(o->msgSubZero);
      HashedTbl_destroy(o->msgSubscribers);
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);
      return(NULL);
    }
    /* init fd structs */
    for(i=0;i<o->numfds;i++){
      o->fdlist[i].fd = -1;
      o->fdlist[i].events = 0;
      o->fdlist[i].revents = 0;
    }
    o->timerlist = dlist_create();
    if(o->timerlist == NULL){
      /* failed to create */
      free(o->fdlist);
      destroy_hashDlist(o->msgSubZeroNoStrm);
      destroy_hashableHash(o->msgSubZero);
      HashedTbl_destroy(o->msgSubscribers);
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);
      return(NULL);
    }
    /* now that everything is created lets set some
     * of the relationships up.
     */
  
    /* we first insert into the msgSubZero Hash table the
     * default stream list.
     */
    ret = HashedTbl_enterKeyed(o->msgSubZero->hash,
  		       o->msgSubZeroNoStrm->streamNo,
  		       (void*)o->msgSubZeroNoStrm,
  		       (void*)&o->msgSubZeroNoStrm->streamNo,
  		       sizeof(int));
    if(ret != LIB_STATUS_GOOD){
      /* We have a problem. bail. */
      dlist_destroy(o->timerlist);
      free(o->fdlist);
      destroy_hashDlist(o->msgSubZeroNoStrm);
      destroy_hashableHash(o->msgSubZero);
      HashedTbl_destroy(o->msgSubscribers);
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);
      return(NULL);
    }
  
    /* Now we must put the default PID hash into
     * the master hashtable.
     */
    ret = HashedTbl_enterKeyed(o->msgSubscribers,
  			     o->msgSubZero->protocolId,
  		       (void*)o->msgSubZero,
  		       (void*)&o->msgSubZero->protocolId,
  		       sizeof(int));
    if(ret != LIB_STATUS_GOOD){
      /* We have a problem. bail. */
      dlist_destroy(o->timerlist);
      free(o->fdlist);
      destroy_hashDlist(o->msgSubZeroNoStrm);
      destroy_hashableHash(o->msgSubZero);
      HashedTbl_destroy(o->msgSubscribers);
      dlist_destroy(o->msgSubscriberList);
      HashedTbl_destroy(o->fdSubscribers);
      dlist_destroy(o->fdSubscribersList);
      dlist_destroy(o->lazyClockTickList);
      dlist_destroy(o->startStopList);
      free(o);
      return(NULL);
    }
    /* set the I am not done flag */
    o->notdone = 1;
  
    /* setup clock tick */
    o->idleClockTick.tv_sec = DIST_IDLE_CLOCK_SEC;
    o->idleClockTick.tv_usec = DIST_IDLE_CLOCK_USEC;
    /* get the last known time setup */
    gettimeofday(&o->lastKnownTime,(struct timezone *)NULL);
  
    return(o);
  }
  
  void
  destroyDistributor(distributor *o)
  {
    timerEntry *te;
    fdEntry *fe;
    hashableHash *hh;
    hashableDlist *he;
    msgConsumer *mc;
    auditTick *at;
    StartStop *sat;
    void *keyP;
    /* we must get each element out of the
     * multi-keyed tables before destroying.
     * since both hashable lists that we
     * keep a pointer to for shortcuts in
     * our main structure are in the individual lists
     * and will be found that way there is no need
     * to do a :
     * destroy_hashDlist(o->msgSubZeroNoStrm);
     * <or>
     * destroy_hashableHash(o->msgSubZero);
     */
  
    /* the list of timers first. */
    dlist_reset(o->timerlist);
    while((te = (timerEntry *)dlist_getNext(o->timerlist)) != NULL){
      free(te);
    }
    dlist_destroy(o->timerlist);
  
    /* now free audit tick list */
    dlist_reset(o->lazyClockTickList);
    while((at = (auditTick *)dlist_getNext(o->lazyClockTickList)) != NULL){
      free(at);
    }
    dlist_destroy(o->lazyClockTickList);
  
    /* now free the start/stop list */
    dlist_reset(o->startStopList);
    while((sat = (StartStop *)dlist_getNext(o->startStopList)) != NULL){
      free(sat);
    }
    dlist_destroy(o->startStopList);
  
    /* get rid of all the file descriptors */
    free(o->fdlist);
    /* here we dump the hash table since all entries
     * are also maintained on the dlist too
     */
    HashedTbl_destroy(o->fdSubscribers);
  
    /* Now go through and free up all on our list */
    dlist_reset(o->fdSubscribersList);
    while((fe = (fdEntry *)dlist_getNext(o->fdSubscribersList)) != NULL){
      free(fe);
    }
    dlist_destroy(o->fdSubscribersList);
  
    /* Now we must go through the hash table
     * of hashableHashes. And for each hashableHash
     * pull all the hashableLists out. Each list is
     * just destroyed since we will cleanup the underlying
     * msgConsumers when we destroy the global msgSubscriberList.
     *
     */
    HashedTbl_rewind(o->msgSubscribers);
    while((hh = (hashableHash *)HashedTbl_getNext(o->msgSubscribers,&keyP)) != NULL){
      /* we have a hashable hash, clean it up */
      HashedTbl_rewind(hh->hash);
      while((he = (hashableDlist *)HashedTbl_getNext(hh->hash,&keyP)) != NULL){
        /* we purge since list entries are all in msgSubscribers */
        destroy_hashDlist(he);
      }
      destroy_hashableHash(hh);
    }
    HashedTbl_destroy(o->msgSubscribers);
  
    /* Now lets go out and get the msg subscriber entries too. */
    dlist_reset(o->msgSubscriberList);
    while((mc = (msgConsumer *)dlist_getNext(o->msgSubscriberList)) != NULL){
      free(mc);
    }
    dlist_destroy(o->msgSubscriberList);
    /* ok free the base and we are clean */
    free(o);
  }
  
  int
  dist_TimerStart(distributor *o,timerFunc f, u_long a, u_long b, void *c, void *d)
  {
    int ret;
    timerEntry *te,*pe;
    /* allocate a new timer element */
    te = calloc(1,sizeof(timerEntry));
    if(te == NULL){
      return(LIB_STATUS_BAD);
    }
    te->arg1 = c;
    te->arg2 = d;
    te->action = f;
    te->tv_sec = a;
    te->tv_usec = b;
  #ifdef MINIMIZE_TIME_CALLS
    te->started.tv_sec = o->lastKnownTime.tv_sec;
    te->started.tv_usec = o->lastKnownTime.tv_usec;
  #else
    gettimeofday(&te->started,(struct timezone *)NULL);
    /* update the time, which is used before we go
     * back into the poll()/select().
     */
    o->lastKnownTime.tv_sec = te->started.tv_sec;
    o->lastKnownTime.tv_usec = te->started.tv_usec;
  #endif
    /* now setup the expire time */
    te->expireTime.tv_sec = te->started.tv_sec + a;
    te->expireTime.tv_usec = te->started.tv_usec + b;
  
    while(te->expireTime.tv_usec > 1000000){
      te->expireTime.tv_sec++;
      te->expireTime.tv_usec -= 1000000;
    }
    /* put in a special precaution because I am
     * paranoid. Set bad status in case my logic
     * is wrong :/
     */
    ret = LIB_STATUS_BAD;
    /* carefully place it on the list */
    dlist_reset(o->timerlist);
    while((pe = (timerEntry *)dlist_get(o->timerlist)) != NULL){
      if(pe->expireTime.tv_sec > te->expireTime.tv_sec){
        /* this one expires after the new one */
        ret = dlist_insertHere(o->timerlist,(void *)te);
        break;
      }else if(pe->expireTime.tv_sec == te->expireTime.tv_sec){
        /* got to check usec's since equal seconds of expiration */
        if(pe->expireTime.tv_usec > te->expireTime.tv_usec){
  	/* ok we are expiring some time sooner than this
  	 * dude
  	 */
  	ret = dlist_insertHere(o->timerlist,(void *)te);
  	break;
        }
      }
      /* otherwise we expire after this one and fall out 
       * hitting the next get.
       */
    }
    if(pe == NULL){
      /* at the end of the list, and never got in append it. */
      ret = dlist_append(o->timerlist,(void *)te);
    }
    dlist_reset(o->timerlist);
    if(ret != LIB_STATUS_GOOD){
      /* we could not insert it into the list */
      /* clean up and fail */
      free(te);
      return(LIB_STATUS_BAD);
    }
    /* and now return */
    return(LIB_STATUS_GOOD);
  }
  
  int
  dist_TimerStop(distributor *o,timerFunc f, void *a, void *b)
  {
    /* find the first timer with matching function and args */
    /* and remove it from the list */
    timerEntry *pe,*pf;  
    dlist_reset(o->timerlist);
    while((pe = (timerEntry *)dlist_get(o->timerlist)) != NULL){
      if((pe->action == f) &&
         (pe->arg1 == a) &&
         (pe->arg2 == b)){
        pf = dlist_getThis(o->timerlist);
        if(pf != pe){
  	/* tragic list failure, here
  	 * during debug only.
  	 */
  	abort();
  	return(LIB_STATUS_BAD1);
        }else{
  	free(pe);
  	break;
        }
      }
    }
    if(pe == NULL){
      /* never found it on the list */
      return(LIB_STATUS_BAD);
    }
    return(LIB_STATUS_GOOD);
  }
  
  
  int
  dist_addFd(distributor *o,int fd,serviceFunc fdNotify, int mask,void *arg)
  {
    fdEntry *fe;
    int i,fnd,ret;
    /* first allocate a fd service structure and init it */
    fe = calloc(1,sizeof(fdEntry));
    if(fe == NULL){
      return(LIB_STATUS_BAD);
    }
    /* init it */
    fe->onTheList = 0;
    fe->fdItServices = fd;
    fe->fdNotify = fdNotify;
    fe->arg = arg;
  
    /* now place it in the hash table and verify its unique */
    ret = HashedTbl_enterKeyed(o->fdSubscribers,
  			     fe->fdItServices,
  			     (void*)fe,
  			     (void*)&fe->fdItServices,
  			     sizeof(int));  
  
    if(ret != LIB_STATUS_GOOD){
      /* no good, probably already in */
      free(fe);
      return(ret);
    }
    /* now add it to the linked list */
    ret = dlist_append(o->fdSubscribersList,(void *)fe);
    if(ret != LIB_STATUS_GOOD){
      /* now we must back it out of the hash table */
      HashedTbl_removeKeyed(o->fdSubscribers,
  			  fe->fdItServices,
  			  (void*)&fe->fdItServices,
  			  sizeof(int),
  			  (void **)NULL);
      free(fe);
      return(ret);
    }
    /* now plop in a place in the fd select array */
    /* first do we need to grow array? */
    if((o->numused+1) >= o->numfds){
      /* yes, this will push us to have none spare */
      struct pollfd *tp;
      tp = (struct pollfd *)calloc((o->numfds*2),sizeof(struct pollfd));
      if(tp == NULL){
        void *v;
        /* can't get bigger we are out of here */
        HashedTbl_removeKeyed(o->fdSubscribers,
  			    fe->fdItServices,
  			    (void*)&fe->fdItServices,
  			    sizeof(int),
  			    (void **)NULL);
        /* push pointer so next get will get the last one */
        dlist_getToTheEnd(o->fdSubscribersList);
        /* do that so we can do a getThis() */
        v = dlist_get(o->fdSubscribersList);
        /* now pull it */
        v = dlist_getThis(o->fdSubscribersList);
        free(fe);
        return(LIB_STATUS_BAD1);
      }
      memcpy(tp,o->fdlist,(sizeof(struct pollfd) * o->numfds));
      /* init the rest of the struct */
      for(i=o->numfds;i<(o->numfds*2);i++){
        tp[i].fd = -1;
        tp[i].events = 0;
        tp[i].revents = 0;
      }
      /* adjust the size */
      o->numfds *= 2;
      /* free old list */
      free(o->fdlist);
      /* place in our new one */
      o->fdlist = tp;
    }
    fnd = 0;
    /* now lets find a entry and use it */
    for(i=o->numused;i<o->numfds;i++){
      if(o->fdlist[i].fd == -1){
        /* found a empty spot */
        o->fdlist[i].fd = fd;
        o->fdlist[i].events = mask;
        o->numused++;
        fnd = 1;
        break;
      }
    }
    if(!fnd){
      for(i=0;i<o->numused;i++){
        if(o->fdlist[i].fd == -1){
  	/* found a empty spot */
  	o->fdlist[i].fd = fd;
  	o->fdlist[i].events = mask;
  	o->numused++;
  	fnd = 1;
  	break;
        }
      }
    }
    if(!fnd){
      /* TSNH */
      void *v;
      HashedTbl_removeKeyed(o->fdSubscribers,
  			  fe->fdItServices,
  			  (void*)&fe->fdItServices,
  			  sizeof(int),
  			  (void **)NULL);
      /* push pointer so next get will get the last one */
      dlist_getToTheEnd(o->fdSubscribersList);
      /* do that so we can do a getThis() */
      v = dlist_get(o->fdSubscribersList);
      /* now pull it */
      v = dlist_getThis(o->fdSubscribersList);
      free(fe);
      return(LIB_STATUS_BAD);
    }
    /* now add it to the list of service functions */
    return(LIB_STATUS_GOOD);
  }
  
  
  int
  dist_deleteFd(distributor *o,int fd)
  {
    fdEntry *fe;
    int i,removed;
    /* Look up the fd, and remove */
    removed = -1;
    fe = HashedTbl_removeKeyed(o->fdSubscribers,
  			     fd,
  			     (void*)&fd,
  			     sizeof(int),
  			     (void **)NULL);
    /* find it in the fdSubscribersList and remove */
    dlist_reset(o->fdSubscribersList);
    while((fe = (fdEntry *)dlist_get(o->fdSubscribersList)) != NULL){
      if(fe->fdItServices == fd){
        fe = dlist_getThis(o->fdSubscribersList);
      }
    }
    /* remove it from the fdlist and subtract current used */
    for(i=0;i<o->numfds;i++){
      if(o->fdlist[i].fd == fd){
        o->fdlist[i].fd = -1;
        o->fdlist[i].events = 0;
        o->fdlist[i].revents = 0;
        removed = i;
        o->numused--;
        break;
      }
    }
    /* now free the memory */
    if(fe != NULL){
      free(fe);
    }
    /* For netbsd we need to collapse the list */
    if(removed != -1){
      if(removed != o->numused){
        /* if removed == o->numused 
         * we removed the last one, no
         * action needed.
         */
        o->fdlist[removed].fd = o->fdlist[o->numused].fd;
        o->fdlist[removed].events  = o->fdlist[o->numused].events;
        o->fdlist[removed].revents  = o->fdlist[o->numused].revents;
        o->fdlist[o->numused].fd = -1;
        o->fdlist[o->numused].events = 0;
        o->fdlist[o->numused].revents = 0;
      }
    }
    return(LIB_STATUS_GOOD);
  }
  
  int
  dist_changeFd(distributor *o,int fd,int newmask)
  {
    int i;
  
    for(i=0;i<o->numfds;i++){
      if(o->fdlist[i].fd == fd){
        o->fdlist[i].events = newmask;
        break;
      }
    }
    return(LIB_STATUS_GOOD);
  }
  
  int
  __place_mc_in_dlist(dlist_t *dl,msgConsumer *mc)
  {
    msgConsumer *at;
    dlist_reset(dl);
    while((at = (msgConsumer *)dlist_get(dl)) != NULL){
      /* if this one has a greater precedence our
       * new one slips in ahead of the one we are
       * looking at on the list 
       */
      if(at->precedence > mc->precedence){
        return(dlist_insertHere(dl,(void *)mc));
      }
    }
    /* place it at the end of the list */
    return(dlist_append(dl,(void *)mc));
  }
  
  void
  __pull_from_dlist(dlist_t *dl,msgConsumer *mc)
  {
    msgConsumer *at;
    dlist_reset(dl);
    while((at = (msgConsumer *)dlist_get(dl)) != NULL){
      if(at == mc){
        at = (msgConsumer *)dlist_getThis(dl);
        break;
      }
    }
  }
  
  msgConsumer *
  __pull_from_dlist2(dlist_t *dl,messageFunc mf,int sctp_proto, int stream_no,void *arg)
  {
    msgConsumer *at,*nat;
    nat = NULL;
    dlist_reset(dl);
    while((at = (msgConsumer *)dlist_get(dl)) != NULL){
      if((at->SCTPprotocolId == sctp_proto) &&
         (at->SCTPstreamId == stream_no) &&
         (at->msgNotify == mf) &&
         (at->arg == arg)){
        nat = (msgConsumer *)dlist_getThis(dl);
        if(nat != at){
  	/* help */
  	abort();
        }
        break;
      }
    }
    return(nat);
  }
  
  
  int
  dist_msgSubscribe(distributor *o,messageFunc mf,int sctp_proto, int stream_no,int priority, void *arg)
  {
    int ret;
    msgConsumer *mc;
    hashableHash *hh;
    HashedTbl *hofl;
    hashableDlist *hdl;
    /* Ok, first we look up the hashTable with the sctp_proto number 
     * If we don't find it we add a new hash table for this to
     * the master hash table. Once we have the master
     * hash table and our sub-hash table, we then lookup
     * the stream_no to find the hashDlist structure.
     * With that we add this guy to the list (after allocating a 
     * entry to pop on and fill it in). As a short cut we look to
     * see if the two are the default values, if so dump it right
     * in to our pre-stored list.
     */
    
    /* first get a msgConsumer */
    mc = (msgConsumer *)calloc(1,sizeof(msgConsumer));
    if(mc == NULL){
      /* no memory, out of here */
      return(LIB_STATUS_BAD);
    }
    mc->precedence = priority;
    mc->SCTPprotocolId = sctp_proto;
    mc->SCTPstreamId = stream_no;
    mc->msgNotify = mf;
    mc->arg = arg;
  
    /* now that we have our structure see if we take the
     * short cut.
     */
    if((sctp_proto == DIST_SCTP_PROTO_ID_DEFAULT) &&
       (stream_no == DIST_STREAM_DEFAULT)){
      ret = __place_mc_in_dlist(o->msgSubZeroNoStrm->list,mc);
      if(ret != LIB_STATUS_GOOD){
        free(mc);
      }
      return(ret);
    }
    /* ok if we reach here, we must put it on some
     * other list in all of our hash tables. So much
     * for the short cut :)
     */
    /* first find the right hashableHash.
     */
    hh = HashedTbl_lookupKeyed(o->msgSubscribers,
  			     sctp_proto,
  			     &sctp_proto,
  			     sizeof(int),
  			     (void **)NULL);
    if(hh == NULL){
      /* no hashable hash table exists for the protocol type */
      /* we must create it */
      char string[32];
      sprintf(string,"protocolId:0x%x",sctp_proto);
      hh = create_hashableHash(sctp_proto,string,11);
      if(hh == NULL){
        /* failed to alloc-create */
        free(mc);
        return(LIB_STATUS_BAD);
      }
      /* now we must enter it in the master table */
      ret = HashedTbl_enterKeyed(o->msgSubscribers,hh->protocolId,(void *)hh,&hh->protocolId,sizeof(int));
      if(ret != LIB_STATUS_GOOD){
        /* something bad is happening */
        free(mc);
        destroy_hashableHash(hh);
        return(LIB_STATUS_BAD);
      }
      /* ok, we now have the hasable hash in the table and hofl setup right */
      hofl = hh->hash;
    }else{
      /* we found one */
      hofl = hh->hash;
    }
    /* now lets find the low level hashDlist */
    hdl = (hashableDlist *)HashedTbl_lookupKeyed(hofl,
  					       stream_no,
  					       &stream_no,
  					       sizeof(int),
  					       (void **)NULL);
    if(hdl == NULL){
      /* it does not exist so we must
       * build one and dump it in the hofl table.
       */
      hdl = create_hashdlist(stream_no);
      if(hdl == NULL){
        /* if we are in trouble, we don't kill the
         * previous hash table we built since
         * it may be used again when the memory problem
         * is cleared.. of course we may not even
         * have created it too.
         */
        free(mc);
        return(LIB_STATUS_BAD);
      }
      /* now add it to our hash table */
      ret = HashedTbl_enterKeyed(hofl,hdl->streamNo,(void *)hdl,&hdl->streamNo,sizeof(int));
      if(ret != LIB_STATUS_GOOD){
        /* no way to add the dlist to the table.. we are out of here. */
        free(mc);
        destroy_hashDlist(hdl);
        return(LIB_STATUS_BAD);
      }
      /* ok its in the table and now hdl points to the
       * guy we need to add to.
       */
    }
    /* now lets place it on the global list */
    ret = __place_mc_in_dlist(o->msgSubscriberList,mc);
    if(ret != LIB_STATUS_GOOD){
      free(mc);
      return(ret);
    }
    /* now on the individual list */
    ret = __place_mc_in_dlist(hdl->list,mc);
    if(ret != LIB_STATUS_GOOD){
      __pull_from_dlist(o->msgSubscriberList,mc);
      free(mc);
    }
    return(ret);
  }
  
  int
  dist_msgUnsubscribe(distributor *o,messageFunc mf,int sctp_proto, int stream_no,void *arg)
  {
    msgConsumer *mc,*mc2;
    hashableHash *hh;
    HashedTbl *hofl;
    hashableDlist *hdl;
  
    /* first lets pull it from the master table of whats in the dlist*/
    mc2 = mc = __pull_from_dlist2(o->msgSubscriberList,mf,sctp_proto,stream_no,arg);
    hh = HashedTbl_lookupKeyed(o->msgSubscribers,
  			     sctp_proto,
  			     &sctp_proto,
  			     sizeof(int),
  			     (void **)NULL);
    if(hh != NULL){
      hofl = hh->hash;
      hdl = (hashableDlist *)HashedTbl_lookupKeyed(hofl,
  						 stream_no,
  						 &stream_no,
  						 sizeof(int),
  						 (void **)NULL);
      if(hdl != NULL){
        /* ok lets find the entry here */
        mc2 = __pull_from_dlist2(hdl->list,mf,sctp_proto,stream_no,arg);
      }
    }
    if(mc && mc2){
      if(mc != mc2){
        /* help! */
        abort();
      }
      /* normal case */
      free(mc);
    }else if(mc){
      free(mc);
    }else if(mc2){
      /* strange should have been in other list */
      free(mc2);
    }else{
      return(LIB_STATUS_BAD);
    }
    return(LIB_STATUS_GOOD);
  }
  
  int
  dist_setDone(distributor *o)
  {
    o->notdone = 0;
    return(LIB_STATUS_GOOD);
  }
  
  void
  __check_time_out_list(distributor *o)
  {
    /* some timeout has occured ? */
    timerEntry *te;
    dlist_dlink *dl,*tmpl;
    tmpl = NULL;
    /* first get the time */
    gettimeofday(&o->lastKnownTime,(struct timezone *)NULL);
  
    /* get each one and have a look */
    dlist_reset(o->timerlist);
    while((te = (timerEntry *)dlist_get(o->timerlist)) != NULL){
      if((te->expireTime.tv_sec < o->lastKnownTime.tv_sec) ||
         ((te->expireTime.tv_sec == o->lastKnownTime.tv_sec) &&
  	(te->expireTime.tv_usec < o->lastKnownTime.tv_usec))
         ){
        /* ok this one has expired */
        dl = dlist_getThisSlist(o->timerlist);
        if(dl != NULL){
  	dl->next = tmpl;
  	tmpl = dl;
        }
      }else{
        /* ok the time is beyond this one I am looking at */
        /* we should have no more after this */
        break;
      }
    }
    dl = tmpl;
    /* call every timer that is up on our temp list */
    while(dl != NULL){
      tmpl = dl->next;
      te = (timerEntry *)dl->ent;
      free(dl);	
      (*te->action)(te->arg1,te->arg2);
      free(te);
      dl = tmpl;
    }
  }
  
  void
  __sendClockTick(distributor *o)
  {
    /* distribute a clock tick to all subscribers for ticks */
    auditTick *at;
    dlist_reset(o->lazyClockTickList);
    while((at = (auditTick *)dlist_get(o->lazyClockTickList)) != NULL){
      (*at->tick)(at->arg);
    }
  }
  
  void
  __sendStartStopEvent(distributor *o,int event)
  {
    /* distribute a clock tick to all subscribers for ticks */
    StartStop *at;
    dlist_reset(o->startStopList);
    while((at = (StartStop *)dlist_get(o->startStopList)) != NULL){
      if(at->flags & event)
        (*at->activate)(at->arg,event);
    }
  }
  
  
  int
  dist_startupChange(distributor *o,startStopFunc sf, void *arg, int howmask)
  {
    int chnged;
    StartStop *at;
    chnged = 0;
    dlist_reset(o->startStopList);
    while((at = (StartStop *)dlist_get(o->startStopList)) != NULL){
      if((sf == at->activate) && (arg == at->arg)){
        chnged = 1;
        at->flags = howmask;
        if(howmask == DIST_DONT_CALL_ME){
  	/* pull it from the list too */
  	void *v;
  	v = dlist_getThis(o->startStopList);
  	if(v != at){
  	  printf("Serious error, TSNH!\n");
  	  return(LIB_STATUS_BAD3); 
  	}else{
  	  free(at);
  	}
        }
      }
    }
    if(!chnged){
      /* new entry */
      int ret;
      at = calloc(1,sizeof(StartStop));
      if(at == NULL){
        /* no memory */
        return(LIB_STATUS_BAD); 
      }
      at->activate = sf;
      at->flags = howmask;
      at->arg = arg;
      ret = dlist_append(o->startStopList,(void *)at);
      if(ret != LIB_STATUS_GOOD){
        free(at);
        return(LIB_STATUS_BAD);
      }
    }
    return(LIB_STATUS_GOOD);
  }
  
  
  int
  dist_process(distributor *o)
  {
    int i,ret,errOut;
    int clockTickFlag;
    int doingMultipleReads;
    timerEntry *te;
    fdEntry *fe;
    dlist_t *list1;
  
    int timeout;
  #ifdef USE_SELECT
    struct timeval timeoutt;
    int cnt;
    fd_set readfds,writefds,exceptfds;
  #endif
  
    list1 = dlist_create();
    if(list1 == NULL){
      printf("Error no memory\n");
      return(LIB_STATUS_BAD);
    }
    i = 0;
    doingMultipleReads = 0;
    __sendStartStopEvent(o,DIST_CALL_ME_ON_START);
    while(o->notdone){
      dlist_reset(o->timerlist);
      te = dlist_get(o->timerlist);
  #ifdef USE_SELECT
      /* spin through and setup things.*/
      FD_ZERO(&readfds);
      FD_ZERO(&writefds);
      FD_ZERO(&exceptfds);
      cnt = 0;
      for(i=0;i<o->numfds;i++){
        if(o->fdlist[i].events & POLLIN){
  	if(o->fdlist[i].fd > cnt){
  	  cnt = o->fdlist[i].fd;
  	}
  	FD_SET(o->fdlist[i].fd,&readfds);
        }
        if(o->fdlist[i].events & POLLOUT){
  	if(o->fdlist[i].fd > cnt){
  	  cnt = o->fdlist[i].fd;
  	}
  	FD_SET(o->fdlist[i].fd,&writefds);	
        }
        cnt++;
      }
  #endif
      /* prepare timeout */
  #ifndef MINIMIZE_TIME_CALLS
      gettimeofday(&o->lastKnownTime,(struct timezone *)NULL);
  #endif
      if(te != NULL){
        /* yep, a timer is running use the delta */
        clockTickFlag = 0;
        if(te->expireTime.tv_sec >= o->lastKnownTime.tv_sec){
  	timeout = ((te->expireTime.tv_sec - o->lastKnownTime.tv_sec) * 1000);
  	if(te->expireTime.tv_usec > o->lastKnownTime.tv_usec){
  	  timeout += ((te->expireTime.tv_usec - o->lastKnownTime.tv_usec)/1000);
  	}else if(te->expireTime.tv_usec < o->lastKnownTime.tv_usec){
  	  /* borrow 1 second from the upper seconds 
  	   */
  	  timeout += (((1000000 - o->lastKnownTime.tv_usec) + te->expireTime.tv_usec)/1000);
  	  /* subtract out the borrowed time above */
  	  timeout -= 1000;
  	}
  	/* subtract off 10ms this is due to
  	 * the clock synchronization issues. What
  	 * happens is the O/S will round up leaving us a
  	 * drift upwards. This compensates for that drift,
  	 * we only do this if we have more than 11ms to wait.
  	 */
  	if(timeout > 11){
  	  timeout -= 10;
  	}
        }else{
  	/* expire time is past? */
  	timeout = 0;
        }
        if(timeout <= 0){
  	/* problem, time out has occured */
  	__check_time_out_list(o);
  	continue;
        }
      }else{
        /* no timeout setup default */
        clockTickFlag = 1;
        timeout = (o->idleClockTick.tv_sec * 1000) + o->idleClockTick.tv_usec/1000;
      }
  #ifdef USE_SELECT
      if(doingMultipleReads){
        /* poll for fd only */
        timeoutt.tv_sec = 0;
        timeoutt.tv_usec = 0;
      }else{
        timeoutt.tv_sec = (timeout/1000);
        timeoutt.tv_usec = (timeout%1000) * 1000;
      }
      ret = select(cnt,&readfds,&writefds,&exceptfds,&timeoutt);
  #else
      if(doingMultipleReads){
        /* non-blocking poll */
        ret = poll(o->fdlist,o->numused,0);
      }else{
        ret = poll(o->fdlist,o->numused,timeout);
      }
  #endif
      /* save off the error from poll/select */
      errOut = errno;
      /* we always check the timeout updating the time in the process */
      __check_time_out_list(o);
      if(ret > 0){
        /* ok we have something to do */
  #ifdef USE_SELECT
        /* translate the select terminology into the poll fundamentals */
        for(i=0;i<o->numfds;i++){
  	if(o->fdlist[i].fd == -1)
  	  continue;
  	o->fdlist[i].revents = 0;
  	if(FD_ISSET(o->fdlist[i].fd,&readfds)){
  	  o->fdlist[i].revents |= POLLIN;
  	}
  	if(FD_ISSET(i,&writefds)){
  	  o->fdlist[i].revents |= POLLOUT;
  	}
        }
  #endif
        /* now rip through and call all the appropriate
         * functions
         */
        for(i=0;i<o->numfds;i++){
  	if(o->fdlist[i].fd == -1){
  	  continue;
  	}
  	if(o->fdlist[i].revents == 0){
  	  continue;
  	}
  	/* got a live one */
  	fe = (fdEntry *)HashedTbl_lookupKeyed(o->fdSubscribers,
  					      o->fdlist[i].fd,
  					      &o->fdlist[i].fd,
  					      sizeof(int),
  					      (void **)NULL);
  	if(fe == NULL){
  	  /* huh? TSNH */
  	  o->fdlist[i].fd = -1;
  	  o->numused--;
  	  if(i != o->numused){
  	    o->fdlist[i].fd = o->fdlist[o->numused].fd;
  	    o->fdlist[i].events = o->fdlist[o->numused].events;
  	    o->fdlist[i].revents = o->fdlist[o->numused].revents;
  	    o->fdlist[o->numused].fd = -1;
  	    o->fdlist[o->numused].events = 0;
  	    o->fdlist[o->numused].revents = 0;
  	  }
  	  continue;
  	}
  	if((*fe->fdNotify)(fe->arg,o->fdlist[i].fd,o->fdlist[i].revents) > 0){
  	  if(fe->onTheList == 0){
  	    if(dlist_append(list1,(void *)fe) < LIB_STATUS_GOOD){
  	      printf("Can't append??\n");
  	    }else{
  	      fe->onTheList = 1;
  	    }
  	  }
  	}
  	o->fdlist[i].revents = 0;
        }
      }else if((ret == 0) && (clockTickFlag) && (!doingMultipleReads)){
        __sendClockTick(o);
      }else if((ret == 0) && (doingMultipleReads)){
        dlist_reset(list1);
        fe = (fdEntry *)dlist_get(list1);
        if((*fe->fdNotify)(fe->arg,o->fdlist[i].fd,0) <= 0){
  	void *v;
  	fe->onTheList = 0;
  	v = dlist_getThis(list1);
  	if(v != (void *)fe){
  	  printf("Help TSNH!\n");
  	}
        }
      }
      if(dlist_getCnt(list1)){
        /* if we have some on the list then we need
         * to arrange to just poll non-blocking
         */
        doingMultipleReads = 1;
      }else{
        doingMultipleReads = 0;
      }
    }
    __sendStartStopEvent(o,DIST_CALL_ME_ON_STOP);
    dlist_destroy(list1);
    return(LIB_STATUS_GOOD);
  }
  
  void
  __distribute_msg_o_list(dlist_t *dl,messageEnvolope *msg)
  {
    /* distribute over a list the particular message *msg */
    msgConsumer *at;
    dlist_reset(dl);
    while((at = (msgConsumer *)dlist_get(dl)) != NULL){
      (*at->msgNotify)(at->arg,msg);
      if((msg->data == NULL) || (msg->totData == NULL)){
        break;
      }
    }
  }
  
  
  void
  dist_sendmessage(distributor *o,messageEnvolope *msg)
  {
    hashableHash *hh;
    HashedTbl *hofl;
    hashableDlist *hdl;
  
    /* first lets lookup in the master table for the protocol */
    msg->distrib = (void *)o;
    hdl = NULL;
    hh = HashedTbl_lookupKeyed(o->msgSubscribers,
  			     msg->protocolId,
  			     &msg->protocolId,
  			     sizeof(int),
  			     (void **)NULL);
    if(hh != NULL){
      /* ok do we have a protocolID specific entry? - yes */
      hofl = hh->hash;
      /* is the stream there? */
      hdl = (hashableDlist *)HashedTbl_lookupKeyed(hofl,
  						 msg->streamNo,
  						 &msg->streamNo,
  						 sizeof(int),
  						 (void **)NULL);
      if(hdl == NULL){
        /* no, see if the default stream is there */
        int x;
        x = DIST_STREAM_DEFAULT;
        hdl = (hashableDlist *)HashedTbl_lookupKeyed(hofl,
  						   x,
  						   &x,
  						   sizeof(int),
  						   (void **)NULL);
      }
    }
    if(hdl != NULL){
      /* ok lets send it over this list */
      __distribute_msg_o_list(hdl->list,msg);
    }
    /* if we did not consume and we were on some other list, pass it over the 
     * default list.
     */
    if((msg->data != NULL) && (msg->totData != NULL) && (hdl != o->msgSubZeroNoStrm)){
      __distribute_msg_o_list(o->msgSubZeroNoStrm->list,msg);
    }
    if((msg->totData != NULL) && (msg->data != NULL) && (msg->takeOk)){
      free(msg->totData);
    }
  }
  
  
  
  int
  dist_wants_audit_tick(distributor *o,auditFunc af,void *arg)
  {
    auditTick *at;
    at = calloc(1,sizeof(auditTick));
    if(at == NULL){
      return(LIB_STATUS_BAD);
    }
    at->arg = arg;
    at->tick = af;
    return(dlist_append(o->lazyClockTickList,(void *)at));
  }
  
  int
  dist_no_audit_tick(distributor *o,auditFunc af,void *arg)
  {
    /* find and remove this guy */
    auditTick *at;
    dlist_reset(o->lazyClockTickList);
    while((at = (auditTick *)dlist_get(o->lazyClockTickList)) != NULL){
      if((at->arg == arg) && (at->tick == af)){
        /* got it */
        dlist_getThis(o->lazyClockTickList);
        free(at);
        return(LIB_STATUS_GOOD);
      }
    }
    return(LIB_STATUS_BAD);
  }
  void
  dist_changelazyclock(distributor *o,struct timeval *tm)
  {
    o->idleClockTick.tv_sec = tm->tv_sec;
    o->idleClockTick.tv_usec = tm->tv_usec;
  }
  