Module: Bud

223
224
225

# File 'lib/bud/aggs.rb', line 223

def accum_pair(x, y)
  [AccumPair.new, x, y]
end

810
811
812
813
814
815
816

# File 'lib/bud.rb', line 810

def async_do
  EventMachine::schedule do
    yield if block_given?
    # Do another tick, in case the user-supplied block inserted any data
    tick_internal
  end
end

191
192
193

# File 'lib/bud/aggs.rb', line 191

def avg(x)
  [Avg.new, x]
end

81
82
83

# File 'lib/bud/aggs.rb', line 81

def bool_and(x)
  [BooleanAnd.new, x]
end

95
96
97

# File 'lib/bud/aggs.rb', line 95

def bool_or(x)
  [BooleanOr.new, x]
end

626
627

# File 'lib/bud.rb', line 626

def bootstrap # :nodoc: all
end

1169
1170
1171
1172
1173

# File 'lib/bud.rb', line 1169

def bud_clock
  raise Bud::Error, "bud_clock undefined outside tick" unless @inside_tick
  @tick_clock_time ||= Time.now
  @tick_clock_time
end

770
771
772
773
774
775

# File 'lib/bud.rb', line 770

def cancel_shutdown_cb(id)
  schedule_and_wait do
    raise Bud::Error unless @shutdown_callbacks.has_key? id
    @shutdown_callbacks.delete(id)
  end
end

115
116
117
118
119

# File 'lib/bud/state.rb', line 115

def channel(name, schema=nil, loopback=false)
  define_collection(name)
  @tables[name] = Bud::BudChannel.new(name, self, schema, loopback)
  @channels[name] = @tables[name]
end

114
115
116

# File 'lib/bud/aggs.rb', line 114

def choose(x)
  [Choose.new, x]
end

144
145
146

# File 'lib/bud/aggs.rb', line 144

def choose_rand(x=nil)
  [ChooseOneRand.new, x]
end

64
65
66
67

# File 'lib/bud/state.rb', line 64

def coll_expr(name, expr, schema=nil)
  define_collection(name)
  @tables[name] = Bud::BudCollExpr.new(name, self, expr, schema)
end

1178
1179
1180
1181
1182
1183
1184

# File 'lib/bud.rb', line 1178

def collection_stratum(collection)
  t_stratum.each do |t|
    return t.stratum if t.predicate == collection
  end

  return nil
end

171
172
173

# File 'lib/bud/aggs.rb', line 171

def count(x=nil)
  [Count.new]
end

900
901
902

# File 'lib/bud.rb', line 900

def delta(out_tbl)
  sync_callback(nil, nil, out_tbl)
end

1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066

# File 'lib/bud.rb', line 1044

def do_bootstrap
  # Evaluate bootstrap for imported modules
  @this_rule_context = self
  imported = import_defs.keys
  imported.each do |mod_alias|
    wrapper = import_instance mod_alias
    wrapper.do_bootstrap
  end
  self.class.ancestors.reverse.each do |anc|
    anc.instance_methods(false).each do |m|
      if /^__bootstrap__/.match m
        self.method(m.to_sym).call
      end
    end
  end
  bootstrap

  if toplevel == self
    @tables.each_value {|t| t.bootstrap}
    @lattices.each_value {|l| l.bootstrap}
  end
  @done_bootstrap = true
end

840
841
842
843
844
845
846
847
848
849
850

# File 'lib/bud.rb', line 840

def do_register_callback(tbl_name, &blk)
  unless @tables.has_key? tbl_name
    raise Bud::Error, "no such collection: #{tbl_name}"
  end

  raise Bud::Error if @callbacks.has_key? @callback_id
  @callbacks[@callback_id] = [tbl_name, blk]
  cb_id = @callback_id
  @callback_id += 1
  return cb_id
end

131
132
133
134

# File 'lib/bud/state.rb', line 131

def file_reader(name, filename)
  define_collection(name)
  @tables[name] = Bud::BudFileReader.new(name, filename, self)
end

202
203
204
205

# File 'lib/bud.rb', line 202

def import_instance(name)
  name = "@" + name.to_s
  instance_variable_get(name) if instance_variable_defined? name
end

111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175

# File 'lib/bud.rb', line 111

def initialize(options={})
  # capture the binding for a subsequent 'eval'. This ensures that local
  # variable names introduced later in this method don't interfere with
  # table names used in the eval block.
  options[:dump_rewrite] ||= ENV["BUD_DUMP_REWRITE"].to_i > 0
  options[:dump_ast]     ||= ENV["BUD_DUMP_AST"].to_i > 0
  options[:print_wiring] ||= ENV["BUD_PRINT_WIRING"].to_i > 0
  @qualified_name = ""
  @tables = {}
  @lattices = {}
  @channels = {}
  @dbm_tables = {}
  @zk_tables = {}
  @stratified_rules = []
  @push_elems = {}
  @callbacks = {}
  @callback_id = 0
  @shutdown_callbacks = {}
  @shutdown_callback_id = 0
  @post_shutdown_callbacks = []
  @timers = []
  @app_tables = []
  @inside_tick = false
  @tick_clock_time = nil
  @budtime = 0
  @inbound = {}
  @done_bootstrap = false
  @done_wiring = false
  @instance_id = ILLEGAL_INSTANCE_ID # Assigned when we start running
  @metrics = {}
  @endtime = nil
  @this_stratum = 0
  @push_sorted_elems = nil
  @running_async = false
  @bud_started = false

  # Setup options (named arguments), along with default values
  @options = options.clone
  @options[:ip] ||= "127.0.0.1"
  @ip = @options[:ip]
  @options[:port] ||= 0
  @options[:port] = @options[:port].to_i
  # NB: If using an ephemeral port (specified by port = 0), the actual port
  # number won't be known until we start EM

  load_lattice_defs
  builtin_state
  resolve_imports
  call_state_methods

  @declarations = self.class.instance_methods.select {|m| m =~ /^__bloom__.+$/}.map {|m| m.to_s}

  @viz = VizOnline.new(self) if @options[:trace]
  @rtracer = RTrace.new(self) if @options[:rtrace]

  do_rewrite
  if toplevel == self
    # initialize per-stratum state
    @num_strata = @stratified_rules.length
    @scanners = @num_strata.times.map{{}}
    @push_sources = @num_strata.times.map{{}}
    @push_joins = @num_strata.times.map{[]}
    @merge_targets = @num_strata.times.map{Set.new}
  end
end

42
43
44

# File 'lib/bud/state.rb', line 42

def input # :nodoc: all
  true
end

905
906
907

# File 'lib/bud.rb', line 905

def inspect
  "#{self.class}:#{self.object_id.to_s(16)}"
end

1033
1034
1035
1036

# File 'lib/bud.rb', line 1033

def int_ip_port
  raise Bud::Error, "int_ip_port called before port defined" if @port.nil? and @options[:port] == 0
  @port.nil? ? "#{@ip}:#{@options[:port]}" : "#{@ip}:#{@port}"
end

51
52
53
54
55

# File 'lib/bud/state.rb', line 51

def interface(mode, name, schema=nil)
  define_collection(name)
  t_provides << [name.to_s, mode]
  @tables[name] = (mode ? BudInputInterface : BudOutputInterface).new(name, self, schema)
end

157
158
159
160
161
162
163
164
165
166
167

# File 'lib/bud/state.rb', line 157

def interfaces(direction, collections)
  mode = case direction
    when :input then true
    when :output then false
  else
    raise Bud::CompileError, "unrecognized interface type #{direction}"
  end
  collections.each do |tab|
    t_provides << [tab.to_s, mode]
  end 
end

1022
1023
1024

# File 'lib/bud.rb', line 1022

def ip
  options[:ext_ip] ? "#{@options[:ext_ip]}" : "#{@ip}"
end

1017
1018
1019
1020

# File 'lib/bud.rb', line 1017

def ip_port
  raise Bud::Error, "ip_port called before port defined" if port.nil?
  ip.to_s + ":" + port.to_s
end

171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231

# File 'lib/bud/state.rb', line 171

def load_lattice_defs
  Bud::Lattice.global_mfuncs.each do |m|
    next if RuleRewriter::MONOTONE_WHITELIST.include? m
    if Bud::BudCollection.instance_methods.include? m.to_s
      puts "monotone method #{m} conflicts with non-monotonic method in BudCollection"
    end
  end

  Bud::Lattice.global_morphs.each do |m|
    next if RuleRewriter::MONOTONE_WHITELIST.include? m
    if Bud::BudCollection.instance_methods.include? m.to_s
      puts "morphism #{m} conflicts with non-monotonic method in BudCollection"
    end
  end

  # Sanity-check lattice definitions
  # XXX: We should do this only once per lattice
  Bud::Lattice.lattice_kinds.each do |wrap_name, klass|
    unless klass.method_defined? :merge
      raise Bud::CompileError, "lattice #{wrap_name} does not define a merge function"
    end

    # If a method is marked as monotone in any lattice, every lattice that
    # declares a method of that name must also mark it as monotone.
    meth_list = klass.instance_methods(false).to_set
    Bud::Lattice.global_mfuncs.each do |m|
      next unless meth_list.include? m.to_s
      unless klass.mfuncs.include? m
        raise Bud::CompileError, "method #{m} in #{wrap_name} must be monotone"
      end
    end

    # Apply a similar check for morphs
    Bud::Lattice.global_morphs.each do |m|
      next unless meth_list.include? m.to_s
      unless klass.morphs.include? m
        raise Bud::CompileError, "method #{m} in #{wrap_name} must be a morph"
      end
    end

    # Similarly, check for non-monotone lattice methods that are found in the
    # builtin list of monotone operators. The "merge" method is implicitly
    # monotone (XXX: should it be declared as a morph or monotone function?)
    meth_list.each do |m_str|
      m = m_str.to_sym
      next unless RuleRewriter::MONOTONE_WHITELIST.include? m
      # XXX: ugly hack. We want to allow lattice class implementations to
      # define their own equality semantics.
      next if m == :==
      unless klass.mfuncs.include?(m) || klass.morphs.include?(m) || m == :merge
        raise Bud::CompileError, "method #{m} in #{wrap_name} must be monotone"
      end
    end

    # XXX: replace "self" with toplevel?
    self.singleton_class.send(:define_method, wrap_name) do |lat_name|
      define_lattice(lat_name)
      @lattices[lat_name] = Bud::LatticeWrapper.new(lat_name, klass, self)
    end
  end
end

125
126
127
128

# File 'lib/bud/state.rb', line 125

def loopback(name, schema=nil)
  schema ||= {[:key] => [:val]}
  channel(name, schema, true)
end

67
68
69

# File 'lib/bud/aggs.rb', line 67

def max(x)
  [Max.new, x]
end

50
51
52

# File 'lib/bud/aggs.rb', line 50

def min(x)
  [Min.new, x]
end

177
178
179
180
181
182
183
184
185
186
187
188

# File 'lib/bud.rb', line 177

def module_wrapper_class(mod)
  class_name = "#{mod}__wrap"
  begin
    klass = Module.const_get(class_name.to_sym)
    unless klass.is_a? Class
      raise Bud::Error, "internal error: #{class_name} is in use"
    end
  rescue NameError # exception if class class_name doesn't exist
  end
  klass ||= eval "class #{class_name}; include Bud; include #{mod}; end"
  klass
end

758
759
760
761
762
763
764
765
766
767
768

# File 'lib/bud.rb', line 758

def on_shutdown(&blk)
  # Start EM if not yet started
  start_reactor
  rv = nil
  schedule_and_wait do
    rv = @shutdown_callback_id
    @shutdown_callbacks[@shutdown_callback_id] = blk
    @shutdown_callback_id += 1
  end
  return rv
end

46
47
48

# File 'lib/bud/state.rb', line 46

def output # :nodoc: all
  false
end

704
705
706
707
708

# File 'lib/bud.rb', line 704

def pause
  schedule_and_wait do
    @running_async = false
  end
end

138
139
140
141
142
143

# File 'lib/bud/state.rb', line 138

def periodic(name, period=1)
  define_collection(name)
  raise Bud::Error if @periodics.has_key? [name]
  @periodics << [name, period]
  @tables[name] = Bud::BudPeriodic.new(name, self)
end

1026
1027
1028
1029
1030

# File 'lib/bud.rb', line 1026

def port
  return nil if @port.nil? and @options[:port] == 0 and not @options[:ext_port]
  return @options[:ext_port] ? @options[:ext_port] :
    (@port.nil? ? @options[:port] : @port)
end

779
780
781
782
783
784
785

# File 'lib/bud.rb', line 779

def post_shutdown(&blk)
  # Start EM if not yet started
  start_reactor
  schedule_and_wait do
    @post_shutdown_callbacks << blk
  end
end

102
103
104
105

# File 'lib/bud/state.rb', line 102

def readonly(name, schema=nil)
  define_collection(name)
  @tables[name] = Bud::BudReadOnly.new(name, self, schema)
end

829
830
831
832
833
834
835
836
837
838

# File 'lib/bud.rb', line 829

def register_callback(tbl_name, &blk)
  # We allow callbacks to be added before or after EM has been started. To
  # simplify matters, we start EM if it hasn't been started yet.
  start_reactor
  cb_id = nil
  schedule_and_wait do
    cb_id = do_register_callback(tbl_name, &blk)
  end
  return cb_id
end

643
644
645
646
647
648
649
650
651
652
653
654
655
656
657

# File 'lib/bud.rb', line 643

def run_bg
  start

  schedule_and_wait do
    if @running_async
      raise Bud::Error, "run_bg called on already-running Bud instance"
    end
    @running_async = true

    # Consume any events received while we weren't running async
    tick_internal
  end

  @rtracer.sleep if options[:rtrace]
end

714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734

# File 'lib/bud.rb', line 714

def run_fg
  # If we're called from the EventMachine thread (and EM is running), blocking
  # the current thread would imply deadlocking ourselves.
  if Thread.current == EventMachine::reactor_thread and EventMachine::reactor_running?
    raise Bud::Error, "cannot invoke run_fg from inside EventMachine"
  end

  q = Queue.new
  # Note that this must be a post-shutdown callback: if this is the only
  # thread, then the program might exit after run_fg() returns. If run_fg()
  # blocked on a normal shutdown callback, the program might exit before the
  # other shutdown callbacks have a chance to run.
  post_shutdown do
    q.push(true)
  end

  run_bg
  # Block caller's thread until Bud has shutdown
  q.pop
  report_metrics if options[:metrics]
end

97
98
99
100

# File 'lib/bud/state.rb', line 97

def scratch(name, schema=nil)
  define_collection(name)
  @tables[name] = Bud::BudScratch.new(name, self, schema)
end

631
632
633

# File 'lib/bud.rb', line 631

def singleton_class # :nodoc: all
  class << self; self; end
end

663
664
665
666
667
668
669

# File 'lib/bud.rb', line 663

def start(do_tick=false)
  start_reactor
  schedule_and_wait do
    do_startup unless @bud_started
    tick_internal if do_tick
  end
end

741
742
743
744
745
746
747
748
749
750
751

# File 'lib/bud.rb', line 741

def stop(stop_em=false, do_shutdown_cb=true)
  schedule_and_wait do
    do_shutdown(do_shutdown_cb)
  end

  if stop_em
    Bud.stop_em_loop
    EventMachine::reactor_thread.join
  end
  report_metrics if options[:metrics]
end

81
82
83
84
85
86
87
88
89
90
91
92
93
94

# File 'lib/bud/state.rb', line 81

def store(name, storage, schema=nil)
  define_collection(name)
  case storage
  when :zookeeper
    # treat "schema" as a hash of options
    options = schema
    raise Bud::Error, "Zookeeper tables require a :path option" if options[:path].nil?
    options[:addr] ||= "localhost:2181"
    @tables[name] = Bud::BudZkTable.new(name, options[:path], options[:addr], self)
    @zk_tables[name] = @tables[name]
  else
    raise Bud::Error, "unknown async storage engine #{storage.to_s}"
  end
end

156
157
158

# File 'lib/bud/aggs.rb', line 156

def sum(x)
  [Sum.new, x]
end

70
71
72
73
74
75
76
77
78
79

# File 'lib/bud/state.rb', line 70

def sync(name, storage, schema=nil)
  define_collection(name)
  case storage
  when :dbm
    @tables[name] = Bud::BudDbmTable.new(name, self, schema)
    @dbm_tables[name] = @tables[name]
  else
    raise Bud::Error, "unknown synchronous storage engine #{storage.to_s}"
  end
end

864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897

# File 'lib/bud.rb', line 864

def sync_callback(in_tbl, tupleset, out_tbl)
  q = Queue.new

  # If the runtime shuts down before we see anything in the output collection,
  # make sure we hear about it so we can raise an error
  # XXX: Using two separate callbacks here is ugly.
  shutdown_cb = on_shutdown do
    q.push :callback
  end

  cb = nil
  sync_do {
    if in_tbl
      t = @tables[in_tbl]
      if t.class <= Bud::BudChannel or t.class <= Bud::BudZkTable
        t <~ tupleset
      else
        t <+ tupleset
      end
    end

    cb = do_register_callback(out_tbl) do |c|
      q.push c.to_a
    end
  }
  result = q.pop
  if result == :callback
    # Don't try to unregister the callbacks first: runtime is already shutdown
    raise Bud::ShutdownWithCallbacksError, "Bud instance shutdown before sync_callback completed"
  end
  unregister_callback(cb)
  cancel_shutdown_cb(shutdown_cb)
  return result
end

799
800
801
802
803
804
805

# File 'lib/bud.rb', line 799

def sync_do
  schedule_and_wait do
    yield if block_given?
    # Do another tick, in case the user-supplied block inserted any data
    tick_internal
  end
end

58
59
60
61

# File 'lib/bud/state.rb', line 58

def table(name, schema=nil)
  define_collection(name)
  @tables[name] = Bud::BudTable.new(name, self, schema)
end

108
109
110
111
112

# File 'lib/bud/state.rb', line 108

def temp(name)
  define_collection(name)
  # defer schema definition until merge
  @tables[name] = Bud::BudTemp.new(name, self, nil, true)
end

145
146
147
148
149
150
151
152
153
154

# File 'lib/bud/state.rb', line 145

def terminal(name) # :nodoc: all
  if defined?(@terminal) && @terminal != name
    raise Bud::Error, "can't register IO collection #{name} in addition to #{@terminal}"
  else
    @terminal = name
  end
  define_collection(name)
  @tables[name] = Bud::BudTerminal.new(name, self)
  @channels[name] = @tables[name]
end

1039
1040
1041

# File 'lib/bud.rb', line 1039

def tick
  start(true)
end

1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164

# File 'lib/bud.rb', line 1070

def tick_internal
  puts "#{object_id}/#{port} : ============================================= (#{@budtime})" if $BUD_DEBUG
  begin
    starttime = Time.now if options[:metrics]
    if options[:metrics] and not @endtime.nil?
      @metrics[:betweentickstats] ||= initialize_stats
      @metrics[:betweentickstats] = running_stats(@metrics[:betweentickstats],
                                                  starttime - @endtime)
    end
    @inside_tick = true

    unless @done_bootstrap
      do_bootstrap
      do_wiring
    else
      # inform tables and elements about beginning of tick.
      @app_tables.each {|t| t.tick}
      @default_rescan.each {|elem| elem.rescan = true}
      @default_invalidate.each {|elem|
        elem.invalidated = true
        # Call tick on tables here itself. The rest below
        elem.invalidate_cache unless elem.class <= PushElement
      }

      # The following loop invalidates additional (non-default) elements and
      # tables that depend on the run-time invalidation state of a table.
      # Loop once to set the flags.
      each_scanner do |scanner, stratum|
        if scanner.rescan
          scanner.rescan_set.each {|e| e.rescan = true}
          scanner.invalidate_set.each {|e|
            e.invalidated = true
            e.invalidate_cache unless e.class <= PushElement
          }
        end
      end

      # Loop a second time to actually call invalidate_cache.  We can't merge
      # this with the loops above because some versions of invalidate_cache
      # (e.g., join) depend on the rescan state of other elements.
      @num_strata.times do |stratum|
        @push_sorted_elems[stratum].each {|e| e.invalidate_cache if e.invalidated}
      end
    end

    receive_inbound
    # compute fixpoint for each stratum in order
    @stratified_rules.each_with_index do |rules,stratum|
      fixpoint = false
      first_iter = true
      until fixpoint
        @scanners[stratum].each_value {|s| s.scan(first_iter)}
        fixpoint = true
        first_iter = false
        # flush any tuples in the pipes
        @push_sorted_elems[stratum].each {|p| p.flush}
        # tick deltas on any merge targets and look for more deltas
        # check to see if any joins saw a delta
        @push_joins[stratum].each do |p|
          if p.found_delta
            fixpoint = false
            p.tick_deltas
          end
        end
        @merge_targets[stratum].each do |t|
          fixpoint = false if t.tick_deltas
        end
      end
      # push end-of-fixpoint
      @push_sorted_elems[stratum].each do |p|
        p.stratum_end
      end
      @merge_targets[stratum].each do |t|
        t.flush_deltas
      end
    end
    @viz.do_cards if @options[:trace]
    do_flush

    invoke_callbacks
    @budtime += 1
    @inbound.clear
    @reset_list.each {|e| e.invalidated = false; e.rescan = false}

  ensure
    @inside_tick = false
    @tick_clock_time = nil
  end

  if options[:metrics]
    @endtime = Time.now
    @metrics[:tickstats] ||= initialize_stats
    @metrics[:tickstats] = running_stats(@metrics[:tickstats], @endtime - starttime)
  end
end

190
191
192

# File 'lib/bud.rb', line 190

def toplevel
   @toplevel = (@options[:toplevel] || self)
end

198
199
200

# File 'lib/bud.rb', line 198

def toplevel?
  toplevel.object_id == self.object_id
end

853
854
855
856
857
858

# File 'lib/bud.rb', line 853

def unregister_callback(id)
  schedule_and_wait do
    raise Bud::Error, "missing callback: #{id.inspect}" unless @callbacks.has_key? id
    @callbacks.delete(id)
  end
end

212
213
214

# File 'lib/bud.rb', line 212

def wiring?
  toplevel? ? (@done_bootstrap && !@done_wiring) : toplevel.wiring?
end