diff options
| -rw-r--r-- | lib/rbot/ircbot.rb | 7 | ||||
| -rw-r--r-- | lib/rbot/ircsocket.rb | 18 | ||||
| -rw-r--r-- | lib/rbot/timer.rb | 335 |
3 files changed, 164 insertions, 196 deletions
diff --git a/lib/rbot/ircbot.rb b/lib/rbot/ircbot.rb index 17d7ffe7..889daed0 100644 --- a/lib/rbot/ircbot.rb +++ b/lib/rbot/ircbot.rb @@ -436,7 +436,7 @@ class Bot @registry = BotRegistry.new self - @timer = Timer::Timer.new(1.0) # only need per-second granularity + @timer = Timer.new @save_mutex = Mutex.new if @config['core.save_every'] > 0 @save_timer = @timer.add(@config['core.save_every']) { save } @@ -770,7 +770,6 @@ class Bot begin quit if $interrupted > 0 connect - @timer.start quit_msg = nil while @socket.connected? @@ -1052,8 +1051,8 @@ class Bot @save_mutex.synchronize do @plugins.cleanup end - debug "\tstopping timers ..." - @timer.stop + # debug "\tstopping timers ..." + # @timer.stop # debug "Closing registries" # @registry.close debug "\t\tcleaning up the db environment ..." diff --git a/lib/rbot/ircsocket.rb b/lib/rbot/ircsocket.rb index c1bc3611..4c6c74bb 100644 --- a/lib/rbot/ircsocket.rb +++ b/lib/rbot/ircsocket.rb @@ -285,10 +285,8 @@ module Irc # host:: optional local host to bind to (ruby 1.7+ required) # create a new IrcSocket def initialize(server_list, host, sendq_delay=2, sendq_burst=4, opts={}) - @timer = Timer::Timer.new - @timer.add(0.2) do - spool - end + @timer = Timer.new + @act_id = @timer.add(0.2, :blocked => true) { spool } @server_list = server_list.dup @server_uri = nil @conn_count = 0 @@ -370,9 +368,9 @@ module Irc @sendq_delay = newfreq if newfreq == 0 clearq - @timer.stop + @timer.block(@act_id) else - @timer.start + @timer.unblock(@act_id) end end end @@ -424,7 +422,7 @@ module Irc if @sendq_delay > 0 @qmutex.synchronize do @sendq.push msg, chan, ring - @timer.start + @timer.unblock(@act_id) end else # just send it if queueing is disabled @@ -438,7 +436,7 @@ module Irc begin debug "in spooler" if @sendq.empty? - @timer.stop + @timer.block(@act_id) return end now = Time.new @@ -448,7 +446,7 @@ module Irc elsif (@burst > @sendq_burst) # nope. can't send anything, come back to us next tick... debug "can't send yet" - @timer.start + @timer.unblock(@act_id) return end @flood_send = now if @flood_send < now @@ -458,7 +456,7 @@ module Irc puts_critical(@sendq.shift, true) end if @sendq.empty? - @timer.stop + @timer.block(@act_id) end rescue Exception => e error "Spooling failed: #{e.pretty_inspect}" diff --git a/lib/rbot/timer.rb b/lib/rbot/timer.rb index bebe41f2..8939b672 100644 --- a/lib/rbot/timer.rb +++ b/lib/rbot/timer.rb @@ -1,226 +1,197 @@ -module Timer - - # timer event, something to do and when/how often to do it +# changes: +# 1. Timer::Timer ---> Timer +# 2. timer id is now the object_id of the action +# 3. Timer resolution removed, we're always arbitrary precision now +# 4. I don't see any obvious races [not that i did see any in old impl, though] +# 5. We're tickless now, so no need to jerk start/stop +# 6. We should be pretty fast now, wrt old impl +# 7. reschedule/remove/block now accept nil as an action id (meaning "current") +# 8. repeatability is ignored for 0-period repeatable timers +# 9. configure() method superceeds reschedule() [the latter stays as compat] + +require 'thread' +require 'monitor' + +class Timer class Action + attr_accessor :next + + def initialize(options = {}, &block) + opts = { + :period => 1, + :blocked => false, + :args => [], + :repeat => false + }.merge(options) + + @block = nil + debug("adding timer #{self} :period => #{opts[:period]}, :repeat => #{opts[:repeat].inspect}") + self.configure(opts, &block) + debug("added #{self}") + end + + def configure(opts = {}, &block) + @period = opts[:period] if opts.include? :period + @blocked = opts[:blocked] if opts.include? :blocked + @repeat = opts[:repeat] if opts.include? :repeat + + if block_given? + @block = block + elsif opts[:code] + @block = opts[:code] + end - # when this action is due next (updated by tick()) - attr_reader :in + raise 'huh?? blockless action?' unless @block + if opts.include? :args + @args = Array === opts[:args] ? opts[:args] : [opts[:args]] + end - # is this action blocked? if so it won't be run - attr_accessor :blocked + if opts[:start] and (Time === opts[:start]) + self.next = opts[:start] + else + self.next = Time.now + (opts[:start] || @period) + end + end - # period:: how often (seconds) to run the action - # data:: optional data to pass to the proc - # once:: optional, if true, this action will be run once then removed - # func:: associate a block to be called to perform the action - # - # create a new action - def initialize(period, data=nil, once=false, &func) - @blocked = false - @period = period - @in = period - @func = func - @data = data - @once = once - @last_tick = Time.new + def reschedule(period, &block) + self.configure(:period => period, &block) end - def tick - diff = Time.new - @last_tick - @in -= diff - @last_tick = Time.new + def block + @blocked = true end - def inspect - "#<#{self.class}:#{@period}s:#{@once ? 'once' : 'repeat'}>" + def unblock + @blocked = false end - def due? - @in <= 0 + def blocked? + @blocked end - # run the action by calling its proc - def run - @in += @period - # really short duration timers can overrun and leave @in negative, - # for these we set @in to @period - @in = @period if @in <= 0 + def run(now = Time.now) + raise 'inappropriate time to run()' unless self.next && self.next <= now + self.next = nil begin - if(@data) - @func.call(@data) - else - @func.call - end + @block.call(*@args) rescue Exception => e - error "Timer action #{self.inspect} with function #{@func.inspect} failed!" + error "Timer action #{self.inspect}: block #{@block.inspect} failed!" error e.pretty_inspect - # TODO maybe we want to block this Action? end - return @once - end - - # reschedule the Action to change its period - def reschedule(new_period) - @period = new_period - @in = new_period - end - end - - # timer handler, manage multiple Action objects, calling them when required. - # The timer must be ticked by whatever controls it, i.e. regular calls to - # tick() at whatever granularity suits your application's needs. - # - # Alternatively you can call run(), and the timer will spawn a thread and - # tick itself, intelligently shutting down the thread if there are no - # pending actions. - class Timer - def initialize(granularity = 0.1) - @granularity = granularity - @timers = Hash.new - @handle = 0 - @lasttime = 0 - @should_be_running = false - @thread = false - @next_action_time = 0 - end - # period:: how often (seconds) to run the action - # data:: optional data to pass to the action's proc - # func:: associate a block with add() to perform the action - # - # add an action to the timer - def add(period, data=nil, &func) - debug "adding timer, period #{period}" - @handle += 1 - @timers[@handle] = Action.new(period, data, &func) - start_on_add - return @handle - end - - # period:: how long (seconds) until the action is run - # data:: optional data to pass to the action's proc - # func:: associate a block with add() to perform the action - # - # add an action to the timer which will be run just once, after +period+ - def add_once(period, data=nil, &func) - debug "adding one-off timer, period #{period}" - @handle += 1 - @timers[@handle] = Action.new(period, data, true, &func) - start_on_add - return @handle - end + if @repeat && @period > 0 + self.next = now + @period + end - # remove action with handle +handle+ from the timer - def remove(handle) - @timers.delete(handle) + return self.next end + end - # block action with handle +handle+ - def block(handle) - raise "no such timer #{handle}" unless @timers[handle] - @timers[handle].blocked = true - end + def initialize + self.extend(MonitorMixin) + @tick = self.new_cond + @thread = nil + @actions = Hash.new + @current = nil + self.start + end - # unblock action with handle +handle+ - def unblock(handle) - raise "no such timer #{handle}" unless @timers[handle] - @timers[handle].blocked = false + def add(period, opts = {}, &block) + a = Action.new({:repeat => true, :period => period}.merge(opts), &block) + self.synchronize do + @actions[a.object_id] = a + @tick.signal end + return a.object_id + end - # reschedule action with handle +handle+ to change its period - def reschedule(handle, period) - raise "no such timer #{handle}" unless @timers[handle] - @timers[handle].reschedule(period) - tick - end + def add_once(period, opts = {}, &block) + self.add(period, {:repeat => false}.merge(opts), &block) + end - # you can call this when you know you're idle, or you can split off a - # thread and call the run() method to do it for you. - def tick - if(@lasttime == 0) - # don't do anything on the first tick - @lasttime = Time.now - return - end - @next_action_time = 0 - diff = (Time.now - @lasttime).to_f - @lasttime = Time.now - @timers.each { |key,timer| - timer.tick - next if timer.blocked - if(timer.due?) - if(timer.run) - # run once - @timers.delete(key) - end - end - if @next_action_time == 0 || timer.in < @next_action_time - @next_action_time = timer.in - end - } - #debug "ticked. now #{@timers.length} timers remain" - #debug "next timer due at #{@next_action_time}" - end + def block(aid) + debug "blocking #{aid}" + self.synchronize { self[aid].block } + end - # for backwards compat - this is a bit primitive - def run(granularity=0.1) - while(true) - sleep(granularity) - tick - end + def unblock(aid) + debug "unblocking #{aid}" + self.synchronize do + self[aid].unblock + @tick.signal end + end - def running? - @thread && @thread.alive? + def remove(aid) + self.synchronize do + @actions.delete(aid) # or raise "nonexistent action #{aid}" end + end - # return the number of seconds until the next action is due, or 0 if - # none are outstanding - will only be accurate immediately after a - # tick() - def next_action_time - @next_action_time - end + alias :delete :remove - # start the timer, it spawns a thread to tick the timer, intelligently - # shutting down if no events remain and starting again when needed. - def start - return if running? - @should_be_running = true - start_thread unless @timers.empty? + def configure(aid, opts = {}, &block) + self.synchronize do + self[aid].configure(opts, &block) + @tick.signal end + end - # stop the timer from ticking - def stop - @should_be_running = false - stop_thread - end + def reschedule(aid, period, &block) + self.configure(aid, :period => period, &block) + end - private + protected - def start_on_add - if running? - stop_thread - start_thread - elsif @should_be_running - start_thread + def start + raise 'double-started timer' if @thread + @thread = Thread.new do + loop do + tmout = self.run_actions + self.synchronize { @tick.wait(tmout) } end end + end - def stop_thread - return unless running? - @thread.kill - end + def [](aid) + aid ||= @current + raise "no current action" unless aid + raise "nonexistent action #{aid}" unless @actions.include? aid + @actions[aid] + end - def start_thread - return if running? - @thread = Thread.new do - while(true) - tick - exit if @timers.empty? - sleep(@next_action_time) + def run_actions(now = Time.now) + nxt = nil + @actions.keys.each do |k| + a = @actions[k] + next if (!a) or a.blocked? + + if a.next <= now + begin + @current = k + v = a.run(now) + ensure + @current = nil + end + + unless v + @actions.delete k + next end + else + v = a.next end + + nxt = v if v and ((!nxt) or (v < nxt)) end + if nxt + delta = nxt - now + delta = 0 if delta < 0 + return delta + else + return nil + end end + end |