/**\file n_thread_pool.c
 *  Thread pool functions
 *\author Castagnier Mickael
 *\version 1.0
 *\date 30/04/2014
 */
#include "n_thread_pool.h"


/*!\fn void *thread_pool_processing_function( void *param )
 * \brief Internal thread pool processing function
 * \param param Teh eventual param of a process
 * \return NULL when exiting
 */
void *thread_pool_processing_function( void *param )
{
   THREAD_POOL_NODE *node = (THREAD_POOL_NODE *)param ;

   pthread_mutex_lock( &node -> lock );
   int state =  node -> state ;
   pthread_mutex_unlock( &node -> lock );
   
   while( state != EXITING_THREAD )
   {
      sem_wait( &node -> th_start );
      
      pthread_mutex_lock( &node -> lock );
      state = node -> state ;

      if( node -> state != IDLE_THREAD && node -> param != NULL )
      {
         node -> state = RUNNING_THREAD ;
         pthread_mutex_unlock( &node -> lock );
         node -> func( node -> param ) ;
         pthread_mutex_lock( &node -> lock );
      }	

      node -> func = NULL ;
      node -> param = NULL ;
      node -> state = IDLE_THREAD ;
      int type = node -> type ;
      node -> type = -1 ;

      pthread_mutex_unlock( &node -> lock );
      
      if( type == SYNCED_PROC )
         sem_post( &node -> th_end );
   }
   
   n_log( LOG_NOTICE , "Thread %d exited" , node -> thr );

   return NULL ;
}



/*!\fn THREAD_POOL *new_thread_pool( int nbmaxthr )
 * \brief Create a new pool of nbmaxthr threads
 * \param nbmaxthr number of waiting threads in the pool
 * \return NULL or a new trhead pool object
 */
THREAD_POOL *new_thread_pool( int nbmaxthr )
{
   THREAD_POOL *thread_pool = NULL ;

   Malloc( thread_pool , THREAD_POOL , 1 );
   if( !thread_pool )
      return NULL ;

   thread_pool -> max_threads = nbmaxthr ;

   thread_pool -> thread_list =  (THREAD_POOL_NODE **)malloc( nbmaxthr * sizeof(THREAD_POOL_NODE *) );
   if( !thread_pool -> thread_list )
   {
      Free( thread_pool );
      return NULL ;
   }

   for( int it = 0 ; it < nbmaxthr ; it ++ )
   {
      thread_pool -> thread_list[ it ] =  (THREAD_POOL_NODE *)malloc( sizeof(THREAD_POOL_NODE ) );
      thread_pool -> thread_list[ it ] -> type = -1 ;
      thread_pool -> thread_list[ it ] -> state = IDLE_THREAD ;
      if( sem_init( &thread_pool -> thread_list[ it ] -> th_start , 0 , 0 ) == -1 )
      {
         fprintf( stderr , "sem_init failed : %s on &thread_pool -> thread_list[ %d ] -> th_start\n" , strerror ( errno ) , it );
         return NULL;
      }
      if( sem_init( &thread_pool -> thread_list[ it ] -> th_end   , 0 , 0 ) == -1 )
      {
         fprintf( stderr , "sem_init failed : %s on &thread_pool -> thread_list[ %d] -> th_end\n" , strerror ( errno ) , it );
         return NULL;
      }
      thread_pool -> thread_list[ it ] -> func = NULL ;
      thread_pool -> thread_list[ it ] -> param = NULL ;

      pthread_mutex_init( &thread_pool -> thread_list[ it ] -> lock , NULL );

      if( pthread_create( &thread_pool -> thread_list[ it ] -> thr , NULL, thread_pool_processing_function , (void *)thread_pool -> thread_list[ it ] ) != 0 )
      {
         fprintf( stderr , "pthread_create failed : %s for it %d\n" ,  strerror ( errno ) , it );
         return NULL;
      }
   }
   return thread_pool ;
}





/*!\fn int add_threaded_process( THREAD_POOL *thread_pool , void *(*func_ptr)(void *param) , void *param , int mode )
 *\brief add a function and params to a thread pool
 *\param thread_pool The target thread pool  
 *\param func_ptr The function pointer to launch
 *\param param Eventual parameter struct to pass to the function
 *\param mode SYNCED_PROC: added to pool, but wait for a start call with other waiting procs. DIRECT_PROC: the function actually starts as quick as possible
 *\return TRUE or FALSE 
 */
int add_threaded_process( THREAD_POOL *thread_pool , void *(*func_ptr)(void *param) , void * param , int mode )
{

   if( !thread_pool )
      return FALSE ;

   if( !thread_pool -> thread_list )
      return FALSE ;

   int it = 0 ;
   while( it < thread_pool -> max_threads )
   {
      pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
      if( thread_pool -> thread_list[ it ] -> state == IDLE_THREAD )
      {
         break ;
      }
      pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
      it ++ ;
   }
   if( it == thread_pool -> max_threads )
   {
      n_log( LOG_ERR , "Thread pool is full: %d process already running" , it );
      return FALSE ;
   }

   switch( mode )
   {
      case DIRECT_PROC :
         thread_pool -> thread_list[ it ] -> func = func_ptr ;
         thread_pool -> thread_list[ it ] -> param = param ;
         thread_pool -> thread_list[ it ] -> state = BUSY_THREAD ;
         thread_pool -> thread_list[ it ] -> type = DIRECT_PROC ;
         pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
         sem_post( &thread_pool -> thread_list[ it ] -> th_start );
         break ;
      case SYNCED_PROC :
         thread_pool -> thread_list[ it ] -> func = func_ptr ;
         thread_pool -> thread_list[ it ] -> param = param ;
         thread_pool -> thread_list[ it ] -> state = BUSY_THREAD ;
         thread_pool -> thread_list[ it ] -> type = SYNCED_PROC ;
         pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
         break ;
      default:
         n_log( LOG_ERR , "Unknown mode %d for thread %d" , mode , it );
         pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
         return FALSE ;
   }
   return TRUE ;
}



/*!\fn int start_threaded_pool( THREAD_POOL *thread_pool )
 * \brief Launch the process waiting for exectution in the thread pool
 * \param thread_pool The thread pool to launche
 * \return TRUE or FALSE
 */
int start_threaded_pool( THREAD_POOL *thread_pool )
{
   if( !thread_pool )
      return FALSE ;

   if( !thread_pool -> thread_list )
      return FALSE ;

   for( int it = 0 ; it < thread_pool -> max_threads ; it ++ )
   {
      int to_run = 0 ;
      pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
      if( thread_pool -> thread_list[ it ] -> type == SYNCED_PROC && thread_pool -> thread_list[ it ] -> state == BUSY_THREAD )
      {
         to_run = 1 ;
      }
      pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
      if( to_run == 1 )
      {
         if( sem_post( &thread_pool -> thread_list[ it ] -> th_start ) != 0 )
            return FALSE ;
      }
   }
   return TRUE ;
}



/*!\fn int wait_for_threaded_pool( THREAD_POOL *thread_pool )
 * \brief Wait for all the launched process in the thread pool to terminate
 * \param thread_pool The thread pool to wait
 * \return TRUE or FALSE
 */
int wait_for_threaded_pool(  THREAD_POOL *thread_pool )
{
   if( !thread_pool )
      return FALSE ;

   if( !thread_pool -> thread_list )
      return FALSE ;

   for( int it = 0 ; it < thread_pool -> max_threads ; it ++ )
   {
      int to_check = 0 ;
      pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
      if( thread_pool -> thread_list[ it ] -> type == SYNCED_PROC && thread_pool -> thread_list[ it ] -> state != IDLE_THREAD )
      {
         to_check = 1 ;
      }
      pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
      if( to_check == 1 )
      {
         sem_wait( &thread_pool -> thread_list[ it ] -> th_end );
      }
      pthread_mutex_lock( &thread_pool -> thread_list[ it ] -> lock );
      thread_pool -> thread_list[ it ] -> state = IDLE_THREAD ;
      pthread_mutex_unlock( &thread_pool -> thread_list[ it ] -> lock );
   }
   return TRUE ;
}
