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

#include <stdlib.h>
#include <stdio.h>

#include <iostream.h>
#include <pthread.h>
#include <signal.h>
#include <sys/time.h>

#include "TCmThread.hxx"

#include <vector>

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


//-------------------------------------------------------
//
// Threads
//
//-------------------------------------------------------

typedef void* (*pthread_function_t) (void*);

static int DebugLevel = 0;

//-------------------------------------------------------
void TCmThread::set_debug_level (int level)
//-------------------------------------------------------
{
  DebugLevel = level;
}

//-------------------------------------------------------
int TCmThread::debug_level ()
//-------------------------------------------------------
{
  return (DebugLevel);
}

//-------------------------------------------------------
TCmThread::TCmThread ()
//-------------------------------------------------------
{
  _running = false;
}

//-------------------------------------------------------
TCmThread::~TCmThread ()
//-------------------------------------------------------
{
  void* value;
  int status;

  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "~TCmThread> " << _thread_id << " running " << _running << std::endl;
      TCmMutex::iounlock ();
    }

  if (_running)
    {
      status = pthread_kill ((pthread_t) _thread_id, SIGINT);

      if (debug_level () > 0)
        {
          TCmMutex::iolock ();
          std::cout << "~TCmThread> " << _thread_id << " status after kill=" << status << std::endl;
          TCmMutex::iounlock ();
        }

      status = pthread_cancel ((pthread_t) _thread_id);

      if (debug_level () > 0)
        {
          TCmMutex::iolock ();
          std::cout << "~TCmThread> " << _thread_id << " status after cancel=" << status << std::endl;
          TCmMutex::iounlock ();
        }

      status = pthread_join ((pthread_t) _thread_id, &value);

      if (debug_level () > 0)
        {
          TCmMutex::iolock ();
          std::cout << "~TCmThread> " << _thread_id << " status after join=" << status << std::endl;
          TCmMutex::iounlock ();
        }

    }

  status = pthread_detach ((pthread_t) _thread_id);

  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "~TCmThread> " << _thread_id << " status after detach=" << status << std::endl;
      TCmMutex::iounlock ();
    }
 
  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "~TCmThread> " << _thread_id << " finished " << std::endl;
      TCmMutex::iounlock ();
    }
}

//-------------------------------------------------------
bool TCmThread::start ()
//-------------------------------------------------------
{
  /*
    On lance le thread qui fait fonctionner la fonction Run
    */

  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "TCmThread::start> " << this << std::endl;
      TCmMutex::iounlock ();
    }

#ifdef PTHREAD_USE_VALUE_ATTR
  pthread_attr_t attr;
  pthread_attr_create (&attr);

  int status = pthread_create (&((pthread_t) _thread_id),
                               attr,
                               (pthread_function_t) run_it,
                               this);
#else
  int status = pthread_create (&((pthread_t) _thread_id),
                               NULL,
                               (pthread_function_t) run_it,
                               this);
#endif

  if (status != 0)
    {
      TCmMutex::iolock ();
      std::cout << "TCmThread::start> status create = " << status << std::endl;
      TCmMutex::iounlock ();
    }

  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "TCmThread::start> " << _thread_id << " running " << _running << std::endl;
      TCmMutex::iounlock ();
    }

  if (status == 0) return (true);
  else return (false);
}

//-------------------------------------------------------
bool TCmThread::stop ()
//-------------------------------------------------------
{
  void* value;
  int status;

  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "TCmThread::stop> " << _thread_id << " running " << _running << std::endl;
      TCmMutex::iounlock ();
    }

  status = pthread_cancel ((pthread_t) _thread_id);
  if (status == 0)
    {
      status = pthread_join ((pthread_t) _thread_id, &value);

      if (status != 0)
        {
          TCmMutex::iolock ();
          perror ("join error");
          std::cout << "TCmThread::stop join error> " << _thread_id << " status=" << status << std::endl;
          TCmMutex::iounlock ();
        }

      if (debug_level () > 0)
        {
          TCmMutex::iolock ();
          std::cout << "TCmThread::stopped> " << _thread_id << " running " << _running << std::endl;
          TCmMutex::iounlock ();
        }

      status = pthread_detach ((pthread_t) _thread_id);

      if (status != 0)
        {
          TCmMutex::iolock ();
          perror ("detach error");
          std::cout << "TCmThread::stop detach error> " << _thread_id << " status=" << status << std::endl;
          TCmMutex::iounlock ();
        }

      if (debug_level () > 0)
        {
          TCmMutex::iolock ();
          std::cout << "TCmThread::stopped> " << _thread_id << " running " << _running << std::endl;
          TCmMutex::iounlock ();
        }

      return ((status == 0));
    }
  else
    {
      //if (debug_level () > 0)
        {
          TCmMutex::iolock ();
          perror ("cancel error");
          std::cout << "TCmThread::stop cancel error> " << _thread_id << " status=" << status << std::endl;
          TCmMutex::iounlock ();
        }
      return (false);
    }
}

//-------------------------------------------------------
void TCmThread::run ()
//-------------------------------------------------------
{
  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "TCmThread::run>" << std::endl;
      TCmMutex::iounlock ();
    }
}

//-------------------------------------------------------
void TCmThread::cleanup ()
//-------------------------------------------------------
{
  TCmMutex::iolock ();
  std::cout << "TCmThread::cleanup>" << std::endl;
  TCmMutex::iounlock ();
}

//-------------------------------------------------------
void TCmThread::do_cleanup (void* thread)
//-------------------------------------------------------
{
  if (thread != 0)
    {
      ((TCmThread*) thread)->cleanup ();
    }
}

//-------------------------------------------------------
void TCmThread::run_it (TCmThread* thread)
//-------------------------------------------------------
{
  //int status = 0;
  //pthread_cleanup_push (do_cleanup, thread);

  thread->_thread_id = pthread_self ();
  thread->_running = true;
  thread->_should_die = false;

  thread->run ();
  thread->_running = false;
  
  if (debug_level () > 0)
    {
      TCmMutex::iolock ();
      std::cout << "TCmThread::finished> " << thread->_thread_id << std::endl;
      TCmMutex::iounlock ();
    }
  
  pthread_detach (thread->_thread_id);
  
  //if (thread->_should_die) delete (thread);

  //pthread_exit (&status);
  //pthread_cleanup_pop (0);

  do_cleanup (thread);
}

//-------------------------------------------------------
//
// Mutex
//
//-------------------------------------------------------

//-------------------------------------------------------
TCmMutex::TCmMutex ()
//-------------------------------------------------------
{
  initialize ();
}

//-------------------------------------------------------
TCmMutex::TCmMutex (const std::string& info) : m_info (info)
//-------------------------------------------------------
{
  initialize ();
}

//-------------------------------------------------------
TCmMutex::~TCmMutex ()
//-------------------------------------------------------
{
  pthread_mutex_t* mutex = (pthread_mutex_t*) _mutex;

  int status = pthread_mutex_destroy (mutex);
  if (status != 0)
    {
      std::cout << "Error in mutex_destroy " << _mutex << " status=" << status << std::endl;
    }

  delete mutex;
}

//-------------------------------------------------------
void TCmMutex::set_info (const std::string& info)
//-------------------------------------------------------
{
  m_info = info;
}

//-------------------------------------------------------
std::string TCmMutex::show () const
//-------------------------------------------------------
{
  std::string s;
  char buff[20];

  sprintf (buff, "%d", m_count);

  s = m_info;
  s += " ";
  s += buff;

  return (s);
}

//-------------------------------------------------------
bool TCmMutex::trylock (const std::string& /*info*/)
//-------------------------------------------------------
{
  int status;

  status = pthread_mutex_trylock ((pthread_mutex_t*) _mutex);

  if (status == 0) 
    {
      m_count++;
      return (true);
    }
  else return (false);
}

//-------------------------------------------------------
void TCmMutex::lock (const std::string& info)
//-------------------------------------------------------
{
  int status;

  /*
  int retries = 0;

  for (retries = 0; retries < 1000000; retries++)
    {
      status = pthread_mutex_trylock ((pthread_mutex_t*) _mutex);

      if (status == 0)
        {
          m_count++;
          break;
        }
      else
        {
          if ((retries % 100000) == 0)
            {
              std::cout << "Error in lock " << _mutex << " " << m_info << " " << info << " " << m_context << " status=" << status << " " << retries << std::endl;
            }
        }
    }

  if (status != 0)
    {
      std::cout << "Final error in lock " << _mutex << " " << m_info << " " << info << " " << m_context << " status=" << status << std::endl;
      exit (1);
    }
  else
    {
      m_context = info;
    }
  */

  status = pthread_mutex_lock ((pthread_mutex_t*) _mutex);

  if (status == 0)
    {
      m_count++;
      //m_context = info;
    }
  else
    {
      std::cout << "Final error in lock " << _mutex << " " << m_info << " " << info << " " << m_context << " status=" << status << std::endl;
      exit (1);
    }
}

//-------------------------------------------------------
void TCmMutex::unlock ()
//-------------------------------------------------------
{
  //m_context = "";
  m_count--;
  int status = pthread_mutex_unlock ((pthread_mutex_t*) _mutex);
  if (status != 0)
    {
      std::cout << "Error in unlock " << _mutex << " " << m_info << " status=" << status << std::endl;
    }
}

//-------------------------------------------------------
void TCmMutex::iolock ()
//-------------------------------------------------------
{
  static TCmMutex& iolock = get_iolock ();

  iolock.lock ("IO");
}

//-------------------------------------------------------
void TCmMutex::iounlock ()
//-------------------------------------------------------
{
  static TCmMutex& iolock = get_iolock ();

  iolock.unlock ();
}

//-------------------------------------------------------
TCmMutex& TCmMutex::get_iolock ()
//-------------------------------------------------------
{
  static TCmMutex iolock ("iolock");

  return (iolock);
}

//-------------------------------------------------------
void TCmMutex::initialize ()
//-------------------------------------------------------
{
  pthread_mutex_t* mutex = new pthread_mutex_t;
  //pthread_mutexattr_t* attr = new pthread_mutexattr_t;

  _mutex = (void*) mutex;
  m_count = 0;

  int status;

  /*
  status = pthread_mutexattr_init (attr);
  if (status != 0)
    {
      std::cout << "Error in mutexattr_init " << attr << " status=" << status << std::endl;
    }

  status = pthread_mutexattr_settype (attr, PTHREAD_MUTEX_RECURSIVE_NP);
  if (status != 0)
    {
      std::cout << "Error in mutexattr_settype " << attr << " status=" << status << std::endl;
    }

  status = pthread_mutex_init (mutex, attr);
  */

  status = pthread_mutex_init (mutex, 0);
  if (status != 0)
    {
      std::cout << "Error in mutex_init " << _mutex << " " << m_info << " status=" << status << std::endl;
    }
}