<commit_before>require 'eventmachine'

module PushyClient
  module Handler
    class Heartbeat

      attr_reader :received
      attr_accessor :monitor

      def initialize(monitor, client)
        @monitor = monitor
        @client = client
      end

      def on_readable(socket, parts)
        unless valid?(parts)
          PushyClient::Log.error "[#{node_name}] Received Invalid Heartbeat : #{parts[1].copy_out_string}}"
          return
        end

        data = Utils.parse_json(parts[1].copy_out_string)
        monitor.checkin!(data)
      end

      private

      def valid?(parts)
        Utils.valid?(parts, @client.server_public_key, @client.session_key)
      end

    end

    class Command

      attr_reader :worker

      def initialize(worker)
        @worker = worker
      end

      def node_name
        worker.node_name
      end

      def on_readable(socket, parts)
        unless valid?(parts)
          PushyClient::Log.error "[#{node_name}] Received Invalid Message : #{parts[1].copy_out_string}}"
          return
        end
        command_hash = Utils.parse_json(parts[1].copy_out_string)

        PushyClient::Log.debug "[#{node_name}] Received command #{command_hash}"
        if command_hash['type'] == "commit"
          ack_nack(command_hash['job_id'], command_hash['command'])
        elsif command_hash['type'] == "run"
          run_command(command_hash['job_id'], command_hash['command'])
        elsif command_hash['type'] == "abort"
          abort
        else
          PushyClient::Log.error "[#{node_name}] Received unknown command #{command_hash}"
        end

      end

      private

      def ack_nack(job_id, command)
        # If we are ready or have started this job, do nothing.
        if worker.job.ready?(job_id) || worker.job.running?(job_id)
          PushyClient::Log.warn "Received command request for job #{job_id} after twice: doing nothing."
          worker.send_command(:ack_commit, job_id)

        # If we are idle, ack.
        elsif worker.job.idle?
          worker.send_command(:ack_commit, job_id)
          worker.change_job(JobState.new(job_id, command, :ready))

        # Otherwise, we're involved with some other job.  nack.
        else
          worker.send_command(:nack_commit, job_id)
        end
      end

      def run_command(job_id, command)
        # If we are already running this job, do nothing.
        if worker.job.running?(job_id)
          PushyClient::Log.warn "[#{node_name}] Received execute request for job #{job_id} twice: doing nothing."

        # If we are ready for this job, or are idle, start.
        elsif worker.job.ready?(job_id) || worker.job.idle?
          worker.send_command(:ack_run, job_id)
          worker.change_job(JobState.new(job_id, command, :running))

          worker.job.pid = pid = Process.spawn(command)

          # Wait for the job to complete and close it out.
          Thread.new do
            Process.wait(pid)
            # This runs on the EM thread and handles a potential race condition
            # between complete and aborted by checking whether the job is still
            # active before sending the completed response
            EM.schedule do
              if worker.job.running?(job_id)
                worker.send_command(:complete, job_id)
                worker.clear_job
              end
            end
          end

        # Otherwise, we're clearly working on another job.  Ignore this request.
        # TODO perhaps remind the server of our state with respect to this job.
        else
          worker.send_command(:nack_run, job_id)
          PushyClient::Log.warn "[#{node_name}] Received execute request for job #{job_id} when we are already #{worker.job}: Doing nothing."
        end
      end

      def abort
        worker.job.cancel
        worker.send_command(:aborted, worker.job.job_id)
        worker.clear_job
      end

      def valid?(parts)
        Utils.valid?(parts, worker.server_public_key, worker.session_key)
      end

    end
<commit_msg>return failed or succeeded based on process exit status<commit_after>require 'eventmachine'

module PushyClient
  module Handler
    class Heartbeat

      attr_reader :received
      attr_accessor :monitor

      def initialize(monitor, client)
        @monitor = monitor
        @client = client
      end

      def on_readable(socket, parts)
        unless valid?(parts)
          PushyClient::Log.error "[#{node_name}] Received Invalid Heartbeat : #{parts[1].copy_out_string}}"
          return
        end

        data = Utils.parse_json(parts[1].copy_out_string)
        monitor.checkin!(data)
      end

      private

      def valid?(parts)
        Utils.valid?(parts, @client.server_public_key, @client.session_key)
      end

    end

    class Command

      attr_reader :worker

      def initialize(worker)
        @worker = worker
      end

      def node_name
        worker.node_name
      end

      def on_readable(socket, parts)
        unless valid?(parts)
          PushyClient::Log.error "[#{node_name}] Received Invalid Message : #{parts[1].copy_out_string}}"
          return
        end
        command_hash = Utils.parse_json(parts[1].copy_out_string)

        PushyClient::Log.debug "[#{node_name}] Received command #{command_hash}"
        if command_hash['type'] == "commit"
          ack_nack(command_hash['job_id'], command_hash['command'])
        elsif command_hash['type'] == "run"
          run_command(command_hash['job_id'], command_hash['command'])
        elsif command_hash['type'] == "abort"
          abort
        else
          PushyClient::Log.error "[#{node_name}] Received unknown command #{command_hash}"
        end

      end

      private

      def ack_nack(job_id, command)
        # If we are ready or have started this job, do nothing.
        if worker.job.ready?(job_id) || worker.job.running?(job_id)
          PushyClient::Log.warn "Received command request for job #{job_id} after twice: doing nothing."
          worker.send_command(:ack_commit, job_id)

        # If we are idle, ack.
        elsif worker.job.idle?
          worker.send_command(:ack_commit, job_id)
          worker.change_job(JobState.new(job_id, command, :ready))

        # Otherwise, we're involved with some other job.  nack.
        else
          worker.send_command(:nack_commit, job_id)
        end
      end

      def run_command(job_id, command)
        # If we are already running this job, do nothing.
        if worker.job.running?(job_id)
          PushyClient::Log.warn "[#{node_name}] Received execute request for job #{job_id} twice: doing nothing."

        # If we are ready for this job, or are idle, start.
        elsif worker.job.ready?(job_id) || worker.job.idle?
          worker.send_command(:ack_run, job_id)
          worker.change_job(JobState.new(job_id, command, :running))

          worker.job.pid = pid = Process.spawn(command)

          # Wait for the job to complete and close it out.
          Thread.new do
            Process.wait(pid)
            exit_status = $?.exitstatus == 0 ? :succeeded : :failed
            # This runs on the EM thread and handles a potential race condition
            # between complete and aborted by checking whether the job is still
            # active before sending the completed response
            EM.schedule do
              if worker.job.running?(job_id)
                worker.send_command(exit_status, job_id)
                worker.clear_job
              end
            end
          end

        # Otherwise, we're clearly working on another job.  Ignore this request.
        # TODO perhaps remind the server of our state with respect to this job.
        else
          worker.send_command(:nack_run, job_id)
          PushyClient::Log.warn "[#{node_name}] Received execute request for job #{job_id} when we are already #{worker.job}: Doing nothing."
        end
      end

      def abort
        worker.job.cancel
        worker.send_command(:aborted, worker.job.job_id)
        worker.clear_job
      end

      def valid?(parts)
        Utils.valid?(parts, worker.server_public_key, worker.session_key)
      end

    end

{'fraction_non_alphanumeric': 0.09096459096459096, 'fraction_numerical': 0.001098901098901099, 'mean_word_length': 2.22353404171586, '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': '4313302', 'n_tokens_mistral': 2361, 'n_tokens_neox': 2321, 'n_words': 639}