scheduler.cc

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

      double prev_exp=0;
      delta_list_entry *prev=0;
      delta_list_entry*next = 0;
      for(traverse=delta_list.begin();traverse!=delta_list.end();traverse++)
	{
	  prev = next;
	  next = *traverse;
	  if (prev_exp > next->expiration)
	    {
	      perror("expiration times out of order");
	      printf("Expiration times are out of order next %x with time %f, prev %x with time %f and %f (diff %f comp %d)\n", next, next->expiration, prev, prev_exp, prev->expiration, prev_exp - next->expiration, prev_exp > next->expiration);  fflush(stdout);
        ASSERT(prev_exp <= next->expiration);
	      exit(58);
	    }
	  prev_exp = next->expiration;

#if NSPORT_DEBUG > 3
	  printf(" %d examining %lf\n",pthread_self(),next->expiration);
#endif
	  if (!next)
	    {
	      perror("null delta entry");
	      printf("Got a null entry in delta list! next %x, prev %x\n", next, prev);  fflush(stdout);
	      exit(52);
	    }
	  
	  if (entry->expiration < next->expiration)
	    {
#if NSPORT_DEBUG > 1
	      if (!prev)
		printf("inserting %x %lf before %x %lf\n", entry, entry->expiration, next, next->expiration); 
	      else
		printf("inserting %x %lf before %x %lf and after %x %lf\n", entry, entry->expiration, next, next->expiration, prev, prev->expiration); 
#endif	
	      delta_list. insert (traverse, entry);	      
	      printlist();
	      inserted = 1;
	      break;
	    } 
	}
    }

//   if (!  inserted && !delta_list.empty())
//     // there was a last entry, I need to check its value
//     {
//       traverse=delta_list.end();
//       delta_list_entry *lastone = *traverse;
//       if (entry->expiration < lastone->expiration)
// 	{
// #if NSPORT_DEBUG > 2
// 	  printf("inserting %lf before last one %lf \n", entry->expiration, lastone->expiration); 
// #endif	
// 	  delta_list. insert (traverse, entry);	      
// 	  printlist();
// 	  inserted = 1;
// 	} 
//     }
  
  if (!  inserted )
    {
#if NSPORT_DEBUG > 1	  
      printf("thread %d inserting entry %x %lf at the end of list\n",pthread_self(), entry, entry->expiration);
#endif

      delta_list. push_back (entry);
      printlist();
    }      
  
  void *arg;
  int ret;

  if (!tid)   // timer thread was already created, so leave
    {
#if NSPORT_DEBUG > 1
      printf("Starting timer thread.\n");
#endif
      arg = (void *)NULL;

      if ((ret = pthread_create(&tid, NULL, startthread, arg)) != 0)
	{
	  perror("pthread_create");
	  printf("Error %d in creating timer thread.\n", ret);
	  exit(76);
	  //sure_sleep(0.5);
	}
    }
  if (lock_acquired)
    {
      globalbase->Unlock();

    }
  return;

}

void protect()
{
  int old_type = 0;
  int old_state = 0;

  pthread_testcancel();
  pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, & old_type);
  pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, & old_state);
  globalbase->Lock();
}

void unprotect()
{
  int old_type = 0;
  int old_state = 0;
  globalbase->Unlock();

  pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, & old_type);
  pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, & old_state);
  pthread_testcancel();
}
  

void *
startthread(void *arg)
{  
  TimerHandler *mytimer;
#if NSPORT_DEBUG > 1
  printf("startthread thread %d\n",pthread_self());
#endif  
  extern Scheduler *globalsched;  
  if (globalsched)
    globalsched->main_thread = pthread_self();
  threads_created++;
  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);
  
  //  printf("TimerHandler:startthread  running\n");
  int premature = 0;

  while(1)
    {
      double  service_started = Scheduler::instance().clock();

      extern int global_exit;
      if (global_exit == 1)
	return 0;

      protect();

       if ((TimerHandler::tobeKilled.find( pthread_self())!=TimerHandler::tobeKilled.end()))
	 {
	   //	   printf("I was told to die %d\n",pthread_self());
	   unprotect();
	   threads_exited++;
	   //	   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_exit(NULL);
	   return 0;
	 }     
      if (TimerHandler::delta_list.empty())
	{
	  printf("TimerHandler:startthread delta_list.empty\n");
	  unprotect();
	  threads_exited++;
	  //	  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_exit(NULL);
	  return 0;
	}
      delta_list_entry*entry =TimerHandler::delta_list.front();
  
      mytimer=entry->timer;
      double expiration = entry->expiration;

      unprotect();

      premature = 0;
      double sure_sleep_result=0;
      if(expiration > Scheduler::instance().clock() + CLOCK_RESOLUTION)
	{
#if NSPORT_DEBUG > 2
	  printf("Timer: sleeping for expire %x entry %x , time=%lf\n", mytimer, entry, expiration);
#endif
	  fflush(stdout);
	  sure_sleep_result=sure_sleep(expiration-Scheduler::instance().clock()); 

	  extern int global_exit;
	  if (global_exit == 1)
	    return 0;

#if NSPORT_DEBUG > 2	
	  printf("Timer: waking for expire %x entry %x, lock = %x, sure_sleep_result=%lf, expiration =%f\n", mytimer, entry, globalbase->agentlock, sure_sleep_result, expiration);
#endif
	  fflush(stdout);
	}

      protect();

      if ((TimerHandler::tobeKilled.find(mytimer->tid)!=TimerHandler::tobeKilled.end()))
	{
#if NSPORT_DEBUG > 2
	  printf("I was told to die %d\n",pthread_self());
#endif
	  unprotect();
	  threads_exited++;
	  //	  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_exit(NULL);
	  return NULL;
	}      
      entry= TimerHandler::delta_list.front();

      if ( entry->timer != mytimer)
	{
	  printf("%d front of timer list switched, exp was %f, exp now %f, curtime %f \n",pthread_self(), expiration, entry->expiration, Scheduler::instance().clock());
	  if(entry->expiration > Scheduler::instance().clock() + CLOCK_RESOLUTION)
	    premature = 1;
	}   

      if (!premature)
	{
	  mytimer=entry->timer;
	  
#if NSPORT_DEBUG > 2
	  printf("Timer: %x entry %x locked for expire %d\n", mytimer,entry, pthread_self());
	  fflush(stdout);
#endif

	  mytimer->expire(NULL);

	  int removed=0;
	  list<delta_list_entry*>::iterator traverse;   
      
	  for(traverse=TimerHandler::delta_list.begin();traverse!=TimerHandler::delta_list.end();traverse++)
	    {
	      if (*traverse==entry)
		{
		  mytimer->printlist();
#if NSPORT_DEBUG > 2
		  printf("Timer: %d removing expired entry %x %lf \n", pthread_self(),entry,entry->expiration); 
	  fflush(stdout);
#endif
		  TimerHandler::delta_list.erase(traverse);
		  removed=1;
		  mytimer->printlist();
		  break;
		}
	    }
	  if (!removed)
	    {
#if NSPORT_DEBUG > 2	
	      printf("Timer: %d can't find expired entry %x %lf \n", pthread_self(),entry,entry->expiration); 
	      fflush(stdout);
#endif
	    }
	  else
	    {
#if NSPORT_DEBUG > 2	
	      printf("Timer: %d deleting expired entry %x %lf \n", pthread_self(),entry,entry->expiration); 
	      fflush(stdout);
#endif
	      delete entry;
	    }

#if NSPORT_DEBUG > 2
	  printf("Timer: %d back from expire, ready to unlock \n",pthread_self());
	  fflush(stdout);
#endif
	}
      unprotect();
      double service_ended= Scheduler::instance().clock();
#if NSPORT_DEBUG > 2
      printf("Timer: service for %lf  \n",service_ended - service_started);
      fflush(stdout);
#endif      
    }
}

void TimerHandler::printlist() {
  double prev_exp = 0;
#if NSPORT_DEBUG > 2
      printf(" \n");

  list<delta_list_entry*>::iterator printer;
  delta_list_entry* sample;
  delta_list_entry* prev;
  for(printer=TimerHandler::delta_list.begin();printer!=TimerHandler::delta_list.end();printer++)
    {
      sample=*printer;
      printf("List: sample = %x timer = %x", sample,sample->timer); fflush(stdout);
      printf(" %lf\n", sample->expiration);
      if (sample->expiration < prev_exp)
	{
	  perror("printlist says expiration times out of order");
	  printf("Printlist says xpiration times are out of order next %x with time %f, prev %x with time %f and %f\n", sample, sample->expiration, prev, prev_exp, prev->expiration);  fflush(stdout);
	  exit(59);
	}
      prev = sample;
      prev_exp = sample->expiration;
    }
#endif

}