<commit_before># -*- coding: utf-8 -*-
require "bundler"
Bundler.setup

require 'fileutils'
require 'json'
require 'set'

## setup thread
require 'thread'
MAX_THREAD = 10
# STOP queue
STOP = 0x01
threads = []
queue = Queue.new

require 'logger'
$log = Logger.new(STDOUT)

### external
require 'bitcask'
require 'bert'
require 'riak'

### load local
require './lib/arr_ex'

# CONST
BASETIME = Time.now
BACK_DIR = "./dump/" + BASETIME.strftime("%Y-%m-%dT%H%M")

## define expire date
# > 60 * 60 * 24 * 60 #=> 5184000
EXPIRE = 5184000
C_EXPIRE = BASETIME.to_i - EXPIRE


### change STDOUT to logfile
if !ENV["DEBUG"] then
  File.open("./log/" + BASETIME.strftime("%Y-%m-%dT%H%M") + "_throw.log", "a") do |file|
    puts "STDOUT redirect to #{file.path}.."
    STDOUT.reopen(file)
  end
end

@bits_root = "/var/db/riak/bitcask/"
# @bits_root = "/Users/sawanoboriyu/github/local/bitcask_dumper/bits/bitcask/"


MAX_THREAD.times do
  threads << Thread.start do
    $log.info "-- Stating thread --"

    # queue.popでenqueueされるまでwaitする。
    while q = queue.pop

      if q == STOP
        $log.info "-- Receive STOP Queue --"
        # 全てのthreadを止めるため、STOPを受け取ったらqueueに再度STOPを入れる
        queue << STOP
        break
      end
      

      # setup riak client
      riak = Riak::Client.new(:host => '127.0.0.1', :protocol => "pbc")

      ### load bitcask
      b = Bitcask.new File.join(@bits_root, q)
      b.load
      
      b.each do |key, value|
        next if value == Bitcask::TOMBSTONE
      
        bucket, key = BERT.decode key
        value = BERT.decode value
  
  
        # exclude expired data
        if bucket =~ /_net[0-9]$/ then
          if key.to_i / 1000 < C_EXPIRE
            # log to 
            $log.info "expired data : " + bucket + "/" + key
            next
          end
        end
  

        # check if exist before restore key-values.
        begin
          riak[bucket][key]
        rescue Riak::HTTPFailedRequest
          # throw to riak
          $log.info "throw to riak : " + bucket + "/" + key

          ## create set of Riak::Link
          links = value.find_array("Links")[0]

          ss = Set.new  
          links.each do |s|
            if BERT::Tuple === s then
              ss<< Riak::Link.new(s[0][0],s[0][1],s[1])
            end
          end

          ## Store to riak.
          begin
            ob = riak.bucket(bucket)
            o = ob.get_or_new(key)
            o.raw_data = value.last
            o.content_type = "application/json"
            o.links = ss
            o.store
          rescue => e
            ## failed key name
            $log.error "store failed  : " + bucket + "/" + key
            ## logging backtrace
            $log.error e.exception
          end
        else
          ## store skiped
          $log.info "skip exist key: " + bucket + "/" + key
        end
      end
    end
  end
end


Dir.entries(@bits_root).each do |bit_dir|
  ### skip . and ..
  next if bit_dir =~ /^\./

  queue << bit_dir
end


queue << STOP

# 全てのthreadをmain threadに合流させる（終わるのを待つ）
threads.map(&:join)
<commit_msg>stop risk client initialize each threads.<commit_after># -*- coding: utf-8 -*-
require "bundler"
Bundler.setup

require 'fileutils'
require 'json'
require 'set'

## setup thread
require 'thread'
MAX_THREAD = 10
# STOP queue
STOP = 0x01
threads = []
queue = Queue.new

require 'logger'
$log = Logger.new(STDOUT)

### external
require 'bitcask'
require 'bert'
require 'riak'

### load local
require './lib/arr_ex'

# CONST
BASETIME = Time.now
BACK_DIR = "./dump/" + BASETIME.strftime("%Y-%m-%dT%H%M")

## define expire date
# > 60 * 60 * 24 * 60 #=> 5184000
EXPIRE = 5184000
C_EXPIRE = BASETIME.to_i - EXPIRE


### change STDOUT to logfile
if !ENV["DEBUG"] then
  File.open("./log/" + BASETIME.strftime("%Y-%m-%dT%H%M") + "_throw.log", "a") do |file|
    puts "STDOUT redirect to #{file.path}.."
    STDOUT.reopen(file)
  end
end

@bits_root = "/var/db/riak/bitcask/"
# @bits_root = "/Users/sawanoboriyu/github/local/bitcask_dumper/bits/bitcask/"


# setup riak client
riak = Riak::Client.new(:host => '127.0.0.1')
# riak = Riak::Client.new(:host => '127.0.0.1', :protocol => "pbc")


MAX_THREAD.times do
  threads << Thread.start do
    $log.info "-- Stating thread --"

    # queue.popでenqueueされるまでwaitする。
    while q = queue.pop

      if q == STOP
        $log.info "-- Receive STOP Queue --"
        # 全てのthreadを止めるため、STOPを受け取ったらqueueに再度STOPを入れる
        queue << STOP
        break
      end
      

      ### load bitcask
      b = Bitcask.new File.join(@bits_root, q)
      b.load
      
      b.each do |key, value|
        next if value == Bitcask::TOMBSTONE
      
        bucket, key = BERT.decode key
        value = BERT.decode value
  
  
        # exclude expired data
        if bucket =~ /_net[0-9]$/ then
          if key.to_i / 1000 < C_EXPIRE
            # log to 
            $log.info "expired data : " + bucket + "/" + key
            next
          end
        end
  

        # check if exist before restore key-values.
        begin
          riak[bucket][key]
        rescue Riak::HTTPFailedRequest
          # throw to riak
          $log.info "throw to riak : " + bucket + "/" + key

          ## create set of Riak::Link
          links = value.find_array("Links")[0]

          ss = Set.new  
          links.each do |s|
            if BERT::Tuple === s then
              ss<< Riak::Link.new(s[0][0],s[0][1],s[1])
            end
          end

          ## Store to riak.
          begin
            ob = riak.bucket(bucket)
            o = ob.get_or_new(key)
            o.raw_data = value.last
            o.content_type = "application/json"
            o.links = ss
            o.store
          rescue => e
            ## failed key name
            $log.error "store failed  : " + bucket + "/" + key
            ## logging backtrace
            $log.error e.exception
          end
        else
          ## store skiped
          $log.info "skip exist key: " + bucket + "/" + key
        end
      end
    end
  end
end


Dir.entries(@bits_root).each do |bit_dir|
  ### skip . and ..
  next if bit_dir =~ /^\./

  queue << bit_dir
end


queue << STOP

# 全てのthreadをmain threadに合流させる（終わるのを待つ）
threads.map(&:join)

{'fraction_non_alphanumeric': 0.13513079762477934, 'fraction_numerical': 0.015727812550152463, 'mean_word_length': 1.837887067395264, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, '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': '7078543', 'n_tokens_mistral': 2285, 'n_tokens_neox': 2177, 'n_words': 522}