ShimShtein · January 13, 2018 04:27
diff --git a/Dynflow bottlenecks.md b/Dynflow bottlenecks.md
diff --git a/heap_analayzer.rb b/heap_analayzer.rb
 require 'json'

 class Heap
  attr_reader :objects, :obj_index, :parents_index
  def initialize(filename, classes_map)
    @filename = filename

    @classes_map = JSON.parse(File.read(classes_map))
  end

  def load
    @objects = []
    @obj_index = {}
    @parents_index = {}

    read do |parsed|
      parsed['class_name'] = @classes_map[parsed['class'].to_i(16).to_s] if parsed['class']
      address = parsed['address']
      @obj_index[address] = parsed
      @objects << parsed

      (parsed['references'] || []).each do |child_address|
        parents = @parents_index[child_address] || []
        parents << address
        @parents_index[child_address] = parents
      end
    end
  end

  private

  def read
    File.open(@filename) do |f|
      f.each_line do |line|
        begin
          parsed = JSON.parse(line)
          yield parsed
        rescue JSON::ParserError
          puts "Could not parse #{line}"
        end
      end
    end
  end
 end

 def draw_puml(roots, objects, prefix)
  visited = Set.new
  edges = render_edges(roots, visited)
  obj_hash = objects.select{ |k,v| visited.member?(k) || roots.member?(v) }

  if obj_hash.count > 2
    File.open("/tmp/objects_map_#{prefix}.puml", "w") do |f|
      f.write("@startuml\n")
      f.write(render_classes(obj_hash))
      f.write(edges)
      f.write("\n@enduml\n")
    end
    puts 'puml done'
  end
 end

 def render_classes(objects)
  objects.values.map{ |o| render_object(o) }.join("\n")
 end

 def render_object(obj)
  "class #{obj['address']} << #{obj['class_name']} >>" +
  if obj['value']
    " {\n  value = #{obj['value']}\n}\n"
  else
    "\n"
  end
 end

 def render_edges(roots, visited = Set.new)
  return "" if roots.empty?
  next_roots = []
  roots.map do |root|
    refs = root['references']
    if refs
      next_roots += refs.map { |child| child if visited.add?(child) }.compact
      refs.map { |child| "#{root['address']} -> #{child}" }.join("\n")
    else
      nil
    end
  end.compact.join("\n") + render_edges(next_roots, visited).to_s
 end

 before_second = Heap.new '/tmp/dump_before_gc_2', '/tmp/classes_map_2'
 before_second.load
 after_second = Heap.new '/tmp/dump_after_gc_2', '/tmp/classes_map_2'
 after_second.load

 cleaned = before_second.obj_index.reject { |k, v| after_second.obj_index.key?(k) }
 roots = cleaned.reject { |k, v| before_second.parents_index.key?(k) }
 by_file = cleaned.values.group_by { |obj| "#{obj['file']}:#{obj['line']}" }
 most_created = by_file.sort_by { |k, v| v.length }.last(10).map { |k, v| [k, v.first['class_name'], v.length] }

 # i = 0
 # roots.values.each do |root|
 #   draw_puml([root], cleaned, i+=1)
 # end
 binding.pry
diff --git a/modified_remote_executor.rb b/modified_remote_executor.rb
 #!/usr/bin/env ruby
 # -*- coding: utf-8 -*-

 require_relative 'example_helper'
 require_relative 'orchestrate_evented'
 require 'tmpdir'
 require 'csv'

 class SampleAction < Dynflow::Action
  def plan
    # number = rand(1e10)
    @@number ||= 0
    @@number += 1

    puts "Plannin action: #{@@number}"
    plan_self(number: @@number)
  end

  def run
    # require 'pry'
    # binding.pry
    #
    puts "Running task number #{input[:number]}"

    # puts GC.stat
    # classes_map = {}
    # ObjectSpace.each_object do |o|
    #   begin
    #     classes_map[(o.object_id << 1)] = o.name if o.is_a?(Class) && o.object_id.is_a?(Integer)
    #   rescue Exception => ex
    #     binding.pry
    #     i=0
    #   end
    # end

    # File.open("/tmp/classes_map_#{input[:number]}", "w") do |f|
    #   f.write(classes_map.to_json)
    # end

    # io=File.open("/tmp/dump_before_gc_#{input[:number]}", "w")
    # ObjectSpace.dump_all(output: io)
    # io.close

    # GC.start

    # io=File.open("/tmp/dump_after_gc_#{input[:number]}", "w")
    # ObjectSpace.dump_all(output: io)
    # io.close

    # require 'pry'
    # binding.pry

    puts 'Done'
  end
 end

 class RemoteExecutorExample
  class << self

    def run_observer
      world               = ExampleHelper.create_world do |config|
        config.persistence_adapter = persistence_adapter
        config.connector           = connector
        config.executor            = false
      end
      run(world)
    end

    def run_server
      require 'gc_tracer'
      GC::Tracer.start_logging
      # require 'allocation_tracer'
      # ObjectSpace::AllocationTracer.trace
      world               = ExampleHelper.create_world do |config|
        config.persistence_adapter = persistence_adapter
        config.connector           = connector
      end
      run(world)
      # results = ObjectSpace::AllocationTracer.stop
      #
      # puts ObjectSpace::AllocationTracer.header.join("\t")
      # results.sort_by{|k, v| -v.last}.first(30).each{|k, v|
      #   puts (k+v).join("\t")
      # }
      #
      # puts GC.stat
    end

    def run(world)
      begin
        ExampleHelper.run_web_console(world)
      rescue Errno::EADDRINUSE
        require 'io/console'
        puts "Running without a web console. Press q<enter> to quit."
        until STDIN.gets.chomp == 'q'
        end
      end
    end

    def db_path
      File.expand_path("/tmp/remote_executor_db.sqlite", __FILE__)
    end

    def persistence_conn_string
      ENV['DB_CONN_STRING'] || "sqlite://#{db_path}"
    end

    def persistence_adapter
      Dynflow::PersistenceAdapters::Sequel.new persistence_conn_string
    end

    def connector
      Proc.new { |world| Dynflow::Connectors::Database.new(world) }
    end

    def run_client
      world    = ExampleHelper.create_world do |config|
        config.persistence_adapter = persistence_adapter
        config.executor            = false
        config.connector           = connector
      end

      # world.trigger(OrchestrateEvented::CreateInfrastructure)
      # world.trigger(OrchestrateEvented::CreateInfrastructure, true)

      10.times do
        world.trigger(SampleAction).finished.wait
        # sleep 0.1
      end
    end

  end
 end

 command = ARGV.first || 'server'

 if $0 == __FILE__
  case command
  when 'observer'
    puts <<MSG
 The observer starting…. You can see what's going on there
 MSG
    RemoteExecutorExample.run_observer
  when 'server'
    puts <<MSG
 The server is starting…. You can send the work to it by running:

   #{$0} client

 MSG
    RemoteExecutorExample.run_server
  when 'client'
    RemoteExecutorExample.run_client
  else
    puts "Unknown command #{comment}"
    exit 1
  end
 end
	require 'json'

	class Heap
	attr_reader :objects, :obj_index, :parents_index
	def initialize(filename, classes_map)
	@filename = filename

	@classes_map = JSON.parse(File.read(classes_map))
	end

	def load
	@objects = []
	@obj_index = {}
	@parents_index = {}

	read do \|parsed\|
	parsed['class_name'] = @classes_map[parsed['class'].to_i(16).to_s] if parsed['class']
	address = parsed['address']
	@obj_index[address] = parsed
	@objects << parsed

	(parsed['references'] \|\| []).each do \|child_address\|
	parents = @parents_index[child_address] \|\| []
	parents << address
	@parents_index[child_address] = parents
	end
	end
	end

	private

	def read
	File.open(@filename) do \|f\|
	f.each_line do \|line\|
	begin
	parsed = JSON.parse(line)
	yield parsed
	rescue JSON::ParserError
	puts "Could not parse #{line}"
	end
	end
	end
	end
	end

	def draw_puml(roots, objects, prefix)
	visited = Set.new
	edges = render_edges(roots, visited)
	obj_hash = objects.select{ \|k,v\| visited.member?(k) \|\| roots.member?(v) }

	if obj_hash.count > 2
	File.open("/tmp/objects_map_#{prefix}.puml", "w") do \|f\|
	f.write("@startuml\n")
	f.write(render_classes(obj_hash))
	f.write(edges)
	f.write("\n@enduml\n")
	end
	puts 'puml done'
	end
	end

	def render_classes(objects)
	objects.values.map{ \|o\| render_object(o) }.join("\n")
	end

	def render_object(obj)
	"class #{obj['address']} << #{obj['class_name']} >>" +
	if obj['value']
	" {\n value = #{obj['value']}\n}\n"
	else
	"\n"
	end
	end

	def render_edges(roots, visited = Set.new)
	return "" if roots.empty?
	next_roots = []
	roots.map do \|root\|
	refs = root['references']
	if refs
	next_roots += refs.map { \|child\| child if visited.add?(child) }.compact
	refs.map { \|child\| "#{root['address']} -> #{child}" }.join("\n")
	else
	nil
	end
	end.compact.join("\n") + render_edges(next_roots, visited).to_s
	end

	before_second = Heap.new '/tmp/dump_before_gc_2', '/tmp/classes_map_2'
	before_second.load
	after_second = Heap.new '/tmp/dump_after_gc_2', '/tmp/classes_map_2'
	after_second.load

	cleaned = before_second.obj_index.reject { \|k, v\| after_second.obj_index.key?(k) }
	roots = cleaned.reject { \|k, v\| before_second.parents_index.key?(k) }
	by_file = cleaned.values.group_by { \|obj\| "#{obj['file']}:#{obj['line']}" }
	most_created = by_file.sort_by { \|k, v\| v.length }.last(10).map { \|k, v\| [k, v.first['class_name'], v.length] }

	# i = 0
	# roots.values.each do \|root\|
	# draw_puml([root], cleaned, i+=1)
	# end
	binding.pry
	#!/usr/bin/env ruby
	# -- coding: utf-8 --

	require_relative 'example_helper'
	require_relative 'orchestrate_evented'
	require 'tmpdir'
	require 'csv'

	class SampleAction < Dynflow::Action
	def plan
	# number = rand(1e10)
	@@number \|\|= 0
	@@number += 1

	puts "Plannin action: #{@@number}"
	plan_self(number: @@number)
	end

	def run
	# require 'pry'
	# binding.pry
	#
	puts "Running task number #{input[:number]}"

	# puts GC.stat
	# classes_map = {}
	# ObjectSpace.each_object do \|o\|
	# begin
	# classes_map[(o.object_id << 1)] = o.name if o.is_a?(Class) && o.object_id.is_a?(Integer)
	# rescue Exception => ex
	# binding.pry
	# i=0
	# end
	# end

	# File.open("/tmp/classes_map_#{input[:number]}", "w") do \|f\|
	# f.write(classes_map.to_json)
	# end

	# io=File.open("/tmp/dump_before_gc_#{input[:number]}", "w")
	# ObjectSpace.dump_all(output: io)
	# io.close

	# GC.start

	# io=File.open("/tmp/dump_after_gc_#{input[:number]}", "w")
	# ObjectSpace.dump_all(output: io)
	# io.close

	# require 'pry'
	# binding.pry

	puts 'Done'
	end
	end

	class RemoteExecutorExample
	class << self

	def run_observer
	world = ExampleHelper.create_world do \|config\|
	config.persistence_adapter = persistence_adapter
	config.connector = connector
	config.executor = false
	end
	run(world)
	end

	def run_server
	require 'gc_tracer'
	GC::Tracer.start_logging
	# require 'allocation_tracer'
	# ObjectSpace::AllocationTracer.trace
	world = ExampleHelper.create_world do \|config\|
	config.persistence_adapter = persistence_adapter
	config.connector = connector
	end
	run(world)
	# results = ObjectSpace::AllocationTracer.stop
	#
	# puts ObjectSpace::AllocationTracer.header.join("\t")
	# results.sort_by{\|k, v\| -v.last}.first(30).each{\|k, v\|
	# puts (k+v).join("\t")
	# }
	#
	# puts GC.stat
	end

	def run(world)
	begin
	ExampleHelper.run_web_console(world)
	rescue Errno::EADDRINUSE
	require 'io/console'
	puts "Running without a web console. Press q<enter> to quit."
	until STDIN.gets.chomp == 'q'
	end
	end
	end

	def db_path
	File.expand_path("/tmp/remote_executor_db.sqlite", __FILE__)
	end

	def persistence_conn_string
	ENV['DB_CONN_STRING'] \|\| "sqlite://#{db_path}"
	end

	def persistence_adapter
	Dynflow::PersistenceAdapters::Sequel.new persistence_conn_string
	end

	def connector
	Proc.new { \|world\| Dynflow::Connectors::Database.new(world) }
	end

	def run_client
	world = ExampleHelper.create_world do \|config\|
	config.persistence_adapter = persistence_adapter
	config.executor = false
	config.connector = connector
	end

	# world.trigger(OrchestrateEvented::CreateInfrastructure)
	# world.trigger(OrchestrateEvented::CreateInfrastructure, true)

	10.times do
	world.trigger(SampleAction).finished.wait
	# sleep 0.1
	end
	end

	end
	end

	command = ARGV.first \|\| 'server'

	if $0 == __FILE__
	case command
	when 'observer'
	puts <<MSG
	The observer starting…. You can see what's going on there
	MSG
	RemoteExecutorExample.run_observer
	when 'server'
	puts <<MSG
	The server is starting…. You can send the work to it by running:

	#{$0} client

	MSG
	RemoteExecutorExample.run_server
	when 'client'
	RemoteExecutorExample.run_client
	else
	puts "Unknown command #{comment}"
	exit 1
	end
	end