<commit_before>from nose.tools import *
import rpyc
from multiprocessing import Process
import time
from os.path import abspath, dirname, join
from shellstreaming.config import Config
from shellstreaming.comm.inputstream import InputStreamDispatcher, InputStreamExecutorService


TEST_CONFIG   = join(abspath(dirname(__file__)), '..', 'data', 'shellstreaming.cnf')
TEST_TEXTFILE = join(abspath(dirname(__file__)), '..', 'data', 'comm_inputstream_input01.txt')


process = None


def _start_worker_thread():
    from rpyc.utils.server import ThreadPoolServer as Server
    Server(InputStreamExecutorService, port=int(Config.instance().get('worker', 'port'))).start()


def setup():
    # setting up worker
    global process
    process = Process(target=_start_worker_thread)
    process.daemon = True  # child process is killed when parent process ends
    process.start()
    # wait for worker process to really start
    while True:
        try:
            conn = rpyc.connect('localhost', int(Config.instance().get('worker', 'port')))
            conn.close()
            break
        except:  # connection refused
            time.sleep(0.1)
            continue
    print('worker server has been started')


def teardown():
    print('worker process is being killed')


def test_inputstream_dispatcher():
    # master's code
    stream = InputStreamDispatcher(
        'localhost',  # Config.instance().get('worker', 'worker1')
        'TextFile',
        (TEST_TEXTFILE, 20),
    )

    # do everything master needs to do

    stream.join()<commit_msg>correctly close worker server Signed-off-by: <NAME> <<EMAIL>><commit_after>from nose.tools import *
import rpyc
from multiprocessing import Process
import time
import signal
from os import kill
from os.path import abspath, dirname, join
from shellstreaming.config import Config
from shellstreaming.comm.inputstream import InputStreamDispatcher, InputStreamExecutorService


TEST_CONFIG   = join(abspath(dirname(__file__)), '..', 'data', 'shellstreaming.cnf')
TEST_TEXTFILE = join(abspath(dirname(__file__)), '..', 'data', 'comm_inputstream_input01.txt')


process = None  # used by master process
server  = None  # used by worker process


def _sigusr1_handler(signum, stack):
    # close worker server
    global server
    server.close()
    exit(0)


def _start_worker_process():
    # register SIGUSR1 handler
    import signal
    signal.signal(signal.SIGUSR1, _sigusr1_handler)

    # start worker server
    global server
    from rpyc.utils.server import ThreadedServer as Server
    server = Server(InputStreamExecutorService, port=int(Config.instance().get('worker', 'port')))
    server.start()


def setup():
    # setting up worker
    global process
    process = Process(target=_start_worker_process)
    process.start()
    # wait for worker process to really start
    while True:
        try:
            conn = rpyc.connect('localhost', int(Config.instance().get('worker', 'port')))
            conn.close()
            break
        except:  # connection refused
            time.sleep(0.1)
            continue
    print('worker server has been started')


def teardown():
    print('worker process is being killed')
    kill(process.pid, signal.SIGUSR1)


def test_inputstream_dispatcher():
    # master's code
    stream = InputStreamDispatcher(
        'localhost',  # Config.instance().get('worker', 'worker1')
        'TextFile',
        (TEST_TEXTFILE, 20),
    )

    # do everything master needs to do

    stream.join()
{'fraction_non_alphanumeric': 0.1043032202906811, 'fraction_numerical': 0.005699629524080934, 'mean_word_length': 3.45997458703939, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, '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': '4143162', 'n_tokens_mistral': 1013, 'n_tokens_neox': 984, 'n_words': 275}