<commit_before>require 'rubygems'
require 'bud'
require 'backports'
require 'timeout'
require 'bfs/bfs_client_proto'
require 'bfs/data_protocol'

# The BFS client and shell stand between ruby and BUD.  BSFShell provides dispatch_command() as a synchronous functional interface
# for FS operations


module BFSClient
  include BFSClientProtocol

  state {
    table :master, [] => [:master]
  }

  declare
  def cglue
    # every request involves some communication with the master.
    #stdio <~ request.map{|r| ["REQUEST: #{r.inspect}"] }
    #stdio <~ response.map{|r| ["response: #{r.inspect}"] }
    request_msg <~ join([request, master]).map{|r, m| [m.master, ip_port, r.reqid, r.rtype, r.arg] }
  
    response <= response_msg.map do |r|
      [r.reqid, r.status, r.response]
    end 
  end
end

class BFSShell
  include Bud
  include BFSClient
  
 def initialize(master)
    @master = master
    @queue = Queue.new
    # bootstrap?

    super(:dump => true)
    my_queue << [@queue]
  end

  
  state {
    table :remember_response, response.schema
    table :watched_ids, [:reqid]
    table :my_queue, [] => [:queue]
    scratch :hollow, [:straw]
  }

  declare
  def synchronization
    remember_response <= response
    snc = join [remember_response, watched_ids, my_queue], [remember_response.reqid, watched_ids.reqid]
    hollow <= snc.map do |r, w, q|
      [q.queue.push r]
    end
    watched_ids <- snc.map{ |r, w| w }
    remember_response <- snc.map{ |r, w| r }
  end
  
  bootstrap do
    master << [@master]
  end

  def dispatch_command(args, filehandle=nil)
    op = args.shift
    case op
      when "ls" then
        do_ls(args)
      when "create"
        do_createfile(args)
      when "append"
        do_append(args, (filehandle.nil? ? STDIN : filehandle))
      when "mkdir"
        do_mkdir(args)
      when "read"
        do_read(args, (filehandle.nil? ? STDOUT : filehandle))
      when "rm"
        do_rm(args)
      else
        raise "unknown op: #{op}"
    end
  end
  
  def get_base_and_path(path)
    items = path.split("/")
    return [items.pop, items.length == 1 ? "/" : items.join("/")]
  end 

  def synchronous_request(op, args)
    reqid = 1 + rand(10000000)
    async_do{ request <+ [[reqid, op, args]] }
    return timed_sync(reqid)
  end 

  def do_createfile(args)
    do_create(args, false)
  end

  def do_create(args, is_dir)
    (file, path) = get_base_and_path(args[0])
    if is_dir
      synchronous_request(:mkdir, [file, path])
    else 
      synchronous_request(:create, [file, path])
    end
  end

  def do_mkdir(args)
    do_create(args, true)
  end

  def do_rm(args)
    (file, path) = get_base_and_path(args[0])
    synchronous_request(:rm, [file, path]) 
  end

  def do_read(args, fh)
    res = synchronous_request(:getchunks, args[0])
    res.response.sort{|a, b| a <=> b}.each do |chk|
      res = synchronous_request(:getchunklocations, chk)
      chunk = DataProtocolClient.read_chunk(chk, res[2])
      fh.write chunk
    end
  end

  def do_append(args, fh)
    ret = true
    while ret
      ret = do_a_chunk(args, fh)
    end
  end
  
  def do_a_chunk(args, fh)
    ret = synchronous_request(:append, args[0])
    raise "add chunk metadata failed: #{ret.inspect}" unless ret[1]
    chunkid = ret[2][0]
    preflist = ret[2][1]
    sendlist = preflist.sort_by{rand}[0..REP_FACTOR-1]
    DataProtocolClient.send_stream(chunkid, sendlist, DataProtocolClient.chunk_from_fh(fh))
  end

  def timed_sync(reqid)
    async_do { watched_ids <+ [[reqid]] }
    res = nil
    Timeout::timeout(5) do
      res = @queue.pop
    end
    # the bud code frequenty (but not always!) enqueues items 2X
    if res.reqid != reqid
      res = timed_sync(reqid)
    end
    return res
  end

  def do_ls(args)
    res = synchronous_request(:ls, args[0])
    return res.response
  end
end<commit_msg>Fix typo.<commit_after>require 'rubygems'
require 'bud'
require 'backports'
require 'timeout'
require 'bfs/bfs_client_proto'
require 'bfs/data_protocol'

# The BFS client and shell stand between ruby and BUD.  BSFShell provides dispatch_command() as a synchronous functional interface
# for FS operations


module BFSClient
  include BFSClientProtocol

  state {
    table :master, [] => [:master]
  }

  declare
  def cglue
    # every request involves some communication with the master.
    #stdio <~ request.map{|r| ["REQUEST: #{r.inspect}"] }
    #stdio <~ response.map{|r| ["response: #{r.inspect}"] }
    request_msg <~ join([request, master]).map{|r, m| [m.master, ip_port, r.reqid, r.rtype, r.arg] }
  
    response <= response_msg.map do |r|
      [r.reqid, r.status, r.response]
    end 
  end
end

class BFSShell
  include Bud
  include BFSClient
  
 def initialize(master)
    @master = master
    @queue = Queue.new
    # bootstrap?

    super(:dump => true)
    my_queue << [@queue]
  end

  
  state {
    table :remember_response, response.schema
    table :watched_ids, [:reqid]
    table :my_queue, [] => [:queue]
    scratch :hollow, [:straw]
  }

  declare
  def synchronization
    remember_response <= response
    snc = join [remember_response, watched_ids, my_queue], [remember_response.reqid, watched_ids.reqid]
    hollow <= snc.map do |r, w, q|
      [q.queue.push r]
    end
    watched_ids <- snc.map{ |r, w| w }
    remember_response <- snc.map{ |r, w| r }
  end
  
  bootstrap do
    master << [@master]
  end

  def dispatch_command(args, filehandle=nil)
    op = args.shift
    case op
      when "ls" then
        do_ls(args)
      when "create"
        do_createfile(args)
      when "append"
        do_append(args, (filehandle.nil? ? STDIN : filehandle))
      when "mkdir"
        do_mkdir(args)
      when "read"
        do_read(args, (filehandle.nil? ? STDOUT : filehandle))
      when "rm"
        do_rm(args)
      else
        raise "unknown op: #{op}"
    end
  end
  
  def get_base_and_path(path)
    items = path.split("/")
    return [items.pop, items.length == 1 ? "/" : items.join("/")]
  end 

  def synchronous_request(op, args)
    reqid = 1 + rand(10000000)
    async_do{ request <+ [[reqid, op, args]] }
    return timed_sync(reqid)
  end 

  def do_createfile(args)
    do_create(args, false)
  end

  def do_create(args, is_dir)
    (file, path) = get_base_and_path(args[0])
    if is_dir
      synchronous_request(:mkdir, [file, path])
    else 
      synchronous_request(:create, [file, path])
    end
  end

  def do_mkdir(args)
    do_create(args, true)
  end

  def do_rm(args)
    (file, path) = get_base_and_path(args[0])
    synchronous_request(:rm, [file, path]) 
  end

  def do_read(args, fh)
    res = synchronous_request(:getchunks, args[0])
    res.response.sort{|a, b| a <=> b}.each do |chk|
      res = synchronous_request(:getchunklocations, chk)
      chunk = DataProtocolClient.read_chunk(chk, res[2])
      fh.write chunk
    end
  end

  def do_append(args, fh)
    ret = true
    while ret
      ret = do_a_chunk(args, fh)
    end
  end
  
  def do_a_chunk(args, fh)
    ret = synchronous_request(:append, args[0])
    raise "add chunk metadata failed: #{ret.inspect}" unless ret[1]
    chunkid = ret[2][0]
    preflist = ret[2][1]
    sendlist = preflist.sort_by{rand}[0..REP_FACTOR-1]
    DataProtocolClient.send_stream(chunkid, sendlist, DataProtocolClient.chunk_from_fh(fh))
  end

  def timed_sync(reqid)
    async_do { watched_ids <+ [[reqid]] }
    res = nil
    Timeout::timeout(5) do
      res = @queue.pop
    end
    # the bud code frequently (but not always!) enqueues items 2X
    if res.reqid != reqid
      res = timed_sync(reqid)
    end
    return res
  end

  def do_ls(args)
    res = synchronous_request(:ls, args[0])
    return res.response
  end
end
{'fraction_non_alphanumeric': 0.12160934458144063, 'fraction_numerical': 0.006489292667099286, 'mean_word_length': 2.8938858009095503, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '4692875', 'n_tokens_mistral': 2844, 'n_tokens_neox': 2723, 'n_words': 688}