lib/pthreads/threadpool.nit

   1 # This file is part of NIT (http://www.nitlanguage.org).
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 # Introduces a minimal ThreadPool implementation using Tasks
  16 module threadpool
  17
  18 intrude import pthreads
  19 import concurrent_collections
  20
  21 # A simple ThreadPool implemented with an array
  22 class ThreadPool
  23         private var queue = new ConcurrentList[Task]
  24         private var mutex = new Mutex
  25         private var cond = new NativePthreadCond
  26         private var threads = new Array[PoolThread]
  27
  28         # Number of threads used, can only grow after the first call to `execute`
  29         var nb_threads = 5 is optional, writable
  30
  31         private fun create_threads do
  32                 while threads.length < nb_threads do
  33                         var t = new PoolThread(queue, mutex, cond)
  34                         t.start
  35                         threads.add t
  36                 end
  37         end
  38
  39         # Adds a Task into the queue
  40         fun execute(task: Task) do
  41                 create_threads
  42                 queue.push(task)
  43                 cond.signal
  44         end
  45
  46         # Join all threads, waiting for all tasks to be completed
  47         fun join_all do
  48                 # Wait
  49                 for t in threads do t.join
  50
  51                 # Reset
  52                 threads.clear
  53         end
  54 end
  55
  56 # A Thread running in a threadpool
  57 private class PoolThread
  58         super Thread
  59
  60         var queue: ConcurrentList[Task]
  61         var mutex: Mutex
  62         var cond : NativePthreadCond
  63
  64         redef fun main do
  65                 loop
  66                         var t = null
  67                         mutex.lock
  68                         if queue.is_empty then cond.wait(mutex.native.as(not null))
  69                         if not queue.is_empty then
  70                                 t = queue.shift
  71                         end
  72                         mutex.unlock
  73                         if t != null then
  74                                 t.main
  75                                 t.after_main
  76                         end
  77                 end
  78         end
  79 end
  80
  81 redef class Task
  82         # Additional work executed after `main` from a `ThreadPool`
  83         private fun after_main do end
  84 end
  85
  86 # A Task which is joinable, meaning it can return a value and if the value is not set yet, it blocks the execution
  87 class JoinTask
  88         super Task
  89
  90         # Is `self` done?
  91         var is_done = false
  92
  93         private var mutex = new Mutex
  94         private var cond: nullable NativePthreadCond = null
  95
  96         # Return immediatly if the task terminated, or block waiting for `self` to terminate
  97         fun join do
  98                 mutex.lock
  99                 if not is_done then
 100                         var cond = new NativePthreadCond
 101                         self.cond = cond
 102                         cond.wait(mutex.native.as(not null))
 103                 end
 104                 mutex.unlock
 105         end
 106
 107         redef fun after_main do
 108                 # TODO move this at the end of main so all `JoinTask` can be joined
 109                 # no matter what calls `main`.
 110
 111                 mutex.lock
 112                 is_done = true
 113                 var tcond = cond
 114                 if tcond != null then tcond.signal
 115                 mutex.unlock
 116         end
 117 end