<commit_before>#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

from __future__ import absolute_import
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import with_statement

import numpy as _numpy
import os as _os
import subprocess as _subprocess
import sys as _sys
import tempfile as _tempfile
import threading as _threading
import time as _time
import traceback as _traceback

class QuiverCommand(object):
    def __init__(self, home_dir, output_dir, impl, mode, address,
                 operation, messages, bytes_, credit, timeout):
        self.home_dir = home_dir
        self.output_dir = output_dir
        self.impl = impl
        self.mode = mode
        self.address = address
        self.operation = operation
        self.messages = messages
        self.bytes_ = bytes_
        self.credit = credit
        self.timeout = timeout

        self.quiet = False
        self.debug = False

        impl_name = "quiver-{}".format(self.impl)

        if self.output_dir is None:
            self.output_dir = _tempfile.mkdtemp(prefix="quiver-")
            
        self.impl_file = _os.path.join(self.home_dir, "exec", impl_name)
        self.transfers_file = _os.path.join(self.output_dir, "transfers.csv")

        self.start_time = None
        self.end_time = None
        
        self.started = _threading.Event()
        self.stop = _threading.Event()
        self.ended = _threading.Event()

        self.periodic_status_thread = _PeriodicStatusThread(self)

    def init(self):
        if not _os.path.exists(self.output_dir):
            _os.makedirs(self.output_dir)
        
    def check(self):
        if not _os.path.exists(self.impl_file):
            msg = "No impl at '{}'".format(self.impl_file)
            raise QuiverError(msg)

        if not _os.path.isdir(self.output_dir):
            msg = "Invalid output dir at '{}'".format(self.output_dir)
            raise QuiverError(msg)

    def run(self):
        if self.address.startswith("//"):
            domain, path = self.address[2:].split("/", 1)
        else:
            domain, path = "localhost", self.address

        args = [
            self.impl_file,
            self.output_dir,
            self.mode,
            domain,
            path,
            self.operation,
            str(self.messages),
            str(self.bytes_),
            str(self.credit),
        ]

        self.dprint("Calling '{}'", " ".join(args))

        if not self.quiet:
            self.print_config()

        if self.operation == "receive":
            self.periodic_status_thread.start()
        
        with open(self.transfers_file, "w") as fout:
            self.started.set()
            self.start_time = _time.time()

            proc = _subprocess.Popen(args, stdout=fout)

            while proc.poll() == None:
                if self.stop.wait(1):
                    proc.terminate()

            if proc.returncode != 0:
                msg = "Subprocess {} exited with code {}".format \
                      (proc.pid, proc.returncode)
                raise QuiverError(msg)
                    
            self.end_time = _time.time()
            self.ended.set()

        if self.operation == "receive":<commit_msg>Use a shorter check interval<commit_after>#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

from __future__ import absolute_import
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import with_statement

import numpy as _numpy
import os as _os
import subprocess as _subprocess
import sys as _sys
import tempfile as _tempfile
import threading as _threading
import time as _time
import traceback as _traceback

class QuiverCommand(object):
    def __init__(self, home_dir, output_dir, impl, mode, address,
                 operation, messages, bytes_, credit, timeout):
        self.home_dir = home_dir
        self.output_dir = output_dir
        self.impl = impl
        self.mode = mode
        self.address = address
        self.operation = operation
        self.messages = messages
        self.bytes_ = bytes_
        self.credit = credit
        self.timeout = timeout

        self.quiet = False
        self.debug = False

        impl_name = "quiver-{}".format(self.impl)

        if self.output_dir is None:
            self.output_dir = _tempfile.mkdtemp(prefix="quiver-")
            
        self.impl_file = _os.path.join(self.home_dir, "exec", impl_name)
        self.transfers_file = _os.path.join(self.output_dir, "transfers.csv")

        self.start_time = None
        self.end_time = None
        
        self.started = _threading.Event()
        self.stop = _threading.Event()
        self.ended = _threading.Event()

        self.periodic_status_thread = _PeriodicStatusThread(self)

    def init(self):
        if not _os.path.exists(self.output_dir):
            _os.makedirs(self.output_dir)
        
    def check(self):
        if not _os.path.exists(self.impl_file):
            msg = "No impl at '{}'".format(self.impl_file)
            raise QuiverError(msg)

        if not _os.path.isdir(self.output_dir):
            msg = "Invalid output dir at '{}'".format(self.output_dir)
            raise QuiverError(msg)

    def run(self):
        if self.address.startswith("//"):
            domain, path = self.address[2:].split("/", 1)
        else:
            domain, path = "localhost", self.address

        args = [
            self.impl_file,
            self.output_dir,
            self.mode,
            domain,
            path,
            self.operation,
            str(self.messages),
            str(self.bytes_),
            str(self.credit),
        ]

        self.dprint("Calling '{}'", " ".join(args))

        if not self.quiet:
            self.print_config()

        if self.operation == "receive":
            self.periodic_status_thread.start()
        
        with open(self.transfers_file, "w") as fout:
            self.started.set()
            self.start_time = _time.time()

            proc = _subprocess.Popen(args, stdout=fout)

            while proc.poll() == None:
                if self.stop.wait(0.1):
                    proc.terminate()

            if proc.returncode != 0:
                msg = "Subprocess {} exited with code {}".format \
                      (proc.pid, proc.returncode)
                raise QuiverError(msg)
                    
            self.end_time = _time.time()
            self.ended.set()

        if self.operation == "receive":
{'fraction_non_alphanumeric': 0.09156596486856858, 'fraction_numerical': 0.00211785224866077, 'mean_word_length': 2.1188811188811187, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1498151', 'n_tokens_mistral': 2291, 'n_tokens_neox': 2212, 'n_words': 651}