<commit_before>#!/usr/bin/env ruby
# encoding: utf-8

require 'drb/drb' 
require 'fileutils'
begin
  require 'kyotocabinet'
  NO_KYOTO = false
rescue LoadError
  require 'pstore'
  class PStore
    def each
      self.roots.each do |key|
        yield(key, self[key])
      end
    end
  end
  NO_KYOTO = true
end

module WorkflowManager
  # default parameters
  LOG_DIR  = 'logs'
  DB_DIR   = 'dbs'
  INTERVAL = 30
  RESUBMIT = 0

  class Server
    @@config = nil
    class Config
      attr_accessor :log_dir
      attr_accessor :db_dir
      attr_accessor :interval
      attr_accessor :resubmit
      attr_accessor :cluster
    end
    def self.config=(config)
      @@config = config
    end
    def self.config
      @@config
    end
    def config
      @@config ||= WorkflowManager.configure{}
    end
    def self.configure
      @@config = Config.new
      # default values
      @@config.log_dir  = LOG_DIR
      @@config.db_dir   = DB_DIR
      @@config.interval = INTERVAL  # interval to check jobs, [s]
      @@config.resubmit = RESUBMIT  # how many times at maximum to resubmit when job fails
      yield(@@config)
      if @@config.cluster
        @@config.cluster.log_dir = File.expand_path(@@config.log_dir)
      end
      @@config
    end

    class KyotoDB
      def initialize(db_file)
        @file = db_file
        @db = KyotoCabinet::DB.new
      end
      def transaction
        @db.open(@file)
        yield(@db)
        @db.close
      end
    end
    class PStoreDB 
      def initialize(db_file)
        @db = PStore.new(db_file)
      end
      def transaction
        @db.transaction do 
          yield(@db)
        end
      end
    end

    def initialize
      @interval = config.interval
      @resubmit = config.resubmit
      extension = NO_KYOTO ? '.pstore' : '.kch'
      db_mode = NO_KYOTO ? 'PStore' : 'KyotoCabinet'
      @db_stat = File.join(config.db_dir, 'statuses'+extension)
      @db_logs  = File.join(config.db_dir, 'logs'+extension)

      @log_dir = File.expand_path(config.log_dir)
      @db_dir  = File.expand_path(config.db_dir)
      FileUtils.mkdir_p @log_dir unless File.exist?(@log_dir)
      FileUtils.mkdir_p @db_dir unless File.exist?(@db_dir)
      #@statuses = KyotoCabinet::DB.new
      @statuses = NO_KYOTO ? PStoreDB.new(@db_stat) : KyotoDB.new(@db_stat)
      #@logs = KyotoCabinet::DB.new 
      @logs = NO_KYOTO ? PStoreDB.new(@db_logs) : KyotoDB.new(@db_logs)
      @system_log = File.join(@log_dir, "system.log")
      @mutex = Mutex.new
      @cluster = config.cluster
      puts("DB = #{db_mode}")
      puts("Cluster = #{@cluster.name}")
      log_puts("DB = #{db_mode}")
      log_puts("Cluster = #{@cluster.name}")
      log_puts("Server starts")
    end
    def hello
      'hello, '+ @cluster.name
    end
    def copy_commands(org_dir, dest_parent_dir)
      @cluster.copy_commands(org_dir, dest_parent_dir)
    end
    def kill_job(job_id)
      status = `#{@cluster.kill_command(job_id)}`
    end
    def log_puts(str)
      time = Time.now.strftime("[%Y.%m.%d %H:%M:%S]")
      @mutex.synchronize do
         open(@system_log, "a") do |out|
           out.print time + " " + str + "\n"
         end
      end
    end<commit_msg>Updated status to fail when kill_job is called<commit_after>#!/usr/bin/env ruby
# encoding: utf-8

require 'drb/drb' 
require 'fileutils'
begin
  require 'kyotocabinet'
  NO_KYOTO = false
rescue LoadError
  require 'pstore'
  class PStore
    def each
      self.roots.each do |key|
        yield(key, self[key])
      end
    end
  end
  NO_KYOTO = true
end

module WorkflowManager
  # default parameters
  LOG_DIR  = 'logs'
  DB_DIR   = 'dbs'
  INTERVAL = 30
  RESUBMIT = 0

  class Server
    @@config = nil
    class Config
      attr_accessor :log_dir
      attr_accessor :db_dir
      attr_accessor :interval
      attr_accessor :resubmit
      attr_accessor :cluster
    end
    def self.config=(config)
      @@config = config
    end
    def self.config
      @@config
    end
    def config
      @@config ||= WorkflowManager.configure{}
    end
    def self.configure
      @@config = Config.new
      # default values
      @@config.log_dir  = LOG_DIR
      @@config.db_dir   = DB_DIR
      @@config.interval = INTERVAL  # interval to check jobs, [s]
      @@config.resubmit = RESUBMIT  # how many times at maximum to resubmit when job fails
      yield(@@config)
      if @@config.cluster
        @@config.cluster.log_dir = File.expand_path(@@config.log_dir)
      end
      @@config
    end

    class KyotoDB
      def initialize(db_file)
        @file = db_file
        @db = KyotoCabinet::DB.new
      end
      def transaction
        @db.open(@file)
        yield(@db)
        @db.close
      end
    end
    class PStoreDB 
      def initialize(db_file)
        @db = PStore.new(db_file)
      end
      def transaction
        @db.transaction do 
          yield(@db)
        end
      end
    end

    def initialize
      @interval = config.interval
      @resubmit = config.resubmit
      extension = NO_KYOTO ? '.pstore' : '.kch'
      db_mode = NO_KYOTO ? 'PStore' : 'KyotoCabinet'
      @db_stat = File.join(config.db_dir, 'statuses'+extension)
      @db_logs  = File.join(config.db_dir, 'logs'+extension)

      @log_dir = File.expand_path(config.log_dir)
      @db_dir  = File.expand_path(config.db_dir)
      FileUtils.mkdir_p @log_dir unless File.exist?(@log_dir)
      FileUtils.mkdir_p @db_dir unless File.exist?(@db_dir)
      #@statuses = KyotoCabinet::DB.new
      @statuses = NO_KYOTO ? PStoreDB.new(@db_stat) : KyotoDB.new(@db_stat)
      #@logs = KyotoCabinet::DB.new 
      @logs = NO_KYOTO ? PStoreDB.new(@db_logs) : KyotoDB.new(@db_logs)
      @system_log = File.join(@log_dir, "system.log")
      @mutex = Mutex.new
      @cluster = config.cluster
      puts("DB = #{db_mode}")
      puts("Cluster = #{@cluster.name}")
      log_puts("DB = #{db_mode}")
      log_puts("Cluster = #{@cluster.name}")
      log_puts("Server starts")
    end
    def hello
      'hello, '+ @cluster.name
    end
    def copy_commands(org_dir, dest_parent_dir)
      @cluster.copy_commands(org_dir, dest_parent_dir)
    end
    def kill_job(job_id)
      status(job_id, 'fail')
      status = `#{@cluster.kill_command(job_id)}`
    end
    def log_puts(str)
      time = Time.now.strftime("[%Y.%m.%d %H:%M:%S]")
      @mutex.synchronize do
         open(@system_log, "a") do |out|
           out.print time + " " + str + "\n"
         end
      end
    end
{'fraction_non_alphanumeric': 0.11475662353666051, 'fraction_numerical': 0.0012322858903265558, 'mean_word_length': 2.3297435897435896, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1982239', 'n_tokens_mistral': 2329, 'n_tokens_neox': 2242, 'n_words': 486}