<commit_before># TODO: THREAD POOL
#
# 1) Exception handling in threads, what's the right thing to do
# 2) General count of failed records in a thread safe way, so we can report
#    it back from 'close', so process can report it back, and non-zero exit
#    code can be emited from command-line.
# 3) back pressure on thread pool. give it a bounded blocking queue instead,
#    to make sure thousands of add tasks don't build up, waiting until the end.
#    or does that even matter? So what if they build up in the queue and only
#    get taken care of at the end, is that okay? I do emit a warning right now
#    if it takes more than 60 seconds to process remaining thread pool task queue
#    at end.
# 4) No tests yet that actually test thread pool stuff; additionally, may make
#    some of the batch tests fail in non-deterministic ways, since batch tests
#    assume order of add (and our Mock solr server is not thread safe yet!)

require 'yell'

require 'traject'
require 'traject/util'
require 'traject/qualified_const_get'
require 'traject/thread_pool'

require 'uri'
require 'thread' # for Mutex

#
# Writes to a Solr using SolrJ, and the SolrJ HttpSolrServer.
#  (sub-class later for the ConcurrentUpdate server?)
#
# After you call #close, you can check #skipped_record_count if you want
# for an integer count of skipped records.
#
# For fatal errors that raise... async processing with thread_pool means that
# you may not get a raise immediately after calling #put, you may get it on
# a FUTURE #put or #close. You should get it eventually though.
#
# settings:
#   [solr.url] Your solr url (required)
#   [solrj_writer.server_class_name]  Defaults to "HttpSolrServer". You can specify
#                                   another Solr Server sub-class, but it has
#                                   to take a one-arg url constructor. Maybe
#                                   subclass this writer class and overwrite
#                                   instantiate_solr_server! otherwise
#   [solrj.jar_dir] Custom directory containing all of the SolrJ jars. All
#                   jars in this dir will be loaded. Otherwise,
#                   we load our own packaged solrj jars. This setting
#                   can't really be used differently in the same app instance,
#                   since jars are loaded globally.
#   [solrj_writer.parser_class_name] A String name of a class in package
#                                    org.apache.solr.client.solrj.impl,
#                                    we'll instantiate one with a zero-arg
#                                    constructor, and pass it as an arg to setParser on
#                                    the SolrServer instance, if present.
#                                    NOTE: For contacting a Solr 1.x server, with the
#                                    recent version of SolrJ used by default, set to
#                                    "XMLResponseParser"
#   [solrj_writer.commit_on_close]  If true (or string 'true'), send a commit to solr
#                                   at end of #process.
#   [solrj_writer.batch_size]       If non-nil and more than 1, send documents to
#                                   solr in batches of solrj_writer.batch_size. If nil/1,
#                                   however, an http transaction with solr will be done
#                                   per doc. DEFAULT to 100, which seems to be a sweet spot.
#   [solrj_writer.thread_pool]      Defaults to 4. A thread pool is used for submitting docs
#                                   to solr. Set to 0 or nil to disable threading. Set to 1,
#                                   there will still be a single bg thread doing the adds.
#                                   May make sense to set higher than number of cores on your
#                                   indexing machine, as these threads will mostly be waiting
#                                   on Solr. Speed/capacity of your solr is more relevant.
class Traject::SolrJWriter
  # just a tuple of a SolrInputDocument
  # and a Traject::Indexer::Context it came from
  class UpdatePackage
    attr_accessor :solr_document, :context
    def initialize(doc, ctx)
      self.solr_document = doc
      self.context = ctx
    end
  end

  include Traject::QualifiedConstGet

  attr_reader :settings

  attr_reader :batched_queue

  def initialize(argSettings)
    @settings = Traject::Indexer::Settings.new(argSettings)
    settings_check!(settings)

    ensure_solrj_loaded!

    solr_server # init

    @batched_queue = java.util.concurrent.LinkedBlockingQueue.new

    # when multi-threaded exceptions raised in threads are held here
    # we need a HIGH performance queue here to try and avoid slowing things down,
    # since we need to check it frequently.
    @async_exception_queue = java.util.concurrent.ConcurrentLinkedQueue.new

    # Store error count in an AtomicInteger, so multi threads can increment
    # it safely, if we're threaded.
    @skipped_record_incrementer = java.util.concurrent.atomic.AtomicInteger.new(0)

    # if our thread pool settings are 0, it'll just create a null threadpool that
    # executes in calling context.
    @thread_pool = Traject::ThreadPool.new( @settings["solrj_writer.thread_pool"].to_i )

    @debug_ascii_progress = (@settings["debug_ascii_progress"].to_s == "true")

    logger.info("   SolrJWriter writing to '#{settings['solr.url']}'")
  end

  # Loads solrj if not already loaded. By loading all jars found
  # in settings["solrj.jar_dir"]
  def ensure_solrj_loaded!
    unless defined?(HttpSolrServer) && defined?(SolrInputDocument)
      Traject::Util.require_solrj_jars(settings)
    end

    # And for now, SILENCE SolrJ logging
    org.apache.log4j.Logger.getRootLogger().addAppender(org.apache.log4j.varia.NullAppender.new)
  end

  # Method IS thread-safe, can be called concurrently by multi-threads.
  #
  # Why? If not using batched add, we just use the SolrServer, which is already
  # thread safe itself.
  #
  # If we are using batch add, we surround all access to our shared state batch queue
  # in a mutex -- just a naive implementation. May be able to improve performance<commit_msg>SolrJWriter logging, complete class name<commit_after># TODO: THREAD POOL
#
# 1) Exception handling in threads, what's the right thing to do
# 2) General count of failed records in a thread safe way, so we can report
#    it back from 'close', so process can report it back, and non-zero exit
#    code can be emited from command-line.
# 3) back pressure on thread pool. give it a bounded blocking queue instead,
#    to make sure thousands of add tasks don't build up, waiting until the end.
#    or does that even matter? So what if they build up in the queue and only
#    get taken care of at the end, is that okay? I do emit a warning right now
#    if it takes more than 60 seconds to process remaining thread pool task queue
#    at end.
# 4) No tests yet that actually test thread pool stuff; additionally, may make
#    some of the batch tests fail in non-deterministic ways, since batch tests
#    assume order of add (and our Mock solr server is not thread safe yet!)

require 'yell'

require 'traject'
require 'traject/util'
require 'traject/qualified_const_get'
require 'traject/thread_pool'

require 'uri'
require 'thread' # for Mutex

#
# Writes to a Solr using SolrJ, and the SolrJ HttpSolrServer.
#  (sub-class later for the ConcurrentUpdate server?)
#
# After you call #close, you can check #skipped_record_count if you want
# for an integer count of skipped records.
#
# For fatal errors that raise... async processing with thread_pool means that
# you may not get a raise immediately after calling #put, you may get it on
# a FUTURE #put or #close. You should get it eventually though.
#
# settings:
#   [solr.url] Your solr url (required)
#   [solrj_writer.server_class_name]  Defaults to "HttpSolrServer". You can specify
#                                   another Solr Server sub-class, but it has
#                                   to take a one-arg url constructor. Maybe
#                                   subclass this writer class and overwrite
#                                   instantiate_solr_server! otherwise
#   [solrj.jar_dir] Custom directory containing all of the SolrJ jars. All
#                   jars in this dir will be loaded. Otherwise,
#                   we load our own packaged solrj jars. This setting
#                   can't really be used differently in the same app instance,
#                   since jars are loaded globally.
#   [solrj_writer.parser_class_name] A String name of a class in package
#                                    org.apache.solr.client.solrj.impl,
#                                    we'll instantiate one with a zero-arg
#                                    constructor, and pass it as an arg to setParser on
#                                    the SolrServer instance, if present.
#                                    NOTE: For contacting a Solr 1.x server, with the
#                                    recent version of SolrJ used by default, set to
#                                    "XMLResponseParser"
#   [solrj_writer.commit_on_close]  If true (or string 'true'), send a commit to solr
#                                   at end of #process.
#   [solrj_writer.batch_size]       If non-nil and more than 1, send documents to
#                                   solr in batches of solrj_writer.batch_size. If nil/1,
#                                   however, an http transaction with solr will be done
#                                   per doc. DEFAULT to 100, which seems to be a sweet spot.
#   [solrj_writer.thread_pool]      Defaults to 4. A thread pool is used for submitting docs
#                                   to solr. Set to 0 or nil to disable threading. Set to 1,
#                                   there will still be a single bg thread doing the adds.
#                                   May make sense to set higher than number of cores on your
#                                   indexing machine, as these threads will mostly be waiting
#                                   on Solr. Speed/capacity of your solr is more relevant.
class Traject::SolrJWriter
  # just a tuple of a SolrInputDocument
  # and a Traject::Indexer::Context it came from
  class UpdatePackage
    attr_accessor :solr_document, :context
    def initialize(doc, ctx)
      self.solr_document = doc
      self.context = ctx
    end
  end

  include Traject::QualifiedConstGet

  attr_reader :settings

  attr_reader :batched_queue

  def initialize(argSettings)
    @settings = Traject::Indexer::Settings.new(argSettings)
    settings_check!(settings)

    ensure_solrj_loaded!

    solr_server # init

    @batched_queue = java.util.concurrent.LinkedBlockingQueue.new

    # when multi-threaded exceptions raised in threads are held here
    # we need a HIGH performance queue here to try and avoid slowing things down,
    # since we need to check it frequently.
    @async_exception_queue = java.util.concurrent.ConcurrentLinkedQueue.new

    # Store error count in an AtomicInteger, so multi threads can increment
    # it safely, if we're threaded.
    @skipped_record_incrementer = java.util.concurrent.atomic.AtomicInteger.new(0)

    # if our thread pool settings are 0, it'll just create a null threadpool that
    # executes in calling context.
    @thread_pool = Traject::ThreadPool.new( @settings["solrj_writer.thread_pool"].to_i )

    @debug_ascii_progress = (@settings["debug_ascii_progress"].to_s == "true")

    logger.info("   #{self.class.name} writing to '#{settings['solr.url']}'")
  end

  # Loads solrj if not already loaded. By loading all jars found
  # in settings["solrj.jar_dir"]
  def ensure_solrj_loaded!
    unless defined?(HttpSolrServer) && defined?(SolrInputDocument)
      Traject::Util.require_solrj_jars(settings)
    end

    # And for now, SILENCE SolrJ logging
    org.apache.log4j.Logger.getRootLogger().addAppender(org.apache.log4j.varia.NullAppender.new)
  end

  # Method IS thread-safe, can be called concurrently by multi-threads.
  #
  # Why? If not using batched add, we just use the SolrServer, which is already
  # thread safe itself.
  #
  # If we are using batch add, we surround all access to our shared state batch queue
  # in a mutex -- just a naive implementation. May be able to improve performance
{'fraction_non_alphanumeric': 0.0626103708460427, 'fraction_numerical': 0.00305024883608926, 'mean_word_length': 2.3021468327590777, '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': 46, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2983540', 'n_tokens_mistral': 3467, 'n_tokens_neox': 3286, 'n_words': 1411}