Class: Pool

Inherits:
Object
  • Object
show all
Defined in:
lib/lumix/thread_pool.rb

Overview

‘Pool` is our thread pool class. It will allow us to do three operations:

  • ‘.new(size)` creates a thread pool of a given size

  • ‘#schedule(*args, &job)` schedules a new job to be executed

  • ‘#shutdown` shuts down all threads (after letting them finish working, of course)

Instance Method Summary collapse

Constructor Details

#initialize(size) ⇒ Pool

### initialization, or ‘Pool.new(size)` Creating a new `Pool` involves a certain amount of work. First, however, we need to define its’ `size`. It defines how many threads we will have working internally.

Which size is best for you is hard to answer. You do not want it to be too low, as then you won’t be able to do as many things concurrently. However, if you make it too high Ruby will spend too much time switching between threads, and that will also degrade performance!



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/lumix/thread_pool.rb', line 39

def initialize(size)
  # Before we do anything else, we need to store some information about
  # our pool. `@size` is useful later, when we want to shut our pool down,
  # and `@jobs` is the heart of our pool that allows us to schedule work.
  @size = size
  @jobs = Queue.new

  # #### Creating our pool of threads
  # Once preparation is done, it’s time to create our pool of threads.
  # Each thread store its’ index in a thread-local variable, in case we
  # need to know which thread a job is executing in later on.
  @pool = Array.new(@size) do |i|
    Thread.new do
      Thread.current[:id] = i

      # We start off by defining a `catch` around our worker loop. This
      # way we’ve provided a method for graceful shutdown of our threads.
      # Shutting down is merely a `#schedule { throw :exit }` away!
      catch(:exit) do
        # The worker thread life-cycle is very simple. We continuously wait
        # for tasks to be put into our job `Queue`. If the `Queue` is empty,
        # we will wait until it’s not.
        loop do
          # Once we have a piece of work to be done, we will pull out the
          # information we need and get to work.
          job, args = @jobs.pop
          job.call(*args)
        end
      end
    end
  end
end

Instance Method Details

#schedule(*args, &block) ⇒ Object

To schedule a piece of work to be done is to say to the ‘Pool` that you want something done.



76
77
78
79
80
 
# File 'lib/lumix/thread_pool.rb', line 76

def schedule(*args, &block)
  # Your given task will not be run immediately; rather, it will be put
  # into the work `Queue` and executed once a thread is ready to work.
  @jobs << [block, args]
end

#shutdown ⇒ Object

If you ever wish to close down your application, I took the liberty of making it easy for you to wait for any currently executing jobs to finish before you exit.



87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/lumix/thread_pool.rb', line 87

def shutdown
  # A graceful shutdown involves threads exiting cleanly themselves, and
  # since we’ve defined a `catch`-handler around the threads’ worker loop
  # it is simply a matter of throwing `:exit`. Thus, if we throw one `:exit`
  # for each thread in our pool, they will all exit eventually!
  @size.times do
    schedule { throw :exit }
  end

  # And now one final thing: wait for our `throw :exit` jobs to be run on
  # all our worker threads. This call will not return until all worker threads
  # have exited.
  @pool.map(&:join)
end