Skip to content
Snippets Groups Projects
Commit 0f765851 authored by Manuela Kuhn's avatar Manuela Kuhn
Browse files

Fix signal exchange

parent 42468bf3
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
src/sender/FileMover.py
+ 45
− 14
View file @ 0f765851
...@@ -6,6 +6,7 @@ import logging ...@@ -6,6 +6,7 @@ import logging
import os import os
import sys import sys
import traceback import traceback
import copy
from multiprocessing import Process from multiprocessing import Process
from WorkerProcess import WorkerProcess from WorkerProcess import WorkerProcess
...@@ -89,7 +90,7 @@ class FileMover(): ...@@ -89,7 +90,7 @@ class FileMover():
else: else:
self.receiverWhiteList.append(host) self.receiverWhiteList.append(host)
self.parallelDataStreams = parallelDataStreams self.parallelDataStreams = int(parallelDataStreams)
self.chunkSize = chunkSize self.chunkSize = chunkSize
self.freeWorker = [] # list of workers which sent a ready signal (again) during signal distributing process self.freeWorker = [] # list of workers which sent a ready signal (again) during signal distributing process
...@@ -163,7 +164,7 @@ class FileMover(): ...@@ -163,7 +164,7 @@ class FileMover():
incomingMessageCounter = 0 incomingMessageCounter = 0
#start worker-processes. each will have its own PushSocket. #start worker-processes. each will have its own PushSocket.
numberOfWorkerProcesses = int(self.parallelDataStreams) numberOfWorkerProcesses = self.parallelDataStreams
for processNumber in range(numberOfWorkerProcesses): for processNumber in range(numberOfWorkerProcesses):
self.log.debug("instantiate new workerProcess (nr " + str(processNumber) + " )") self.log.debug("instantiate new workerProcess (nr " + str(processNumber) + " )")
newWorkerProcess = Process(target=WorkerProcess, args=(processNumber, newWorkerProcess = Process(target=WorkerProcess, args=(processNumber,
...@@ -256,6 +257,8 @@ class FileMover(): ...@@ -256,6 +257,8 @@ class FileMover():
else: else:
version, signal, host, port = incomingMessage version, signal, host, port = incomingMessage
host = host.split(',')
port = port.split(',')
if version: if version:
if helperScript.checkVersion(version, self.log): if helperScript.checkVersion(version, self.log):
...@@ -269,13 +272,26 @@ class FileMover(): ...@@ -269,13 +272,26 @@ class FileMover():
# Checking signal sending host # Checking signal sending host
self.log.debug("Check if signal sending host is in WhiteList...") self.log.debug("Check if signal sending host is in WhiteList...")
if helperScript.checkHost(host, self.receiverWhiteList): if helperScript.checkHost(host, self.receiverWhiteList, self.log):
self.log.debug("Host " + str(host) + " is allowed to connect.") self.log.debug("One of the hosts is allowed to connect.")
self.log.debug("hosts: " + str(host))
else: else:
self.log.debug("Host " + str(host) + " is not allowed to connect.") self.log.debug("One of the hosts is not allowed to connect.")
self.log.debug("hosts: " + str(host))
self.sendResponse("NO_VALID_HOST") self.sendResponse("NO_VALID_HOST")
return False, None, None, None return False, None, None, None
if signal in ["START_QUERY_NEXT", "STOP_QUERY_NEXT"]:
if len(host) != self.parallelDataStreams:
self.log.debug("Not enough hosts specified.")
self.sendResponse("INCORRECT_NUMBER_OF_HOSTS")
return False, None, None, None
if len(port) != self.parallelDataStreams:
self.log.debug("Not enough ports specified.")
self.sendResponse("INCORRECT_NUMBER_OF_PORTS")
return False, None, None, None
return True, signal, host, port return True, signal, host, port
...@@ -285,14 +301,17 @@ class FileMover(): ...@@ -285,14 +301,17 @@ class FileMover():
def reactToSignal(self, signal, host, port): def reactToSignal(self, signal, host, port):
message = signal + "," + host + "," + port
# React to signal # React to signal
if signal == "START_STREAM": if signal == "START_STREAM":
#FIXME
host = host[0]
port = port[0]
self.log.info("Received signal to start stream to host " + str(host) + " on port " + str(port)) self.log.info("Received signal to start stream to host " + str(host) + " on port " + str(port))
if [host, port] not in self.openConnections["streams"]: if [host, port] not in self.openConnections["streams"]:
self.openConnections["streams"].append([host, port]) self.openConnections["streams"].append([host, port])
# send signal to workerProcesses and back to receiver # send signal to workerProcesses and back to receiver
message = signal + "," + host + "," + port
self.sendSignalToWorker(message) self.sendSignalToWorker(message)
self.sendResponse(signal) self.sendResponse(signal)
else: else:
...@@ -301,10 +320,13 @@ class FileMover(): ...@@ -301,10 +320,13 @@ class FileMover():
return return
elif signal == "STOP_STREAM": elif signal == "STOP_STREAM":
host = host[0]
port = port[0]
self.log.info("Received signal to stop stream to host " + str(host) + " on port " + str(port)) self.log.info("Received signal to stop stream to host " + str(host) + " on port " + str(port))
if [host, port] in self.openConnections["streams"]: if [host, port] in self.openConnections["streams"]:
self.openConnections["streams"].remove([host, port]) self.openConnections["streams"].remove([host, port])
# send signal to workerProcesses and back to receiver # send signal to workerProcesses and back to receiver
message = signal + "," + host + "," + port
self.sendSignalToWorker(message) self.sendSignalToWorker(message)
self.sendResponse(signal) self.sendResponse(signal)
else: else:
...@@ -315,9 +337,9 @@ class FileMover(): ...@@ -315,9 +337,9 @@ class FileMover():
elif signal == "START_QUERY_NEXT" or signal == "START_REALTIME_ANALYSIS": elif signal == "START_QUERY_NEXT" or signal == "START_REALTIME_ANALYSIS":
self.log.info("Received signal from host " + str(host) + " to enable querying for data") self.log.info("Received signal from host " + str(host) + " to enable querying for data")
if [host, port] not in self.openConnections["queryNext"]: if [host, port] not in self.openConnections["queryNext"]:
self.openConnections["queryNext"].append([host, port]) self.openConnections["queryNext"].append([copy.deepcopy(host), copy.deepcopy(port)])
# send signal to workerProcesses and back to receiver # send signal to workerProcesses and back to receiver
self.sendSignalToWorker(message) self.sendSignalToWorker(signal, host, port)
self.sendResponse(signal) self.sendResponse(signal)
else: else:
self.log.info("Query connection to host " + str(host) + " on port " + str(port) + " is already started") self.log.info("Query connection to host " + str(host) + " on port " + str(port) + " is already started")
...@@ -326,10 +348,11 @@ class FileMover(): ...@@ -326,10 +348,11 @@ class FileMover():
elif signal == "STOP_QUERY_NEXT" or signal == "STOP_REALTIME_ANALYSIS": elif signal == "STOP_QUERY_NEXT" or signal == "STOP_REALTIME_ANALYSIS":
self.log.info("Received signal from host " + str(host) + " to disable querying for data") self.log.info("Received signal from host " + str(host) + " to disable querying for data")
print self.openConnections["queryNext"]
if [host, port] in self.openConnections["queryNext"]: if [host, port] in self.openConnections["queryNext"]:
self.openConnections["queryNext"].remove([host, port]) self.openConnections["queryNext"].remove([host, port])
# send signal to workerProcesses and back to receiver # send signal to workerProcesses and back to receiver
self.sendSignalToWorker(message) self.sendSignalToWorker(signal, host, port)
self.log.debug("Send signal to worker: " + str(signal)) self.log.debug("Send signal to worker: " + str(signal))
self.sendResponse(signal) self.sendResponse(signal)
self.log.debug("Setime.sleep(0.1)nd response back: " + str(signal)) self.log.debug("Setime.sleep(0.1)nd response back: " + str(signal))
...@@ -343,8 +366,9 @@ class FileMover(): ...@@ -343,8 +366,9 @@ class FileMover():
self.sendResponse("NO_VALID_SIGNAL") self.sendResponse("NO_VALID_SIGNAL")
def sendSignalToWorker(self, signal, individual = False): def sendSignalToWorker(self, signal, individualHost = False, individualPort = False):
numberOfWorkerProcesses = int(self.parallelDataStreams)
numberOfWorkerProcesses = self.parallelDataStreams
alreadySignalled = [] alreadySignalled = []
# for processNumber in range(numberOfWorkerProcesses): # for processNumber in range(numberOfWorkerProcesses):
...@@ -354,19 +378,26 @@ class FileMover(): ...@@ -354,19 +378,26 @@ class FileMover():
self.log.debug("Available workerProcess detected.") self.log.debug("Available workerProcess detected.")
if address not in alreadySignalled: if address not in alreadySignalled:
self.log.debug("Send signal " + str(signal) + " to " + str(address))
if individualHost and individualPort:
signalToSend = signal + "," + individualHost.pop() + "," + individualPort.pop()
else:
signalToSend = signal
self.log.debug("Send signal " + str(signalToSend) + " to " + str(address))
# address == "worker-0" # address == "worker-0"
# empty == b'' # as delimiter # empty == b'' # as delimiter
# signal == b'START_STREAM' # signal == b'START_STREAM'
self.routerSocket.send_multipart([ self.routerSocket.send_multipart([
address, address,
b'', b'',
signal, signalToSend,
]) ])
alreadySignalled.append(address) alreadySignalled.append(address)
else: else:
self.log.debug("Signal " + str(signal) + " already sent to " + str(address) + ". Mark this worker as ready/.") self.log.debug("Signal " + str(signalToSend) + " already sent to " + str(address) + ". Mark this worker as ready/.")
self.freeWorker.append(address) self.freeWorker.append(address)
......
src/shared/LiveViewCommunicator.py
+ 8
− 4
View file @ 0f765851
...@@ -228,6 +228,8 @@ class LiveViewCommunicator: ...@@ -228,6 +228,8 @@ class LiveViewCommunicator:
else: else:
version, signal, host, port = message version, signal, host, port = message
host = host.split(',')
port = port.split(',')
if version: if version:
if helperScript.checkVersion(version, self.log): if helperScript.checkVersion(version, self.log):
...@@ -240,11 +242,13 @@ class LiveViewCommunicator: ...@@ -240,11 +242,13 @@ class LiveViewCommunicator:
if signal and host and port : if signal and host and port :
# Checking signal sending host # Checking signal sending host
self.log.debug("Check if signal sending host is in WhiteList...") self.log.debug("Check if hosts is in WhiteList...")
if helperScript.checkHost(host, self.receiverWhiteList): if helperScript.checkHost(host, self.receiverWhiteList, self.log):
self.log.debug("Host " + str(host) + " is allowed to connect.") self.log.debug("One of the hosts is allowed to connect.")
self.log.debug("hosts: " + str(host))
else: else:
self.log.debug("Host " + str(host) + " is not allowed to connect.") self.log.debug("One of the hosts is not allowed to connect.")
self.log.debug("hosts: " + str(host))
self.sendResponse("NO_VALID_HOST") self.sendResponse("NO_VALID_HOST")
return False, None, None, None return False, None, None, None
......
src/shared/helperScript.py
+ 22
− 6
View file @ 0f765851
...@@ -222,17 +222,33 @@ def checkVersion(version, log): ...@@ -222,17 +222,33 @@ def checkVersion(version, log):
return True return True
def checkHost(hostname, whiteList ): def checkHost(hostname, whiteList, log):
if hostname and whiteList: if hostname and whiteList:
if hostname.endswith(".desy.de"): if type(hostname) == list:
hostnameModified = hostname[:-8] temp = True
for host in hostname:
if host.endswith(".desy.de"):
hostModified = host[:-8]
else:
hostModified = host
if host not in whiteList and hostModified not in whiteList:
log.info("Host " + str(host) + " is not allowed to connect")
temp = False
return temp
else: else:
hostnameModified = hostname if hostname.endswith(".desy.de"):
hostnameModified = hostname[:-8]
else:
hostnameModified = hostname
if hostname in whiteList or hostnameModified in whiteList: if hostname in whiteList or hostnameModified in whiteList:
return True return True
return False return False
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment