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].wrap(new List[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: Task) 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: nullable Task = 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.pop
  61                         end
  62                         mutex.unlock
  63                         if t != null then t.main
  64                 end
  65         end
  66 end