-
Notifications
You must be signed in to change notification settings - Fork 0
Home
Im Kontrast zu Meilenstein A hat diese Implementierung nichts mehr mit der ursprünglichen Idee zu tun. Unser neuer Ansatz bezieht sich auf den, in der Besprechung diskutierten, Vorschlag. Jede Unit wird dabei wie folgt parallel abgearbeitet: Eine Unit wird in logische Segmente unterteilt. Ein Segment besteht aus mehreren Paragraphen, bis ein Seitenumbruch erzwungen wird. Das hat den Vorteil, dass ein Segment vorzeitig gerendert und gedruckt werden kann. Der erzwungene Seitenumbruch ermöglicht es dem Algorithmus die Paragraphen entsprechend auf den Abschnitt der Seiten zu verteilen, da folgende Paragraphen definitiv nicht mehr auf diese Seiten abgebildet werden. Das Verarbeiten der Segmente wird den Hauptteil der Nebenläufigkeit darstellen.
Für jede Unit wird zunächst ein eigener Thread erstellt. In diesem wird als Erstes ein Parser-Thread erstellt und gestartet, damit andere Threads nebenläufig aus dem so entstehenden Dokument lesen können während dieses noch zusammengesetzt wird. Hier haben wir einen festen Pool von Threads, denen Aufgaben wiederholt zugeteilt werden. Diese Aufgaben bestehen aus dem Abarbeiten der einzelnen Paragraphen und Seitenumbrüche, sowie dem Rendern und Drucken der Segmente. Sobald die nach dem Beenden des Parser-Threads feststehende Segmentanzahl der Anzahl der gedruckten Segmente entspricht, kann die Unit fertiggestellt werden.
Diese Klasse hat die Aufgabe alle Threads für diese Unit zu erzeugen und auf die Beendigung der Abarbeitung zu warten, bzw. auf einen frühzeitigen Abbruch angemessen zu reagieren. Für jedes übergebene Dokument wird ein Thread dieser Klasse erstellt. Zunächst wird ein Parser-Thread erzeugt und gestartet. Danach wird ein ThreadPool fester Größe erzeugt und auf die Erfüllung einer Abbruchbedingung mittels wait() und notify() gewartet. Der ThreadPool wird durch einen Java ExecutorService implementiert, welchen wir im Folgenden nur noch Executor nennen werden. Die Bedingung ist erfüllt, wenn alle Segmente des Dokuments gedruckt wurden oder eine UnableToBreak-Exception geworfen wird. Sollte durch eine Exception das Warten unterbrochen worden sein, wird eine Fehlerseite gedruckt. Schlussendlich wird der Executor und dann der UnitHandler beendet.
Diese Klasse erhält ihren Namen dadurch, dass hier die wesentlichen Daten für die Unit gespeichert werden. Dazu zählen hauptsächlich eine Datenstruktur für alle gerenderten Seiten, sowie eine Datenstruktur für alle vorkommenden Segmente.
closeJob(Job)
Ruft die add()-Methode des jeweiligen Segments auf. Falls noch kein entsprechendes Segment in der dafür vorgesehenen Datenstruktur existiert, wird ein Neues erstellt.
addPages(int, List<Page>)
Diese Methode fügt eine Liste aus Seiten an die richtige Stelle in unserer Datenstruktur, um so die Seiten in der korrekten Reihenfolge zu haben. Sollten die Seiten die nächst folgenden Seiten sein in Bezug auf die richtige Reihenfolge, als jene die schon gedruckt worden sind, so werden sie dem Executor als Druckauftrag übergeben. Sollten dabei auch schon unmittelbar folgende Seiten in der Datenstruktur vorhanden sein, so werden für diese ebenfalls Druckaufträge erstellt.
Diese Klasse implementiert den DocumentBuilder und ersetzt damit die ursprüngliche Document-Klasse. Der Parser ruft die Methoden des Interface auf und signalisiert so, dass er einen Paragraphen oder Seitenumbruch gelesen hat. In diesen Fällen wird sofort ein Job an den Executor vergeben. Durch diese Änderung müssen in unserer Implementierung keine Daten intern zwischengespeichert werden.
appendForcedPageBreak(Position)
appendParagraph(Position)
Während der Verarbeitung der Unit durch den Parser werden diese Methoden aufgerufen. Dabei wird dem Executor jeweils ein ParagraphThread mit dem jeweiligen Job, welcher alle später benötigten Informationen enthält, übergeben.
finish()
Diese Methode wird vom Parser aufgerufen, sobald dieser mit dem Verarbeiten der Unit fertig ist. Dabei wird auch noch für das letzte Segment der Seitenumbruch hinzugefügt und eine Flag gesetzt.
Ein Klasse die Runnable und IBlockVisitor implementiert. Instanzen dieser Klasse werden dem Executor durch das ConcurrentDocument zugewiesen, wenn ein Paragraph oder ein Seitenumbruch verarbeitet werden soll. Alle dazu nötigen Informationen sind in dem übergebenen Job-Objekt enthalten. Intern wird die jeweilige visit() Methode aufgerufen und das Ergebnis wieder in das Job-Objekt geschrieben. Dann wird dieses der UnitData übergeben, welche sich in closeJob() um die Verarbeitung kümmert.
Hier werden Informationen über die SegmentID und die Paragraphreihenfolge innerhalb des Segments und das zu verarbeitende Element gespeichert. Ebenso wird auch das Ergebnis des verarbeiteten Elements, als Liste gespeichert, sofern diese schon vorhanden ist. Instanzen dieser Klasse dienen zur einfachen und standardisierten Übermittlung von Daten zwischen ThreadPool und UnitData. Da nur Daten zwischengespeichert werden enthält die Klasse nur Getter und Setter Methoden.
Die Segment-Klasse beinhaltet eine eigene ID und eine Liste von Items. Diese Liste speichert alle verarbeiteten Paragraphen eines logischen Segmentabschnitts. Eine Instanz dieser Klasse repräsentiert ein Segment wie oben beschrieben.
add(int, List<Item<Renderable>>, int)
Fügt die Liste der Items an die entsprechende Position der Datenstruktur ein. Falls das Segment nach Aufruf der add-Methode vollständig gefüllt ist (alle Elemente sind vorhanden), so wird ein Renderauftrag an den Executor für das Segment gestellt.
Das Programm startet mit der Main-Methode in der Rocket. Für jedes übergebene Dokument wird ein neuer UnitHandler-Thread gestartet und warten bis sich alle Threads beendet haben.
Jeder UnitHandler erzeugt einen eigenen ThreadPool aus der Anzahl an Threads, wie logische Kerne zur Verfügung stehen. Außerdem wird zu Beginn ein separater Thread, unabhängig des Executors, für das Parses des Inputs ins bereitgestellte ConcurrentDocument gestartet. Der separate Thread wird allein deshalb gebraucht, dass bei nur einem vorhanden Kern nicht ein sequentieller Ablauf entsteht, bei dem zuerst das gesamte Dokument erstellt wird. Während der UnitHandler auf seine Bedingung wartet, dass alle Seiten gedruckt wurden oder dass ein Fehler zum frühzeitigen Abbruch führt, werden im ConcurrentDocument die ersten Jobs vergeben:
Für jeden geparsten Paragraphen wird appendParagraph() aufgerufen. Ab diesem Zeitpunkt ist der neue Paragraph aber noch nicht vollständig gefüllt, weshalb immer der zuvor erstellte Paragraph in ein Job-Objekt gepackt wird und dann dem Executor ein neuer ParagraphThread übergeben wird, welcher diesen Job zugeteilt bekommt. Ebenso wird beim Aufruf von appendForcePageBreak() analog der letzte Paragraph als Job übermittelt und ein weiterer für den Seitenumbruch hinzugefügt. Das ConcurrentDocument zählt mit, bei welchem Segment der Parser angekommen ist und der wievielte Paragraph dieses Segments der aktuell hinzugefügte Paragraph ist. In appendForcePageBreak() wird dementsprechend das Segment erhöht und der Paragraph-Zähler zurückgesetzt. Beide Informationen werden dem Job ebenfalls übergeben.
Der Executor fängt nun an seine Aufgaben der Reihe nach auszuführen und neue Aufgaben an fertig werdende Threads zu verteilen. Dabei werden zunächst vermehrt ParagraphThreads ausgeführt.
Ein ParagraphThread nimmt sich das BlockElement aus dem Job und führt die jeweilige visit()-Methode aus. Die dadurch resultierende List aus Item wird dem Job-Object übergeben. Danach wird die closeJob()-Methode der UnitData aufgerufen.
In der closeJob()-Methode wird zunächst überprüft, ob schon ein entsprechendes Segment gemäß der Angabe im Job-Objekt existiert. Falls dies nicht der Fall ist, wird zunächst ein solchen erstellt. Dann wird auf dieses Segment die add()-Methode aufgerufen mit der Sequenz-Nummer, welche dem Paragraph-Zähler im Job-Objekt entspricht, und die zuvor resultierte List. Die add()-Methode bekommt zusätzlich die Anzahl an Paragraphen, die im fertigen Segment enthalten sein sollten. Da dies natürlich nicht von Anfang an bekannt ist, wird diese Anzahl erst korrekt übergeben, wenn ein ForcedPageBreak eingefügt wird. Ansonsten wird '-1' kontinuierlich übergeben, da dieser Wert nie mit der aktuellen Anzahl an Paragraphen übereinstimmen wird und als Platzhalter dient.
In der add()-Methode der Segment-Klasse wird anhand der Sequenz-Nummer diese Liste aus Items an die richtige Stelle in eine Map geschrieben. Sobald die erwartete Anzahl der Paragraphen im Segment erreicht ist, hier also die Anzahl an Schlüssel in der Map – da jeder Schlüssel einem Paragraphen im Segment entspricht – mit dieser übereinstimmt, wird dem Executor ein Auftrag zum Rendern dieses Segments übergeben. Sollte die erwartete Anzahl noch nicht feststehen wird dieser Abschnitt übersprungen.
Wenn der Executor diesen Auftrag abarbeitet, wird breakIntoPieces()-Methode auf die Items der Paragraphen im Segment aufgerufen und die so erhaltene Liste aus Pieces zunächst gerendert und dann als Liste an Pages der addPages()-Methode der UnitData übergeben.
Die addPages()-Methode fügt die gerenderten Seiten in eine Map ein, mit der SegmentID als Schlüssel für die korrekte Reihenfolge. Der Executor bekommt als nächstes den Auftrag die Seiten des Segments zu drucken, unter der Voraussetzung, dass die aktuell hinzugefügten Seiten die nächst höhere SegmentID besitzen als die zuletzt gedruckten Seiten. Dieser Vorgang wird wiederholt ausgeführt, falls die jeweils folgenden Seiten bezüglich ihrer SegmentID in der Map schon vorhanden sind.
Sollte der Executor diesen Auftrag abarbeiten, dann gibt er dem Printer die entsprechenden Seiten in printPages() mit und benachrichtigt anschließend den UnitHandler, dass dieser seine Wartebedingung überprüfen soll.
Sobald die Wartebedingung des UnitHandlers nicht mehr erfüllt ist, sind alle Paragraphen abgearbeitet und alle Seiten gedruckt. So wird dem Executor signalisiert sich zu beenden und das Document mittels finishDocument() des Printers fertiggestellt. Danach beendet sich der UnitHandler und damit die Unit erfolgreich.
Sollte ein ParagraphThread in seiner visit()-Methode auf eine "UnableToBreak-Exception" stoßen, die durch breakIntoPieces() geworfen wird, so wird in der UnitData die entsprechende Flag gesetzt. Unmittelbar danach wird die WaitCondition des UnitHandlers benachrichtigt, welche daraufhin sofort dem Executor einen Terminierungsbefehl mittels shutdownNow() gibt. Danach wird ohne auf die Terminierung zu warten der Parser beendet, eine Fehlerseite gedruckt und schließlich die Unit abgeschlossen.
Beim Erstellen des Executers mit newFixedThreadPool() wird intern ein neuer ThreadPoolExecutor mit einer LinkedBlockingQueue erstellt. Wird nun ein neuer Task submittet, wird intern ein neues Task-Objekt erstellt und der execute()-Methode übergeben. Diese wiederum offert dieses Objekt der workQueue, also der LinkedBlockingQueue, welche wiederum ein internes eigenes ReentrantLock verwendet und zum Einfügen setzt. Somit können mehrere Threads gleichzeitig an unseren Executor einen Task submitten.
Die boolean Variable isFinished wird im ConcurrentDocument initial auf False gesetzt. Der UnitHandler prüft in seiner Condition mit Hilfe der Methode isFinished() des ConcurrentDocuments, ob diese Variable auf True gesetzt wurde. Der Parser ruft genau ein einziges Mal die Methode finish() auf, welche den isFinished auf True setzt. Da hier ein Read-Write-Datarace unterschiedlicher Threads entstehen würde, ist diese Methoden mit dem synchronized Keyword versehen. Dadurch müssen die beiden Threads um ein Lock kämpfen und die Variable kann nur von jeweils einem Thread gleichzeitig gelesen bzw. verändert werden. Da nicht alle Methoden in dem ConcurrentDocument mit synchronize versehen sind, sollte erwähnt werden, dass diese Methoden isFinished weder lesend noch schreibend verwenden!
Die segmentCounter-Variable wird während dem Parsen nur von diesem erhöht. Erst danach wird der Wert der Variable mit Hilfe der getSegmentCounter()-Methode abgefragt. Es gibt keine weiteren Möglichkeiten auf diese Variable von außerhalb zuzugreifen. Dies passiert jedoch nur in der WaitCondition des UnitHandlers. Aufgrund der "short-circuit evaluation" wird sichergestellt, dass getSegmentCounter() nur aufgerufen wird, falls der erste Teil der Condition, also u.a. isFinished(), zu True auswertet. Das geschieht allerdings nur, wenn der Parser selbst finish() aufruft und somit das Parsen beendet. Wie bereits erwähnt wird danach der segmentCounter nicht mehr erhöht.
The && and || operators perform Conditional-AND and Conditional-OR operations on two boolean expressions. These operators exhibit "short-circuiting" behavior, which means that the second operand is evaluated only if needed. ~ siehe javaDoc
Auf ein Segment wird nur durch die add()-Methode zugegriffen. Die einzige Stelle, an der dies aufgerufen wird, ist die closeJob()-Methode der UnitData. Da diese synchronized ist, kann jede Segment-Instanz jeweils nur von einem Thread beansprucht werden. Innerhalb dieses Segments kann es deshalb nicht zu einem Datarace kommen!
ParagraphThreads beenden sich mit closeJob() der UnitData. Wegen "synchronize" erhält jeweils nur einer dieser Threads ein Lock und führt die add()-Methode auf einem Segment aus. Segmente werden so nie gleichzeitig beschrieben. Erst nachdem die jeweilige add()-Methode abgeschlossen ist und somit die closeJob() beendet ist, wird das Lock wieder freigegeben und der nächste ParagraphThread hat exklusiven Zugriff auf EIN Segment.
Die einzige Methode die auf die Map "segments" zugreift, ist die closeJob()-Methode. Da diese Methode synchronized ist, kann immer nur ein Thread gleichzeitig auf diese Map zugreifen, sie lesen und/oder sie verändern. Hier kann also kein Datarace entstehen und eine normale HashMap ist ausreichend.
Auf die Map pages wird an zwei Stellen zugegriffen: In der addPages()-Methode der UnitData selber und in einem separaten Thread, der die jeweiligen Seiten drucken soll. Die addPages()-Methode selbst ist synchronized und ruft nur die put()-Methode der Map auf. Es ist die einzige Stelle, an der etwas in die Map eingefügt wird. Dementsprechend kann in unserem Fall nur jeweils ein Thread in die Map schreiben. Threads die Seiten drucken, rufen nur die get()-Methode auf "pages" auf. Es ist sichergestellt, dass an jeden Schlüssel der Map nur ein einziges Mal geschrieben wird (aufgrund unserer Job-Logik) und erst zeitlich danach davon gelesen wird.
Wodurch ist sichergestellt, dass erst gelesen wird, nachdem geschrieben wurde?
Zuerst wird in addPages() – z.B. an die Stelle x – in die Map eingefügt. Erst danach wird überhaupt der Druckauftrag dem Executor übergeben, der aus Stelle x wieder liest. Zudem wird das Übergeben von weiteren Aufträgen an den Executor in einer While-Schleife getätigt, die prüft, ob die zu übergebenden Seiten in der Map bereits vorhanden sind. Dafür müssen sie natürlich vorher einmal geschrieben worden sein. Da wir diese NIE wieder verändern, können diese Threads darauf zugreifen und es entsteht an dieser Stelle kein Read-Write-Datarace.
Die einzige Methode, die auf printQueuePages zugreift, ist die addPages()-Methode der UnitData. Da diese Methode synchronized ist, entsteht hier kein Datarace.
Diese Variable hat ihr eigenes ReentrantLock. Sie wird in den Threads, die für das Drucken von Seiten zuständig sind, gelesen und geschrieben und dort dementsprechend gelockt. Zusätzlich wird sie im UnitHandler durch die getPrintedPages()-Methode abgefragt, da sie in der WaitCondition wichtig ist. Dementsprechend ist die getPrintedPages()-Methode ebenfalls durch das ReentrantLock geschützt. Weitere Variablennutzungen gibt es nicht.
Diese boolesche Variable kann ausschließlich mit isUnableToBreak() gelesen und mit setUnableToBreak() geschrieben werden. Beide dieser Methoden sind synchronized und somit können nicht zwei Threads gleichzeitig auf unableToBreak zugreifen. Es kann also nicht zu einem Datarace kommen.
Um frühzeitig zu Drucken ist die Einteilung in Segmente essenziell. Ein Segment ist eine Abfolge von Paragraphen, die mit einem ForcedPageBreak beendet wird. Ab dieser Stelle wird eine neue Seite erzwungen und der Algorithmus kann die Paragraphen des Segments auf Seiten verteilen, da die folgenden Paragraphen unabhängig davon sowieso auf neue Seiten verteilt werden müssen. Mit anderen Worten: Segmente müssen folgende Segmente nicht in ihrer Berechnung – im eigentlichen Algorithmus – beachten. Jedes Segment wird unmittelbar nach seiner vollständigen Abarbeitung gerendert. Sollten nach dem Rendern Seiten zum Drucken zur Verfügung stehen und diese die Reihenfolge der gedruckten Seiten nicht verletzen, werden Druckaufträge sofort an den ThreadPool übergeben. Seiten, die zwar gerendert sind, für die aber eine Lücke in der Seitenreihenfolge entstehen würde, das heißt vorherige Seiten noch nicht gedruckt wurden, können und dürfen allerdings nicht gedruckt werden. Deshalb überprüfen wir jeweils ob der Nachfolger der zuletzt gedruckte Seitennummer bereits gerendert vorliegt. Sollte dies der Fall sein, so geben wir dazu ebenfalls einen Druckauftrag auf und wiederholen diesen Vorgang. Andernfalls kann nicht sofort gedruckt werden und es passiert nichts weiter.
In den Threads die Seiten drucken sollen, muss der Drucker gelockt werden, damit nicht mehrere Threads versuchen gleichzeitig zu drucken und so zeitgleich auf den OutputWriter schreiben. Der so resultierende Text wäre eine wirre Mischung aus verschiedenen Paragraphen und kein zusammenhängender Text der Sinn ergibt.
Sollten noch Fragen offen sein, die sich auf die genau Implementierung beziehen bietet unsere Codebasis eine ausführliche Grundlage an Kommentaren, um diese zu befriedigen.