This is a thread pool and the single management "main" thread will spawn out free threads in the pool to perform specific work. If there are no free threads then it will block until one becomes available. It uses ForThreads, which is a wrapper around pthreads. The thread pool works by creating a number of threads and then passing the work to these threads, rather than creating a new thread for each piece of work. More...

Data Types
interface	thread_procedure
	The thread call procedure interface. More...

type	threaded_procedure_container_type
	Wraps the thread procedure with the call itself and the data to pass to it. More...

Functions/Subroutines
subroutine, public	threadpool_init (io_configuration)
	Initialises the thread pool and marks each thread as idle. More...

subroutine, public	threadpool_lock_netcdf_access ()
	Aquires the NetCDF thread lock, NetCDF is not thread safe so we need to manage thread calls to it. More...

subroutine, public	threadpool_unlock_netcdf_access ()
	Releases the NetCDF thread lock, NetCDF is not thread safe so we need to manage thread calls to it. More...

subroutine, public	threadpool_start_thread (proc, arguments, data_buffer)
	Starts an idle thread from the pool to execute a specific procedure with some data. If there is no thread available then this will block until one becomes idle. More...

subroutine	threadpool_thread_entry_procedure (thread_id)
	Entry point called by each thread creation in the pool, this calls out to the actual procedure to execute and doing it this way allows us to perform some actions before or after which can help with the management of the pool. More...

logical function, public	threadpool_is_idle ()
	Determines whether the thread pool is idle or not (i.e. all threads are idle and waiting for work) More...

subroutine, public	threadpool_deactivate ()
	This waits for all busy threads to complete and then shuts all the pthreads down. The deactivation and finalisation procedures are split out as we want to deactivate the pool (to ensure no threads are running actions), finalise these actions which might involve destroying mutexes, and then destroying the threading environment in finalisation. More...

subroutine, public	threadpool_finalise ()
	Finalises the thread pool. More...

integer function	find_idle_thread ()
	Finds an idle thread, if one is not available then will block until one becomes free. More...

integer function	get_index_of_idle_thread ()
	Specifically gets the index of the next idle thread or -1 if they are all busy. This starts from a next suggested idle thread and will wrap around, as often the next thread will be idle rather than searching from the beginning again. More...

subroutine, public	check_thread_status (ierr)
	Checks the error status of any thread operation and reports an error if it failed. More...

Variables
integer, parameter	default_thread_pool_size =10
	Number of threads in the pool. More...

logical, dimension(:), allocatable, volatile	thread_busy

logical, dimension(:), allocatable, volatile	thread_start

integer, dimension(:), allocatable, volatile	thread_ids

integer, dimension(:), allocatable, volatile	thread_pass_data

integer, dimension(:), allocatable, volatile	activate_thread_condition_variables

integer, dimension(:), allocatable, volatile	activate_thread_mutex

type(threaded_procedure_container_type), dimension(:), allocatable, volatile	thread_entry_containers

integer, volatile	netcdfmutex
	Mutex used for controling NetCDF access. More...

integer, volatile	next_suggested_idle_thread

logical, volatile	threadpool_active

integer, volatile	active_threads

integer, volatile	total_number_of_threads

integer, volatile	active_scalar_mutex

Detailed Description

This is a thread pool and the single management "main" thread will spawn out free threads in the pool to perform specific work. If there are no free threads then it will block until one becomes available. It uses ForThreads, which is a wrapper around pthreads. The thread pool works by creating a number of threads and then passing the work to these threads, rather than creating a new thread for each piece of work.

Function/Subroutine Documentation

◆ check_thread_status()

subroutine, public threadpool_mod::check_thread_status ( integer, intent(in) ierr )

Checks the error status of any thread operation and reports an error if it failed.

Parameters

ierr	The error/success flag returned from the ForThreads library, which itself is returned from pthreads

Definition at line 229 of file threadpool.F90.

     integer, intent(in) :: ierr
 
     if (ierr .ne. 0) then
       call log_log(log_error, "Pthreads error in IO server, error code="//conv_to_string(ierr))
     end if    

Here is the call graph for this function:

◆ find_idle_thread()

integer function threadpool_mod::find_idle_thread ( )

private

Finds an idle thread, if one is not available then will block until one becomes free.

Returns: The id of the idle thread which can be used

Definition at line 202 of file threadpool.F90.

     find_idle_thread=get_index_of_idle_thread()
     do while (find_idle_thread == -1)
       find_idle_thread=get_index_of_idle_thread()
     end do

Here is the call graph for this function:

Here is the caller graph for this function:

◆ get_index_of_idle_thread()

integer function threadpool_mod::get_index_of_idle_thread ( )

private

Specifically gets the index of the next idle thread or -1 if they are all busy. This starts from a next suggested idle thread and will wrap around, as often the next thread will be idle rather than searching from the beginning again.

Returns: The index of the next idle thread or -1 if there is none

Definition at line 212 of file threadpool.F90.

     integer :: i
 
     do i=next_suggested_idle_thread, total_number_of_threads
       if (.not. thread_busy(i)) then
         get_index_of_idle_thread=i
         next_suggested_idle_thread=i+1
         if (next_suggested_idle_thread .gt. total_number_of_threads) next_suggested_idle_thread=1
         return
       end if          
     end do
     next_suggested_idle_thread=1
     get_index_of_idle_thread=-1

Here is the caller graph for this function:

◆ threadpool_deactivate()

subroutine, public threadpool_mod::threadpool_deactivate ( )

This waits for all busy threads to complete and then shuts all the pthreads down. The deactivation and finalisation procedures are split out as we want to deactivate the pool (to ensure no threads are running actions), finalise these actions which might involve destroying mutexes, and then destroying the threading environment in finalisation.

Definition at line 174 of file threadpool.F90.

     integer :: i
     integer, pointer :: retval
 
     allocate(retval)
 
     threadpool_active=.false.
     do i=1, total_number_of_threads
       call check_thread_status(forthread_mutex_lock(activate_thread_mutex(i)))
       call check_thread_status(forthread_cond_signal(activate_thread_condition_variables(i)))
       call check_thread_status(forthread_mutex_unlock(activate_thread_mutex(i)))
       call check_thread_status(forthread_join(thread_ids(i),retval))
       call check_thread_status(forthread_mutex_destroy(activate_thread_mutex(i)))
       call check_thread_status(forthread_cond_destroy(activate_thread_condition_variables(i)))
     end do

Here is the call graph for this function:

Here is the caller graph for this function:

◆ threadpool_finalise()

subroutine, public threadpool_mod::threadpool_finalise ( )

Finalises the thread pool.

Definition at line 192 of file threadpool.F90.

       call check_thread_status(forthread_mutex_destroy(netcdfmutex))
       call check_thread_status(forthread_mutex_destroy(active_scalar_mutex))
       deallocate(thread_busy, thread_start, thread_ids, thread_pass_data, activate_thread_condition_variables, &
            activate_thread_mutex, thread_entry_containers)
     call check_thread_status(forthread_destroy())

Here is the call graph for this function:

Here is the caller graph for this function:

◆ threadpool_init()

subroutine, public threadpool_mod::threadpool_init ( type(io_configuration_type), intent(inout) io_configuration )

Initialises the thread pool and marks each thread as idle.

Definition at line 51 of file threadpool.F90.

     type(io_configuration_type), intent(inout) :: io_configuration
 
     integer :: n
 
     call check_thread_status(forthread_init())
     call check_thread_status(forthread_mutex_init(netcdfmutex, -1))
     if (io_configuration%number_of_threads .ge. 1) then
       total_number_of_threads=io_configuration%number_of_threads
     else
       if (io_configuration%my_io_rank==0) then
         call log_log(log_warn, "No setting for IO server thread pool size which must be 1 or more so using default size")
       end if
       total_number_of_threads=default_thread_pool_size
     end if
     allocate(thread_busy(total_number_of_threads), thread_start(total_number_of_threads), &
          thread_ids(total_number_of_threads), thread_pass_data(total_number_of_threads), &
          activate_thread_condition_variables(total_number_of_threads), activate_thread_mutex(total_number_of_threads), &
          thread_entry_containers(total_number_of_threads))
     threadpool_active=.true.
     active_threads=total_number_of_threads
     next_suggested_idle_thread=1
     call check_thread_status(forthread_mutex_init(active_scalar_mutex, -1))
     do n=1, total_number_of_threads
       call check_thread_status(forthread_cond_init(activate_thread_condition_variables(n), -1))
       call check_thread_status(forthread_mutex_init(activate_thread_mutex(n), -1))
       thread_busy(n)=.false.
       thread_start(n)=.false.        
       thread_pass_data(n)=n
       call check_thread_status(forthread_create(thread_ids(n), -1, threadpool_thread_entry_procedure, thread_pass_data(n)))
     end do

Here is the call graph for this function:

Here is the caller graph for this function:

◆ threadpool_is_idle()

logical function, public threadpool_mod::threadpool_is_idle ( )

Determines whether the thread pool is idle or not (i.e. all threads are idle and waiting for work)

Returns: Whether the thread pool is idle

Definition at line 164 of file threadpool.F90.

 
     call check_thread_status(forthread_mutex_lock(active_scalar_mutex))
     threadpool_is_idle = active_threads==total_number_of_threads
     call check_thread_status(forthread_mutex_unlock(active_scalar_mutex))

Here is the call graph for this function:

Here is the caller graph for this function:

◆ threadpool_lock_netcdf_access()

subroutine, public threadpool_mod::threadpool_lock_netcdf_access ( )

Aquires the NetCDF thread lock, NetCDF is not thread safe so we need to manage thread calls to it.

Definition at line 85 of file threadpool.F90.

85 #ifdef ENFORCE_THREAD_SAFETY

86 call check_thread_status(forthread_mutex_lock(netcdfmutex))

87 #endif

Here is the call graph for this function:

◆ threadpool_start_thread()

subroutine, public threadpool_mod::threadpool_start_thread	(	procedure(thread_procedure)	proc,
		integer, dimension(:), intent(in)	arguments,
		character, dimension(:), intent(in), optional, allocatable	data_buffer
	)

Starts an idle thread from the pool to execute a specific procedure with some data. If there is no thread available then this will block until one becomes idle.

Parameters

proc	The procedure for the thread to execute
data	Data to pass into the thread

Definition at line 102 of file threadpool.F90.

     procedure(thread_procedure) :: proc
     integer, dimension(:), intent(in) :: arguments
     character, dimension(:), allocatable, intent(in), optional :: data_buffer
 
     integer :: idle_thread_id
 
     if (.not. threadpool_active) call log_log(log_error, "Attemping to start IO thread on deactivated thread pool")
 
     idle_thread_id=find_idle_thread()
     if (idle_thread_id .ne. -1) then
       thread_busy(idle_thread_id)=.true.              
       thread_entry_containers(idle_thread_id)%proc=>proc
       allocate(thread_entry_containers(idle_thread_id)%arguments(size(arguments)))
       thread_entry_containers(idle_thread_id)%arguments=arguments
       if (present(data_buffer)) allocate(thread_entry_containers(idle_thread_id)%data_buffer(size(data_buffer)), &
            source=data_buffer)
       ! Send the signal to the thread to wake up and start
       call check_thread_status(forthread_mutex_lock(activate_thread_mutex(idle_thread_id)))
       thread_start(idle_thread_id)=.true.
       call check_thread_status(forthread_cond_signal(activate_thread_condition_variables(idle_thread_id)))
       call check_thread_status(forthread_mutex_unlock(activate_thread_mutex(idle_thread_id)))      
     end if

Here is the call graph for this function:

Here is the caller graph for this function:

◆ threadpool_thread_entry_procedure()

subroutine threadpool_mod::threadpool_thread_entry_procedure ( integer thread_id )

private

Entry point called by each thread creation in the pool, this calls out to the actual procedure to execute and doing it this way allows us to perform some actions before or after which can help with the management of the pool.

Parameters

thread_id The thread pool id (index) of this thread

Definition at line 130 of file threadpool.F90.

     integer :: thread_id
 
     do while (threadpool_active)
       call check_thread_status(forthread_mutex_lock(activate_thread_mutex(thread_id)))
       do while (.not. thread_start(thread_id) .and. threadpool_active)
         call check_thread_status(forthread_cond_wait(activate_thread_condition_variables(thread_id), &
              activate_thread_mutex(thread_id)))
       end do
       call check_thread_status(forthread_mutex_unlock(activate_thread_mutex(thread_id)))      
       if (.not. threadpool_active) return
       thread_busy(thread_id)=.true.
       thread_start(thread_id)=.false.
       
       call check_thread_status(forthread_mutex_lock(active_scalar_mutex))
       active_threads=active_threads-1
       call check_thread_status(forthread_mutex_unlock(active_scalar_mutex))
       if (allocated(thread_entry_containers(thread_id)%data_buffer)) then
         call thread_entry_containers(thread_id)%proc(thread_entry_containers(thread_id)%arguments, &
              data_buffer=thread_entry_containers(thread_id)%data_buffer)
         deallocate(thread_entry_containers(thread_id)%data_buffer)
       else
         call thread_entry_containers(thread_id)%proc(thread_entry_containers(thread_id)%arguments)
       end if
       deallocate(thread_entry_containers(thread_id)%arguments)
       call check_thread_status(forthread_mutex_lock(active_scalar_mutex))
       active_threads=active_threads+1
       call check_thread_status(forthread_mutex_unlock(active_scalar_mutex))
       thread_busy(thread_id)=.false.
     end do

Here is the call graph for this function:

Here is the caller graph for this function:

◆ threadpool_unlock_netcdf_access()

subroutine, public threadpool_mod::threadpool_unlock_netcdf_access ( )

Releases the NetCDF thread lock, NetCDF is not thread safe so we need to manage thread calls to it.

Definition at line 92 of file threadpool.F90.

92 #ifdef ENFORCE_THREAD_SAFETY

93 call check_thread_status(forthread_mutex_unlock(netcdfmutex))

94 #endif

Here is the call graph for this function:

Variable Documentation

◆ activate_thread_condition_variables

integer, dimension(:), allocatable, volatile threadpool_mod::activate_thread_condition_variables

private

Definition at line 37 of file threadpool.F90.

37 integer, volatile, dimension(:), allocatable :: activate_thread_condition_variables, activate_thread_mutex

◆ activate_thread_mutex

integer, dimension(:), allocatable, volatile threadpool_mod::activate_thread_mutex

private

Definition at line 37 of file threadpool.F90.

◆ active_scalar_mutex

integer, volatile threadpool_mod::active_scalar_mutex

private

Definition at line 43 of file threadpool.F90.

◆ active_threads

integer, volatile threadpool_mod::active_threads

private

Definition at line 43 of file threadpool.F90.

43 integer, volatile :: active_threads, total_number_of_threads, active_scalar_mutex

◆ default_thread_pool_size

integer, parameter threadpool_mod::default_thread_pool_size =10

private

Number of threads in the pool.

Definition at line 34 of file threadpool.F90.

34 integer, parameter :: default_thread_pool_size=10

◆ netcdfmutex

integer, volatile threadpool_mod::netcdfmutex

private

Mutex used for controling NetCDF access.

Definition at line 39 of file threadpool.F90.

39 integer, volatile :: netcdfmutex

◆ next_suggested_idle_thread

integer, volatile threadpool_mod::next_suggested_idle_thread

private

Definition at line 40 of file threadpool.F90.

40 integer, volatile :: next_suggested_idle_thread

◆ thread_busy

logical, dimension(:), allocatable, volatile threadpool_mod::thread_busy

private

Definition at line 35 of file threadpool.F90.

35 logical, volatile, dimension(:), allocatable :: thread_busy, thread_start

◆ thread_entry_containers

type(threaded_procedure_container_type), dimension(:), allocatable, volatile threadpool_mod::thread_entry_containers

private

Definition at line 38 of file threadpool.F90.

38 type(threaded_procedure_container_type), volatile, dimension(:), allocatable :: thread_entry_containers

◆ thread_ids

integer, dimension(:), allocatable, volatile threadpool_mod::thread_ids

private

Definition at line 36 of file threadpool.F90.

36 integer, volatile, dimension(:), allocatable :: thread_ids, thread_pass_data

◆ thread_pass_data

integer, dimension(:), allocatable, volatile threadpool_mod::thread_pass_data

private

Definition at line 36 of file threadpool.F90.

◆ thread_start

logical, dimension(:), allocatable, volatile threadpool_mod::thread_start

private

Definition at line 35 of file threadpool.F90.

◆ threadpool_active

logical, volatile threadpool_mod::threadpool_active

private

Definition at line 41 of file threadpool.F90.

41 logical, volatile :: threadpool_active

◆ total_number_of_threads

integer, volatile threadpool_mod::total_number_of_threads

private

Definition at line 43 of file threadpool.F90.

Data Types

Functions/Subroutines

Variables

Detailed Description

Function/Subroutine Documentation

◆ check_thread_status()

◆ find_idle_thread()

◆ get_index_of_idle_thread()

◆ threadpool_deactivate()

◆ threadpool_finalise()

◆ threadpool_init()

◆ threadpool_is_idle()

◆ threadpool_lock_netcdf_access()

◆ threadpool_start_thread()

◆ threadpool_thread_entry_procedure()

◆ threadpool_unlock_netcdf_access()

Variable Documentation

◆ activate_thread_condition_variables

◆ activate_thread_mutex

◆ active_scalar_mutex

◆ active_threads

◆ default_thread_pool_size

◆ netcdfmutex

◆ next_suggested_idle_thread

◆ thread_busy

◆ thread_entry_containers

◆ thread_ids

◆ thread_pass_data

◆ thread_start

◆ threadpool_active

◆ total_number_of_threads