Parent

BackgrounDRb::ThreadPool

Attributes

logger[RW]

(Not documented)

master[RW]

(Not documented)

result_queue[RW]

(Not documented)

size[RW]

(Not documented)

threads[RW]

(Not documented)

work_queue[RW]

(Not documented)

Public Class Methods

new(master,size,logger) click to toggle source

(Not documented)

    # File server/lib/bdrb_thread_pool.rb, line 16
16:     def initialize(master,size,logger)
17:       @master = master
18:       @logger = logger
19:       @size = size
20:       @threads = []
21:       @work_queue = Queue.new
22:       @size.times { add_thread }
23:     end

Public Instance Methods

add_thread() click to toggle source

Start worker threads

    # File server/lib/bdrb_thread_pool.rb, line 52
52:     def add_thread
53:       @threads << Thread.new do
54:         Thread.current[:job_key] = nil
55:         Thread.current[:persistent_job_id] = nil
56:         while true
57:           task = @work_queue.pop
58:           Thread.current[:job_key] = task.job_key
59:           Thread.current[:persistent_job_id] = task.persistent_job_id
60:           block_result = run_task(task)
61:         end
62:       end
63:     end
defer(method_name,args = nil) click to toggle source

can be used to make a call in threaded manner passed block runs in a thread from thread pool for example in a worker method you can do:

  def user_tags url
    thread_pool.defer(:fetch_url,url)
  end
  def fetch_url(url)
    begin
      data = Net::HTTP.get(url,'/')
      File.open("#{RAILS_ROOT}/log/pages.txt","w") do |fl|
        fl.puts(data)
      end
    rescue
      logger.info "Error downloading page"
    end
  end

you can invoke above method from rails as:

  MiddleMan.worker(:rss_worker).async_user_tags(:arg => "en.wikipedia.org")

assuming method is defined in rss_worker

    # File server/lib/bdrb_thread_pool.rb, line 45
45:     def defer(method_name,args = nil)
46:       job_key = Thread.current[:job_key]
47:       persistent_job_id = Thread.current[:persistent_job_id]
48:       @work_queue << WorkData.new(args,job_key,method_name,persistent_job_id)
49:     end
run_task(task) click to toggle source

run tasks popped out of queue

    # File server/lib/bdrb_thread_pool.rb, line 66
66:     def run_task task
67:       block_arity = master.method(task.job_method).arity
68:       begin
69:         ActiveRecord::Base.verify_active_connections!
70:         t_data = task.args
71:         result = nil
72:         if block_arity != 0
73:           result = master.send(task.job_method,task.args)
74:         else
75:           result = master.send(task.job_method)
76:         end
77:         return result
78:       rescue
79:         logger.info($!.to_s)
80:         logger.info($!.backtrace.join("\n"))
81:         return nil
82:       end
83:     end

secsequence

--- SEC00036

seccomment

--- ""

attributes

--- 
- name: logger
  rw: RW
  a_desc: ""
- name: master
  rw: RW
  a_desc: ""
- name: result_queue
  rw: RW
  a_desc: ""
- name: size
  rw: RW
  a_desc: ""
- name: threads
  rw: RW
  a_desc: ""
- name: work_queue
  rw: RW
  a_desc: ""

method_list

--- 
- methods: 
  - visibility: public
    aref: M000129
    name: new
    sourcecode: "    <span class=\"ruby-comment cmt\"># File server/lib/bdrb_thread_pool.rb, line 16</span>\n\
      16:     <span class=\"ruby-keyword kw\">def</span> <span class=\"ruby-identifier\">initialize</span>(<span class=\"ruby-identifier\">master</span>,<span class=\"ruby-identifier\">size</span>,<span class=\"ruby-identifier\">logger</span>)\n\
      17:       <span class=\"ruby-ivar\">@master</span> = <span class=\"ruby-identifier\">master</span>\n\
      18:       <span class=\"ruby-ivar\">@logger</span> = <span class=\"ruby-identifier\">logger</span>\n\
      19:       <span class=\"ruby-ivar\">@size</span> = <span class=\"ruby-identifier\">size</span>\n\
      20:       <span class=\"ruby-ivar\">@threads</span> = []\n\
      21:       <span class=\"ruby-ivar\">@work_queue</span> = <span class=\"ruby-constant\">Queue</span>.<span class=\"ruby-identifier\">new</span>\n\
      22:       <span class=\"ruby-ivar\">@size</span>.<span class=\"ruby-identifier\">times</span> { <span class=\"ruby-identifier\">add_thread</span> }\n\
      23:     <span class=\"ruby-keyword kw\">end</span>"
    params: (master,size,logger)
  category: Class
  type: Public
- methods: 
  - visibility: public
    aref: M000131
    name: add_thread
    sourcecode: "    <span class=\"ruby-comment cmt\"># File server/lib/bdrb_thread_pool.rb, line 52</span>\n\
      52:     <span class=\"ruby-keyword kw\">def</span> <span class=\"ruby-identifier\">add_thread</span>\n\
      53:       <span class=\"ruby-ivar\">@threads</span> <span class=\"ruby-operator\">&lt;&lt;</span> <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">new</span> <span class=\"ruby-keyword kw\">do</span>\n\
      54:         <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">current</span>[<span class=\"ruby-identifier\">:job_key</span>] = <span class=\"ruby-keyword kw\">nil</span>\n\
      55:         <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">current</span>[<span class=\"ruby-identifier\">:persistent_job_id</span>] = <span class=\"ruby-keyword kw\">nil</span>\n\
      56:         <span class=\"ruby-keyword kw\">while</span> <span class=\"ruby-keyword kw\">true</span>\n\
      57:           <span class=\"ruby-identifier\">task</span> = <span class=\"ruby-ivar\">@work_queue</span>.<span class=\"ruby-identifier\">pop</span>\n\
      58:           <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">current</span>[<span class=\"ruby-identifier\">:job_key</span>] = <span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">job_key</span>\n\
      59:           <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">current</span>[<span class=\"ruby-identifier\">:persistent_job_id</span>] = <span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">persistent_job_id</span>\n\
      60:           <span class=\"ruby-identifier\">block_result</span> = <span class=\"ruby-identifier\">run_task</span>(<span class=\"ruby-identifier\">task</span>)\n\
      61:         <span class=\"ruby-keyword kw\">end</span>\n\
      62:       <span class=\"ruby-keyword kw\">end</span>\n\
      63:     <span class=\"ruby-keyword kw\">end</span>"
    m_desc: |-
      <p>
      Start worker threads
      </p>
    params: ()
  - visibility: public
    aref: M000130
    name: defer
    sourcecode: "    <span class=\"ruby-comment cmt\"># File server/lib/bdrb_thread_pool.rb, line 45</span>\n\
      45:     <span class=\"ruby-keyword kw\">def</span> <span class=\"ruby-identifier\">defer</span>(<span class=\"ruby-identifier\">method_name</span>,<span class=\"ruby-identifier\">args</span> = <span class=\"ruby-keyword kw\">nil</span>)\n\
      46:       <span class=\"ruby-identifier\">job_key</span> = <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">current</span>[<span class=\"ruby-identifier\">:job_key</span>]\n\
      47:       <span class=\"ruby-identifier\">persistent_job_id</span> = <span class=\"ruby-constant\">Thread</span>.<span class=\"ruby-identifier\">current</span>[<span class=\"ruby-identifier\">:persistent_job_id</span>]\n\
      48:       <span class=\"ruby-ivar\">@work_queue</span> <span class=\"ruby-operator\">&lt;&lt;</span> <span class=\"ruby-constant\">WorkData</span>.<span class=\"ruby-identifier\">new</span>(<span class=\"ruby-identifier\">args</span>,<span class=\"ruby-identifier\">job_key</span>,<span class=\"ruby-identifier\">method_name</span>,<span class=\"ruby-identifier\">persistent_job_id</span>)\n\
      49:     <span class=\"ruby-keyword kw\">end</span>"
    m_desc: |-
      <p>
      can be used to make a call in threaded manner passed block runs in a thread
      from thread pool for example in a worker method you can do:
      </p>
      <pre>
        def user_tags url
          thread_pool.defer(:fetch_url,url)
        end
        def fetch_url(url)
          begin
            data = Net::HTTP.get(url,'/')
            File.open(&quot;#{RAILS_ROOT}/log/pages.txt&quot;,&quot;w&quot;) do |fl|
              fl.puts(data)
            end
          rescue
            logger.info &quot;Error downloading page&quot;
          end
        end
      </pre>
      <p>
      you can invoke above method from rails as:
      </p>
      <pre>
        MiddleMan.worker(:rss_worker).async_user_tags(:arg =&gt; &quot;en.wikipedia.org&quot;)
      </pre>
      <p>
      assuming method is defined in rss_worker
      </p>
    params: (method_name,args = nil)
  - visibility: public
    aref: M000132
    name: run_task
    sourcecode: "    <span class=\"ruby-comment cmt\"># File server/lib/bdrb_thread_pool.rb, line 66</span>\n\
      66:     <span class=\"ruby-keyword kw\">def</span> <span class=\"ruby-identifier\">run_task</span> <span class=\"ruby-identifier\">task</span>\n\
      67:       <span class=\"ruby-identifier\">block_arity</span> = <span class=\"ruby-identifier\">master</span>.<span class=\"ruby-identifier\">method</span>(<span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">job_method</span>).<span class=\"ruby-identifier\">arity</span>\n\
      68:       <span class=\"ruby-keyword kw\">begin</span>\n\
      69:         <span class=\"ruby-constant\">ActiveRecord</span><span class=\"ruby-operator\">::</span><span class=\"ruby-constant\">Base</span>.<span class=\"ruby-identifier\">verify_active_connections!</span>\n\
      70:         <span class=\"ruby-identifier\">t_data</span> = <span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">args</span>\n\
      71:         <span class=\"ruby-identifier\">result</span> = <span class=\"ruby-keyword kw\">nil</span>\n\
      72:         <span class=\"ruby-keyword kw\">if</span> <span class=\"ruby-identifier\">block_arity</span> <span class=\"ruby-operator\">!=</span> <span class=\"ruby-value\">0</span>\n\
      73:           <span class=\"ruby-identifier\">result</span> = <span class=\"ruby-identifier\">master</span>.<span class=\"ruby-identifier\">send</span>(<span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">job_method</span>,<span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">args</span>)\n\
      74:         <span class=\"ruby-keyword kw\">else</span>\n\
      75:           <span class=\"ruby-identifier\">result</span> = <span class=\"ruby-identifier\">master</span>.<span class=\"ruby-identifier\">send</span>(<span class=\"ruby-identifier\">task</span>.<span class=\"ruby-identifier\">job_method</span>)\n\
      76:         <span class=\"ruby-keyword kw\">end</span>\n\
      77:         <span class=\"ruby-keyword kw\">return</span> <span class=\"ruby-identifier\">result</span>\n\
      78:       <span class=\"ruby-keyword kw\">rescue</span>\n\
      79:         <span class=\"ruby-identifier\">logger</span>.<span class=\"ruby-identifier\">info</span>(<span class=\"ruby-identifier\">$!</span>.<span class=\"ruby-identifier\">to_s</span>)\n\
      80:         <span class=\"ruby-identifier\">logger</span>.<span class=\"ruby-identifier\">info</span>(<span class=\"ruby-identifier\">$!</span>.<span class=\"ruby-identifier\">backtrace</span>.<span class=\"ruby-identifier\">join</span>(<span class=\"ruby-value str\">&quot;\\n&quot;</span>))\n\
      81:         <span class=\"ruby-keyword kw\">return</span> <span class=\"ruby-keyword kw\">nil</span>\n\
      82:       <span class=\"ruby-keyword kw\">end</span>\n\
      83:     <span class=\"ruby-keyword kw\">end</span>"
    m_desc: |-
      <p>
      run tasks popped out of queue
      </p>
    params: (task)
  category: Instance
  type: Public

sectitle

--- 

[Validate]

Generated with the Darkfish Rdoc Generator.