<commit_before>from nose.tools import *
import rpyc
from multiprocessing import Process
import time
import os

from shellstreaming.inputstream.textfile import TextFile


class InputStreamStarterService(rpyc.Service):
    class exposed_InputStreamStarter(object):
        def __init__(self, callback, inputstream, args):
            self._callback    = callback
            self._inputstream = inputstream
            self._args        = args

        def exposed_start(self):
            module = __import__('shellstreaming.inputstream.textfile', globals(), locals(), [self._inputstream], -1)
            stream = eval('%s %s' % (self._inputstream, self._args))

            n_records = 0
            for batch in stream:
                for record in batch:
                    eq_(len(record), 1)
                    line = record[0]
                    eq_('line ', line[0:5])
                    ok_(0 <= int(line[5:]) < 100)  # record order in a batch is not always 'oldest-first'
                    n_records += 1
            self._callback(n_records)


process = None


def _start_worker_thread():
    from rpyc.utils.server import ThreadedServer as Server  # can be ThreadedServer or ThreadPoolServer for performance
    print('[pid=%d] strating worker server' % (os.getpid()))
    Server(InputStreamStarterService, port=18871).start()


def setup():
    global process
    # setting up worker
    print('[pid=%d] starting worker process' % (os.getpid()))
    process = Process(target=_start_worker_thread)
    process.start()
    # wait for worker process to start server
    while True:
        try:
            conn = rpyc.connect('localhost', 18871)
            conn.close()
            break
        except:  # connection refused
            time.sleep(0.1)
            continue


def teardown():
    global process
    print('[pid=%d] terminating worker process' % (os.getpid()))
    process.terminate()


bgsrv = conn = None
stop = False


def f(number_of_lines):  # masterの呼び出すcallback
    global bgsrv, conn, stop
    print '[pid=%d] finish!!! (%d lines)' % (os.getpid(), number_of_lines)
    stop = True
    bgsrv.stop()
    conn.close()


def test_jobdispatcher_makes_worker_input_file():
    global bgsrv, conn
    conn = rpyc.connect("localhost", 18871)  # conn to worker  (arg of config={'allow_pickle': True}) does not fasten execution
    bgsrv = rpyc.BgServingThread(conn)

    starter = conn.root.InputStreamStarter(
        f,
        'TextFile',
        ('/home/nakatani/git/shellstreaming/shellstreaming/test/inputstream/test_textfile_input01.txt', )
    )

    # async
    astart = rpyc.async(starter.start)
    astart()

    global stop
    while not stop:
        print('[pid=%d] waiting for worker\'s inputstream to finish' % (os.getpid()))
        time.sleep(0.1)<commit_msg>using `AsyncResult.wait()` Signed-off-by: <NAME> <<EMAIL>><commit_after>from nose.tools import *
import rpyc
from multiprocessing import Process
import time
import os

from shellstreaming.inputstream.textfile import TextFile


class InputStreamStarterService(rpyc.Service):
    class exposed_InputStreamStarter(object):
        def __init__(self, callback, inputstream, args):
            self._callback    = callback
            self._inputstream = inputstream
            self._args        = args

        def exposed_start(self):
            module = __import__('shellstreaming.inputstream.textfile', globals(), locals(), [self._inputstream], -1)
            stream = eval('%s %s' % (self._inputstream, self._args))

            n_records = 0
            for batch in stream:
                for record in batch:
                    eq_(len(record), 1)
                    line = record[0]
                    eq_('line ', line[0:5])
                    ok_(0 <= int(line[5:]) < 100)  # record order in a batch is not always 'oldest-first'
                    n_records += 1
            self._callback(n_records)
            return 'hoge'


process = None


def _start_worker_thread():
    from rpyc.utils.server import ThreadPoolServer as Server  # can be ThreadedServer or ThreadPoolServer for performance
    print('[pid=%d] strating worker server' % (os.getpid()))
    Server(InputStreamStarterService, port=18871).start()


def setup():
    global process
    # setting up worker
    print('[pid=%d] starting worker process' % (os.getpid()))
    process = Process(target=_start_worker_thread)
    process.start()
    # wait for worker process to start server
    while True:
        try:
            conn = rpyc.connect('localhost', 18871)
            conn.close()
            break
        except:  # connection refused
            time.sleep(0.1)
            continue


def teardown():
    global process
    print('[pid=%d] terminating worker process' % (os.getpid()))
    process.terminate()


bgsrv = conn = None
stop = False


def f(number_of_lines):  # masterの呼び出すcallback
    global bgsrv, conn, stop
    print '[pid=%d] finish!!! (%d lines)' % (os.getpid(), number_of_lines)


def test_jobdispatcher_makes_worker_input_file():
    global bgsrv, conn
    conn = rpyc.connect("localhost", 18871)  # conn to worker  (arg of config={'allow_pickle': True}) does not fasten execution
    bgsrv = rpyc.BgServingThread(conn)

    starter = conn.root.InputStreamStarter(
        f,
        'TextFile',
        ('/home/nakatani/git/shellstreaming/shellstreaming/test/inputstream/test_textfile_input01.txt', )
    )

    # async
    astart = rpyc.async(starter.start)
    async_res = astart()

    # do everything master want to do

    async_res.wait()

    bgsrv.stop()
    conn.close()
{'fraction_non_alphanumeric': 0.1076017130620985, 'fraction_numerical': 0.011420413990007138, 'mean_word_length': 2.53182104599874, 'pattern_counts': {'":': 0, '<': 10, '<?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': '2550634', 'n_tokens_mistral': 1717, 'n_tokens_neox': 1632, 'n_words': 421}