__author__ = 'Manuela Kuhn <manuela.kuhn@desy.de>'
import socket
import zmq
import os
import logging
import sys
import trace
import cPickle
from logutils.queue import QueueHandler
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
EVENTDETECTOR_PATH = BASE_PATH + os.sep + "src" + os.sep + "sender" + os.sep + "eventDetectors"
if not EVENTDETECTOR_PATH in sys.path:
sys.path.append ( EVENTDETECTOR_PATH )
del EVENTDETECTOR_PATH
import helpers
#
# -------------------------- class: TaskProvider --------------------------------------
#
class TaskProvider():
def __init__ (self, eventDetectorConfig, requestFwPort, routerPort, logQueue, context = None):
global BASE_PATH
#eventDetectorConfig = {
# eventDetectorType : ... ,
# monDir : ... ,
# monEventType : ... ,
# monSubdirs : ... ,
# monSuffixes : ... ,
#}
self.log = self.getLogger(logQueue)
self.log.debug("TaskProvider started (PID " + str(os.getpid()) + ").")
self.dataDetectorModule = None
self.eventDetector = None
self.config = eventDetectorConfig
self.log.debug("Configuration for event detector: " + str(self.config))
eventDetectorModule = self.config["eventDetectorType"]
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
self.log.info("Loading eventDetector: " + eventDetectorModule)
self.eventDetectorModule = __import__(eventDetectorModule)
self.eventDetector = self.eventDetectorModule.EventDetector(self.config, logQueue)
self.createSockets()
try:
self.run()
except KeyboardInterrupt:
self.log.debug("Keyboard interruption detected. Shuting down")
self.stop()
except:
self.log.info("Stopping TaskProvider due to unknown error condition.", exc_info=True)
self.stop()
# Send all logs to the main process
# The worker configuration is done at the start of the worker process run.
# Note that on Windows you can't rely on fork semantics, so each process
# will run the logging configuration code when it starts.
def getLogger (self, queue):
# Create log and set handler to queue handle
h = QueueHandler(queue) # Just the one handler needed
logger = logging.getLogger("TaskProvider")
logger.propagate = False
logger.addHandler(h)
logger.setLevel(logging.DEBUG)
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:
self.log.error("Failed to start requestFwSocket (connect): '" + connectionStr + "'", exc_info=True)
# 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:
self.log.error("Failed to start router Socket (bind): '" + connectionStr + "'", exc_info=True)
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 KeyboardInterrupt:
break
except:
self.log.error("Invalid fileEvent message received.", exc_info=True)
#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.", exc_info=True)
# 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:
self.log.error("Unable to assemble message dict.", exc_info=True)
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:
self.log.error("Sending message...failed.", exc_info=True)
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(0)
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, logQueue, context = None):
# Send all logs to the main process
self.log = self.getLogger(logQueue)
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)
self.log.info("[requestResponder] requestFwSocket started (bind) for '" + connectionStr + "'")
self.run()
# Send all logs to the main process
# The worker configuration is done at the start of the worker process run.
# Note that on Windows you can't rely on fork semantics, so each process
# will run the logging configuration code when it starts.
def getLogger (self, queue):
# Create log and set handler to queue handle
h = QueueHandler(queue) # Just the one handler needed
logger = logging.getLogger("requestResponder")
logger.propagate = False
logger.addHandler(h)
logger.setLevel(logging.DEBUG)
return logger
def run (self):
hostname = socket.gethostname()
self.log.info("[requestResponder] Start run")
openRequests = [[hostname + ':6003', 1], [hostname + ':6004', 0]]
while True:
request = self.requestFwSocket.recv()
self.log.debug("[requestResponder] Received request: " + str(request) )
self.requestFwSocket.send(cPickle.dumps(openRequests))
self.log.debug("[requestResponder] Answer: " + str(openRequests) )
def __exit__(self):
self.requestFwSocket.close(0)
self.context.destroy()
if __name__ == '__main__':
from multiprocessing import Process, freeze_support, Queue
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"
logsize = 10485760
# 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"
logQueue = Queue(-1)
# Get the log Configuration for the lisener
h1, h2 = helpers.getLogHandlers(logfile, logsize, verbose=True, onScreenLogLevel="debug")
# Start queue listener using the stream handler above
logQueueListener = helpers.CustomQueueListener(logQueue, h1, h2)
logQueueListener.start()
# Create log and set handler to queue handle
root = logging.getLogger()
root.setLevel(logging.DEBUG) # Log level = DEBUG
qh = QueueHandler(logQueue)
root.addHandler(qh)
taskProviderPr = Process ( target = TaskProvider, args = (eventDetectorConfig, requestFwPort, routerPort, logQueue) )
taskProviderPr.start()
requestResponderPr = Process ( target = requestResponder, args = ( requestFwPort, logQueue) )
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)
logQueue.put_nowait(None)
logQueueListener.stop()