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[JoinTask].wrap(new List[JoinTask])
  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[JoinTask]
  51         var mutex: Mutex
  52         var cond : NativePthreadCond
  53
  54         redef fun main do
  55                 loop
  56                         var t: nullable JoinTask = 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.mutex.lock
  66                                 t.is_done = true
  67                                 var tcond = t.cond
  68                                 if tcond != null then tcond.signal
  69                                 t.mutex.unlock
  70                         end
  71                 end
  72         end
  73 end
  74
  75 # A Task which is joinable, meaning it can return a value and if the value is not set yet, it blocks the execution
  76 class JoinTask
  77         super Task
  78
  79         # Is `self` done ?
  80         var is_done = false
  81
  82         private var mutex = new Mutex
  83         private var cond: nullable NativePthreadCond = null
  84
  85         # Return immediatly if the task terminated, or block waiting for `self` to terminate
  86         fun join do
  87                 mutex.lock
  88                 if not is_done then
  89                         var cond = new NativePthreadCond
  90                         self.cond = cond
  91                         cond.wait(mutex.native.as(not null))
  92                 end
  93                 mutex.unlock
  94         end
  95 end