Module: MasterSlaveSynchroniserMixin

Included in:
MasterSlaveSynchroniser
Defined in:
lib/buzzcore/extra/thread_utils.rb

Overview

This module decouples multiple master threads from multiple slave (worker) threads It provides two main methods : + master_attempt_command to be called by multiple clients with a command, returning a result or rasing an exception + slave_do_command to be called by worker threads with a block which processes the command.

see ProviderWorker in pay_server/app/pay_server.rb

Instance Method Summary collapse

Instance Method Details

#logger ⇒ Object 



327
328
329
 
# File 'lib/buzzcore/extra/thread_utils.rb', line 327

def logger
  @logger || (@logger = Logger.new(STDERR))
end

#master_attempt_command(aCommand) ⇒ Object 



339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
 
# File 'lib/buzzcore/extra/thread_utils.rb', line 339

def master_attempt_command(aCommand)
  @semaphore.exclusive do
    command_sent = false
    begin
      before = Time.now
      logger.debug { "master sending aCommand:"+aCommand.inspect }
      @mvCommand.value = aCommand
      command_sent = true
      logger.debug { "master waiting for result" }
      result = @mvResponse.value
      logger.debug { "master received result:"+result.inspect }
    rescue Exception => e
      # exception causes thread critical status to be lost
      logger.debug { "master exception:"+e.inspect }
      if command_sent
        logger.debug { "rejecting" }
        @mvResponse.reject_value    #!!! this doesn't seem to return
      end
      raise e
    ensure
      logger.debug { "master_attempt_command: command_sent=#{command_sent.to_s} elapsed:"+(Time.now-before).to_s }
    end
    result
  end
end

#ms_synchronizer_initialize(aTimeout = nil, aLogger = nil) ⇒ Object 



331
332
333
334
335
336
337
 
# File 'lib/buzzcore/extra/thread_utils.rb', line 331

def ms_synchronizer_initialize(aTimeout=nil,aLogger=nil)
  @logger = aLogger
  @semaphore = CountingSemaphore.new(1)
  timeout = aTimeout && aTimeout/2.0
  @mvCommand = MultiThreadVariable.new(nil,timeout)
  @mvResponse = MultiThreadVariable.new(timeout,nil)
end

#shutdown ⇒ Object 



378
379
380
381
 
# File 'lib/buzzcore/extra/thread_utils.rb', line 378

def shutdown
  @mvCommand.unblock()
  @mvResponse.unblock()
end

#slave_do_command(&block) ⇒ Object 



365
366
367
368
369
370
371
372
373
374
375
376
 
# File 'lib/buzzcore/extra/thread_utils.rb', line 365

def slave_do_command(&block)
  Thread.exclusive do
    logger.debug { "slave waiting for command" }
    command = @mvCommand.value
    logger.debug { "slave received command:"+command.inspect }
    result = yield(command)
    logger.debug { "slave sending result:"+result.inspect }
    @mvResponse.value = result
    logger.debug { "slave finished" }
    result
  end
end