invoke_process.py

# -*- coding: utf-8 -*-

"""
invoke_process.py
launch multiple subprocesses running SmallfileWorkload instance
Copyright 2012 -- Ben England
Licensed under the Apache License at http://www.apache.org/licenses/LICENSE-2.0
See Appendix on this page for instructions pertaining to license.
"""

import multiprocessing
import os
import shutil
import time

import smallfile
from smallfile import SMFRunException, unittest_module
from sync_files import touch

# this class launches multiple threads with SmallfileWorkload instances
# we do this because we can use > 1 core this way, with python threading,
# it doesn't really use > 1 core because of the GIL (global lock)
# occasional status reports could be sent back using pipe as well


class subprocess(multiprocessing.Process):
    def __init__(self, invocation):
        multiprocessing.Process.__init__(self)
        (conn1, conn2) = multiprocessing.Pipe(False)
        self.receiver = conn1  # master process receives test result data here
        self.sender = conn2  # slave process sends test result data here
        invocation.buf = None
        invocation.biggest_buf = None
        invocation.log = None
        self.invoke = invocation  # all workload generated by this object

    def run(self):
        try:
            self.invoke.do_workload()
            self.invoke.log.debug(
                "exiting subprocess and returning invoke " + str(self.invoke)
            )
        except Exception as e:
            print(
                "Exception seen in thread %s host %s (tail %s) "
                % (self.invoke.tid, self.invoke.onhost, self.invoke.log_fn())
            )
            self.invoke.log.error(str(e))
            self.status = self.invoke.NOTOK
        finally:
            self.rsptimes = None  # response time array already saved to file
            # reduce amount of data returned from this thread
            # by eliminating references objects that are no longer needed
            self.invoke.log = None  # log objects cannot be serialized
            self.invoke.buf = None
            self.invoke.biggest_buf = None
            self.invoke.rsptimes = None
            self.invoke.loggers = None
            self.invoke.file_dirs = None
            self.sender.send(self.invoke)


# below are unit tests for SmallfileWorkload
# including multi-threaded test
# to run, just do "python invoke_process.py"


class Test(unittest_module.TestCase):
    def setUp(self):
        self.invok = smallfile.SmallfileWorkload()
        self.invok.debug = True
        self.invok.verbose = True
        self.invok.tid = "regtest"
        self.invok.start_log()
        shutil.rmtree(self.invok.src_dirs[0], ignore_errors=True)
        os.makedirs(self.invok.src_dirs[0])

    def test_multiproc_stonewall(self):
        self.invok.log.info("starting stonewall test")
        thread_ready_timeout = 4
        thread_count = 4
        for tree in self.invok.top_dirs:
            shutil.rmtree(tree)
            os.mkdir(tree)
        for dir in self.invok.src_dirs:
            os.mkdir(dir)
        for dir in self.invok.dest_dirs:
            os.mkdir(dir)
        os.mkdir(self.invok.network_dir)
        self.invok.starting_gate = os.path.join(self.invok.network_dir, "starting-gate")
        sgate_file = self.invok.starting_gate
        invokeList = []
        for j in range(0, thread_count):
            s = smallfile.SmallfileWorkload()

            # s.log_to_stderr = True

            s.verbose = True
            s.tid = str(j)
            s.prefix = "thr_"
            s.suffix = "foo"
            s.iterations = 10
            s.stonewall = False
            s.starting_gate = sgate_file
            invokeList.append(s)
        threadList = []
        for s in invokeList:
            threadList.append(subprocess(s))
        for t in threadList:
            t.start()
        threads_ready = True
        for i in range(0, thread_ready_timeout):
            threads_ready = True
            for s in invokeList:
                thread_ready_file = s.gen_thread_ready_fname(s.tid)
                if not os.path.exists(thread_ready_file):
                    threads_ready = False
            if threads_ready:
                break
            time.sleep(1)
        if not threads_ready:
            raise SMFRunException(
                "threads did not show up within %d seconds" % thread_ready_timeout
            )
        time.sleep(1)
        touch(sgate_file)
        for t in threadList:
            rtnd_invok = t.receiver.recv()
            t.join()
            self.invok.log.info(str(rtnd_invok))
            assert rtnd_invok.elapsed_time is not None
            assert rtnd_invok.rq_final is not None
            assert rtnd_invok.filenum_final is not None
            if rtnd_invok.status != rtnd_invok.OK:
                raise SMFRunException(
                    "subprocess failure for %s invocation %s: "
                    % (str(t), str(rtnd_invok))
                )


# so you can just do "python invoke_process.py" to test it

if __name__ == "__main__":
    unittest_module.main()