scheduler.cc

这是一个著名的应用层组播中间件的源码

//Copyright (c) 2004, Charles Killian, Adolfo Rodriguez, Dejan Kostic, Sooraj Bhat, and Amin Vahdat
//All rights reserved.
//
//Redistribution and use in source and binary forms, with or without
//modification, are permitted provided that the following conditions are met:
//
//   * Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//   * 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.
//   * Neither the names of Duke University nor The University of
//     California, San Diego, 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 COPYRIGHT HOLDERS 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 COPYRIGHT OWNER 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 <LEDA/set.h>

#include <set>
#include <iterator>
#include "scheduler.h"
#include "nsport.h"
#include <pthread.h>
#include <errno.h>
#undef NSPORT_DEBUG
#define NSPORT_DEBUG 0

int threads_created=0;
int threads_joined=0;
int threads_exited=0;
int threads_cancelled=0;
int threads_timer=0;
int threads_tfrc=0;
int threads_tcp=0;

extern class Agent *globalbase;

//  has to initialize static members

set<pthread_t> TimerHandler::tobeKilled;
list < delta_list_entry *> TimerHandler::delta_list;  
pthread_t TimerHandler::tid= 0;

double sure_sleep( double amount);
pthread_t start_cancellation();

Scheduler::Scheduler()
{
  clock_.tv_sec = 0;
  killer= start_cancellation();
}

/* double */
/* Scheduler::clock() */
/* { */
/*   struct timezone tz_; */
/*   gettimeofday(&clock_, &tz_); */
/*   return (clock_.tv_sec+(double)clock_.tv_usec/1000000.0); */
/* } */

TimerHandler::TimerHandler()
{
  sched_time = 0;
  sched_first = 0;
  exited = 0;
  //  tid = 0;
  rescheded = 0;
  outside = 0;
}

void* killerthread(void *temp)
{
  // printf("killerthread thread %d.\n",pthread_self());

  threads_created++;
  //  printf("Number of threads is %d,  cr %d, join %d, exit %d, can %d time %d, tfrc %d, tcp %d\n", threads_created-threads_joined, threads_created, threads_joined, threads_exited, threads_cancelled, threads_timer, threads_tfrc, threads_tcp);
  
  pthread_t tid;
  set <pthread_t> mytobeKilled;

  while(1)
    {
      sleep(1+(int)(3*drand48()));
     
      extern int global_exit;
      if (global_exit == 1)
	return 0;

      if(!globalbase)
	{
	  continue;
	}
      globalbase->Lock();
      int count = 0;

      swap(mytobeKilled,TimerHandler::tobeKilled);
      globalbase->Unlock();   

      set<pthread_t>::iterator pos;   

       for(pos=mytobeKilled.begin();pos!=mytobeKilled.end();pos++)
	 {
	   extern int global_exit;
	   if (global_exit == 1)
	     return 0;
#if NSPORT_DEBUG > 2
	   printf("thread %d about to kill %d\n",pthread_self(),*pos);
#endif

	   if (pthread_join(*pos,NULL)!=0)
	     {
	       perror("Join Failed");
	       //exit(79);
	     }
	   else 
	     {
	       threads_joined++;
	       threads_timer--;
	       //	       printf("Number of threads is %d,  cr %d, join %d, exit %d, can %d time %d, tfrc %d, tcp %d\n", threads_created-threads_joined, threads_created, threads_joined, threads_exited, threads_cancelled, threads_timer, threads_tfrc, threads_tcp);

#if NSPORT_DEBUG > 1
	       printf("thread %d join thread finished %d.\n",pthread_self(), *pos);
#endif
	     }   
	 }
       mytobeKilled.clear();
    }
}


pthread_t start_cancellation()
{
  int ret;
  pthread_t ctid;
  if ((ret = pthread_create(&ctid, NULL, killerthread, NULL)) != 0)
    {
      perror("pthread_create");
      printf("Error %d in creating cancel thread.\n", ret);
      exit(80);
      //sure_sleep(0.5);
    }
  else 
    {
#if NSPORT_DEBUG > 1
      printf("created %d cancel thread\n",ctid );
#endif
    }
  return ctid;
}
//assumption: global lock is held before calling this function.

void
TimerHandler::force_cancel()
{
globalbase->Lock();
#if NSPORT_DEBUG > 2
 printf("%d force_cancel %x.\n",pthread_self(), this);
#endif
 
 list<delta_list_entry*>::iterator traverse; 
   
  for(traverse=delta_list.begin();traverse!=delta_list.end();traverse++)
    {
      delta_list_entry*entry = *traverse;
#if NSPORT_DEBUG > 3
	  printf(" %d examining %lf %x for %x  \n",pthread_self(),entry->expiration,entry->timer, this);
#endif
	  if (entry->timer==this)
	    {
	      if ( delta_list. front() == entry)
		{
#if NSPORT_DEBUG > 2
		  printf("stopping sleeping thread\n");
#endif
		}
      
	      delta_list.erase(traverse);
#if NSPORT_DEBUG > 2
	      printf("stopping delta %lf\n", entry->expiration);
	      printlist();
#endif
	      delete entry;
	      break;
	    } 
	}
  fflush(stdout);
globalbase->Unlock();
}
//assumption: global lock is held before calling this function.

void
TimerHandler::cancel_thread()
{
globalbase->Lock();
#if NSPORT_DEBUG > 1	  
      printf("cancel_thread %d thread %d.\n",pthread_self(), tid);
#endif
  if (pthread_self()== tid)
    {
#if NSPORT_DEBUG > 1  
      printf("thread %d killing itself.\n",pthread_self());
#endif
    }
  else if (tid) // somebody wants to kill thread 0
    {    
#if NSPORT_DEBUG >= 1	  
      printf("thread %d cancelling thread  %d.\n",pthread_self(), tid);
#endif
      if (pthread_cancel(tid)!=0)
	{
	  perror("Cancel Failed");
	  printf("Tid of %d\n",tid);
	  //      exit(78);
	}
      threads_cancelled++;
      //      printf("Number of threads is %d,  cr %d, join %d, exit %d, can %d time %d, tfrc %d, tcp %d\n", threads_created-threads_joined, threads_created, threads_joined, threads_exited, threads_cancelled,	threads_timer, threads_tfrc, threads_tcp);
      tobeKilled.insert(tid);
	
#if NSPORT_DEBUG > 1	  
      printf("thread %d asking for join on thread  %d.\n",pthread_self(), tid);
#endif
      tid = 0;
      globalbase->Unlock();
      return;
    }
  
  fflush(stdout);
  globalbase->Unlock();
}

//assumption: global lock is held before calling this function.
//warning: this may not be the case for constructors
void
TimerHandler::resched(double time)
{
  ASSERT(time >= 0);
  int lock_acquired = 0;
  
  extern int global_exit;
  if (global_exit == 1)
    return;

  if (globalbase==NULL) 
    {
      printf("global pointer is null in resched\n");  fflush(stdout);
      perror("null global pointer");
      exit(60);
    }
  
  globalbase->Lock();
  lock_acquired = 1;
  
#if NSPORT_DEBUG > 1	  
	  printf("%d rescheduling for %lf\n",pthread_self(), time);
	  printlist();
#endif
  list<delta_list_entry*>::iterator traverse;   
  double now = Scheduler::instance().clock();
  double first = 0;
  int inserted = 0;
  delta_list_entry* entry = new delta_list_entry ();
  entry->timer = this;
  entry->expiration = now + time;
 
  if (!delta_list.empty() && entry->expiration<delta_list.front()->expiration)
    {
#if NSPORT_DEBUG > 1	  
      printf("thread %d inserting the earliest entry %x %lf before %x %lf %x \n",pthread_self(), entry, entry->expiration, delta_list.front(), delta_list.front()->expiration, this);
#endif  
      delta_list. push_front(entry);      
      printlist();
      inserted = 1;
      if (pthread_self()== tid)
	{
#if NSPORT_DEBUG > 1	  
	  printf("thread %d rescheduling itself at front of list. %x\n",pthread_self(), this);
#endif
//  	  exit( 203);
	}
      else
	{
	  cancel_thread();
	}
    }
  else if (!delta_list.empty())
    {