Class: Barney::Share

Inherits:

Object

• Object
• Barney::Share

show all

Defined in:

lib/barney/share.rb

Defined Under Namespace

Classes: StreamPair

Class Attribute Summary (collapse)

• + (Mutex) mutex readonly private

  Returns a Mutex that is used when Barney::Share.value is being accessed by #synchronize.

• + (Object) value private

  Returns the latest value read from a spawned child process.

Instance Attribute Summary (collapse)

• - (Hash<Fixnum, Hash<Symbol, Object>>) history readonly

  Returns a history of changes made in multiple forks for a single instance of Share.

• - (Fixnum) pid readonly

  Returns the Process ID of the last forked child process.

• - (Array<Symbol>) shared readonly

  Returns a list of all variables or constants being shared for this instance of Barney::Share.

Instance Method Summary (collapse)

• - (Fixnum) fork(&blk)

  Serves as a method to spawn a new child process.

• - (Barney::Share) initialize {|self| ... } constructor
• - (Array<Symbol>) share(*variables)

  Serves as a method to mark a variable or constant to be shared between two processes.

• - synchronize (also: #sync)

  Serves as a method that synchronizes data between the parent and child process.

• - (Array<Symbol>) unshare(*variables)

  Serves as a method to remove a variable or constant from being shared between two processes.

40
41
42
43
44
45
46

# File 'lib/barney/share.rb', line 40

def initialize
  @shared  = {}
  @history = {}
  @context = nil
  @seq     = 0
  yield self if block_given? 
end

22
23
24

# File 'lib/barney/share.rb', line 22

def mutex
  @mutex
end

17
18
19

# File 'lib/barney/share.rb', line 17

def value
  @value
end

36
37
38

# File 'lib/barney/share.rb', line 36

def history
  @history
end

32
33
34

# File 'lib/barney/share.rb', line 32

def pid
  @pid
end

27

# File 'lib/barney/share.rb', line 27

def shared; @shared.keys; end

79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

# File 'lib/barney/share.rb', line 79

def fork &blk
  raise ArgumentError, "A block or Proc object is expected" unless block_given?
  share *@shared.keys if @pid

  @context = blk.binding
  @pid     = Kernel.fork do
    blk.call
    @shared.each do |variable, history|
      stream = history[-1]
      stream.in.close  
      stream.out.write Marshal.dump(eval("#{variable}", @context))
      stream.out.close
    end
  end
  
  @seq += 1
  @pid
end

51
52
53
54
55
56
57
58

# File 'lib/barney/share.rb', line 51

def share *variables
  variables.map(&:to_sym).each do |variable|
    if (@shared[variable].nil?) || (not @shared[variable].find { |struct| struct.seq == @seq })
      @shared.store variable, (@shared[variable] || []) << StreamPair.new(@seq, *IO.pipe)
    end
  end
  @shared.keys
end

101
102
103
104
105
106
107
108
109
110
111
112
113
114

# File 'lib/barney/share.rb', line 101

def synchronize 
  Barney::Share.mutex.synchronize do
    @shared.each do |variable, history|
      history.each do |stream|
        stream.out.close
        Barney::Share.value = Marshal.load stream.in.read
        stream.in.close
        object = eval "#{variable} = Barney::Share.value", @context 
        @history[stream.seq] = (@history[stream.seq] || {}).merge!({ variable => object })
      end
      history.clear
    end
  end
end

63
64
65
66
67
68

# File 'lib/barney/share.rb', line 63

def unshare *variables
  variables.map(&:to_sym).each do |variable|
    @shared.delete variable
  end
  @shared.keys
end