A Thread that doesn't hold long enough

[lucky6969b]

I've got a process called Arrivals, this process has mean arrival time of 8 seconds.
I've created an ExponentialStream like this,

#ifndef _ARRIVALS_H_
#define _ARRIVALS_H_

//#ifndef PROCESS_H_
#  include "Process.h"
//#endif

//#ifndef RANDOM_H_
#  include "SimuDLL\Random.h"
//#endif
 
 

class Arrivals : public Process {
public:
    Arrivals (double);
    virtual ~Arrivals ();

    virtual void Body (); 

private:
    ExponentialStream* InterArrivalTime;

	
};

#endif

/*
 * Copyright (C) 1994-1998,
 *
 * Department of Computing Science,
 * The University,
 * Newcastle upon Tyne,
 * UK.
 */

#ifndef ARRIVALS_H_
#  include "Arrivals.h"
#endif

#ifndef JOB_H_
#  include "Job.h"
#endif

#include "CMinMax.h" 


Arrivals::Arrivals (double mean)
		   : InterArrivalTime(new ExponentialStream(mean)) 
{
}

Arrivals::~Arrivals () { delete InterArrivalTime; }

void Arrivals::Body ()
{
 
	for (;;)
	{
		double arrivalTime = (*InterArrivalTime)();
	
		// Hold a number of intervals, before a new lorry arrives		 
		Hold(arrivalTime);

		CMinMax<int> lorryRand(1,3);
		int lorryId = lorryRand.GetRandomNumber();

		Job* work = new Job(lorryId);
	
	}
}

I've triggered this process in the main routine of the program

Sim::Sim()
{ 
	m_Arrivals = new Arrivals(8);

#ifndef NO_RESOURCE
    Resources::ref(m_Arrivals);     
#endif

	m_Arrivals->Activate();
	m_Arrivals->Resume();
	 

	Scheduler::scheduler().Resume();
}

But the arrivals of different events came so fast so that I even can't identify which is which.
(They actually came in under 0.1 s, not around 8 seconds), I conclude that the Suspend method doesn't even hold the thread. I wonder why.... I don't know this code would be where the core problem is, I will post more later.
Thanks
Jack

#ifndef __GNUG__
static double SimulatedTime = 0.0;
#else
double SimulatedTime = 0.0;
#endif

#ifndef TESTQUEUE
static Queue_Type ReadyQueue;  // Queue_Type is replaced by cpp
#else
Queue_Type ReadyQueue;
#endif

static Mutex* _theMutex = Mutex::create();

Scheduler* Scheduler::theScheduler = (Scheduler*) 0;
Boolean Scheduler::schedulerRunning = FALSE;
Process* Process::Current = (Process*) 0;
const double Process::Never = -1; // Process will never awaken.

/*
 * Note: unlike in SIMULA, an active process is removed from the simulation
 * queue prior to being activated.
 */

//
// Class Scheduler
//

// scheduler is just an object, with no associated thread.

Scheduler::Scheduler ()
{
}

Scheduler::~Scheduler ()
{
}

/*
 * This routine resets the simulation time to zero and removes all
 * entries from the scheduler queue (as their times may no longer
 * be valid). Such entries are suspended as though Cancel had been
 * called on them by a "user" process.
 *
 * Note: if this is to be used to reset the simulation environment
 * for another "run" then some means of re-initializing these queue
 * entries will be required. This is up to the user to provide and
 * invoke prior to their being used again.
 */

void Scheduler::reset () const
{
    Process* tmp = (Process*) 0;

    do
    {
	tmp = ReadyQueue.Remove();
	
	if (tmp)
	{
	    tmp->Cancel();
	    tmp->reset();  // call user-definable reset routine.
	}
	
    } while (tmp);

    SimulatedTime = 0.0;
}

double Scheduler::CurrentTime () const { return SimulatedTime; }

void Scheduler::print (std::ostream& strm)
{
    strm << "Scheduler queue:\n" << std::endl;
    
    ReadyQueue.print(strm);

    strm << "End of scheduler queue." << std::endl;
}

//
// Class Process
//

/*
 * Before this process is deleted, make sure it
 * is removed from the queue, or we could have
 * problems with pointer dereferencing!
 */

Process::~Process ()
{
    /*
     * We don't call Cancel of terminate here since they will
     * attempt to suspend this process if it is running, and
     * we do not want that to occur - garbage could quickly
     * build up. So, we let the destructor run to completion
     * which will implicitly suspend the process anyway, and
     * let the thread specific destructor then decide whether
     * it needs to do anything specific (e.g., reactivate the
     * scheduler) before it finishes.
     */

    if (!Terminated)
    {
	Terminated = TRUE;
	Passivated = TRUE;

	unschedule(); // remove from scheduler queue

	wakeuptime = Process::Never;

	if (this == Process::Current)
	{
	    schedule();
	    _theMutex->unlock();
	}
    }
}

double Process::CurrentTime () { return SimulatedTime; }

void Process::set_evtime (double time)
{
    if (!idle())  // error if we are not scheduled for activation
    {
	if (time >= Process::CurrentTime())
	    wakeuptime = time;
	else
	    error_stream << WARNING
			 << "Process::set_evtime - time " << time
			 << " invalid" << std::endl;
    }
    else
        error_stream << WARNING
		     << "Process::set_evtime called for idle process." << std::endl;
}

// return process to be activated after this process.

const Process* Process::next_ev () const
{
    return ((!idle()) ? ReadyQueue.getNext(this) : (Process*) 0);
}

/*
 * These routines are slightly different to the SIMULA
 * equivalents in that a process must be given, and that
 * process must be scheduled. Complete compatibility
 * may be provided in future releases.
 */

void Process::ActivateBefore (Process &p)
{
    // No op if already scheduled
    if (Terminated || !idle()) return;

    Passivated = FALSE;
    if (ReadyQueue.InsertBefore(*this, p))
	wakeuptime = p.wakeuptime;
    else
	error_stream << WARNING << "ActivateBefore failed because 'before' process is not scheduled" << std::endl;
}

void Process::ActivateAfter (Process &p)
{
    // No op if already scheduled
    if (Terminated || !idle()) return;

    Passivated = FALSE;
    if (ReadyQueue.InsertAfter(*this, p))
	wakeuptime = p.wakeuptime;
    else
	error_stream << WARNING << "ActivateAfter failed because 'after' process is not scheduled" << std::endl;
}

/*
 * These routines differ from their SIMULA counterparts in
 * that a negative AtTime is considered an error, and will
 * not cause the process to be activated at the current time.
 * This may change in a future release, when we may provide
 * a "SIMULA compatibility" mode.
 */

void Process::ActivateAt (double AtTime, Boolean prior)
{
    // No op if already scheduled
    if ((AtTime < Process::CurrentTime()) || Terminated || !idle()) return;

    Passivated = FALSE;
    wakeuptime = AtTime;

    ReadyQueue.Insert(*this, prior);
}

void Process::ActivateDelay (double Delay, Boolean prior)
{
    // No op if already scheduled
    if (!checkTime(Delay) || Terminated || !idle()) return;

    Passivated = FALSE;
    wakeuptime = SimulatedTime + Delay;
    ReadyQueue.Insert(*this, prior);
}

void Process::Activate ()
{
    // No op if already scheduled
    if (Terminated || !idle()) return;

    Passivated = FALSE;
    wakeuptime = CurrentTime();
    ReadyQueue.Insert(*this, TRUE);
}

/*
 * Similarly, there are four ways to reactivate
 * Note that if a process is already scheduled, the reactivate
 * will simply re-schedule the process.
 */

void Process::ReActivateBefore (Process &p)
{
    if (Terminated) return;
    
    unschedule();
    ActivateBefore(p);
    if (Process::Current == this)
        Suspend();
}

void Process::ReActivateAfter (Process &p)
{
    if (Terminated) return;
    
    unschedule();
    ActivateAfter(p);
    if (Process::Current == this)
        Suspend();
}

void Process::ReActivateAt (double AtTime, Boolean prior)
{
    if (Terminated) return;

    unschedule();
    ActivateAt(AtTime, prior);
    if (Process::Current == this)
        Suspend();
}

void Process::ReActivateDelay (double Delay, Boolean prior)
{
    if (Terminated) return;
    
    unschedule();
    ActivateDelay(Delay, prior);
    if (Process::Current == this)
        Suspend();
}

void Process::ReActivate ()
{
    if (Terminated) return;
    
    unschedule();
    Activate();
    if (Process::Current == this)
        Suspend();
}

Boolean Process::schedule ()
{
    if (Scheduler::simulationStarted())
    {
	Boolean doSuspend = TRUE;
    
	Process::Current = ReadyQueue.Remove();

	if (Process::Current == (Process*) 0)    // all done
	{
	    std::cout << "Scheduler queue is empty. Simulation ending" << std::endl;
	    Thread::Exit();
	}

	if (Process::Current->evtime() < 0)
	{
	    error_stream << WARNING << "Scheduler Error: Process WakeupTime "
			 << Process::Current->evtime() << " invalid.\n";
	}
	else
	    SimulatedTime = Process::Current->evtime();

#ifdef DEBUG
	debug_stream << FUNCTIONS << FAC_SCHEDULER << VIS_PUBLIC;
	debug_stream << "Simulated time is now " << SimulatedTime << std::endl;
#endif

	if (Process::Current != this)
	    Process::Current->Resume();
	else
	    doSuspend = FALSE;

	return doSuspend;
    }
    else
	return FALSE;
}

// only called if process is running or on queue to be run

void Process::unschedule ()
{
    if (!idle())
    {
	if (this != Process::Current)
	    (void) ReadyQueue.Remove(this); // remove from queue
	wakeuptime = Process::Never;
	Passivated = TRUE;
    }
}
// suspend current process for simulated time t

void Process::Hold (double t)
{
    if (checkTime(t))
    {
	if ((this == Process::Current) || (!Process::Current))
	{
	    wakeuptime = Process::Never;
	    ActivateDelay(t);
	    Suspend();
	}
	else
	    error_stream << WARNING
			 << "Process::Hold - can only be applied to active object."
			 << std::endl;
    }
    else
	error_stream << WARNING << "Process::Hold - time " << t
		     << " invalid." << std::endl;
}

/*
 * We add these routines because we may need to do some
 * Process specific manipulations prior to calling the
 * thread specific implementations.
 */

void Process::Suspend ()
{
    if (schedule())
    {
	_theMutex->unlock();
	Thread::Suspend();
	_theMutex->lock();
    }
}