source: tbroadcast/v1/src/tbroadcast.cxx @ 231

#include <TCmThread.hxx>

#include <iostream>
#include <fstream>
#include <strstream>
#include <vector>
#include <map>
#include <cstdlib>
#include <cstdio>
#include <sys/time.h>


/**
 *
 * Transactions
 *
 */

class Transaction
{
public:
  Transaction ();
  Transaction (const std::string& info);
  ~Transaction ();

  void lock ();
  void unlock ();
  const std::string& get_info () const;
  void do_signal ();
  bool do_wait_signal (int milliseconds);
  void do_wait ();
  bool do_wait (int milliseconds);

  std::string _info;
  pthread_cond_t _var;
  pthread_mutex_t _mutex;
  int _count;
};

#include <time.h>
#include <unistd.h>

/**
 * Local cleanup function 
 */
static void UnlockMutex (void* arg)
{
  int status;
  
  status = pthread_mutex_unlock ((pthread_mutex_t*) arg);
}

Transaction::Transaction ()
{
  pthread_mutex_init (&_mutex, 0);
  pthread_cond_init (&_var, 0);
  _count = 0;
}

Transaction::Transaction (const std::string& info)
{
  _info   = info;
  pthread_mutex_init (&_mutex, 0);
  pthread_cond_init (&_var, 0);
  _count = 0;
}

Transaction::~Transaction ()
{
  pthread_mutex_destroy (&_mutex);
  pthread_cond_destroy (&_var);
}

void Transaction::lock ()
{
  pthread_mutex_lock (&_mutex);
}

void Transaction::unlock ()
{
  pthread_cond_broadcast (&_var);
  pthread_mutex_unlock (&_mutex);
}

void Transaction::do_signal ()
{
  pthread_mutex_lock (&_mutex);
  if (TCmThread::debug_level () > 0)
    {
      std::cout << "Transaction::do_signal>" << _info << " " << _count << std::endl;
    }
  _count++;
  pthread_cond_broadcast (&_var);
  pthread_mutex_unlock (&_mutex);
}

const std::string& Transaction::get_info () const
{
  return (_info);
}

bool Transaction::do_wait_signal (int milliseconds)
{
  int status = 0;

  pthread_cleanup_push (UnlockMutex, &_mutex);
  pthread_mutex_lock (&_mutex);

  struct timeval now;

  TCmMutex::iolock ();
  gettimeofday (&now, 0);
  TCmMutex::iounlock ();

  struct timespec delta;
  struct timespec* timeout;

  delta.tv_sec = now.tv_sec + milliseconds / 1000;
  delta.tv_nsec = now.tv_usec + milliseconds * 1000;

  timeout = &delta;

  if (TCmThread::debug_level () > 0)
    {
      std::cout << "Transaction::do_wait_signal(t)> " << _info << " " << _count << std::endl;
    }
  
  status = pthread_cond_timedwait (&_var, &_mutex, timeout);
  
  //std::cout << "Transaction::do_wait(t)> " << _info << " status=" << status << std::endl;
  
  pthread_cleanup_pop (1);

  return (status == 0);
}

void Transaction::do_wait ()
{
  pthread_cleanup_push (UnlockMutex, &_mutex);
  pthread_mutex_lock (&_mutex);

  if (TCmThread::debug_level () > 0)
    {
      std::cout << "Transaction::do_wait> " << _info << std::endl;
    }

  pthread_cond_wait (&_var, &_mutex);
  pthread_cleanup_pop (1);
}

bool Transaction::do_wait (int milliseconds)
{
  int status = 0;

  pthread_cleanup_push (UnlockMutex, &_mutex);
  pthread_mutex_lock (&_mutex);

  struct timeval now;

  TCmMutex::iolock ();
  gettimeofday (&now, 0);
  TCmMutex::iounlock ();

  struct timespec delta;
  struct timespec* timeout;

  delta.tv_sec = now.tv_sec + milliseconds / 1000;
  delta.tv_nsec = now.tv_usec + milliseconds * 1000;

  timeout = &delta;

  if (TCmThread::debug_level () > 0)
    {
      std::cout << "Transaction::do_wait(t)> " << _info << std::endl;
    }
  
  status = pthread_cond_timedwait (&_var, &_mutex, timeout);
  
  //std::cout << "Transaction::do_wait(t)> " << _info << " status=" << status << std::endl;
  
  pthread_cleanup_pop (1);
  
  return (status == 0);
}

/**
 * The configuration parameters for the program
 */
int MaxThreads = 20;

/**
 * Global variables
 */
int LastId = 0;

/// A transaction to wake up the scheduler
Transaction QueueTr ("QueueTr");

/// Count the number of objects still pending
int ObjectCount = 0;

/// Accumulating the run time
double TotalTime = 0;

/// A mutex for ObjectCount and TotalTime
TCmMutex CountLock ("CountLock");

class A;

/// All pending objects (and associated mutex)
std::vector <A*> ThreadQueue;
TCmMutex ThreadLock ("ThreadLock");

/// Counting the running threads (and associated mutex)
int ThreadCount = 0;
TCmMutex ThreadCountLock ("ThreadCountLock");

class Package;

/**
 * Dictionaries for packages and for runner objects
 */
std::map <std::string, Package*> Packages;
std::map <std::string, Project*> Projects;
std::map <std::string, A*> As;
int PackageCount = 0;

/**
 *  The shell command to run in every runner 
 */
std::string Command = "";

/**
 *  The expanded CMTPATH
 */
std::vector<std::string> Cmtpath_items;

/**
 *  Global parameters for selection or exclusion
 */
std::vector <std::string> Selections;
std::vector <std::string> Exclusions;
std::string Begin;
TCmMutex BeginLock ("BeginLock");
std::vector <std::string> Cmtpath_selections;

/**
 * Helper to get a flat random integer value
 */
unsigned int get_random (unsigned int range)
{
  double x = (((double) range) * (double) rand ()) / ((double) RAND_MAX);

  return ((unsigned int) x);
}

std::string p_pwd ()
{
  char buffer[256] = "";
  char* ptr = 0;
  ptr = getcwd (buffer, sizeof (buffer));

  const char* t = &buffer[0];
  return ((std::string) t);
}

void p_basename (const std::string& file_name, std::string& result)
{
  unsigned int pos = file_name.find_last_of ('/');
  if (pos == std::string::npos)
    {
      pos = file_name.find_last_of ('\\');
    }

  if (pos == std::string::npos)
    {
      result = file_name;
    }
  else
    {
      result = file_name.substr (pos + 1);
    }
}

//--------------------------------------------------
void p_dirname (const std::string& file_name, std::string& result)
{
  unsigned int pos = file_name.find_last_of ('/');
  if (pos == std::string::npos)
    {
      pos = file_name.find_last_of ('\\');
    }

  if (pos == std::string::npos)
    {
      result = "";
    }
  else
    {
      result = file_name;
      result.erase (pos);
    }
}


/**
 * Execute a shell command and accumulate its output into a string
 */
static int execute (const std::string& command, std::string& output)
{
  output = "";

  FILE* f = popen (command.c_str (), "r"); 
  
  if (f != 0) 
    { 
      char line[256]; 
      char* ptr;

      while ((ptr = fgets (line, sizeof (line), f)) != NULL) 
        {
          output += ptr;
        } 
      pclose (f);

      return (1);
    }

  return (0);
}

/**
 * The Package class stores information about a package
 */
class Package
{
public:
  Package (const std::string& name,
           const std::string& version,
           const std::string& offset,
           const std::string& cmtpath) :
    m_name (name),
    m_version (version),
    m_offset (offset),
    m_cmtpath (cmtpath)
  {
  }

  void set (const std::string& name,
            const std::string& version,
            const std::string& offset,
            const std::string& cmtpath)
  {
    m_name = name;
    m_version = version;
    m_offset = offset;
    m_cmtpath = cmtpath;

    m_location = m_cmtpath;

    if (m_cmtpath != "")
      {
        if (m_offset != "") 
          {
            m_location += m_offset;
          }
        m_location += "/";
      }

    m_location += m_name;
    m_location += "/";
    m_location += m_version;
    m_location += "/cmt";
  }

  const std::string& name () const
  {
    return (m_name);
  }

  const std::string& version () const
  {
    return (m_version);
  }

  const std::string& offset () const
  {
    return (m_offset);
  }

  const std::string& cmtpath () const
  {
    return (m_cmtpath);
  }

  const std::string& location () const
  {
    return (m_location);
  }

private:
  std::string m_name;
  std::string m_version;
  std::string m_offset;
  std::string m_cmtpath;
  std::string m_location;
};

/**
 * A runner object. It is associated with a package.
 */
class A : public TCmThread
{
public:
  A () : m_package (0),
         m_started (false), 
         m_finished (false), 
         m_ready (false),
         m_queued (false),
         m_launched (false),
         m_run_status (0)
  {
    ObjectCount++;
    LastId++;
    m_id = LastId;
  }

  A (Package* package) : m_package (package),
                         m_started (false), 
                         m_finished (false), 
                         m_ready (false),
                         m_queued (false),
                         m_launched (false),
                         m_run_status (0)
  {
    ObjectCount++;
    LastId++;
    m_id = LastId;
    m_mutex.set_info (name ());
  }

  ~A ()
  {
  }

  /**
   * Called when a runner terminates.
   *
   *   1) Push back the parent to the activation queue when it's ready to start
   *   2) wake up the scheduler
   */
  void terminate ()
  {
    if (m_parents.size () > 0)
      {
        std::vector <A*>::iterator it;

        for (it = m_parents.begin (); it != m_parents.end (); ++it)
          {
            A* a = *it;

            a->terminate_child (this);

            if (a->should_start ())
              {
                a->queue ();
              }
          }
      }

    QueueTr.do_signal ();
  }

  /**
   * Callback called after a thread terminates.
   *
   *   1) count down the number of active threads
   *   2) Push back the parent to the activation queue when it's ready to start
   *   3) wake up the scheduler
   */
  void cleanup ()
  {
    /*
    TCmMutex::iolock ();
    std::cout << "cleanup " << name () << std::endl;
    TCmMutex::iounlock ();
    */

    ThreadCountLock.lock ("cleanup");
    ThreadCount--;
    ThreadCountLock.unlock ();

    terminate ();
  }

  /**
   * Construct the name of this runner from the package name
   */
  std::string name () const
  {
    if (m_package != 0)
      {
        return (m_package->name ());
      }
    else
      {
        char buff [80];
        sprintf (buff, "A%d", m_id);

        return (buff);
      }
  }

  const std::string& version () const
  {
    if (m_package != 0)
      {
        return (m_package->version ());
      }
    else
      {
        static std::string empty;
        return (empty);
      }
  }

  const std::string& offset () const
  {
    if (m_package != 0)
      {
        return (m_package->offset ());
      }
    else
      {
        static std::string empty;
        return (empty);
      }
  }

  const std::string& cmtpath () const
  {
    if (m_package != 0)
      {
        return (m_package->cmtpath ());
      }
    else
      {
        static std::string empty;
        return (empty);
      }
  }

  const std::string& location () const
  {
    if (m_package != 0)
      {
        return (m_package->location ());
      }
    else
      {
        static std::string empty;
        return (empty);
      }
  }

  int id () const
  {
    return (m_id);
  }

  /**
   * Recursively find if an object is already in the parent list
   * (check for cycles)
   */
  bool has_parent (A* a)
  {
    if (a == this) return (true);

    std::vector <A*>::iterator it;

    for (it = m_parents.begin (); it != m_parents.end (); ++it)
      {
        A* p = *it;

        if (p == a) return (true);

        if (p->has_parent (a)) return (true);
      }

    return (false);
  }

  /**
   *  Adds a unique child (should first check for cycles)
   *  Avoid duplications.
   */
  void add_child (A* a)
  {
    if (a != 0)
      {
        std::vector <A*>::iterator it;

        for (it = m_children.begin (); it != m_children.end (); ++it)
          {
            A* c = *it;
            if (c == a) return;
          }

        //std::cout << name () << " add child " << a->name () << std::endl;
        m_children.push_back (a);
        a->add_parent (this);
      }
  }

  /**
   * Adds a parent without duplication
   */
  void add_parent (A* a)
  {
    std::vector <A*>::iterator it;

    for (it = m_parents.begin (); it != m_parents.end (); ++it)
      {
        A* p = *it;
        if (p == a) return;
      }

    m_parents.push_back (a);
  }

  void dequeue ()
  {
    m_mutex.lock ("dequeue");
    m_queued = false;
    m_mutex.unlock ();
  }

  bool is_finished ()
  {
    bool result;

    m_mutex.lock ("is_finished");
    result = m_finished;
    m_mutex.unlock ();

    return (result);
  }

  /**
   *  A runner can be queued if it was never started yet and
   *  it has no more children
   */
  bool can_be_queued ()
  {
    bool result;

    m_mutex.lock ("can_be_queued");
    result = !m_finished && !m_started && (m_children.size () == 0);
    m_mutex.unlock ();

    return (result);
  }

  /**
   *  A runner should start when all children are finished
   */
  bool should_start ()
  {
    bool result = (check_state () == ready_to_start);

    return (result);
  }

  int get_run_status ()
  {
    int result;

    m_mutex.lock ("get_run_status");
    result = m_run_status;
    m_mutex.unlock ();

    return (result);
  }

  typedef enum
  {
    already_started,
    already_finished,
    not_yet_completed,
    ready_to_start,
    really_started,
    queued
  } lock_status;

  /**
   *  Analyze the state of a runner
   */
  lock_status check_state ()
  {
    bool done = false;

    m_mutex.lock ("check_state-1");
    done = m_finished;
    m_mutex.unlock ();

    if (done)
      {
        return (already_finished);
      }

    m_mutex.lock ("check_state-2");
    done = m_started;
    m_mutex.unlock ();

    if (done)
      {
        return (already_started);
      }

    // not yet started

    /*
    TCmMutex::iolock ();
    std::cout << "check_state-1> " << name () << std::endl;
    TCmMutex::iounlock ();
    */

    bool should_check_children = false;

    m_mutex.lock ("check_state-3");
    should_check_children = !m_ready;
    m_mutex.unlock ();

    if (should_check_children)
      {
        // check for unfinished children

        std::vector <A*>::iterator it;

        bool is_ready = true;

        m_mutex.lock ("check_state-4");
        for (it = m_children.begin (); it != m_children.end (); ++it)
          {
            A* c = *it;

            if (!c->is_finished ()) 
              {
                is_ready = false;
                break;
              }
          }
        m_mutex.unlock ();

        if (!is_ready)
          {
            return (not_yet_completed);
          }

        // all children are finished
        
        m_mutex.lock ("check_state-4");
        m_ready = true;
        m_mutex.unlock ();
      }

    return (ready_to_start);
  }

  /**
   *  Actually ask the runner to run its thread
   */
  bool trigger ()
  {
    /*
    TCmMutex::iolock ();
    std::cout << "trigger> " << name ();
    std::cout << std::endl;
    TCmMutex::iounlock ();
    */

    ThreadCountLock.lock ("trigger");
    ThreadCount++;
    ThreadCountLock.unlock ();
    
    m_mutex.lock ("trigger");
    m_started = true;
    m_mutex.unlock ();

    bool result = start ();

    return (result);
  }

  /**
   *  Abort the planned activity. Must emulate all
   *  mechanisms of trigger-start-cleanup-wakeup
   */
  void abort ()
  {
    CountLock.lock ("abort");
    ObjectCount--;
    int oc = ObjectCount;
    CountLock.unlock ();

    TCmMutex::iolock ();

    std::cout << "# Aborting " << name ()
              << " (" << PackageCount - oc << "/" << PackageCount << ")" 
              << std::endl;

    TCmMutex::iounlock ();

    m_mutex.lock ("abort");
    m_finished = true;
    m_run_status = 1;
    m_mutex.unlock ();

    terminate ();
  }

  /**
   *  Ignore the planned activity. Must emulate all
   *  mechanisms of trigger-start-cleanup-wakeup
   */
  void ignore ()
  {
    CountLock.lock ("ignore");
    ObjectCount--;
    //int oc = ObjectCount;
    CountLock.unlock ();

    /*
    TCmMutex::iolock ();

    std::cout << "# Ignoring " << name ()
              << " (" << PackageCount - oc << "/" << PackageCount << ")" 
              << std::endl;

    TCmMutex::iounlock ();
    */

    m_mutex.lock ("ignore");
    m_finished = true;
    m_run_status = 0;
    m_mutex.unlock ();

    terminate ();
  }

  /**
   *  Construct a text representation of the runner state
   */
  std::string get_state ()
  {
    std::string s = "(";

    m_mutex.lock ("get_state");
    if (m_started) s += "s";
    if (m_finished) s += "f";
    if (m_ready) s += "r";
    if (m_queued) s += "q";
    if (m_launched) s += "l";
    m_mutex.unlock ();

    s += ")";

    return (s);
  }

  /**
   *  Debugging function
   */
  void special_show ()
  {
    std::string s = "";

    s = get_state ();

    s += " p[";
    std::vector<A*>::iterator it;

    for (it = m_parents.begin (); it != m_parents.end (); ++it)
      {
        A* a = *it;
        s += a->name ();
        s += a->get_state ();
        s += " ";
      }
    
    s += "]";
    s += " c[";
    
    for (it = m_children.begin (); it != m_children.end (); ++it)
      {
        A* a = *it;
        s += a->name ();
        s += a->get_state ();
        s += " ";
      }
    
    s += "]";

    TCmMutex::iolock ();
    std::cout << "### " << name () << " " << s << std::endl;
    TCmMutex::iounlock ();
  }

  /**
   *   A child has just terminated (possibly with error)
   *   The parent removes it from its children and inherit error.
   */
  void terminate_child (A* child)
  {
    if (child == 0) return;

    std::vector <A*>::iterator cit;

    m_mutex.lock ("terminate_child");

    for (cit = m_children.begin (); cit != m_children.end (); ++cit)
      {
        A* c = *cit;
        if (c == child)
          {
            int status = c->get_run_status ();
            if (status != 0)
              {
                m_run_status = status;
              }
            m_children.erase (cit);
            break;
          }
      }

    m_mutex.unlock ();

  }

  bool selected ()
  {
    if ((Begin == "") && 
        (Cmtpath_selections.size () == 0) && 
        (Selections.size () == 0) && 
        (Exclusions.size () == 0)) return (true);

    const std::string& loc = location ();

    bool result = true;

    BeginLock.lock ("selected");
    if (Begin != "")
      {
        if (loc.find (Begin) == std::string::npos)
          {
            result = false;
          }
        else
          {
            Begin = "";
          }
      }
    BeginLock.unlock ();

    if (!result)
      {
        return (result);
      }

    if (Cmtpath_selections.size () != 0)
      {
        result = false;

        std::vector<std::string>::iterator sit;

        for (sit = Cmtpath_selections.begin (); sit != Cmtpath_selections.end (); ++sit)
          {
            const std::string& s = *sit;
            
            if (loc.find (s) == 0)
              {
                result = true;
                break;
              }
          }
      }

    if (Selections.size () != 0)
      {
        result = false;

        std::vector<std::string>::iterator sit;

        for (sit = Selections.begin (); sit != Selections.end (); ++sit)
          {
            const std::string& s = *sit;
            
            if (loc.find (s) != std::string::npos)
              {
                result = true;
                break;
              }
          }
      }

    if (result)
      {
        std::vector<std::string>::iterator eit;

        for (eit = Exclusions.begin (); eit != Exclusions.end (); ++eit)
          {
            const std::string& e = *eit;
            
            if (loc.find (e) != std::string::npos)
              {
                result = false;
                break;
              }
          }
      }

    return (result);
  }

  /**
   *  Send a runner to the activation queue.
   *  to optimize the next checks
   */
  void queue ()
  {
    /*
    TCmMutex::iolock ();
    std::cout << "Queue-begin> " << name ();
    std::cout << std::endl;
    TCmMutex::iounlock ();
    */

    /*
    ThreadLock.lock ("queue-1");
    std::vector <A*>::iterator it;
    TCmMutex::iolock ();
    std::cout << "Q[";
    for (it = ThreadQueue.begin (); it != ThreadQueue.end (); ++it)
      {
        A* a = *it;
        std::cout << a->name () << " ";
      }
    std::cout << "]" << std::endl;
    TCmMutex::iounlock ();
    ThreadLock.unlock ();
    */

    if (!selected ())
      {
        ignore ();
        return;
      }

    m_mutex.lock ("queue-1");

    if (!m_queued)
      {
        /*
          std::string s = get_state ();

          TCmMutex::iolock ();
          std::cout << "Sending " << name () << " to queue";
          std::cout << " " << s << std::endl;
          TCmMutex::iounlock ();
        */

        ThreadLock.lock ("queue-2");
        m_queued = true;
        ThreadQueue.push_back (this);
        ThreadLock.unlock ();
      }
    else
      {
        /*
          std::string s = get_state ();

          TCmMutex::iolock ();
          std::cout << "Keeping " << name () << " in queue";
          std::cout << " " << s << std::endl;
          TCmMutex::iounlock ();
        */
      }

    m_mutex.unlock ();
  }

  /**
   * Initialize the activation queue by selectively pushing leaves
   * of the graph
   */
  void launch ()
  {
    std::map<std::string, A*>::iterator ait;
    for (ait = As.begin (); ait != As.end (); ++ait)
      {
        A* a = (*ait).second;
        if (a->can_be_queued ())
          {
            a->queue ();
          }
      }

    /*
    if (m_launched) return;

    m_launched = true;

    if (m_children.size () > 0)
      {
        for (unsigned int i = 0; i < m_children.size (); i++)
          {
            A* child = m_children[i];
            
            child->launch ();
          }
      }
    else
      {
        queue ();
      }
    */
  }

  /**
   *  The runner threaded activity.
   *
   *   1) execute the command (and accumulates its output in a text)
   *   2) display the output and statistics
   *   3) set the finished state
   *
   *   (the cleanup callback will manage the post-run activities)
   */
  void run ()
  {
    int status = 0;
    std::string output;

    std::string cp = cmtpath ();

    std::string where = location ();

    if (cp == "")
      {
        output = "# Package ";
        output += name ();
        output += " not found";
      }
    else
      {
        
        std::string cmd = "(cd ";
        cmd += where;
        if (Command != "")
          {
            cmd += "; ";
            cmd += Command;
          }
        cmd += "; echo cmtbroadcaststatus=$?) 2>&1";

        execute (cmd, output);

        unsigned int status_pos = output.find ("cmtbroadcaststatus=");
        if (status_pos != std::string::npos)
          {
            std::string s = output.substr (status_pos);
            output.erase (status_pos);
            sscanf (s.c_str (), "cmtbroadcaststatus=%d", &status);
          }
      }

    CountLock.lock ("run");
    ObjectCount--;
    int oc = ObjectCount;
    CountLock.unlock ();

    TCmMutex::iolock ();

    std::cout << "#--------------------------------------------------------------" << std::endl;
    std::cout << "# Now trying [" << Command << "] in " << where
              << " (" << PackageCount - oc << "/" << PackageCount << ")" 
              << std::endl;
    std::cout << "#--------------------------------------------------------------" << std::endl;

    std::cout << output << std::endl;

    TCmMutex::iounlock ();

    m_mutex.lock ("run");
    m_finished = true;
    m_run_status = status;
    m_mutex.unlock ();
  }

private:
  int m_id;
  std::vector <A*> m_children;
  std::vector <A*> m_parents;
  Package* m_package;
  bool m_started;
  bool m_finished;
  bool m_ready;
  bool m_queued;
  bool m_launched;
  TCmMutex m_mutex;
  int m_run_status;
};

/**
 *  Analyze one line of the "cmt show uses" result
 */
void parse_line (const std::string& line, A& top)
{
  static std::vector <A*> Hierarchy;

  std::istrstream s (line.c_str ());

  if (line[0] == '#')
    {
      /**
       * First section : the hierarchy
       *  o create all Packages and Runners
       *  o construct the graph (avoiding cycles)
       */

      std::string s1;
      std::string s2;
      s >> s1 >> s2;

      if (s2 != "use") return;

      unsigned int pos = 2;

      pos = line.find ("use") - 2;
      
      unsigned int level = (pos / 2);

      std::string n;
      std::string v;
      std::string o;

      s >> n >> v >> o;

      if (o[0] == '(')
        {
          o = "";
        }

      std::map <std::string, Package*>::iterator p_it;

      p_it = Packages.find (n);

      Package* package;

      if (p_it == Packages.end ())
        {
          package = new Package (n, v, o, "");
          Packages[n] = package;
        }
      else
        {
          package = (*p_it).second;
        }

      //std::cout << "Level=" << level << " n=" << n << std::endl;

      A* parent = 0;

      if (level > 0)
        {
          parent = Hierarchy[level - 1];
        }
      else
        {
          parent = &top;
        }

      std::map <std::string, A*>::iterator a_it;

      a_it = As.find (n);

      A* a;

      if (a_it == As.end ())
        {
          a = new A (package);
          As[n] = a;
        }
      else
        {
          a = (*a_it).second;
        }

      if (Hierarchy.size () <= level)
        {
          Hierarchy.push_back (a);
        }
      else
        {
          Hierarchy[level] = a;
        }

      if (parent != 0)
        {
          if (parent->has_parent (a)) 
            {
              std::cout << "Cycle for package " << n << std::endl;
            }
          else
            {
              parent->add_child (a);
            }
        }
    }
  else
    {
      /**
       *  Second section : get physical package information
       */

      std::string s1;
      std::string n;
      std::string v;
      std::string o;
      std::string p;
      s >> s1;
      if (s1 != "use") return;
      s >> n >> v >> o;
      if (o[0] == '(')
        {
          p = o.substr (1, o.length() - 2);
          o = "";
        }
      else
        {
          s >> p;
          p = p.substr (1, p.length() - 2);
        }
      
      //std::cout << "n=" << n << " v=" << v << " o=" << o << " p=" << p << std::endl;

      std::map <std::string, Package*>::iterator it;

      it = Packages.find (n);

      if (it != Packages.end ())
        {
          Package* pa = (*it).second;
          pa->set (n, v, o, p);
        }
      else
        {
          Packages[n] = new Package (n, v, o, p);
        }
    }
}

class Scheduler
{
public:
  Scheduler ()
  {
  }

  /**
   *  Obtain all packages from a cmt show uses command
   */
  void create_packages_from_command (const std::string& command)
  {
    std::string pwd = p_pwd ();

    std::string v = pwd;
    p_dirname (pwd, v);
    std::string p = v;
    p_basename (v, v);
    p_dirname (p, p);
    std::string c = p;
    p_basename (p, p);

    p_dirname (c, c);

    static Package current (p, v, "", c);

    current.set (p, v, "", c);

    static A top (&current);

    FILE* f = popen (command.c_str (), "r"); 
  
    if (f != 0) 
      { 
        char line[256]; 
        char* ptr;

        std::string s;

        while ((ptr = fgets (line, sizeof (line), f)) != NULL) 
          {
            s = ptr;
            parse_line (s, top);
          } 
        pclose (f);
      }

    PackageCount = Packages.size ();
    top.launch ();
  }
  
  A* get_next_pending_object ()
  {
    A* to_start = 0;

    ThreadLock.lock ("get_next_pending_object");
    if (ThreadCount < MaxThreads)
      {
        if (ThreadQueue.size () > 0)
          {
            std::vector <A*>::iterator it;
            it = ThreadQueue.begin ();
            if (it != ThreadQueue.end ())
              {
                to_start = *it;
                ThreadQueue.erase (it);
                to_start->dequeue ();
              }
          }
      }
    ThreadLock.unlock ();

    return (to_start);
  }

  bool should_wait ()
  {
    bool result = true;

    ThreadCountLock.lock ("main-1");
    int tc = ThreadCount;
    ThreadCountLock.unlock ();

    ThreadLock.lock ("should_wait");
    int ts = ThreadQueue.size ();
    ThreadLock.unlock ();

    if ((tc < MaxThreads) && (ts > 0)) 
      {
        result = false;
      }

    return (result);
  }

  void run ()
  {
    QueueTr.do_signal ();

    for (;;)
      {
        int oc = 0;
        
        ThreadLock.lock ("run");
        oc = ObjectCount;
        ThreadLock.unlock ();

        /// the scheduler stops when all objects have been activated
        if (oc == 0) break;

        A* to_start = get_next_pending_object ();

        if (to_start != 0)
          {
            if (to_start->should_start ())
              {
                int status = to_start->get_run_status ();

                if (status == 0)
                  {
                    // start it now

                    if (!to_start->trigger ())
                      {
                        to_start->queue ();
                      }
                  }
                else
                  {
                    to_start->abort ();
                  }
              }
            else
              {
                // Now that the queue only contains ready_to_start runners, this
                // case should never happen !!
                // however we re-queue it
                
                to_start->queue ();
              }
          }

        if (should_wait ())
          {
            QueueTr.do_wait_signal (10000);
          }
      }
  }
};

void expand_cmtpath ()
{
  const char* cmtpath_str = ::getenv ("CMTPATH");

  if (cmtpath_str != 0)
    {
      static const char path_separator = ':';

      std::string cmtpath_list (cmtpath_str);
      std::string::size_type pos = 0;

      for (;;)
        {
          bool ending = false;

          std::string::size_type next = cmtpath_list.find (path_separator, pos);

          std::string path;

          if (next == std::string::npos)
            {
              path = cmtpath_list.substr (pos);
              ending = true;
            }
          else
            {
              path = cmtpath_list.substr (pos, next - pos);
              pos = next + 1;
            }

          Cmtpath_items.push_back (path);
          if (ending) break;
        }
    }
}

/**
 *  Entry point for the application
 */
int main (int argc, char* argv[])
{
  argc--;
  argv++;

  bool in_options = true;
  bool is_global = false;

  expand_cmtpath ();

  while (argc > 0)
    {
      if (in_options)
        {
          std::string opt = argv[0];

          if (opt[0] == '-')
            {
              if (opt.substr (0, 9) == "-exclude=")
                {
                  std::string exclusion;

                  exclusion = opt.substr (9);

                  unsigned int size = exclusion.size ();
              
                  if (size >= 2)
                    {
                      if (((exclusion[0] == '"') && (exclusion[size - 1] == '"')) ||
                          ((exclusion[0] == '\'') && (exclusion[size - 1] == '\'')))
                        {
                          exclusion.erase (size - 1);
                          exclusion.erase (0, 1);
                        }
                    }

                  std::istrstream buff (exclusion.c_str ());
                  while (!buff.eof ())
                    {
                      std::string s;
                      buff >> s;
                      Exclusions.push_back (s);
                    }
                }
              else if (opt.substr (0, 7) == "-global")
                {
                  is_global = true;
                }
              else if (opt.substr (0, 8) == "-select=")
                {
                  std::string selection;

                  selection = opt.substr (8);

                  unsigned int size = selection.size ();
              
                  if (size >= 2)
                    {
                      if (((selection[0] == '"') && (selection[size - 1] == '"')) ||
                          ((selection[0] == '\'') && (selection[size - 1] == '\'')))
                        {
                          selection.erase (size - 1);
                          selection.erase (0, 1);
                        }
                    }

                  std::istrstream buff (selection.c_str ());
                  while (!buff.eof ())
                    {
                      std::string s;
                      buff >> s;
                      Selections.push_back (s);
                    }
                }
              else if (opt.substr (0, 7) == "-begin=")
                {
                  Begin = opt.substr (7);
                }
              else if (opt.substr (0, 9) == "-threads=")
                {
                  std::string o = opt.substr (7);
                  int value = -1;
                  sscanf (o.c_str (), "%d", &value);

                  if (value > 0)
                    {
                      MaxThreads = value;
                    }
                }
            }
          else
            {
              in_options = false;
            }
        }

      if (!in_options)
        {
          Command += argv[0];
          Command += " ";
        }

      argc--;
      argv++;
    }

  if (is_global)
    {
      Cmtpath_selections = Cmtpath_items;
    }
  else
    {
      if (Cmtpath_items.size () != 0)
        {
          std::string s = Cmtpath_items[0];
          Cmtpath_selections.push_back (s);
        }
    }

  //TCmThread::set_debug_level (1);

  Scheduler s;

  s.create_packages_from_command ("cmt show uses 2>&1 ");

  s.run ();

  return (0);
}