mikrocontroller.net

Forum: Mikrocontroller und Digitale Elektronik Knobelei: Nachrichten-Queue funktioniert fehlerhaft - Tipps?


Autor: Robert (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo!

Ich benutze eine Nachrichten-Queue um nachrichten zwischen verschiedenen 
Prozessen in meinem AVR auszutauschen und um lokale Puffer einzusparen.

Funktionsweise:

Es existiert eine Array von Nachrichten-Structs. Die Nachrichten können 
in Form einer verketteteten Liste geordnet werden ("next"-Pointer"). 
Zudem existieren Pointer auf Head und Tail zum einfachen Duchlaufen und 
Hinzufügen von Nachrichten.

Die einzelnen Prozesse holen sich per "new_msg()" den Pointer auf einen 
noch leeren Slot und benutzen ihn. Wenn alle Slots belegt sind wird NULL 
zurückgeliefert.

In der Hauptschleife des Programms werden nun die Nachrichten 
nacheinander durchgegangen und den entsprechenden "process"-Functionen 
der Module übergeben (als Pointer). Möchte ein Modul die Nachricht noch 
Puffern wird "OK" zurückgeliefert und es wird einfach mit der nächsten 
Nachricht weiter gemacht. Wird etwas anderes zurückgeliefert, wird die 
Nachricht gelöscht.

Scheinbar mache ich beim Löschen oder beim Einfügen einen Fehler, denn 
nach ein paar Durchläufen ist zwar "msgs_head == NULL", aber es sind 
wenn man das Array per Zähler durchläuft noch Slots von Nachrichten 
besetzt.
/* container for message */
typedef struct msg_s msg_t;

struct msg_s
{
     uint8_t dest;
     uint8_t data[50];  // msg data
     msg_t* next;    // pointer to next message
};

typedef enum result_e
{
  OK = 0,
  FREE,
  EOVERFLOW,
  EUNKNOWN,
  ENOTSUPP,
} result_t;


#define NUM_MSGS  10

msg_t *msgs_tail;
msg_t *msgs_head;
msg_t msgs[NUM_MSGS];

void msg_init()
{
  uint8_t c;
  for (c = 0; c < NUM_MSGS; c++)
  {
    msgs[c].dest = 0x00; // Nachricht ist frei
    msgs[c].next = NULL;
  }
  msgs_tail = msgs_head = NULL;
}

void msg_task()
{
  msg_t* previous = NULL;
  msg_t* iterator = msgs_head;
  result_t result = EUNKNOWN;
  while ( (iterator) /*&& (result != OK)*/ )
  {
    // direct to appropriate dest
    switch (dest)
    {
    case 0x01:
      result = a_process(iterator);
      break;
    case 0x02:  
      result = b_process(iterator);
      break;
    default:  // error!
      result = EUNKNOWN;
    }

    // delete msgs that were not successfully processed to avoid queue stall
    if (result != OK)
    {
      iterator->dest = 0x00;  // mark as free/available

      if (iterator == msgs_tail)  // iterator was last msg
      {
        msgs_tail = previous;
      }

      if (iterator == msgs_head)  // iterator was first msg
      {
        msgs_head = iterator->next;
      }
      else
      {
        previous->next = iterator->next;
      }
    }
    else
    {
      previous = iterator;
    }

    // move on
    iterator = iterator->next;
  }
}

msg_t *msg_new()
{
  uint8_t c;
  msg_t *free_msg = NULL;
  // look for free msg slot in "msgs"
  for (c = 0; c < NUM_MSGS; c++)
  {
    if (msgs[c].dest == 0x00)
    {
      free_msg = &msgs[c];
      break;
    }
  }

  if (free_msg)
  {
    // set next to NULL
    free_msg->next = NULL;
    // now append
    if (msgs_tail != NULL)
    {
      msgs_tail->next = free_msg;
    }
    else
    {
      msgs_tail = free_msg;
      msgs_head = free_msg;
    }
  }

   return free_msg;
}

Ich würde mich sehr freuen wenn jemand mal schaut wo mein Denkfehler 
ist! Keine der Funktionen werden durch Interrupts aufgerufen! Die 
Pointer Head, Tail, Next etc. werden im sonstigen Code nirgendwo gelesen 
oder gar geschrieben.

Grüße
Robert

Autor: Maxx (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Ja, dir fehlt vollkommen die Zugriffskontrolle.

Stell dir vor:
 if (iterator == msgs_head)  // iterator was first msg
      {
        msgs_head = iterator->next;
      }

Wird genau dann Unterbrochen, wenn iterator->next ausgelesen, aber 
msgs_head noch nicht gesetzt wurde. In der Unterbrechung wurde ein neues 
Element eingefügt direkt nach iterator. Sprich iterator->next war vorher 
NULL, neues elemt wird eingefügt iterator->next ist ein neues element. 
msgs_head wird auf NULL gesetzt.

Und noch viele anderen stellen.

Überleg wann du nicht unterbrochen werden darfst um das zu vermeiden, 
verhindere das und schau dann nochmal.

Autor: Robert (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo!

Wer soll mich denn unterbrechen? Ich schrieb "die Funktionen werden von 
KEINEN Interrupts aufgerufen!" Und KEIN anderer Codeteil außer den 
gezeigten ändert tail, head, msg!

Grüße
Robert

Autor: MaWin (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
>     switch (dest)

Welches dest

>    msgs_tail->next = free_msg;

Tail und Head bedeutungsmässig vertauscht.

>    // delete msgs that were not successfully processed to avoid queue stall

Und erfolgreich bearbeitete verstopfen sie ?

> a_process(

Wenn a-process oder b_process ein msg_new macht, stimmt nichts mehr von 
deinen Annahmen (previous etc.)


Versuch doch einfach mal

while(msgs_head)
{
  iterator=msgs_head;
  msgs_head=iterator->next;
  if(!msgs_head) msgs_tail=NULL;
  switch(iterator->dest)
  {
  :
  }
  iterator->dest=0; // nach Bearbeitung immer freigeben
}


und bei msgs_new Head und Tail richtigrum (obwohl es ja nur Namen sind)

msgs_new()
{
  uint8_t c;

  for(c=0;c<NUM_MSGS;c++)
  {
    if(!msgs[c].dest)
    {
        msgs[c].next=NULL;
        if(!msgs_tail) msgs_tail=msgs+c;
        if(msgs_head) msgs_head->next=msgs+c;
        msgs_head=msgs+c;
        break;
    }
  }
}

Eine free-list wäre wohl auch sinnvoll.

Autor: Maxx (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Sorry, der Teil da unten ist mir nciht aufgefallen :-P

Autor: Robert (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo!

Danke für das Feedback. Habe jetzt des "dest" geändert und head und tail 
getauscht.
/* container for message */
typedef struct msg_s msg_t;

struct msg_s
{
     uint8_t dest;
     uint8_t data[50];  // msg data
     msg_t* next;    // pointer to next message
};

typedef enum result_e
{
  OK = 0,
  FREE,
  EOVERFLOW,
  EUNKNOWN,
  ENOTSUPP,
} result_t;

#define NUM_MSGS  10

msg_t *msgs_head;
msg_t *msgs_tail;
msg_t msgs[NUM_MSGS];

void msg_init()
{
  uint8_t c;
  for (c = 0; c < NUM_MSGS; c++)
  {
    msgs[c].dest = 0x00; // Nachricht ist frei
    msgs[c].next = NULL;
  }
  msgs_head = msgs_tail = NULL;
}

result_t a_process(msg_t* msg)
{
  if (externe_bedingung)
    return OK;
  else
    return FREE;
}

result_t b_process(msg_t* msg)
{
  if (externe_bedingung2)
    return OK;
  else
    return FREE;
}

void msg_task()
{
  msg_t* previous = NULL;
  msg_t* iterator = msgs_tail;
  result_t result = EUNKNOWN;
  while ( (iterator) )
  {
    // direct to appropriate dest
    switch (iterator->dest)
    {
    case 0x01:
      result = a_process(iterator);
      break;
    case 0x02:  
      result = b_process(iterator);
      break;
    default:  // error!
      result = EUNKNOWN;
    }

    // delete msgs that were not successfully processed to avoid queue stall
    if (result != OK)
    {
      iterator->dest = 0x00;  // mark as free/available

      if (iterator == msgs_head)  // iterator was last msg
      {
        msgs_head = previous;
      }

      if (iterator == msgs_tail)  // iterator was first msg
      {
        msgs_tail = iterator->next;
      }
      else
      {
        previous->next = iterator->next;
      }
    }
    else
    {
      previous = iterator;
    }

    // move on
    iterator = iterator->next;
  }
}

msg_t *msg_new()
{
  uint8_t c;
  msg_t *free_msg = NULL;
  // look for free msg slot in "msgs"
  for (c = 0; c < NUM_MSGS; c++)
  {
    if (msgs[c].dest == 0x00)
    {
      free_msg = &msgs[c];
      break;
    }
  }

  if (free_msg)
  {
    // set next to NULL
    free_msg->next = NULL;
    // now append
    if (msgs_head != NULL)
    {
      msgs_head->next = free_msg;
    }
    else
    {
      msgs_head = free_msg;
      msgs_tail = free_msg;
    }
  }

   return free_msg;
}

Folgende Anmerkungen zu MaWin's Vorschlägen:
1) behoben - sorry
2) behoben

3) Nein, dann werden die Nachrichten gepuffert. Genau das ist ja der 
Sinn der Sache. Sonst brauch ich in jedem Modul einen Puffer. Zudem 
meine Schleife ALLE Nachrichten durchiteriert - d.h. andere Nachrichten 
werden soweit möglich verarbeitet. Wenn ein Prozess eine Nachricht 
partout nicht freigibt (was nicht passiert - das sehe ich beim Debuggen) 
verringert sich halt der verfügbare Platz um eine Nachricht.

4) Durch ausführen von msg_new() wird ja INNERHALB der Queue nichts 
geändert, es wird etwas angehängtr. D.h. er gibt ein Element AM ENDE 
mehr, das auch sogar von der Schleife durchgelaufen wird. Da ich 
zwischendrin nichts einfüge wird die Abfolge in der Schleife nicht 
geändert. Oder?

5) Vielen Dank für den Code. Allerdings passt die eben nicht zu dem 
benötigten Verhalten was in 3) beschrieben ist. Eine Nachricht kann eben 
eine "interne" Nachricht sein, aber auch eine "externe" die gesendet 
werden soll. Wenn das momentan nicht möglich ist soll diese ja nicht 
gelöscht werden. Auch möchte ich diese nicht umkopieren, sondern direkt 
puffern. Also einfach "ok" zurück und im nächsten Durchlauf gucken ob 
sich etwas geändert hat. Derweil sollen natürlich andere Nachrichten 
weiter hinten in der Queue verarbeitet werden können.

Grüße
Robert

Autor: Sascha Weber (sascha-w)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo,

sag mal ...

wenn jede Nachricht (bzw. der Puffer) eine konstante Größe hat, und in 
jedem Nachrichtenpuffer mit 'dest' der Belegungsstatus gespeichert wird 
dann kannst du dir die Verkettung auch sparen.
Packe deine msg_struct (mit dest und msg_data) in ein Array, dann kannst 
du dir freie Plätze suchen (dest=0) und neu belegen. Oder Nachrichen 
löschen indem du dest auf Null setzt.

Sascha

Autor: Robert (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo Sascha,

da hast du völlig recht. Allerdings möchte ich gerne die zeitliche 
Abfolge der Nachrichten beim Senden beibehalten. Also die älteste 
Nachricht (und das ist eben leider nicht die mit dem kleinsten Index) 
soll in jedem Durchgang als erstes an die Prozesse übergeben werden. Bei 
der Queue können die Pakete zwar zeitlich "aufrücken", aber nie eine 
Nachricht überholen.

Grüße
Robert

Autor: Robert (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo!

Um die Geschichte aufzulösen:

Der Fehler lag in der Allokation neuer Elemente:
msg_t *msg_new()
{
  [...]

    if (msgs_head != NULL)
    {
      msgs_head->next = free_msg;
    }
    else
    {
      msgs_head = free_msg;
      msgs_tail = free_msg;
    }
  }

   return free_msg;
}

muss lauten
msg_t *msg_new()
{
  [...]

    if (msgs_head != NULL)
    {
      msgs_head->next = free_msg;
    }
    else
    {
      msgs_tail = free_msg;
    }
    msgs_head = free_msg;
  }

   return free_msg;
}

Das msg_head muss natürlich IMMER auf die neue Nachricht zeigen. Nur das 
Update von head->next oder eben tail ist konditional.

Grüße
Robert

Autor: Karl Heinz (kbuchegg) (Moderator)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Das sieht immer noch nicht richtig aus.
Ich rate dir:
Mal dir die Situation auf!

Wenn man das erste mal verpointerte Listenstrukturen baut, kommt man 
schnell mal durcheinander.
Mit einer Skizze, wie die einzelnen Einzelknoten über Pointer 
miteinander verbunden werden und welche Operation welchen Pointer wohin 
umsetzt, kann man das ganz gut aussortieren.

Das sei deine Liste, mit 3 Knoten



  msg_head       msg_tail
  +--------+     +--------+
  |   o    |     |   o    |
  +---|----+     +---|----+
      |              |
      |              +--------------------+
      v                                   v
   +---------+   +--->+---------+   +---->+---------+
   |         |   |    |         |   |     |         |
   | next: o-----+    | next: o-----+     | next: 0 |
   +---------+        +---------+         +---------+


Jetzt kommt deine New Funktion zum Zug. Sie hat bereits einen freien 
Knoten identifiziert und einen Pointer darauf in free_msg



  msg_head       msg_tail
  +--------+     +--------+
  |   o    |     |   o    |
  +---|----+     +---|----+
      |              |
      |              +--------------------+
      v                                   v
   +---------+   +--->+---------+   +---->+---------+
   |         |   |    |         |   |     |         |
   | next: o-----+    | next: o-----+     | next: 0 |
   +---------+        +---------+         +---------+




        +-->+----------+
        |   |          |
        |   | next:  0 |
        |   +----------+
        |
        +---------+
                  |
   free_msg       |
   +---------+    |
   |    o---------+
   +---------+


Das ist deine Ausgangssituation, wenn der Code hier
msg_t *msg_new()
{
  [...]

    if (msgs_head != NULL)


angelangt ist. Durchlaufe den Code weiter, ändere die Pointer so wie es 
dir der Code vorschreibt und sieh nach was rauskommt.

Alles was ám Ende nicht so aussieht

   msg_head       msg_tail
  +--------+     +--------+
  |   o    |     |   o    |
  +---|----+     +---|----+
      |              |
 +----+              +--------------------+
 |                                        v
 | +---------+   +--->+---------+   +---->+---------+
 | |         |   |    |         |   |     |         |
 | | next: o-----+    | next: o-----+     | next: 0 |
 | +---------+        +---------+         +---------+
 | ^
 | +-----------------------+
 +----------+              |
            v              |
        +-->+----------+   |
        |   |          |   |
        |   | next:  o-----+
        |   +----------+
        |
        +---------+
                  |
   free_msg       |
   +---------+    |
   |    o---------+
   +---------+

ist fehlerhaft, wenn das Element vorne eingefügt werden soll.


Noch ein guter Tip.

Teste deine Queue Verwaltung erst mal auf dem PC.
Dazu schreibst du dir noch eine Funktion, die dir die Queue komplett 
ausgeben kann.
Und dann schreibst du dir ein main() in dem du in wilder Folge einfügst 
bzw. rauslöschst. Fang aber einfach an! Im ersten Testprogramm genügt 
es, wenn du einen Knoten einfügst. Im zweiten dann 2 und im dritten 3. 
Dann löscht du nach den 3 eingefügten Knoten einen raus.
Wichtig ist aber: Nach jeder, und ich meine ausnahmslos jeder Operation, 
sei es einfügen oder löschen, lässt du dir die Queue ausgeben!
Wenn da in der Verpointerung was nicht stimmt, dann merkt man das 
mittels Ausgabe meistens sehr schnell.

Daher auch auf dem PC: Da hast du bessere Ausgabemöglichkeiten.

Danymische Datenstrukturen sehen einfach aus, können aber tricky sein. 
Überhaupt dann wenn man das noch nie gemacht hat. Da hilft dann nur: 
systematisch rangehen, das Raten einstellen und Fakten (in Form von 
Output) schaffen. Und Mitzeichnen um zu verstehen, welcher Pointer an 
welcher Stelle zu früh überschrieben wurde, noch nicht geschrieben 
wurde, vergessen wurde, etc.

Autor: Karl Heinz (kbuchegg) (Moderator)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Das hier
   // delete msgs that were not successfully processed to avoid queue stall
    if (result != OK)
    {
      iterator->dest = 0x00;  // mark as free/available

      if (iterator == msgs_head)  // iterator was last msg
      {
        msgs_head = previous;
      }

      if (iterator == msgs_tail)  // iterator was first msg
      {
        msgs_tail = iterator->next;
      }
      else
      {
        previous->next = iterator->next;
      }
    }
    else
    {
      previous = iterator;
    }


sieht ebenfalls nach einem grossen Haufen Unsinn aus.

Autor: Karl Heinz (kbuchegg) (Moderator)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Ausserdem hast du einen riesen Bock geschossen.

Bau deine Queue so auf, wie man das normalerweise macht.
Neue Elemente werden hinten angefügt und nicht vorne und die Queue wird 
von vorne nach hinten abgearbeitet.

Fügst du nämlich vorne ein, dann sind die ältesten Messages 
logischerweise hinten. Das heist aber auch, dass du die Queue von hinten 
nach vorne durchgehen musst, wenn du die Knoten in der Reihenfolge ihres 
Eintreffens abarbeiten willst. Nur: Wie machst du denn das mit einer 
einfach verketteten Liste, wenn die Verkettung genau in die andere 
Richtung zeigt?

Und die Vorgabe:
> Möchte ein Modul die Nachricht noch Puffern wird "OK" zurückgeliefert
> und es wird einfach mit der nächsten Nachricht weiter gemacht. Wird
> etwas anderes zurückgeliefert, wird die Nachricht gelöscht.

die würde ich mir noch mal gut überlegen.
Das Problem: Wenn ein Task eine Nachricht zurückstellt, was passiert 
dann mit den restlichen Nachrichten. Oder anders ausgedrückt: Wann wird 
eine zurückgestellte Nachricht erneut dem Task zur Bearbeitung 
vorgelegt? Werden erst alle anderen Nachrichten abgearbeitet, oder wird 
die Nachricht dem Task immer wieder erneut zugestellt, oder ....

Wenn ein Task eine Nachricht nicht sofort bearbeiten will, dann muss er 
selbst dafür sorgen, dass er sich das merkt. Er kann ja zb eine neue 
Nachricht in die Queue stellen. Geht die Kontrolle an den Verteiler 
zurück, dann löscht der auf jeden Fall die Nachricht. Für ihn ist die 
Sache damit erledigt und die Nachricht wird auf jeden Fall gelöscht. 
Alles andere führt nur zu Chaos, weil plötzlich keinerlei Reihenfolgen 
mehr definiert sind.

Autor: MaWin (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
> wird ja INNERHALB der Queue nichts geändert

Aber die von aussen auf die queue verweisenden Variablen
wie previous etc. bekommen eventuell unpassende Inhalte.

Deine Überlegungen scheinen mir unvollständig zu sein.

Du solltest BEVOR du programmierst erst mal eine in sich
konsistente Gedankenlogik haben, sonst wird das auch mit
dem Programm nichts.

Autor: Robert (Gast)
Datum:

Bewertung
0 lesenswert
nicht lesenswert
Hallo Karl heinz Buchegger,

vielen Dank für die hinweise. Kann es sein, dass sich deine Hinweise auf 
das erste Posting mit der noch "falschen" Verwendung der Begriffe Head 
und Tail beziehen? Dann erklärt sich vl. auch zusammen mit der jetzt 
geposteten Korrektur der Einfügrroutine die deiner Meinung nach falsche 
Verkettung mit "Next"-Pointern. Denn ansonsten wird das von Dir 
beschriebene Einfügen felherfrei durchgeführt. Die verschiedenen 
Szenarien beim Einfügen und Löschen (Head&Tail sind null, nur Tail ist 
Null, beide sind verschieden etc.) hab ich bereits durchprobiert und 
keine weiteren Fehler gefunden.

Da die einzelnen Elemente auch Arrays von ca. 50 Byte bestehen möchte 
ich in Falle eines "Pufferns" nicht umkopieren sondern das ELement 
zurückstellen. Dann bleibt es einfach an seiner Stelle, was dazu führt 
das alte Elemente auch eher aufgrufen werden, was für meinennudng Sinn 
macht. Natürlich könnte ich auch auf das Umkopieren verzichten und das 
Element anhand der zeiger nach hinten sortieren - das wird allerdings 
nicht weniger kompliziert und bietet in meinemn Augen keinerlei 
Vorteile.

Was ansonsten irgendwelche Modifikationen angeht: Keine Interrupts und 
die externen Funktionen rufen nur (fein zeitlich hintereinander!) 
msg_new() oder msg_task() auch. Das einzige was die Funktionen dann 
bekommen ist ein zeiger auf die datenstruktur, an der Sie aber auch am 
Next-Pointer nicht herummanipulieren.

Grüße
Robert

Antwort schreiben

Die Angabe einer E-Mail-Adresse ist freiwillig. Wenn Sie automatisch per E-Mail über Antworten auf Ihren Beitrag informiert werden möchten, melden Sie sich bitte an.

Wichtige Regeln - erst lesen, dann posten!

  • Groß- und Kleinschreibung verwenden
  • Längeren Sourcecode nicht im Text einfügen, sondern als Dateianhang

Formatierung (mehr Informationen...)

  • [c]C-Code[/c]
  • [avrasm]AVR-Assembler-Code[/avrasm]
  • [code]Code in anderen Sprachen, ASCII-Zeichnungen[/code]
  • [math]Formel in LaTeX-Syntax[/math]
  • [[Titel]] - Link zu Artikel
  • Verweis auf anderen Beitrag einfügen: Rechtsklick auf Beitragstitel,
    "Adresse kopieren", und in den Text einfügen




Bild automatisch verkleinern, falls nötig
Bitte das JPG-Format nur für Fotos und Scans verwenden!
Zeichnungen und Screenshots im PNG- oder
GIF-Format hochladen. Siehe Bildformate.
Hinweis: der ursprüngliche Beitrag ist mehr als 6 Monate alt.
Bitte hier nur auf die ursprüngliche Frage antworten,
für neue Fragen einen neuen Beitrag erstellen.

Mit dem Abschicken bestätigst du, die Nutzungsbedingungen anzuerkennen.