#!/usr/bin/env python3 # Parallele_Suche.py """ Kapitel CP-SAT: Was mehrere Arbeiter wirklich bringen - und was sie kosten. Das Buch setzt an mehreren Stellen 'num_workers = 1', damit die abgedruckte Ausgabe reproduzierbar ist. Behauptet wird dabei zweierlei: dass parallele Suche schneller ist, und dass sie von Lauf zu Lauf verschiedene, gleich gute Loesungen findet. Beides wird hier gemessen statt geglaubt. Gerechnet wird auf einem Job-Shop wie in JobShop_Intervalle.py, nur gross genug, dass die Suche wirklich arbeitet: 12 Auftraege auf 10 Maschinen, jeder Auftrag besucht jede Maschine. Zwei Messungen: 1. Dieselbe Aufgabe mit 1, 2, 4 und 8 Arbeitern - Laufzeit und Ergebnis. 2. Dieselbe Konfiguration mehrfach, bei FESTEM random_seed - wie oft kommt derselbe Plan heraus? Die zweite ist die wichtigere: Sie entscheidet, wie man Tests schreibt. Achtung, Laufzeiten sind hardwareabhaengig. Die ZAHL der verschiedenen Plaene ist es auch - und genau das ist die Aussage. Benoetigt: numpy, ortools """ from __future__ import annotations import collections import time import numpy as np from ortools.sat.python import cp_model AUFTRAEGE = 12 MASCHINEN = 10 INSTANZ_SEED = 20260908 SOLVER_SEED = 1 # bleibt ueber ALLE Laeufe gleich - das ist der Punkt ARBEITERZAHLEN = (1, 2, 4, 8) WIEDERHOLUNGEN = 4 ZEITLIMIT = 120.0 def baue_instanz() -> list[list[tuple[int, int]]]: """Klassischer Job-Shop: Jeder Auftrag besucht jede Maschine genau einmal, in einer eigenen zufaelligen Reihenfolge.""" rng = np.random.default_rng(INSTANZ_SEED) return [[(int(m), int(rng.integers(2, 20))) for m in rng.permutation(MASCHINEN)] for _ in range(AUFTRAEGE)] def loese(auftraege, arbeiter: int): """Minimiert den Makespan. Gibt Status, Zielwert, Laufzeit und den Plan zurueck - den Plan als Tupel aller Startzeiten, damit sich zwei Laeufe vergleichen lassen.""" horizont = sum(dauer for auftrag in auftraege for _, dauer in auftrag) modell = cp_model.CpModel() Gang = collections.namedtuple("Gang", "start ende intervall") plaene: dict[tuple[int, int], Gang] = {} je_maschine = collections.defaultdict(list) for a, auftrag in enumerate(auftraege): for g, (maschine, dauer) in enumerate(auftrag): start = modell.NewIntVar(0, horizont, f"start_{a}_{g}") ende = modell.NewIntVar(0, horizont, f"ende_{a}_{g}") intervall = modell.NewIntervalVar(start, dauer, ende, f"iv_{a}_{g}") plaene[a, g] = Gang(start, ende, intervall) je_maschine[maschine].append(intervall) for maschine in range(MASCHINEN): modell.AddNoOverlap(je_maschine[maschine]) for a, auftrag in enumerate(auftraege): for g in range(len(auftrag) - 1): modell.Add(plaene[a, g + 1].start >= plaene[a, g].ende) makespan = modell.NewIntVar(0, horizont, "makespan") modell.AddMaxEquality( makespan, [plaene[a, len(auftrag) - 1].ende for a, auftrag in enumerate(auftraege)]) modell.Minimize(makespan) loeser = cp_model.CpSolver() loeser.parameters.num_workers = arbeiter loeser.parameters.random_seed = SOLVER_SEED loeser.parameters.max_time_in_seconds = ZEITLIMIT beginn = time.perf_counter() status = loeser.Solve(modell) dauer = time.perf_counter() - beginn if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE): return loeser.StatusName(status), None, dauer, None plan = tuple(loeser.Value(plaene[a, g].start) for a, auftrag in enumerate(auftraege) for g in range(len(auftrag))) return loeser.StatusName(status), int(loeser.ObjectiveValue()), dauer, plan if __name__ == "__main__": auftraege = baue_instanz() gaenge = sum(len(a) for a in auftraege) print("=" * 78) print(" WAS MEHRERE ARBEITER WIRKLICH BRINGEN") print("=" * 78) print(f"Job-Shop: {AUFTRAEGE} Auftraege, {MASCHINEN} Maschinen, " f"{gaenge} Arbeitsgaenge.") print(f"random_seed = {SOLVER_SEED} in JEDEM Lauf, " f"{WIEDERHOLUNGEN} Wiederholungen je Arbeiterzahl.\n") # Jede Konfiguration mehrfach - ein einzelner Zeitwert sagt nichts. ergebnisse: dict[int, list[tuple]] = {} for arbeiter in ARBEITERZAHLEN: ergebnisse[arbeiter] = [loese(auftraege, arbeiter) for _ in range(WIEDERHOLUNGEN)] print("=" * 78) print(" (1) Laufzeit: was die Arbeiter bringen") print("=" * 78) print(f"{'Arbeiter':>9} {'Status':>9} {'Makespan':>9} " f"{'schnellste':>11} {'langsamste':>11} {'Beschleunigung':>15}") print("-" * 78) basis_zeit = min(d for _, _, d, _ in ergebnisse[1]) for arbeiter in ARBEITERZAHLEN: laeufe = ergebnisse[arbeiter] zeiten = [d for _, _, d, _ in laeufe] ziele = {o for _, o, _, _ in laeufe} status = {s for s, _, _, _ in laeufe} print(f"{arbeiter:>9} {'/'.join(sorted(status)):>9} " f"{'/'.join(str(z) for z in sorted(ziele)):>9} " f"{min(zeiten):>10.2f}s {max(zeiten):>10.2f}s " f"{basis_zeit / min(zeiten):>14.1f}x") beste = min(ARBEITERZAHLEN, key=lambda w: min(d for _, _, d, _ in ergebnisse[w])) faktor = basis_zeit / min(d for _, _, d, _ in ergebnisse[beste]) print(f"\n Am schnellsten sind {beste} Arbeiter: Faktor {faktor:.1f} gegenueber") print(f" einem einzigen - also MEHR, als {beste} Kerne hergeben sollten.") print(f" Das ist kein Messfehler. CP-SAT laesst nicht {beste}-mal dieselbe Suche") print(" laufen, sondern verschiedene Strategien nebeneinander, die einander") print(" ihre Schranken zurufen. Der zweite Arbeiter ist deshalb nicht der") print(" 'zweite Kern', sondern ein anderes Verfahren.") letzte = ARBEITERZAHLEN[-1] if beste != letzte: zeit_letzte = min(d for _, _, d, _ in ergebnisse[letzte]) print(f"\n Und mehr ist nicht immer besser: {letzte} Arbeiter brauchen " f"{zeit_letzte:.2f}s") print(f" gegenueber {min(d for _, _, d, _ in ergebnisse[beste]):.2f}s bei " f"{beste}. Ab einer gewissen Zahl kosten Abstimmung") print(" und Speicherbandbreite mehr, als ein weiterer Suchstrang einbringt.") print("\n" + "=" * 78) print(" (2) Reproduzierbarkeit: derselbe Seed, derselbe Plan?") print("=" * 78) print(f"{'Arbeiter':>9} {'Laeufe':>7} {'verschiedene Plaene':>21} " f"{'Makespan':>10}") print("-" * 78) for arbeiter in ARBEITERZAHLEN: plaene = {p for _, _, _, p in ergebnisse[arbeiter] if p is not None} ziele = {o for _, o, _, _ in ergebnisse[arbeiter]} print(f"{arbeiter:>9} {WIEDERHOLUNGEN:>7} {len(plaene):>21} " f"{'/'.join(str(z) for z in sorted(ziele)):>10}") einer = len({p for _, _, _, p in ergebnisse[1]}) # Ab welcher Arbeiterzahl bricht die Reproduzierbarkeit? Das ist die # eigentliche Zahl - nicht, was acht Arbeiter anrichten. ab = next((w for w in ARBEITERZAHLEN if len({p for _, _, _, p in ergebnisse[w]}) > 1), None) print(f"\n Mit einem Arbeiter: {einer} Plan aus {WIEDERHOLUNGEN} Laeufen.") if ab is not None: viele = len({p for _, _, _, p in ergebnisse[ab]}) print(f" Schon mit {ab} Arbeitern: {viele} verschiedene Plaene aus " f"{WIEDERHOLUNGEN} Laeufen -") print(" bei identischem random_seed und identischem Zielwert.") print(f" Es braucht also keine acht Arbeiter, um die Reproduzierbarkeit") print(f" zu verlieren. {ab} genuegen.") print("\n Der Seed allein macht einen Lauf also NICHT reproduzierbar. Er legt") print(" fest, wie ein einzelner Suchstrang wuerfelt - nicht, welcher von") print(" mehreren zuerst fertig wird. Das entscheidet die Uhr.") print("\n" + "=" * 78) print(" (3) Was daraus fuer Tests folgt") print("=" * 78) print(" Ein Test der Form assert plan == erwarteter_plan besteht mal und") print(" scheitert mal, ohne dass sich am Code etwas geaendert haette.") print(" Zu pruefen sind stattdessen:") print(" * der ZIELWERT (hier in allen Laeufen gleich),") print(" * die Einhaltung aller Regeln (Abnahmepruefung, Kapitel Praxisfallen),") print(" * und der Status - nicht die Gestalt der Loesung.") print("\n Wer doch einen bestimmten Plan braucht - fuer eine abgedruckte") print(" Ausgabe, fuer einen Regressionstest -, setzt num_workers = 1.") print(f" Das kostet hier den Faktor {faktor:.1f} an Laufzeit und ist genau") print(" deshalb eine Entscheidung fuer den Test, nicht fuer den Betrieb.") print("=" * 78)