Thread safety demo on debian

thread-safety-debian.py

#!/usr/bin/env python

import os
import subprocess
import sys

from threading import Event, Thread
from time import sleep


class ThreadSafety:
    def __init__(self):
        self.__running = True

    def start(self, asDaemon):
        try:
            if asDaemon:
                self.__createDaemon()

            # non-ascii/binary file
            sourceFile = "/usr/bin/awk"
            basePort = 64000
            count = 10

            startedEvent = Event()

            mainRecvThread = MainRecvThread(startedEvent, basePort, count)
            mainRecvThread.start()

            startedEvent.wait()

            mainSendThread = MainSendThread(sourceFile, basePort, count)
            mainSendThread.start()

            mainSendThread.join()
            mainRecvThread.join()

            return True
        except Exception, e:
            print(e)
            print("Failure starting ThreadSafety")
            return False


    def __createDaemon(self):
        # http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/
        try:
            pid = os.fork()
            if pid > 0:
                sys.exit(0)
        except OSError, e:
            sys.stderr.write("Fork #1 failed: %d (%s)\n" %
                             (e.errno, e.strerror))
            sys.exit(1)
        os.chdir("/")
        os.setsid()
        os.umask(0)

        try:
            pid = os.fork()
            if pid > 0:
                sys.exit(0)
        except OSError, e:
            sys.stderr.write("fork #2 failed: %d (%s)\n" %
                             (e.errno, e.strerror))
            sys.exit(1)

        print("Created daemon successfully, pid %s" % os.getpid())

        sys.stdout.flush()
        sys.stderr.flush()
        si = file("/dev/null", "r")
        so = file("/dev/null", "a+")
        se = file("/dev/null", "a+", 0)
        os.dup2(si.fileno(), sys.stdin.fileno())
        os.dup2(so.fileno(), sys.stdout.fileno())
        os.dup2(se.fileno(), sys.stderr.fileno())
        pass


class Command:
    def __init__(self, caller):
        self.__caller = caller
        self._subProcess = None

    def executeCommand(self, command):
        self._subProcess = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True
        )
        stdout, stderr = self._subProcess.communicate()
        exitCode = self._subProcess.returncode
        if exitCode:
            print("Failure executing command: %r" % command)
            return "%s::%s" % (stdout, stderr), exitCode
        else:
            return stdout, exitCode


class MainRecvThread(Thread):
    def __init__(self, startedEvent, basePort, count):
        Thread.__init__(self)
        self.__startedEvent = startedEvent
        self.__basePort = basePort
        self.__count = count
        self.__recvThreads = set([])
        self.__counter = 0


    def run(self):
        try:
            mainRecvEvent = Event()
            for i in xrange(self.__count):
                self.__recvThreads.add(
                    WorkerRecvThread((self.__basePort + i), mainRecvEvent))

            for recvThread in self.__recvThreads:
                recvThread.start()
                mainRecvEvent.wait()
                mainRecvEvent.clear()

            self.__startedEvent.set()

            for recvThread in self.__recvThreads:
                if recvThread.isAlive():
                    recvThread.join()
            return True
        except Exception, e:
            print(e)
            print("Fatal Error in Recv Thread")


class WorkerRecvThread(Thread):
    def __init__(self, port, mainRecvEvent):
        Thread.__init__(self)
        self.__port = port
        self.__mainRecvEvent = mainRecvEvent
        self.__command = Command(self)

    def run(self):
        self.__mainRecvEvent.set()
        command = "nc -l 127.0.0.1 -p %s | cat > /tmp/tts_%s" %\
                  (
                  self.__port, self.__port
                  )
        output, exitCode = self.__command.executeCommand(command)
        if exitCode:
            print("Failure with command %r: [%s]: %r" %
                  (command, exitCode, output))
        elif len(output) > 0:
            print("Output: %r" % output)
        return True


class MainSendThread(Thread):
    def __init__(self, sourceFile, basePort, count):
        Thread.__init__(self)
        self.__sourceFile = sourceFile
        self.__basePort = basePort
        self.__count = count
        self.__sendThreads = set([])


    def run(self):
        for i in xrange(self.__count):
            self.__sendThreads.add(
                WorkerSendThread(
                    self.__sourceFile,
                    (self.__basePort + i))
            )
        for sendThread in self.__sendThreads:
            sendThread.start()
        sleep(10)
        for sendThread in self.__sendThreads:
            if sendThread.isAlive():
                sendThread.join()
        return True


class WorkerSendThread(Thread):
    def __init__(self, sourceFile, port):
        Thread.__init__(self)
        self.__sourceFile = sourceFile
        self.__port = port
        self.__command = Command(self)

    def run(self):
        command = "cat %s | nc --send-only 127.0.0.1 %s" %\
                  (
                  self.__sourceFile, self.__port
                  )
        output, exitCode = self.__command.executeCommand(command)
        if exitCode:
            print("Failure command %r: [%s]: %r" %
                  (command, exitCode, output))
        elif len(output) > 0:
            print("Output: %r" % output)
        return True


if __name__ == "__main__":
    threadSafety = ThreadSafety()
    exitCode = threadSafety.start(True)
    #exitCode = threadSafety.start(False)
    if exitCode:
        print("Success")
        sys.exit(0)
    else:
        print("Failure: %s" % exitCode)
        sys.exit(-1)