<commit_before>import logging
import time
import threading
from importlib import import_module


class JobWorker( threading.Thread ):
  def __init__( self, contractor, cookie, job_id, function, paramaters, semaphore ):
    super().__init__()
    self.contractor = contractor
    self.cookie = cookie
    self.job_id = job_id
    self.function = function
    self.paramaters = paramaters
    self.semaphore = semaphore

  def run( self ):
    try:
      logging.debug( 'handler: acquring lock for "{0}"...'.format( self.job_id ) )
      self.semaphore.acquire()
      logging.debug( 'handler: lock acquired for "{0}"...'.format( self.job_id ) )
      logging.debug( 'handler: starting job "{0}" with "{1}"'.format( self.function, self.paramaters ) )

      try:
        data = self.function( self.paramaters )
      except Exception as e:
        logging.exception( 'handler: Exception with function "{0}" paramaters "{1}"'.format( self.function, self.paramaters ) )
        self.contractor.jobError( self.job_id, 'Unhandled Exception "{0}"({1})'.format( e, type( e ).__name__ ), self.cookie )
        return

    finally:
      self.semaphore.release()
      logging.debug( 'handler: log for "{0}" released'.format( self.job_id ) )

    if not isinstance( data, dict ):
      logging.error( 'handler: result from function was not a dict, got "{0}"({1})'.format( str( data )[ 0:50 ], type( data ).__name__ ) )
      self.contractor.jobError( self.job_id, 'result was not a dict, got "{0}"({1})'.format( data, type( data ).__name__ ), self.cookie )

    logging.debug( 'handler: results of "{0}" with "{1}" is "{2}"'.format( self.function, self.paramaters, data ) )
    response = self.contractor.jobResults( self.job_id, data, self.cookie )  # this is after releasing the semaphore so we are not holding things up if it requires retries
    logging.info( 'handler: job "{0}" complete, contractor said "{1}"'.format( self.job_id, response ) )


class Handler():
  def __init__( self, contractor ):
    super().__init__()
    self.contractor = contractor
    self.max_concurent_jobs = 0  # there is not a semaphore to inforce this, this is used to limit the number of jobs being requested<commit_msg>handle non Accepted responses from Contractor<commit_after>import logging
import time
import threading
from importlib import import_module


class JobWorker( threading.Thread ):
  def __init__( self, contractor, cookie, job_id, function, paramaters, semaphore ):
    super().__init__()
    self.contractor = contractor
    self.cookie = cookie
    self.job_id = job_id
    self.function = function
    self.paramaters = paramaters
    self.semaphore = semaphore

  def run( self ):
    try:
      logging.debug( 'handler: acquring lock for "{0}"...'.format( self.job_id ) )
      self.semaphore.acquire()
      logging.debug( 'handler: lock acquired for "{0}"...'.format( self.job_id ) )
      logging.debug( 'handler: starting job "{0}" with "{1}"'.format( self.function, self.paramaters ) )

      try:
        data = self.function( self.paramaters )
      except Exception as e:
        logging.exception( 'handler: Exception with function "{0}" paramaters "{1}"'.format( self.function, self.paramaters ) )
        self.contractor.jobError( self.job_id, 'Unhandled Exception "{0}"({1})'.format( e, type( e ).__name__ ), self.cookie )
        return

    finally:
      self.semaphore.release()
      logging.debug( 'handler: log for "{0}" released'.format( self.job_id ) )

    if not isinstance( data, dict ):
      logging.error( 'handler: result from function was not a dict, got "{0}"({1})'.format( str( data )[ 0:50 ], type( data ).__name__ ) )
      self.contractor.jobError( self.job_id, 'result was not a dict, got "{0}"({1})'.format( data, type( data ).__name__ ), self.cookie )

    logging.debug( 'handler: results of "{0}" with "{1}" is "{2}"'.format( self.function, self.paramaters, data ) )
    response = 'Error'
    while response == 'Error':
      response = self.contractor.jobResults( self.job_id, data, self.cookie )  # this is after releasing the semaphore so we are not holding things up if it requires retries
      logging.info( 'handler: job "{0}" complete, contractor said "{1}"'.format( self.job_id, response ) )
      if response == 'Accepted':
        break

      if response != 'Error':
        raise Exception( 'Unknown jobResults response "{0}"'.format( response ) )

      logging.debug( 'handler: Contractor said it had an error, sleeping before trying again...' )
      time.sleep( 60 )


class Handler():
  def __init__( self, contractor ):
    super().__init__()
    self.contractor = contractor
    self.max_concurent_jobs = 0  # there is not a semaphore to inforce this, this is used to limit the number of jobs being requested
{'fraction_non_alphanumeric': 0.1329976924690581, 'fraction_numerical': 0.009859450388084749, 'mean_word_length': 3.3863845446182155, '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': '5029806', 'n_tokens_mistral': 1451, 'n_tokens_neox': 1416, 'n_words': 459}