IOException.de

Caches dienen in der Informatik als Methode, Zugriffe auf bestimmte Daten zu beschleunigen, in dem diese vorgelagert/gepuffert werden. Sie sind in verschiedensten Bereichen zu finden, unter anderem auf Prozessoren, in Festplatten, aber auch in Technologien wie dem Web. Verschiedene Verdrängungsstrategien ermöglichen es, die beschränkte Kapazität eines Caches zu berücksichtigen, so dass nur wichtige Werte im Cache gelagert werden. Least Recently Used (LRU), ist eine solche Strategie, die häufig angewandt wird. Sie sortiert die Werte im Cache nach der letzten Nutzung. Wird auf ein Element über einen längeren Zeitraum nicht mehr zugegriffen, so wird es aus dem Cache verdrängt.

In Java lässt sich ein solcher LRU-Cache besonders einfach implementieren, da die Klasse java.util.LinkedHashMap bereits die wesentlichen Mechanismen unterstützt. Eine HashMap ist eine Hash-Tabelle, die Zugriffe auf Werte über ihre Schlüssel regelt. Zusätzlich verkettet die LinkedHashMap aber die Werte noch in einer Liste, womit auch eine Traversierung in Einfügereihenfolge ermöglicht wird. Mithilfe eines Flags in einem der Konstruktoren kann dieses Verhalten geändert werden, so dass bei jedem Zugriff das angesprochene Element neu in diese Liste eingereiht wird. Damit verwaltet die Liste die Zugriffe und ist Basis für die LRU-Strategie.

Die Methode removeEldestEntry() der LinkedHashMap wird bei jedem Schreibezugriff auf die Map, also nach Einfügeoperationen über put() oder putAll() automatisch aufgerufen und bietet die Möglichkeit, durch Überschreiben der Methode die Verdrängungsstrategie zu implementieren. Diese Methode gibt ein boolean zurück, ob der älteste Eintrag gelöscht werden soll. Es ist auch möglich, innerhalb der Methode selbst die Liste zu manipulieren, dann sollte allerdings die Methode immer false zurückgeben. Für den LRU-Cache reicht es aus, die Größe der Map mit dem gewünschten Maximum zu vergleichen. Ist der Inhalt der Map zu groß, so soll das letzte Element gelöscht werden.

Im Folgenden nun der Code dazu. Zu beachten ist noch, dass es sich um eine threadsichere Klasse handelt, da die Map explizit synchronisiert wird.

import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * A thread-safe LRU cache implementation based on internal LinkedHashMap.
 * 
 * @author Benjamin Erb
 *
 * @param <K> Entry Key Type
 * @param <V> Entry Value Type
 */
public class LRUCache<K, V>
{
	public static final int DEFAULT_MAX_SIZE = 1000;

	private final Map<K, V> internalMap;

	public LRUCache()
	{
		this(DEFAULT_MAX_SIZE);
	}

	public LRUCache(final int maxSize)
	{
		this.internalMap = (Map<K, V>) Collections.synchronizedMap(new LinkedHashMap<K, V>(maxSize + 1, .75F, true)
		{
			private static final long serialVersionUID = 5369285290965670135L;

			@Override
			protected boolean removeEldestEntry(Map.Entry<K, V> eldest)
			{
				return size() > maxSize;
			}
		});
	}

	public V put(K key, V value)
	{
		return internalMap.put(key, value);
	}

	public V get(K key)
	{
		return internalMap.get(key);
	}

}

Viele Netzwerkprotokolle teilen ihre Kommunikationsentitäten in Pakete oder Nachrichten auf. Manche Protokolle erwarten außerdem eine sequentielle Abarbeitung (z.B. TCP auf Transportebene), auch wenn die darunterliegenden Protokollschichten dass nicht unbedingt unterstützen. Für die Implementierung eines Nachrichtenpuffers für RTSP-Nachrichten, die ungeordnet ankommen können, aber sequentiell abgearbeitet werden müssen, habe ich eine entsprechende Datenstruktur in Java erstellt. RTSP-Nachrichten besitzen ein Cseq-Header-Feld, welches die einzelnen Nachrichten durchnummiert und als ordnendes Element genutzt werden kann.

Die folgende generische Implementierung erwartet, dass alle Nachrichtenelemete eindeutige und miteinander vergleichbare Identifier besitzen. Desweiteren erlaubt diese Implementierung zwar das parallele Schreiben in die Queue, allerdings sollte das Lesen durch einen einzelnen Thread realisiert werden. Ansonsten kann die Ordnung nach dem Entnehmen aus der Queue durch unterschiedlich lange Laufzeiten der Worker-Threads wieder verloren gehen. Außerdem realisiert diese Implementierung die sequentielle Ordnung eines vollständigen Nachrichtenstroms. Nachrichtenverluste oder Timeouts werden nicht behandelt. Hierfür eignen sich eher Automatic repeat request Protokolle.

Interface für MessageQueue
Eine MessageQueue bietet die grundlegenden Funktionen zum Einfügen und Entfernen von Elementen sowie Hilfsmethoden für die Größe der Queue an.

/**
 * Interface for message queues.
 * 
 * @author Benjamin Erb
 * 
 * @param <E>
 *            Element type
 */
public interface MessageQueue<E>
{
	/**
	 * Adds element to queue. Blocks until element can be added.
	 * 
	 * @param e
	 *            new element
	 * @throws InterruptedException
	 */
	void push(E e) throws InterruptedException;

	/**
	 * Takes an element from the queue. Blocks until element becomes available.
	 * 
	 * @return taken element
	 * @throws InterruptedException
	 */
	E pop() throws InterruptedException;

	/**
	 * Gets the size of the queue
	 * 
	 * @return
	 */
	int size();

	/**
	 * Checks whether queue is empty or not.
	 * 
	 * @return
	 */
	boolean empty();

}

Interface für speziellen Comparator
Dieses spezielel Comparator-Interface fügt Methoden hinzu für das Erfragen von vorherigen und nachfolgenden Identifiern an. Außerdem können Nachrichten, Identifier oder eine Kombination aus beidem verglichen werden.

import java.util.Comparator;

/**
 * An enhanced comparator interface for retrieving preceding and subsequent identifiers.  
 * 
 * @author Benjamin Erb
 *
 * @param <E> Element type
 * @param <T> Element identifier
 */
/**
 */
public interface SequentialComparator<E, T extends Comparable<T>> extends Comparator<E>
{
	/**
	 * Returns the subsequent identifier of a given entity.
	 * 
	 * @param t
	 *            entity
	 * @return subsequent identifier
	 */
	T getNext(E e);

	/**
	 * Returns the preceding identifier of a given entity.
	 * 
	 * @param t
	 *            entity
	 * @return preceding identifier
	 */
	T getPrevious(E e);

	/**
	 * Compares two entities.
	 * 
	 * @param t1
	 *            entity 1
	 * @param t2
	 *            entity 2
	 * @return a negative integer, zero, or a positive integer as the first
	 *         argument is less than, equal to, or greater than the second.
	 */
	int compare(T t1, T t2);

	/**
	 * Compares an entity and an identifier.
	 * 
	 * @param t1
	 *            entity
	 * @param e2
	 *            identifier
	 * @return a negative integer, zero, or a positive integer as the first
	 *         argument is less than, equal to, or greater than the second.
	 */
	int compare(T t1, E e2);

	/**
	 * Compares an identifier and an entity.
	 * 
	 * @param e1
	 *            identifier
	 * @param t2
	 *            entity
	 * @return a negative integer, zero, or a positive integer as the first
	 *         argument is less than, equal to, or greater than the second.
	 */
	int compare(E e1, T t2);
}

SequentialMessageQueue Implementierung
Die eigentliche Queue-Implementierung greift intern auf eine PriorityBlockingQueue zurück, allerdings wird durch die Kapselung sichergestellt, dass nur das nächste zu erwartende Element entnommen werden kann.

import java.util.concurrent.PriorityBlockingQueue;

/**
 * A message queue for sequential message processing. This queue orders incoming
 * messages entities by their identifiers and outputs entites in-order. This
 * queue supports more than one writing thread. However, it is designed for one
 * reading/consuming thread in order to prevent out-of-order processing by
 * multiple threads.
 * 
 * @author Benjamin Erb
 * 
 * @param <E>
 *            Message entity
 * @param <T>
 *            Message identifier
 */
public class SequentialMessageQueue<E, T extends Comparable<T>> implements MessageQueue<E>
{
	/**
	 * Lock object for queueing
	 */
	private final Object lock = new Object();

	/**
	 * internal queue
	 */
	private final PriorityBlockingQueue<E> internalQueue;
	/**
	 * comparator
	 */
	private final SequentialComparator<E, T> comparator;

	/**
	 * Represents the identifier of the next expected entity
	 */
	private T expectedIdentifier;

	public SequentialMessageQueue(SequentialComparator<E, T> comparator, T initialIdentifier)
	{
		this.internalQueue = new PriorityBlockingQueue<E>(16, comparator);
		this.comparator = comparator;
		this.expectedIdentifier = initialIdentifier;
	}

	@Override
	public boolean empty()
	{
		return internalQueue.isEmpty();
	}

	@Override
	public E pop() throws InterruptedException
	{
		synchronized (lock)
		{
			E firstElement;
			while ((firstElement = internalQueue.peek()) == null || comparator.compare(firstElement, expectedIdentifier) != 0)
			{
				lock.wait();
			}
			internalQueue.remove();
			expectedIdentifier = comparator.getNext(firstElement);
			return firstElement;
		}
	}

	@Override
	public void push(E e) throws InterruptedException
	{
		synchronized (lock)
		{
			internalQueue.put(e);
			lock.notifyAll();
		}

	}

	@Override
	public int size()
	{
		return internalQueue.size();
	}

	/**
	 * Returns the used comparator
	 * 
	 * @return
	 */
	protected SequentialComparator<E, T> getSequentialComparator()
	{
		return comparator;
	}

}

Die zentrale Frage bei der Implementierung des Schedulers ist die Frage nach der Datenstruktur in der die Tupel gehalten werden. Der Scheduler soll, gemäß der SJF Strategie, immer das Tupel auswählen dessen Joblänge am kürzesten ist. Für solche eine Problemstellung eignet sich die Datenstruktur std::set bzw. std::multiset parametrisiert auf std::pair am Besten. Jedoch ist zu beachten dass die Sortierung bei std::pair normalerweise auf dem ersten Element, also bei uns auf der ThreadId basiert. Daher müssen wir den Vergleichsoperator des std::multiset mit unserem eigenen Operator überschreiben.

/*
Scheduler.h
*/

/*
this struct is used to have a different compare operator for std::pair in my std::multiset. I have to compare the second argument of pair!
*/
struct compareBySecond {
  bool operator() (const std::pair<int,int>& a, const std::pair<int,int>& b) const
  {return (a).second<(b).second;}
};

/*
my scheduler class, schedules the requests
*/

class Scheduler{

  public:
    [  ... ]

  private:
    //multiset holds all pairs with (thread,job)
    std::multiset< std::pair<int,int> , compareBySecond > jobs;

    [  ... ]
    
};

Als Synchronisationselemente kommen normale RW-Locks und Condition-Variablen zur Anwendung, auf diese Standardelemente wird hier nicht weiter eingegangen. Guten Beispielcode für diese Elemente findet sich in den Qt3.3-Quellen (z.B.: QReadWriteLock).

Die einzige Methode die von den Anfragethreads ausgeführt wird ist Scheduler::enqueue(int thread, int length) (siehe Teil 1). Bemerkenswert ist hier nur der letzte Aufruf, da wir ja immer eine möglichst volle Queue gewährleisten wollen wartet der Scheduler darauf dass er Jobs erst auswählt, wenn die Queue voll ist oder eine volle Queue nicht mehr erreicht werden kann. Hierzu ist die Condition-Variable cv_queuefull nötig auf die beim Überprüfen der Queue gewartet wird.

/*
enqueues a job
*/
void  Scheduler::enqueue(int thread, int length){

  //wait if queue is full or thread has already job in queue
  while(isQueueFull() || threadHasRequestInQueue(thread)){
    cv->wait();
  }
  
  //new pair for enqueueing
  std::pair<int,int> tmp(thread,length);
  
  //lock and insert in queue
  rw->lockWrite();
  jobs.insert(tmp);
  rw->unlockWrite();
  
  //we did it!
  std::cout << "requester " << thread << " joblength " << length << std::endl;
  
  //signal scheduler thread that thread is now maybe full
  cv_queuefull->signal();  
}

Die beiden hier noch verwendeten Methoden bool Scheduler::isQueueFull() und bool Scheduler::threadHasRequestInQueue(int thread) sind aus Implementierungssicht trivial.

Die wichtigste Methode in der die eigentlichen Jobs “abgearbeitet” werden ist Scheduler::processNextJob(), hier muss zunächst überprüft werden ob die Queue bereit ist. Dies geschieht in Scheduler::QueueReady() und ist weiter unten erläutert. Für die Auswahl des nächsten Jobs mit der kürzesten Joblänge reicht es das erste Element des Sets zu entnehmen, dies ist nach der von uns definierten Ordnung im Set automatisch der Job mit der kürzesten Laufzeit. Nach dem Abarbeiten wird der Job aus der Queue gelöscht. Falls gerade ein Thread darauf wartet einen Job zu enqueuen dem es vorher nicht möglich war zu enqueuen da er schon einen Job in der Queue hatte wird dieser zum Schluss noch aufgeweckt.

/*
processes the next job
*/
void  Scheduler::processNextJob(){
  //is queue ready? 
  QueueReady();
  
  //get job
  rw->lockRead();
  std::pair<int,int> tmp = *( jobs.begin());
  rw->unlockRead();
  
  //process .......... :-)
  std::cout << "service requester " << tmp.first << " joblength " << tmp.second << std::endl;

  //delete from queue
  rw->lockWrite();
  jobs.erase(jobs.find(tmp)); 
  rw->unlockWrite();

  //wake up all guys that wait on enqueuing
  cv->signalAll();
}

Für die Überprüfung ob die Queue bereit ist, sind zwei Bedingungen nötig: Ersten muss die Queue voll sein, siehe Annahmen. In dem Fall dass schon Threads beendet wurden kann es passieren dass die Queue nicht voll werden kann. In diesem Fall soll der Scheduler trotz nicht voller Queue seine Arbeit aufnehmen. Hier wäre noch Optimierungspotential gegeben, da nicht garantiert wird dass die Queue in momentanen Zustand der Threads immer so voll wie möglich ist.

/*
checks if queue is ready
*/
void Scheduler::QueueReady(){
  while(!isQueueFull() && (jobs.size() < nrOfThreads) ){
    //wait if our queue is not full or some threads are already dead, thus we can proceed if #threads<Queuesize
    cv_queuefull->wait();
  }
}

Anscheinend war es wohl mal wieder Zeit für einen etwas theoretischeren Beitrag. Binärbäume sind in der Informatik eine wichtige Datenstruktur, mit der sich beispielsweise Rechenterme abspeichern lassen. Sie bestehen aus Knoten, die einen linken und/oder rechten Knoten besitzen können, welche selbst wieder Unterknoten besitzen können. Mehr dazu findet man unter anderem bei Wikipedia.

Binärbäume werden über rekursive Aufrufe traversiert, bei (Binär-)Bäumen gibt es hauptsächlich drei wichtige Traversierungen: Pre-Order, In-Order und Post-Order. Ein Traversierungschritt besteht jeweils aus drei Teilschritten: Auswertung des aktuellen Knotenwerts, und der rekursive Abstieg in den linken und rechten Unterknoten. Der Name gibt jeweils an, an welcher Stelle die Auswertung des aktuellen Knotens stattfindet. Bei einer Post-Order-Traversierung wird also erst, der linke und der rechte Knoten rekursiv aufgerufen, und dann erst der aktuelle Knoten selbst ausgewertet. Hier ein Beispiel:

Pre-Order: L A H L O
In-Order: H A L L O
Post-Order: H L A O L
Nun ist es eine beliebte (Prüfungs-)Aufgabe, anhand zweier gegebenen Traversierungen den Originalbaum rekonstruieren zu lassen. In manchen Lehrbüchern ist zu finden, dass dies mit zwei Traversierungen möglich ist. Allerdings misslang mir dies bei einigen Versuchen mit Pre- und Post-Order-Ausagabe und ich sah mir das Ganze genauer an. Die Aussagen ist nämlich nicht ganz korrekt, ohne eine In-Order-Ausgabe ist dies nur in bestimmten Situationen möglich (nämlich wenn der Baum vollständig ist). Doch wieso klappt es nur mit Pre & Post nicht? Hier ein Gegenbeispiel:

Bei beiden Bäumen ergibt sich als Pre-Order-Durchlauf „A B C“ und als Post-Order „C B A“, obwohl die Binärbäume nicht identisch sind.

Nun ein vollständiger Baum:

Hier ergeben sich folgende Traversierungen:
Pre-Order: A B D E C F G
Post-Order: D E B F G C A
Hiermit lässt sich der Baum rekursiv rekonstruieren, indem man die Traversierung rückwärts durchspielt und dabei jeweils Teilbäume wiederherstellt.
Wird nun jedoch der rote F-Knoten entfernt (was die Vollständigkeit aufhebt), so ist nur anhand des Pre- und Post-Order-Durchlaufs nicht mehr eindeutig, ob der Knoten G links oder rechts unterhalb des Knotens C sich befindet. Also lassen sich nur vollständige Bäume mit einer Kombination von Pre und Post rekonstruieren.

Bei allen anderen Bäumen ist dies mit einer Kombination eines In-Order-Druchlaufs zusammen mit einem Pre- oder Post-Order-Durchlaufs möglich.

Update: Nach längerer Suche bin ich im Internet noch auf diese Seite gestoßen, die das Problem ebenfalls anspricht und Rekonstruktionen noch genauer untersucht.

Unter einem Heap (im Deutschen gelegentlich auch als Halde bezeichnet) versteht man eine baumartige Datenstrukur mir einer besonderen Eigenschaft: Die Nachfolgeknoten jedes Knotens besitzen immer kleinere (oder immere größere) Schlüsselwerte als der Knoten selbst. Somit entsteht ein partiell geordneter Baum. Ein Heap wird vor allem für zwei Dinge eingesetzt: Zur Realisierung von gewichteten Prioritätswarteschlangen und zur Sortierung (Heap Sort).

Bei der Sortierung wird eine nicht geordnete Liste in einen binären Heap eingelesen und dabei entsprechend der Heapeigenschaft umstrukturiert. Danach ist die Wurzel genau der Knoten mit dem maximalen Wert, und alle Knoten jeweils kleiner als ihre Elternknoten, aber größer als ihre Kindknoten. Nun beginnt der eigentliche Sortierschritt. Iterativ wird immer der aktuelle Wurzelknoten aus dem Heap entnommen und an den sortieren Bereich angefügt. Danach wird der Heap wieder erneuert und ein neuer Knoten mit dem Maximalwert der übrigen Knoten wandert an die Stelle der Wurzel. Meistens wird diese Datenstruktur auf einem Array realisiert. Das Array wird zunächst in eine gültige Heap-Ordnung überführt. Danach wird jeweils der erste Werte als verbliebenes Maximun in den hinteren Bereich des Arrays einsortiert, wo schrittweise die Ordnung aufgebaut wird. Dabei wird einfach das Wurzelelement mit dem Element an der richtigen Stelle am Ende des Arrays getauscht. Nun wird erneuert der Heap korrigiert.

In der Visualisierung ist der binäre Heap als sogenannter Sunburst Tree dargestellt. Die Wurzel entspricht dem inneren Kreis, die Kindknoten jeweils den konzentrisch abstehenden Kreisbogensegmenten. Die Ordnung entspricht der farblichen Sortierung von Schwarz zu Rot. Im Video sind die einzelnen Phasen gut sichtbar. Zunächst wird die chaotische Ordnung durch eine Ordnung entsprechend des Heaps angepasst. Hier wandern die dunklen Felder nach innen und die roten Felder nach außen. Anschließend beginnt die Sortierung. Im Falle des ersten Elements bedeutet dies das Tauschen des dunkelsten schwarzen Feldes (Kreismitte) mit dem Ende des Arrays. Anschließend wird wieder die Heapstruktur hergestellt. Somit ensteht schrittweise die Ordnung – im Uhrzeigersinn und von außen nach innen.

Realisiert habe ich die animierte Visualisierung mithilfe von Java (zur Grafikgenerierung) und mencoder (Stopmotion). Auf vimeo gibt es außerdem das Video in hoher Qualität.

Originalartikel: Visualisierung Heap / Heap Sort