TaskProvider.py

__author__ = 'Manuela Kuhn <manuela.kuhn@desy.de>'

import socket
import zmq
import os
import logging
import sys
import trace
import cPickle

try:
    BASE_PATH = os.path.dirname ( os.path.dirname ( os.path.dirname ( os.path.realpath ( __file__ ) )))
except:
    BASE_PATH = os.path.dirname ( os.path.dirname ( os.path.dirname ( os.path.abspath ( sys.argv[0] ) )))
SHARED_PATH  = BASE_PATH + os.sep + "src" + os.sep + "shared"

if not SHARED_PATH in sys.path:
    sys.path.append ( SHARED_PATH )
del SHARED_PATH

import helpers


#
#  --------------------------  class: TaskProvider  --------------------------------------
#

class TaskProvider():
    def __init__ (self, eventDetectorConfig, requestFwPort, routerPort, logConfig = None, context = None):
        global BASE_PATH

        #eventDetectorConfig = {
        #        eventDetectorType   : ... ,
        #        monDir       : ... ,
        #        monEventType : ... ,
        #        monSubdirs   : ... ,
        #        monSuffixes  : ... ,
        #}

        if logConfig:
            logfile = BASE_PATH + os.sep + "logs" + os.sep + "taskProvider_.log"
            helpers.initLogging(logfile, verbose=True, onScreenLogLevel="debug")

        self.log               = self.getLogger()
        self.log.debug("TaskProvider: __init__()")

        self.eventDetector     = None

        self.config = eventDetectorConfig
        self.log.debug("Configuration for Event detector: " + str(self.config))

        self.localhost         = "127.0.0.1"
        self.extIp             = "0.0.0.0"
        self.requestFwPort     = requestFwPort
        self.routerPort        = routerPort
        self.requestFwSocket   = None
        self.routerSocket      = None

        self.log.debug("Registering ZMQ context")
        # remember if the context was created outside this class or not
        if context:
            self.context    = context
            self.extContext = True
        else:
            self.context    = zmq.Context()
            self.extContext = False


        if self.config.has_key("eventDetectorType") and self.config["eventDetectorType"] == "inotifyx":

            from InotifyxDetector import InotifyxDetector as EventDetector

            # check format of config
            if ( not self.config.has_key("monDir") or
                    not self.config.has_key("monEventType") or
                    not self.config.has_key("monSubdirs") or
                    not self.config.has_key("monSuffixes") ):
                self.log.error ("Configuration of wrong format")

        elif self.config.has_key("eventDetectorType") and self.config["eventDetectorType"] == "watchdog":

            from WatchdogDetector import WatchdogDetector as EventDetector

            # check format of config
            if ( not self.config.has_key("monDir") or
                    not self.config.has_key("monEventType") or
                    not self.config.has_key("monSubdirs") or
                    not self.config.has_key("monSuffixes") or
                    not self.config.has_key("timeTillClosed") ):
                self.log.error ("Configuration of wrong format")

        else:
            self.log.error("Type of event detector is not supported: " + str( self.config["eventDetectorType"] ))
            return -1

        self.eventDetector     = EventDetector(self.config)

        self.createSockets()

        try:
            self.run()
        except KeyboardInterrupt:
            self.log.debug("Keyboard interruption detected. Shuting down")
        except:
            trace = traceback.format_exc()
            self.log.info("Stopping TaskProvider due to unknown error condition.")
            self.log.debug("Error was: " + str(trace))


    def getLogger (self):
        logger = logging.getLogger("TaskProvider")
        return logger


    def createSockets (self):
        # socket to get requests
        self.requestFwSocket = self.context.socket(zmq.REQ)
        connectionStr  = "tcp://{ip}:{port}".format( ip=self.localhost, port=self.requestFwPort )
        try:
            self.requestFwSocket.connect(connectionStr)
            self.log.info("Start requestFwSocket (connect): '" + str(connectionStr) + "'")
        except Exception as e:
            self.log.error("Failed to start requestFwSocket (connect): '" + connectionStr + "'")
            self.log.debug("Error was:" + str(e))

        # socket to disribute the events to the worker
        self.routerSocket = self.context.socket(zmq.PUSH)
        connectionStr  = "tcp://{ip}:{port}".format( ip=self.localhost, port=self.routerPort )
        try:
            self.routerSocket.bind(connectionStr)
            self.log.info("Start to router socket (bind): '" + str(connectionStr) + "'")
        except Exception as e:
            self.log.error("Failed to start router Socket (bind): '" + connectionStr + "'")
            self.log.debug("Error was:" + str(e))


    def run (self):
        while True:
            try:
                # the event for a file /tmp/test/source/local/file1.tif is of the form:
                # {
                #   "sourcePath" : "/tmp/test/source/"
                #   "relativePath": "local"
                #   "filename"   : "file1.tif"
                # }
                workloadList = self.eventDetector.getNewEvent()
            except Exception, e:
                self.log.error("Invalid fileEvent message received.")
                self.log.debug("Error was: " + str(e))
                #skip all further instructions and continue with next iteration
                continue

            #TODO validate workload dict
            for workload in workloadList:
                # get requests for this event
                try:
                    self.log.debug("Get requests...")
                    self.requestFwSocket.send("")
                    requests = cPickle.loads(self.requestFwSocket.recv())
                    self.log.debug("Get requests... done.")
                    self.log.debug("Requests: " + str(requests))
                except:
                    self.log.error("Get Requests... failed.")
                    trace = traceback.format_exc()
                    self.log.debug("Error was: " + str(trace))


                # build message dict
                try:
                    self.log.debug("Building message dict...")
                    messageDict = cPickle.dumps(workload)  #sets correct escape characters
                    self.log.debug("Building message dict...done.")
                except Exception, e:
                    self.log.error("Unable to assemble message dict.")
                    self.log.debug("Error was: " + stri(e))
                    continue

                # send the file to the fileMover
                try:
                    self.log.debug("Sending message...")
                    message = [messageDict]
                    if requests != ["None"]:
                        message.append(cPickle.dumps(requests))
                    self.log.debug(str(message))
                    self.routerSocket.send_multipart(message)
                    self.log.debug("Sending message...done.")
                except Exception, e:
                    self.log.error("Sending message...failed.")
                    self.log.debug("Error was: " + str(e))



    def stop(self):
        if self.routerSocket:
            self.routerSocket.close(0)
            self.routerSocket = None
        if self.requestFwSocket:
            self.requestFwSocket.close(0)
            self.requestFwSocket = None
        if not self.extContext and self.context:
            self.context.destroy()
            self.context = None


    def __exit__(self):
        self.stop()


    def __del__(self):
        self.stop()


# cannot be defined in "if __name__ == '__main__'" because then it is unbound
# see https://docs.python.org/2/library/multiprocessing.html#windows
class requestResponder():
    def __init__ (self, requestFwPort, logConfig = None, context = None):
        if logConfig:
            logfile = BASE_PATH + os.sep + "logs" + os.sep + "taskProvider__.log"
            helpers.initLogging(logfile, verbose=True, onScreenLogLevel="debug")

        self.context         = context or zmq.Context.instance()
        self.requestFwSocket = self.context.socket(zmq.REP)
        connectionStr   = "tcp://127.0.0.1:" + requestFwPort
        self.requestFwSocket.bind(connectionStr)
        logging.info("[requestResponder] requestFwSocket started (bind) for '" + connectionStr + "'")

        self.run()


    def run (self):
        hostname = socket.gethostname()
        logging.info("[requestResponder] Start run")
        openRequests = [[hostname + ':6003', 1], [hostname + ':6004', 0]]
        while True:
            request = self.requestFwSocket.recv()
            logging.debug("[requestResponder] Received request: " + str(request) )

            self.requestFwSocket.send(cPickle.dumps(openRequests))
            logging.debug("[requestResponder] Answer: " + str(openRequests) )


    def __exit__(self):
        self.requestFwSocket.close(0)
        self.context.destroy()



if __name__ == '__main__':
    from multiprocessing import Process, freeze_support
    import time
    from shutil import copyfile
    from subprocess import call

    freeze_support()    #see https://docs.python.org/2/library/multiprocessing.html#windows

    logfile = BASE_PATH + os.sep + "logs" + os.sep + "taskProvider.log"

    #enable logging
    helpers.initLogging(logfile, verbose=True, onScreenLogLevel="debug")

#    eventDetectorConfig = {
#            "eventDetectorType"   : "inotifyx",
#            "monDir"              : BASE_PATH + os.sep + "data" + os.sep + "source",
#            "monEventType"        : "IN_CLOSE_WRITE",
#            "monSubdirs"          : ["commissioning", "current", "local"],
#            "monSuffixes"         : [".tif", ".cbf"]
#            }

    eventDetectorConfig = {
            "eventDetectorType" : "watchdog",
            "monDir"            : BASE_PATH + os.sep + "data" + os.sep + "source",
            "monEventType"      : "IN_CLOSE_WRITE",
            "monSubdirs"        : ["commissioning", "current", "local"],
            "monSuffixes"       : [".tif", ".cbf"],
            "timeTillClosed"    : 1 #s
            }

    requestFwPort = "6001"
    routerPort    = "7000"

    logConfig = "test"

    taskProviderPr = Process ( target = TaskProvider, args = (eventDetectorConfig, requestFwPort, routerPort, logConfig) )
    taskProviderPr.start()

    requestResponderPr = Process ( target = requestResponder, args = ( requestFwPort, logConfig) )
    requestResponderPr.start()

    context       = zmq.Context.instance()

    routerSocket  = context.socket(zmq.PULL)
    connectionStr = "tcp://localhost:" + routerPort
    routerSocket.connect(connectionStr)
    logging.info("=== routerSocket connected to " + connectionStr)

    sourceFile = BASE_PATH + os.sep + "test_file.cbf"
    targetFileBase = BASE_PATH + os.sep + "data" + os.sep + "source" + os.sep + "local" + os.sep + "raw" + os.sep

    i = 100
    try:
        while i <= 105:
            time.sleep(0.5)
            targetFile = targetFileBase + str(i) + ".cbf"
            logging.debug("copy to " + targetFile)
            copyfile(sourceFile, targetFile)
#            call(["cp", sourceFile, targetFile])
            i += 1

            workload = routerSocket.recv_multipart()
            logging.info("=== next workload " + str(workload))
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    finally:

        requestResponderPr.terminate()
        taskProviderPr.terminate()

        routerSocket.close(0)
        context.destroy()

        for number in range(100, i):
            targetFile = targetFileBase + str(number) + ".cbf"
            logging.debug("remove " + targetFile)
            os.remove(targetFile)