operations_research/OR_HTML_04/programme/Parallele_Suche.py

#!/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)