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
  27         # Number of threads used
  28         var nb_threads: Int is noinit
  29
  30         init do
  31                 for i in [0..nb_threads[ do
  32                         var t = new PoolThread(queue, mutex, cond)
  33                         t.start
  34                 end
  35         end
  36
  37         private fun set_nb_threads(nb: nullable Int) is autoinit do nb_threads = nb or else 5
  38
  39         # Adds a Task into the queue
  40         fun execute(task: JoinTask) do
  41                 queue.push(task)
  42                 cond.signal
  43         end
  44 end
  45
  46 # A Thread running in a threadpool
  47 private class PoolThread
  48         super Thread
  49
  50         var queue: ConcurrentList[Task]
  51         var mutex: Mutex
  52         var cond : NativePthreadCond
  53
  54         redef fun main do
  55                 loop
  56                         var t = null
  57                         mutex.lock
  58                         if queue.is_empty then cond.wait(mutex.native.as(not null))
  59                         if not queue.is_empty then
  60                                 t = queue.shift
  61                         end
  62                         mutex.unlock
  63                         if t != null then
  64                                 t.main
  65                                 t.after_main
  66                         end
  67                 end
  68         end
  69 end
  70
  71 redef class Task
  72         # Additional work executed after `main` from a `ThreadPool`
  73         private fun after_main do end
  74 end
  75
  76 # A Task which is joinable, meaning it can return a value and if the value is not set yet, it blocks the execution
  77 class JoinTask
  78         super Task
  79
  80         # Is `self` done?
  81         var is_done = false
  82
  83         private var mutex = new Mutex
  84         private var cond: nullable NativePthreadCond = null
  85
  86         # Return immediatly if the task terminated, or block waiting for `self` to terminate
  87         fun join do
  88                 mutex.lock
  89                 if not is_done then
  90                         var cond = new NativePthreadCond
  91                         self.cond = cond
  92                         cond.wait(mutex.native.as(not null))
  93                 end
  94                 mutex.unlock
  95         end
  96
  97         redef fun after_main do
  98                 # TODO move this at the end of main so all `JoinTask` can be joined
  99                 # no matter what calls `main`.
 100
 101                 mutex.lock
 102                 is_done = true
 103                 var tcond = cond
 104                 if tcond != null then tcond.signal
 105                 mutex.unlock
 106         end
 107 end