source: Sophya/trunk/SophyaExt/JThreadsC++/JTCSrc/ThreadGroup.cpp@ 2050

// **********************************************************************
//
// Copyright (c) 2000
// Object Oriented Concepts, Inc.
// Billerica, MA, USA
//
// All Rights Reserved
//
// **********************************************************************

#include <JTC/Types.h>
#include <JTC/Syscall.h>
#include <JTC/Mutex.h>
#include <JTC/Handle.h>
#include <JTC/HandleI.h>
#include <JTC/Event.h>
#include <JTC/Cond.h>
#include <JTC/Monitor.h>
#include <JTC/Exception.h>
#include <JTC/Thread.h>
#include <JTC/ThreadGroup.h>
#include <JTC/Runnable.h>
#include <JTC/Sync.h>

#include <string.h>

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class JTCHandleT<JTCThreadGroup>;
#else
#  ifdef HAVE_PRAGMA_DEFINE
#    pragma define(JTCHandleT<JTCThreadGroup>)
#  endif
#endif

#ifdef HAVE_STD_IOSTREAM
using namespace std;
#endif

// ----------------------------------------------------------------------
// JTCThreadGroup::JTCThreadGroupSet public member implementation
// ----------------------------------------------------------------------

//
// Allocate n members.
//
inline void
JTCThreadGroup::JTCThreadGroupSet::allocate(int n)
{
    len_ = n;
    handles_ = new JTCThreadGroupHandle[n];
}

//
// Return the number of elements in the set.
//
inline int
JTCThreadGroup::JTCThreadGroupSet::length() const
{
    return len_;
}

//
// Return a reference to the i'th element.
//
inline JTCThreadGroupHandle&
JTCThreadGroup::JTCThreadGroupSet::operator[](int i)
{
    return handles_[i];
}

// ----------------------------------------------------------------------
// JTCThreadGroup private member implementation
// ----------------------------------------------------------------------

//
// Create a snapshot of current child ThreadGroups.
//
void
JTCThreadGroup::createGroupSnapshot(JTCThreadGroupSet& set) const
{
    set.allocate(ngroups_);
    for(int i = 0; i < ngroups_; ++i)
    {
        set[i] = groups_[i];
    }
}

//
// Add a ThreadGroup to my child list.
//
void
JTCThreadGroup::add(JTCThreadGroup* g)
{
    JTCSynchronized sync(*this);
    if(destroyed_)
    {
        throw JTCIllegalThreadStateException();
    }
    //
    // No children yet?
    //
    if(groups_ == 0)
    {
        groups_ = new JTCThreadGroup*[4];
        groupsLength_ = 4;
    }
    //
    // Time to grow array?
    //
    else if(ngroups_ == groupsLength_)
    {
        //
        // Allocate memory and then copy...  This is more exception
        // safe.
        //
        int newGroupsLength = ngroups_*2;
        JTCThreadGroup** newGroups = new JTCThreadGroup*[newGroupsLength];
        for(int i = 0; i < ngroups_; ++i)
        {
            newGroups[i] = groups_[i];
        }
        
        delete[] groups_;
        groups_ = newGroups;
        groupsLength_ = newGroupsLength;
    }

    //
    // Add the group to the child list.
    //
    groups_[ngroups_] = g;
    ++ngroups_;
}

//
// Remove a ThreadGroup from my child list.
//
void
JTCThreadGroup::remove(JTCThreadGroup* g)
{
    JTCSynchronized sync(*this);
    if(destroyed_)
    {
        throw JTCIllegalThreadStateException();
    }
    
    for(int i =0 ; i < ngroups_; ++i)
    {
        if(groups_[i] == g)
        {
            --ngroups_;
            for(int j = i; j < ngroups_; ++j)
            {
                groups_[j] = groups_[j+1];
            }
            groups_[ngroups_] = 0;
        }
    }

    //
    // If we remove the last thread then notify the ThreadGroups
    // monitor.
    //
    if(nthreads_ == 0)
    {
        notifyAll();
    }

    //
    // If we've gotten rid of all threads and ThreadGroups then
    // destroy the ThreadGroup.
    //
    if(daemon_ && nthreads_ == 0 && ngroups_ == 0)
    {
        destroy();
    }
}

//
// This code is present to deal with internal compilation errors
// in gcc 2.7.2
//
void
copy_it(JTCThreadHandle* from, int len, JTCThreadHandle* to)
{
    for (int i =0; i < len; ++i)
    {
        to[i] = from[i];
    }
}

void
delete_it(JTCThreadHandle* it)
{
    delete[] it;
}

//
// End internal compilation error code
//

//
// Add a thread to my thread set.
//
void
JTCThreadGroup::add(JTCThreadHandle t)
{
    JTCSynchronized sync(*this);

    if(destroyed_)
    {
        throw JTCIllegalThreadStateException();
    }


    //
    // No threads yet?
    //
    if(threads_ == 0)
    {
        threads_ = new JTCThreadHandle[4];
        threadsLength_ = 4;
    }
    //
    // Time to grow the array?
    //
    else if(nthreads_ == threadsLength_)
    {

        //
        // Allocate and then copy.  This is more exception safe.
        //
        int newThreadLength = nthreads_*2;
        JTCThreadHandle* newThreads = new JTCThreadHandle[newThreadLength];


        // This generates an internal error with gcc 2.7.2
        //for(int i = 0; i < nthreads_; ++i)
        //{
        //    newThreads[i] = threads_[i];
        //}
        // Use this instead
        copy_it(threads_, nthreads_, newThreads);
        //
        
        // This generates an internal error with gcc 2.7.2
        //delete[] threads_;
        // Use this instead
        delete_it(threads_);
        //
        threads_ = newThreads;
        threadsLength_ = newThreadLength;
    }
    //
    // Add thread.
    //
    threads_[nthreads_] = t;
    ++nthreads_;
}

//
// Remove a thread from my thread set.
//
void
JTCThreadGroup::remove(JTCThreadHandle t)
{
    JTCSynchronized sync(*this);
    if(destroyed_)
    {
        throw JTCIllegalThreadStateException();
    }
    for(int i = 0; i < nthreads_; ++i)
    {
        if(threads_[i] == t)
        {
            --nthreads_;
            for(int j = i; j < nthreads_; ++j)
            {
                threads_[j]  = threads_[j+1];
            }
            threads_[nthreads_] = JTCThreadHandle();
        }
    }
    if(nthreads_ == 0)
    {
        notifyAll();
    }
    if(daemon_ && nthreads_ == 0 && ngroups_ == 0)
    {
        destroy();
    }
}

//
// TODO:  ensure that exceptions don't cause memory leaks.  Currently
// exceptions can cause the GroupSnapshot to be trashed.  Some sort
// of auto_ptr can fix this.
//

//
// Copy threads from my thread set into the provided array.
//
int
JTCThreadGroup::enumerate(JTCThreadHandle* list, int len, int n,
                          bool recurse) const
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);

        if(destroyed_)
        {
            return 0;
        }

        //
        // Copy threads into the provided list.
        //
        int nt = nthreads_;
        if(nt > len - n)
        {
            nt = len - n;
        }
        if(nt > 0)
        {
            for(int i = 0; i < nt; ++i)
            {
                    list[i+n] = threads_[i];
            }
            n += nt;
        }
        createGroupSnapshot(groupsSnapshot);
    }

    //
    // Recurse into child ThreadGroups?
    //
    if(recurse)
    {
        for(int i = 0; i < groupsSnapshot.length(); ++i)
        {
            n = groupsSnapshot[i] -> enumerate(list, len, n, 1);
        }
    }

    return n;
}

//
// Copy the set of ThreadGroups into the provided array.
//
int
JTCThreadGroup::enumerate(JTCThreadGroupHandle* list, int len, int n,
                           bool recurse) const
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);

        if(destroyed_)
        {
            return 0;
        }
        
        int ng = nthreads_;
        if(ng > len - n)
        {
            ng = len - n;
        }
        if(ng > 0)
        {
            for(int i = 0; i < ng; ++i)
            {
                list[i+n] = groups_[i];
            }
            n += ng;
        }
        createGroupSnapshot(groupsSnapshot);
    }

    //
    // Recurse into child ThreadGroups?
    //
    if(recurse)
    {
        for(int i = 0; i < groupsSnapshot.length(); ++i)
        {
            n = groupsSnapshot[i] -> enumerate(list, len, n, 1);
        }
    }

    return n;
}

//
// Constructor helper.
//
void
JTCThreadGroup::init(JTCThreadGroup* parent, const char* n)
{
    //
    // Initialize state variables.
    //
    ngroups_ = 0;
    groups_ = 0;
    groupsLength_ = 0;
    nthreads_ = 0;
    threads_ = 0;
    threadsLength_ = 0;
    destroyed_ = false;
    daemon_ = parent -> daemon_;

    //
    // Copy name.
    //
    name_ = new char[strlen(n)+1];
    strcpy(name_, n);
    
    maxPriority_ = parent -> maxPriority_;

    //
    // Add myself to my parent.
    //
    parent_ = parent;
    parent -> add(this);
}

// ----------------------------------------------------------------------
// JTCThreadGroup private constructor
// ----------------------------------------------------------------------

//
// This is the constructor that creates the main thread group.
//
JTCThreadGroup::JTCThreadGroup(bool main)
    : parent_(0),
      destroyed_(false),
      nthreads_(0),
      threads_(0),
      threadsLength_(0),
      ngroups_(0),
      groups_(0),
      groupsLength_(0),
      daemon_(false)
{
    //
    // Default name of the system ThreadGroup is system, otherwise
    // the name is adopted.
    //
    const char* defaultName = (main) ? "system" : "adopted";
    name_ = new char[strlen(defaultName)+1];
    strcpy(name_, defaultName);
    
    maxPriority_ = JTCThread::JTC_MAX_PRIORITY;
}

// ----------------------------------------------------------------------
// JTCThreadGroup constructors and destructor
// ----------------------------------------------------------------------

JTCThreadGroup::JTCThreadGroup(const char* name)
{
    init(JTCThread::currentThread() -> getThreadGroup().get(), name);
}

JTCThreadGroup::JTCThreadGroup(JTCThreadGroup* parent, const char* name)
{
    init(parent, name);
}

JTCThreadGroup::~JTCThreadGroup()
{
    try
    {
        JTCSynchronized sync(*this);
        if (!destroyed_)
        {
            destroy();
        }
    }
    catch(const JTCIllegalThreadStateException&)
    {
        cerr << "Unexpected JTCIllegalThreadStateException in ~JTCThreadGroup"
             << endl;
    }
    delete[] name_;
}

// ----------------------------------------------------------------------
// JTCThreadGroup public member implementation
// ----------------------------------------------------------------------

//
// Return name of the this ThreadGroup.
//
const char*
JTCThreadGroup::getName() const
{
    return name_;
}

//
// Return the parent of this ThreadGroup.
//
JTCThreadGroupHandle
JTCThreadGroup::getParent() const
{
    return JTCThreadGroupHandle(parent_);
}

//
// Is this a daemon ThreadGroup?
//
bool
JTCThreadGroup::isDaemon() const
{
    return daemon_;
}

//
// Set the daemon ThreadGroup flag.
//

void
JTCThreadGroup::setDaemon(bool daemon)
{
    daemon_ = daemon;
}

//
// Called on an uncaught exception.
//
void
JTCThreadGroup::uncaughtException(JTCThreadHandle t, const JTCException& e)
{
    if (parent_)
    {
        parent_ -> uncaughtException(t, e);
    }
    else
    {
        cerr << e << endl;
    }
}

//
// Called on an uncaught exception (... style)
//
void
JTCThreadGroup::uncaughtException(JTCThreadHandle t)
{
    if (parent_)
    {
        parent_ -> uncaughtException(t);
    }
    else
    {
        cerr << "unknown exception" << endl;
    }
}

//
// Return the maximum priority of this ThreadGroup.
//
int
JTCThreadGroup::getMaxPriority() const
{
    return maxPriority_;
}

#ifdef HAVE_JTC_STOP
//
// Stop all threads in this ThreadGroup, and child ThreadGroups.
//
void
JTCThreadGroup::stop()
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);

        //
        // Stop all of the threads that this ThreadGroup owns.
        //
        for (int i = 0; i < nthreads_; ++i)
        {
            threads_[i] -> stop();
        }
        createGroupSnapshot(groupsSnapshot);
    }

    //
    // Ask each child ThreadGroup to stop their threads.
    //
    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        groupsSnapshot[i] -> stop();
    }
}

//
// Resume all threads in our ThreadGroup, and child ThreadGroups.
//
void
JTCThreadGroup::resume()
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);
        //
        // Resume all threads in our ThreadGroup.
        //
        for (int i = 0; i < nthreads_; ++i)
        {
            threads_[i] -> resume();
        }
        createGroupSnapshot(groupsSnapshot);
    }

    //
    // Resume all threads in our child ThreadGroups.
    //
    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        groupsSnapshot[i] -> resume();
    }
}

//
// Suspend all threads in our ThreadGroup, and child ThreadGroups.
//
void
JTCThreadGroup::suspend()
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);

        //
        // Suspend all threads in our ThreadGroup.
        //
        for (int i = 0; i < nthreads_; ++i)
        {
            threads_[i] -> suspend();
        }
        createGroupSnapshot(groupsSnapshot);
    }

    //
    // Suspend all threads in our child ThreadGroups.
    //
    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        groupsSnapshot[i] -> suspend();
    }
}
#endif

//
// Has this ThreadGroup been destroyed?
//
bool
JTCThreadGroup::isDestroyed() const
{
    JTCSynchronized sync(*this);
    return destroyed_;
}

//
// Destroy this ThreadGroup.
//
void
JTCThreadGroup::destroy()
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);
        if(destroyed_ || nthreads_ > 0)
        {
            throw JTCIllegalThreadStateException();
        }
        createGroupSnapshot(groupsSnapshot);

        destroyed_ = true;
        ngroups_ = 0;
        delete[] groups_;
        groups_ = 0;
        groupsLength_ = 0;
        nthreads_ = 0;
        delete[] threads_;
        threads_ = 0;
        threadsLength_ = 0;
    }

    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        groupsSnapshot[i] -> destroy();
    }

    if(parent_)
    {
        parent_ -> remove(this);
    }
}

//
// Set maximum priority value for this ThreadGroup, and our child
// ThreadGroups.
//
void
JTCThreadGroup::setMaxPriority(int pri)
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);
        if(pri < JTCThread::JTC_MIN_PRIORITY)
        {
            maxPriority_ = JTCThread::JTC_MIN_PRIORITY;
        }
        else if(pri < maxPriority_)
        {
            maxPriority_ = pri;
        }
        createGroupSnapshot(groupsSnapshot);
    }

    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        groupsSnapshot[i] -> setMaxPriority(pri);
    }
}

//
// Is this ThreadGroup a parent of handle?
//
bool
JTCThreadGroup::parentOf(JTCThreadGroupHandle handle)
{
    JTCThreadGroup* g = handle.get();
    for(;g != 0; g = g -> parent_.get())
    {
        if(g == this)
        {
            return true;
        }
    }
    return false;
}

//
// Determine the number of active threads in this ThreadGroup, and
// child ThreadGroups.
//
int
JTCThreadGroup::activeCount() const
{
    int result = 0;

    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);

        if(destroyed_)
        {
            return 0;
        }
        result = nthreads_;

        createGroupSnapshot(groupsSnapshot);
    }

    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        result += groupsSnapshot[i] -> activeCount();
    }

    return result;
}

//
// Determine the number of ThreadGroups in this group, and child
// ThreadGroups.
//
int
JTCThreadGroup::activeGroupCount() const
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);

        if(destroyed_)
        {
            return 0;
        }
        createGroupSnapshot(groupsSnapshot);
    }

    int n = groupsSnapshot.length();
    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        n += groupsSnapshot[i] -> activeCount();
    }
    return n;
}

//
// Copy the set of threads in this ThreadGroup, and child ThreadGroups
// if recurse is true into array list.
//
int
JTCThreadGroup::enumerate(JTCThreadHandle* list, int len, bool recurse) const
{
    return enumerate(list, len, 0, recurse);
}

//
// Copy the set child ThreadGroups in this ThreadGroup, and child ThreadGroups
// if recurse is true into array list.
//
int
JTCThreadGroup::enumerate(JTCThreadGroupHandle* list, int len,
                          bool recurse) const
{
    return enumerate(list, len, 0, recurse);
}

//
// Display set of threads and ThreadGroups to cout.
//
void
JTCThreadGroup::list()
{
    list(cout, 0);
}

//
// Display the set of threads and ThreadGroups to os, with given indent
// level.
//
void
JTCThreadGroup::list(ostream& os, int indent)
{
    JTCThreadGroupSet groupsSnapshot;

    {
        JTCSynchronized sync(*this);
        for(int j = 0; j < indent; ++j)
        {
            os << " ";
        }
        os << *this << endl;
        indent += 4;
        for(int i = 0; i < nthreads_; ++i)
        {
            for(int i_level = 0; i_level < indent; ++i_level)
            {
                os << " ";
            }
            os << *threads_[i] << endl;
        }
        createGroupSnapshot(groupsSnapshot);
    }
    for(int i = 0; i < groupsSnapshot.length(); ++i)
    {
        groupsSnapshot[i] -> list(os, indent);
    }
}

//
// IOstream insertion operator for a ThreadGroup.
//
ostream&
operator<<(ostream& os, const JTCThreadGroup& g)
{
    os << "ThreadGroup: "
       << "[name=" << g.getName() << ",maxpri=" << g.getMaxPriority() << "]";
    return os; 
}