<commit_before>#!/usr/bin/env ruby
$:<< './lib'

require 'boot'

require 'goliath'
require 'em-synchrony/em-http'

class Simple < Goliath::API
  use Goliath::Rack::Params             # parse query & body params
  # plugin Goliath::Plugin::Latency       # ask eventmachine reactor to track its latency

  QUERIES = [ 1.0, 14.5, 2.5, 0.5, 4.0, 0.25, 5.0, 9.5, 2.5, 0.5, 2.3, 2.5 ]
  TARGET_CONCURRENCY   = QUERIES.length

  def response(env)
    batch_id = "%7.04f" % (env[:start_time].to_f - 100*(env[:start_time].to_f.to_i / 100))

    env.logger.debug "req #{object_id} @#{batch_id}: constructing request group"
    BatchIterator.new(env, batch_id, QUERIES.each_with_index.to_a, TARGET_CONCURRENCY).perform
    env.logger.debug "req #{object_id} @#{batch_id}: constructed request group"
    chunked_streaming_response(200, {'X-Responder' => self.class.to_s, })
  end
end


class BatchIterator < EM::Synchrony::Iterator

  TARGET_URL_BASE = "http://localhost:9002/meta/http/sleepy.json"
  HTTP_REQUEST_OPTIONS = { :connect_timeout => 1.0, :inactivity_timeout => 1.2 }

  attr_reader :requests, :responses

  def initialize env, batch_id, *args
    @env       = env
    @batch_id  = batch_id
    @requests  = []
    @responses = {:results => {}, :errors => {}}
    super *args
  end

  def handle_result req_id, req
    @env.chunked_stream_send([req_id, "\t", req.response, "\n"].join)
  end

  def handle_error req_id, req
    @env.chunked_stream_send([req_id, "\t", %Q{{"error":'#{req.error}'}}, "\n"].join)
  end

  def logline indent, seg
    @env.logger.debug( "req %s @%s: %s%s" % [object_id, @batch_id, " "*indent, seg] )
  end

  def perform
    EM.synchrony do
      logline 0, "synchrony start"

      each(
        proc{|(delay, req_id), iter|
          req = EM::HttpRequest.new(TARGET_URL_BASE, HTTP_REQUEST_OPTIONS).aget(:query => { :delay => delay })
          req.callback{ handle_result(req_id, req) ; iter.next ; logline(4, "request #{[req_id, delay]}\tsuccess") }
          req.errback{  handle_error(req_id, req)  ; iter.next ; logline(4, "request #{[req_id, delay]}\terror")   }

          logline 4, "request #{[req_id, delay]}\tconstructed"

        }, proc{
          EM.next_tick do
            logline 2, "stream closing"
            @env.chunked_stream_close
            end
        }
        )
      logline 0, "synchrony end"
    end
  end

end

<commit_msg>oops pulled some debugging voodoo out that wan't supposed to be there<commit_after>#!/usr/bin/env ruby
$:<< './lib'

require 'boot'

require 'goliath'
require 'em-synchrony/em-http'

class Simple < Goliath::API
  use Goliath::Rack::Params             # parse query & body params
  # plugin Goliath::Plugin::Latency       # ask eventmachine reactor to track its latency

  QUERIES = [ 1.0, 14.5, 2.5, 0.5, 4.0, 0.25, 5.0, 9.5, 2.5, 0.5, 2.3, 2.5 ]
  TARGET_CONCURRENCY   = QUERIES.length

  def response(env)
    batch_id = "%7.04f" % (env[:start_time].to_f - 100*(env[:start_time].to_f.to_i / 100))

    env.logger.debug "req #{object_id} @#{batch_id}: constructing request group"
    BatchIterator.new(env, batch_id, QUERIES.each_with_index.to_a, TARGET_CONCURRENCY).perform
    env.logger.debug "req #{object_id} @#{batch_id}: constructed request group"
    chunked_streaming_response(200, {'X-Responder' => self.class.to_s, })
  end
end


class BatchIterator < EM::Synchrony::Iterator

  TARGET_URL_BASE = "http://localhost:9002/meta/http/sleepy.json"
  HTTP_REQUEST_OPTIONS = { :connect_timeout => 1.0, :inactivity_timeout => 1.2 }

  attr_reader :requests, :responses

  def initialize env, batch_id, *args
    @env       = env
    @batch_id  = batch_id
    @requests  = []
    @responses = {:results => {}, :errors => {}}
    super *args
  end

  def handle_result req_id, req
    @env.chunked_stream_send([req_id, "\t", req.response_header.http_status, "\t", req.response, "\n"].join)
  end

  def handle_error req_id, req
    @env.chunked_stream_send([req_id, "\t", "", "\t", %Q{{"error":'#{req.error}'}}, "\n"].join)
  end

  def logline indent, seg
    @env.logger.debug( "req %s @%s: %s%s" % [object_id, @batch_id, " "*indent, seg] )
  end

  def perform
    EM.synchrony do
      logline 0, "synchrony start"

      each(
        proc{|(delay, req_id), iter|
          req = EM::HttpRequest.new(TARGET_URL_BASE, HTTP_REQUEST_OPTIONS).aget(:query => { :delay => delay })
          req.callback{ handle_result(req_id, req) ; iter.next ; logline(4, "request #{[req_id, delay]}\tsuccess") }
          req.errback{  handle_error(req_id, req)  ; iter.next ; logline(4, "request #{[req_id, delay]}\terror")   }

          logline 4, "request #{[req_id, delay]}\tconstructed"
        }, proc{
          logline 2, "stream closing"
          @env.chunked_stream_close
        }
        )
      logline 0, "synchrony end"
    end
  end

end


{'fraction_non_alphanumeric': 0.15403726708074533, 'fraction_numerical': 0.021532091097308487, 'mean_word_length': 3.1184995737425405, 'pattern_counts': {'":': 2, '<': 11, '<?xml version=': 0, '>': 17, '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': '6452708', 'n_tokens_mistral': 1882, 'n_tokens_neox': 1759, 'n_words': 385}