Die lieben Threads mal wieder, es Fehlert so rum

**Medium**

Aloah!
So ganz so grün bin ich in Sachen Multithreading ja eigentlich nicht - dachte ich - aber in meinem aktuellen Projekt hüpft mir gelegentlich dann doch die gelegentliche AV ins Gesicht.
Der Sinn des Teils ist: Es gibt eine Liste von Anfragen, die via TCP/IP an ein anderes System gesendet werden. Da die Antworten dieses Systems keine Zuordnung zur gemachten Anfrage zulassen, muss auf die Antwort gewartet werden, bevor eine nächste geschickt werden kann. Die Anfragen werden in einem definierten Zyklus verschickt, so dass ich das in einen Thread ausgelagert habe, der immer wieder diese Anfragen abschickt wenn sie anstehen, und zurück gelieferte Antworten verarbeitet (=Daten in meine DB schreibt).
Ich versuche mal die grobe Struktur darzustellen:

zusammenfalten · markieren

Delphi-Quellcode:

			unit MyThreads;

interface

type

  TFetchThread = class(TThread)

  private

    FetchGroups: TFetchGroupList; // Eine List mit Instanzen von TFetchGroup, die sind nichts großartig spannendes. Reine Datenhalter.

    Sock: TClientSocket;

    Con: TUniConnection;

    Qry: TUniQuery;

    CycleThread: TCycleThread;

    NewDataArrived: Boolean;

    WaitingForReply: Boolean;

    SockForm: TForm;

    procedure IssueFetch(fetchGroupIndex: Integer);

    procedure HandleFetchReply(aReplyBuffer: TSockBuffer);

    procedure MakeDBEntriesForGroup(groupID: Integer; buf: TFetchEntryBuffer);

    procedure InitIssueFetch;

  protected

    procedure Execute; override;

  public

    constructor Create(aSocket: TClientSocket; aSocketForm: TForm; aDBConnection: TUniConnection);

    destructor Destroy; override;

    procedure SocketDataReady(aBuffer: TSockBuffer);

  end;

implementation

constructor TFetchThread.Create(aSocket: TClientSocket; aSocketForm: TForm; aDBConnection: TUniConnection);

var

  i, m: Integer;

begin

  inherited Create(true);

  Sock := aSocket;  // Socket kommt vom MainForm

  SockForm := aSocketForm;  // Das Handle brauche ich später auch

  // Eigene Connection für den Thread erstellen, Daten von der übergebenen Connection nehmen

  try

    Con := TUniConnection.Create(nil);

    Con.ProviderName := aDBConnection.ProviderName;

    Con.Server   := aDBConnection.Server;

    Con.Database := aDBConnection.Database;

    Con.Port     := aDBConnection.Port;

    Con.Username := aDBConnection.Username;

    Con.Password := aDBConnection.Password;

    Con.LoginPrompt := false;

    Con.Connect;

  except

    Con.Free;

    raise Exception.Create('Mäh');

  end;

  // Thread-eigene Query

  try

    Qry := TUniQuery.Create(nil);

    Qry.Connection := Con;

  except

    Qry.Free;

    raise Exception.Create('Muh');

  end;

  // Gruppen füllen, dies ist die einzige Stelle, an der an diesen strukturell etwas gemacht wird

  FetchGroups := TFetchFunctions.MakeFetchGroups(Qry);  

  CycleThread := TCycleThread.Create;

  CycleThread.Items := FetchGroups;

  CycleThread.Resume;

  self.Resume;

end;

// WaitingForReply heisst, es wurde eine Anfrage gesendet, die Antwort steht noch aus. In der Zeit darf nichts gemacht werden.

// NewDataArrived wird im SocketRead gesetzt, wenn die erwartete Menge Bytes angekommen und valide ist.

procedure TFetchThread.Execute;

var

  i, k: Integer;

  lowCycle: Integer;

begin

  repeat

    if not WaitingForReply then

    begin

      // Gruppe suchen, die aktualisiert werden muss (die, die am längsten überfällig ist)

      lowCycle := 0;

      k := -1;

      for i := 0 to FetchGroups.Count-1 do

      begin

        if FetchGroups[i].CurrentCycleTime < lowCycle then

        begin

          lowCycle := FetchGroups[i].CurrentCycleTime;

          k := i;

        end;

      end;

      // Wurde eine gefunden, dann via Socket-Kompo auf dem MainForm die Anfrage senden

      if (k >= 0) then

      begin

        IssueFetchID := k;

        Synchronize(InitIssueFetch);

      end;

    end

    else

    begin

      // Wird eine Antwort zu einer Anfrage erwartet, und es ist eine eingetroffen, diese nun verarbeiten.

      if NewDataArrived then

        HandleFetchReply(ReplyBuffer); // Der Buffer ist ein array[0..8191] of Byte, mehr kommt definitiv nicht

    end;

    Sleep(1);

  until Terminated;

end;

procedure TFetchThread.HandleFetchReply(aReplyBuffer: TSockBuffer);

var

  h: TFetchAnswerHeader; // ein record

  isValid: Boolean;

  buf: TFetchEntryBuffer; // array of Byte

begin

  Move(aReplyBuffer[0], h, SizeOf(TFetchAnswerHeader));

  // Header auf Validität prüfen

  isValid := DiverseDinge;

  if isValid then

  begin

    // ReplyGroupID wird in InitIssueFetch() gesetzt, und ist der Index der zuletzt angefragen Fetchgruppe in der Liste dieser

    SetLength(buf, FetchGroups[ReplyGroupID].ByteCount);

    Move(aReplyBuffer[SizeOf(TFetchAnswerHeader)], buf[0], FetchGroups[ReplyGroupID].ByteCount);

    MakeDBEntriesForGroup(ReplyGroupID, buf);

  end;

  NewDataArrived := false;

  WaitingForReply := false;

end;

// Mein Sorgenkind! Das erstellen des TDBEntryThreads knallt sporadisch, und zwar beim Erzeugen selbt.

// Der Konstruktur läuft sauber durch, das hab ich getestet, dennoch gibt's ab und an eine AV an unterschiedlichen

// Adressen. Mal 0, mal $FFFFFFFF, mal irgendwas im Codesegment, und dort immer leicht verschiedene.

// Das try..except hier ist nur ein Workaround, den ich gern los wäre. Zudem greift dies nicht immer, selten springt

// Delphi auch gleich ins CPU-Fenster, obwohl "bei Delphi-Exceptions stoppen" aus ist. Igitt!

procedure TFetchThread.MakeDBEntriesForGroup(groupID: Integer; buf: TFetchEntryBuffer);

var

  group: TFetchGroup;

  threadOkay: Boolean;

begin

  group := FetchGroups[groupID];

  threadOkay := false;

  repeat

    try

      TDBEntryThread.Create(Qry.Connection, group, buf, SockForm.Handle);

      threadOkay := true;

    except

    end;

    Sleep(1);

  until threadOkay;

end;

// Das ist besagter Konstruktor. Ein paar private Felder werden gesetzt, viel mehr nicht.

constructor TDBEntryThread.Create(aCon: TUniConnection; aGroup: TFetchGroup; aBuf: TFetchEntryBuffer; aWnd: HWND);

begin

  inherited Create(true);

  Con := aCon;

  Sql := TUniSQL.Create(nil);

  Sql.Connection := Con;

  Group := aGroup;

  Buf := aBuf;

  Wnd := aWnd;

  FreeOnTerminate := true;

  Resume;

end;

// Und hier der zugehörige Workload

procedure TDBEntryThread.Execute;

begin

  // Hier wird anhand von Infos aus der "Group" der Empfagspuffer interpretiert. Group wird dabei nur lesend angefasst.

  // Da das recht viel ist, durch diesen Kommentar ersetzt. Im wesentlichen wird ein SQL Statement zusammengestückelt.

  while Sql.Connection.InTransaction do

    Sleep(1);

  Sql.Execute;

  // Puffer zur Änderungserkennung für diesen Fetch-Aufruf speichern (einziger schreibender Zugriff auf Group)

  group.Buffer := Buf;

  group.FirstRun := false;

end;

// Wird vom MainForm aufgerufen, wenn OnSocketRead die geforderte Datenmenge gelesen hat

procedure TFetchThread.SocketDataReady(aBuffer: TSockBuffer);

begin

  ReplyBuffer := aBuffer;

  NewDataArrived := true;

end;

// Das hier macht der ominöse Cyclethread: Er zählt in den Fetchgruppen einen Zeitwert runter,

// anhand dessen ermittelt wird welche Anfrage als nächste ansteht.

procedure TCycleThread.Execute;

var

  i: Integer;

  tickDelta: Int64;

begin

  repeat

    tickDelta := GetTickCount-LastTickCount;

    LastTickCount := GetTickCount;

    for i := 0 to Items.Count-1 do

      Items[i].CurrentCycleTime := Items[i].CurrentCycleTime - tickDelta;

    Sleep(1);

  until Terminated;

end;

Was mich fuchst ist, dass das Erzeugen des DBEntryThreads ab und an knallt (so im Schnitt alle 20 Mal ein Mal, die Zykluszeit ist bei 2 Sekunden), obwohl der Konstruktor selbst keine AV erzeugt (getestet via try..except drum rum und bei Exception Logfile machen, was nie geschah). Das try..Except um den Aufruf des Konstruktors dagegen läuft wie gesagt ab und an auf den Hammer, ich komme beim Debuggen nur nicht an die genaue fehlerhafte Zeile, da diese sich nicht explizit in meinem Code zu befinden scheint.
Ich hab mir ein wenig Sorgen um diese Groups gemacht, da diese ja ein Feld des FetchThreads sind, und sowohl vom Cyclethread als auch vom DBEntryThread dort hineingegriffen wird. Jedoch werden in beiden nur elementare Operationen ausgeführt, und die Struktur der Liste bleibt nach dem Konstruktor von TFetchThread für den Rest des Ablaufs immer gleich.
Was hab ich hier nicht bedacht? Wäre prima, wenn trotz der Länge des Teils jemandem was auffällt. Danke schon mal!

\\Edit:
Hier mal ein Auszug aus meinem Fehlerlog (das schreibe ich im except-Teil der Methode MakeDBEntriesForGroup(), das ist hier der Länge wegen nicht im Code):

markieren

Code:

07:2011 12:46:17:625 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:46:31:859 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:47:00:953 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:47:04:562 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:47:21:609 : Zugriffsverletzung bei Adresse 00004244. Lesen von Adresse 00004244 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:48:00:421 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:48:02:859 : Zugriffsverletzung bei Adresse 00004244. Lesen von Adresse 00004244 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:49:27:796 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:49:45:890 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:49:49:937 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:49:53:203 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:49:53:984 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:00:500 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:06:125 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:07:796 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:08:140 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:12:203 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:44:234 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:50:52:703 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:51:52:156 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:52:03:109 : Zugriffsverletzung bei Adresse 00000000. Lesen von Adresse 00000000 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:54:52:859 : Zugriffsverletzung bei Adresse 00004244. Lesen von Adresse 00004244 -> TDBEntryThread.Create(Qry.Connection, group, buf);

07:2011 12:58:24:640 : Zugriffsverletzung bei Adresse 0040399A in Modul 'Server.exe'. Lesen von Adresse FFFFFFFF -> TDBEntryThread.Create(Qry.Connection, group, buf);

Die lieben Threads mal wieder, es Fehlert so rum

Die lieben Threads mal wieder, es Fehlert so rum

Forumregeln