5.5 Synchronisation

Navigator ist der Rover 2 aus dem LEGO Bausatz 9736 Exploration Mars, erweitert um einen Rotations-Sensor zur Navigation und ausgestattet mit zwei Touch-Sensoren. Alternativ könnte man Trusty mit einem Rotations-Sensor ausbauen.

Navigator ist mit Subsumption programmiert und wendet sich wie Trusty von Hindernissen ab. Zusätzlich werden aber Messages ausgewertet: 1 und 3 halten jeweils den linken oder rechten Motor an, bis die Message nochmals eintrifft, das heißt, Navigator dreht sich oder bleibt sogar stehen. Die Messages werden außerdem mit Sounds quittiert. Message 2 macht die im Rotations-Sensor über alle Drehungen akkumulierte Zahl rückgängig, das heißt, Navigator bewegt sich dann wieder in der ursprünglichen Richtung.

Das eigentliche Programmierproblem besteht darin, daß es verschiedene Tasks gibt, die alle -- mit oder ohne Subsumption -- auf eine Message warten. Eine Message muß deshalb zuverlässig allen Interessenten zugestellt werden, bevor die nächste empfangen wird.

Algorithmus

Der Algorithmus für die Zustellung wird zunächst als Java-Klasse beschrieben und später so gut wie möglich in nqc implementiert.

Ein Messenger -Objekt arbeitet als Thread, der Nachrichten an Kunden zustellt:

5/Messenger.java

/** distributes a message to up to 32 clients.
 */
public class Messenger implements Runnable {
 /** loop to distribute messages:
     wait for all clients to post done, mark message as empty,
     wait for next message, post all clients.
   */
 public synchronized void run () {
   try {
     for (;;) {
	empty = true; notifyAll();
	while (empty) wait();
	done = 0; notifyAll();
       while (done != all) wait();
     }
   } catch (InterruptedException e) { }
 }

Ein Auftraggeber wartet durch set() , bis er eine Nachricht hinterlegen kann, die dann sicher an alle Kunden ausgeliefert wird:

5/Messenger.java

  /** ok to set message. */				protected boolean empty;
 /** current message. */				protected int message;
 /** wait until there is room for a message. 
   */
 public synchronized void waitForEmpty () throws InterruptedException {
   while (!empty) wait();
 }
 /** set message: wait until there is room for a message, then set it.
     @param msg to set.
   */
 public synchronized void set (int message) throws InterruptedException {
   waitForEmpty();
   this.message = message; empty = false; notifyAll();
 }

Ein Kunde erhält mit login() ein Identifikations-Bit, und gibt es mit logout() wieder frei:

5/Messenger.java

  /** mask of clients. */				protected int all;
 /** current completions. */				protected int done;
 /** become a client.
     @return handle to be used to wait for a message.
     @throws RuntimeException if no more clients can be accepted.
   */
 public synchronized int login () {
   for (int handle = 1; handle != 0; handle <<= 1)
     if ((handle & all) == 0) {
       all |= handle; done |= handle; return handle;
     }
   throw new RuntimeException("no more clients");
 }
 /** validate a handle: one bit, set in all.
     @param handle to be validated.
     @throws IllegalArgumentException if handle is invalid.
   */
 public synchronized void validate (int handle) {
   if ((all & handle) != 0)
     for (int bit = 1; bit != 0; bit <<= 1)
       if (bit == handle) return;
   throw new IllegalArgumentException("bad handle");
 }
 /** cease to be a client.
     @param handle to be recycled.
     @throws IllegalArgumentException if handle is invalid.
   */
 public synchronized void logout (int handle) {
   validate(handle);
   all &= &#126;handle; done &= &#126;handle;
 }

Mit dem Bit kann er durch get() auf eine Nachricht warten und durch done() anzeigen, daß er die Nachricht fertig bearbeitet hat:

5/Messenger.java

  /** retrieve next message: wait for handle to be cleared in done, take message.
     @param handle obtained from login().
     @return next message.
     @throws IllegalArgumentException if handle is invalid.
   */
 public synchronized int get (int handle) throws InterruptedException {
   validate(handle);
   while ((done & handle) != 0) wait();
   return message;
 }
 /** done with message: post handle in done.
     @param handle obtained from login().
     @throws IllegalArgumentException if handle is invalid.
   */
 public synchronized void done (int handle) {
   validate(handle);
   done |= handle; notifyAll();
 }
}

Entscheidend sind die Identifikations-Bits. Eine Variable all enthält je ein Bit für jeden Kunden. Zu Beginn einer Zustellung wird eine andere Variable done auf 0 gesetzt. get() wartet darauf, daß das relevante Bit in done 0 ist, done() setzt es auf 1. Die Zustellung ist offensichtlich beendet, wenn done und all wieder gleich sind.

Da die verschiedenen Tasks (in Java Threads) parallel ausgeführt werden, muß man den Zugriff auf die Variablen all und done so absichern, daß eine Task eine Änderung komplett durchführen kann, bevor eine andere Zugriff erhält. Messenger wird als Monitor verwendet und alle Zugriffe sind mit synchronized abgesichert.

Messenger kann mit Demo getestet werden. Clickt man auf einen der drei Knöpfe im Sender -Panel, werden alle inaktiviert und eine Zahl wird über einen Thread mit set() an Messenger geschickt, damit der Event-Thread dabei nicht blockiert. Jedes der Receiver -Textfelder hat einen Thread, der mit get() eine Nachricht holt, auf einen Click in das Feld wartet, und die Nachricht mit done() quittiert. Merkt der Sender -Thread mit waitForEmpty() , daß der Messenger frei ist, werden die Knöpfe wieder aktiviert.

Implementierung

In nqc empfängt eine Task die Messages und stellt sie Interessenten zu:

5/navigator.nqc

// message receivers
#define	breakW	(1 << 3)
#define musicW		(1 << 4)
#define fixW		(1 << 5)
				// bits for all message receivers combined
#define	ALL	(breakW|musicW|fixW)
 
int ready = 0;				// bit(s) set for task requiring callback
int message = 0;				// published message
int complete = 0;				// current completions
// distribute messages to up to 16 clients: wait for next message,
// post all clients, wait for all clients to post complete.
task messenger () {
 while (true) {
   ClearMessage();
   do message = Message(); while (message == 0);
   complete = 0;					// atomic
   until (complete == ALL);					// atomic
 }
}
 
// wait for next message.
void get (const int& handle) {
 until ((handle & complete) == 0);					// might loop once too many
}
 
// mark completion.
void done (const int& handle) {
 complete |= handle;					// atomic
}

Man kann sich mit nqc -L davon überzeugen, daß die als atomic markierten Anweisungen in einzelne Bytecodes übersetzt werden. Die Firmware kann die Ausführung eines Bytecodes nicht unterbrechen, das heißt, daß der Zugriff auf complete im Sti von synchronized erfolgt.

Die Bremsen werden unabhängig von Subsumption mit einem trivialen endlichen Automaten bearbeitet:

5/navigator.nqc

// message 1/3: left/right break pedal
task breaks () {
 int left = 1, right = 1;
 while (true) {
   get(breakW);
   switch (message) {
   case 1: doBreak(leftM, left); break;
   case 3: doBreak(rightM, right); break;
   }
   done(breakW);
 }
}
 
void doBreak (const int& motor, int& state) {
 if (state != 0) { Off(motor); state = 0; }
 else { On(motor); state = 1; }
}

Die Musik spielt unabhängig dazu als Klient von messenger :

5/navigator.nqc

// message 1/3: provide accompaniment
task music () {
 while (true) {
   get(musicW);
   if (message == 1) PlaySound(2);
   if (message == 3) PlaySound(3);
   done(musicW);
 }
}

Die Kurskorrektor erfolgt durch Subsumption -- allerdings nur, wenn beide Motoren in Betrieb sind. Merkwürdigerweise kann man in nqc 2.1 den Zustand der Motoren nicht abfragen, aber man kann eigene Makros vereinbaren; @() kombiniert Source und Wert für einen RCX-Zugriff:

5/navigator.nqc

// macros for motor state
#define	MOTOR_A		0
#define	MOTOR_B		1
#define	MOTOR_C		2
#define	isOn(n)		((@(0x30000 + (n)) & 0x80) != 0)
#define	isOff(n)		((@(0x30000 + (n)) & 0xc0) == 0x40)
#define	isFloat(n)		((@(0x30000 + (n)) & 0xc0) == 0x00)
#define	isFwd(n)		((@(0x30000 + (n)) & 0x08) != 0)
#define	isRev(n)		((@(0x30000 + (n)) & 0x08) == 0)
#define	Power(n)		(@(0x30000 + (n)) & 0x07)
 
#define leftM		OUT_C		// drive motors on outputs C and A
#define rightM		OUT_A
#define	moving	(isOn(MOTOR_A) && isOn(MOTOR_C))

Mit moving stellt fix fest, ob die Motoren laufen. Dies erfolgt im Subsumption-Thread, bevor die Nachricht quittiert wird, kann also nicht durch eine andere Nachricht unterbrochen werden.

5/navigator.nqc

// message 2: compensate breaking
task fix () {
 while (true) {
   get(fixW);
   if (message == 2) break;
   done(fixW);
 }
 ready |= fixW;
}
 
void doFix () {
 stop fix;
 if (moving) {
   if (COMPASS > 1) {
     Toggle(rightM); until (abs(COMPASS) <= 1); Toggle(rightM);
   } else if (COMPASS < -1) { 
     Toggle(leftM); until (abs(COMPASS) <= 1); Toggle(leftM);
   }
 }
 done(fixW); ready &= &#126;fixW;
 start fix;
}

Die Kurskorrektur akkumuliert ganz von selbst, wenn man den Rotations-Sensor COMPASS nur beim Start löscht. Man muß allerdings die Drehungen beim Zurücksetzen von einem Hindernis dann relativ berechnen:

5/navigator.nqc

// watches for collision at right
task right () {
 until (rightB != 1);
 ready |= rightW;
}
 
void doRight () {
 stop right;
 backup(COMPASS, COMPASS + turnT/4, rightM, leftM);
 ready &= &#126;rightW;
 start right;
}
 
// backup from bumper: reverse motors a and b,
// reverse motor a until COMPASS shows ticks, then reverse motor b
void backup (int low, int high, const int& a, const int& b) {
 Toggle(a + b); Wait(backupT);
 Toggle(a);
 until (COMPASS < low || COMPASS > high);
 Toggle(b);
}