Tentokrát paralelně. Úkolem je pustit graf úloh, kde jednotlivé větve mají běžet paralelně. Používám zde jen python s balíčkem multiprocessing, což má svůj důvod. Multiprocessing funguje na jedničku (bez hvězdičky), ale procesu nedovoluje paralelně pouštět další procesy. Pokusíte-li se o to přesto, odmění Vás python tímto hlášením ve stylu fantasy románů: "daemonic processes are not allowed to have children". To ale nevadí, protože každý orientovaný graf bez cyklů lze dekomponovat na jednotlivé větve. A přesně to dělám v přiloženém skriptu. Skript pouští tento graf úloh:
Procedura job(x) trvá x sekund. Celý graf úloh by měl trvat 12 sekund, a také trvá. Ale nejsem spokojen s cyklem, který procesy pouští, protože neumím čekat na první konec nějakého procesu jinak než pomocí timoutu. Předpokládám, že jednotlivé úlohy trvají alespoň minuty, ale spíš hodiny, takže to tolik nevadí, ale není to čisté řešení a nelíbí se mi.
# -*- coding: cp1250 -*-
import sys
import copy
from multiprocessing import Process, Pool, Lock, Manager, Queue
def decompose(bash, reservedIDs = [0], parentsConditions = [0]):
processes = []
process = None
conditions = copy.copy(parentsConditions)
for i in range(len(bash)):
command = bash[i][0]
if command == "branch":
conditions2 = copy.copy(conditions)
for j in range(len(bash[i][1])):
for subprocess in decompose(bash[i][1][j], reservedIDs, conditions2):
processes.append(subprocess)
conditions.append(subprocess["ID"])
process = None
else:
if process == None:
processID = max(reservedIDs) + 1
reservedIDs.append(processID)
process = {
"ID": processID,
"bash": [],
"conditions": copy.copy(list(set(conditions)))
}
processes.append(process)
conditions.append(processID)
process["bash"].append(bash[i])
return processes
def doDecomposedBash(bash, jobs):
for [command, argument] in bash:
jobs[command](*argument)
def job(t):
print "Doing some staff ..."
import time
time.sleep(t)
print "... done (%f sec)." % t
if __name__ == "__main__":
manager = Manager()
jobs = {
"job": job,
}
enviroment = manager.dict()
bash = [
["branch",
[
[
["job", [1.0]],
["branch",
[
[
["job", [2.0]],
],
[
["job", [3.0]],
],
],
],
["job", [4.0]],
],
[
["job", [5.0]],
["branch",
[
[
["job", [6.0]],
],
[
["job", [7.0]],
],
],
],
],
],
],
]
processes = decompose(bash)
for process in processes:
process["process"] = Process(target = doDecomposedBash, args = (process["bash"], jobs))
process["state"] = "waiting"
closedProcesses = {0: 1}
# starting processes
while True:
theEnd = True
for process in processes:
if process["state"] == "running" and not process["process"].is_alive():
process["state"] = "closed"
if process["state"] == "closed":
closedProcesses[process["ID"]] = 1
else:
theEnd = False
if theEnd:
break
for process in processes:
if process["state"] == "waiting":
run = True
for condition in process["conditions"]:
if not condition in closedProcesses:
run = False
if run:
process["state"] = "running"
process["process"].start()
for process in processes:
if process["state"] == "running":
process["process"].join(0.01)
break
# All processes are done now.
