Queue.remove() -> no such element exception

[WIng2005]

Hallo zusammen,

ich habe meine Anwendung von Windows auf MAC-OS probiert und stehe vor einem seltsamen Phänomen. Ich lese eine DB aus und generiere daraus Abfragen and div. Controller per Socket. Das ganze läuft in einem getrennten Thread. D.h. die Abfragen sammeln sich in einer Queue und werden in //do something bearbeitet. Im nachten Durchlauf der while wir jeweils der erste Eintrag entfernt.

private static void sendSocketToClient() {
Runnable runnable = new Runnable() {

@Override

        public void run() {
        try {.  

                    while (cmdQueue.size()>0) {

                        try {
                            cmd = cmdQueue.remove();


                        }catch (Exception e) {
                            System.out.println("CmdQueueHandler-> sendSocketToClient -> cmdQueue.remove() : "+e);
                            System.out.println("size: "+cmdQueue.size());
                          
                        }

                        //do something

                    }//while

         }catch (Exception e) {
               System.out.println("CmdQueueHandler-> sendSocketToClient -> : "+e);
         }


...usw.

So, jetzt der spannende Teil: das funktioniert nur bedingt, da ich wenn der MAC ne weile läuft und ich ihn komplett in Ruhe lasse (Ruhezustand ist aber deaktiviert) eine "no such element exception" geworfen bekommen. Aber: ich frage im Catch-Block die Länge der Warteschlange ab... und sie ist nie 0. Heute Morgen wurde mir 6170 ausgegeben. Zudem ist das remove ja auch nur dann relevant, wenn die Länge >0 ist. Das Verhalten zeigt sich unter Windows nicht. Der Inhalt der Queue wird zudem nach jedem Durchlauf in eine Textarea eingetragen, um die Bearbeitung zu visualisieren... diese ist leer.
Jetzt ist es natürlich schwierig, ohne die ganze App zu kennen hier Vermutungen anzustellen, aber ggf. fällt dem Ein oder Anderen ja doch etwas dazu ein....


VG
Steffen

 

[Oneixee5]

Welchen Typ hat cmdQueue. möglicherweise: ConcurrentLinkedQueue?

 

[WIng2005]

Hallo,
LinkedList:

static Queue<String> cmdQueue = new LinkedList<>();

VG
Steffen

 

[Jw456]

Frage wo bei was wird denn die Exception geworfen bist du dir sicher das sie beim remove kommt.

zeige doch mal die ganze Fehlermeldung.

Du löschst ja auch erst und dann machst du was in "do something" ist wirklich zum Zeitpunkt des löschen was in der Query

LinkedList ist auch nicht Threadsichere

Queue<String> cmdQueue = new ConcurrentLinkedQueue<>();

 

[Oneixee5]

LinkedList ist hier ungeeignet und auch ConcurrentLinkedQueue ist nicht blockierend. LinkedBlockingQueue wäre geeigneter, weil sie quasi wartet bis in der Oueue auch etwas drin ist. Dann ist auch die Abfrage von size() nicht notwendig. Das Vorgehen, auf eine LinkedList in einem anderem Thread zuzugreifen, ohne eine Synchronisierung, ist suspekt.
Auch static ist in dem Zusammenhang fragwürdig.

 

[WIng2005]

Hi zusammen und danke für die Hilfe,

kurz zur Erklärung: die Linkedlist wird im Thread initialisiert und von außen greift nichts darauf zu. Der Fehler wird direkt im try/catch der remove-Anweisung geworfen. Mir ist suspekt, warum das Ganze auf Windows über Monate läuft (ohne Fehler) und auf dem MAC nicht. Ich habe jetzt vor dem anhängen an die Queue eine Anfrage zur Prüfung des Kommandos eingebaut. Zusätzlich setze ich die Größe der Queue (durch leeren) auf 0, wenn im Remove ein Fehler auftritt:

public class CmdQueueHandler {
    static Queue<String> cmdQueue = new LinkedList<>();
    static boolean socketRunning;
    static String cmd;


    static void addCmdQueue(String cmd){
        try {
            
            if (cmd.indexOf(".")<1) {
                throw new Exception("illegal cmd: "+cmd);
            }
            cmdQueue.add(cmd);

            sendSocketToClient();
            //writeTextAreaOpen();
        } catch (Exception e) {
            Main.mainFrame.appendTextAreaErrorMessages("->addCmdQueue: "+e);

        }
    }

    private static void writeTextAreaOpen(){
    ...
    }// writeTextAreaOpen

    private static String getNow(){
        ...
    }

    private static void sendSocketToClient() {
        Runnable runnable = new Runnable() {

            @Override

            public void run() {
                try {

                    String resultStr="";

                    while (cmdQueue.size()>0) {

                        try {
                            cmd = cmdQueue.remove();


                        }catch (Exception e) {
                            Main.mainFrame.appendTextAreaErrorMessages("->CmdQH.sendSToC.remove(): "+e);
                            Main.mainFrame.appendTextAreaErrorMessages("--size: "+cmdQueue.size());
                            Main.mainFrame.appendTextAreaErrorMessages("--CMD: "+cmd);
                            break;
                        }

                        String[] cmdArray = cmd.split(">");

                        SocketClient socl = new SocketClient(cmdArray[0], 23);      //to be changed
                        resultStr = socl.send_str(cmdArray[1]);                            // to be changed
                        
                        Main.mainFrame.appendTextAreaLog(getNow()+"     "+cmd+"-->"+resultStr);

                        cmdArray=null;

                        //Write last Value to DB
                        writeLastValueDB(cmd, resultStr);

                        writeTextAreaOpen();

                        try {
                            Thread.sleep(400);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }

                        socl=null;
                        System.gc();

                    }//while

                } catch (Exception e) {
                    e.printStackTrace();
                    //System.out.println("CMD Queue Main: " +e);
                    Main.mainFrame.appendTextAreaErrorMessages("->CMD Queue Main"+e);
                    //falschen Eintrag löschen
                    Main.mainFrame.appendTextAreaLog(getNow()+"     "+cmd+"-->Failed");

                } finally {

                    socketRunning=false;
                    cmd="";
                    cmdQueue.clear();
                    writeTextAreaOpen();
                    System.gc();
                }
            }
        };
        if (!socketRunning) {
            socketRunning=true;
            Thread thread = new Thread(runnable);
            thread.start();
        }
    }

    private static void writeLastValueDB(String cmd, String resultStr) {
        ...
    }


    static void readToLatestTextField(){
        ...
    }


}//Class

Soweit erstmal ein Versuch. Ich wüßte gern die Ursache. Die LinkedBlockingQueue schaue ich mir an, wenngleich eine "erste Sichtung" mich leicht überforderte -> Bin kein Informatiker... alles Udemy.

VG
Steffen

 

[KonradN]

die Linkedlist wird im Thread initialisiert und von außen greift nichts darauf zu.

Das kann doch so nicht stimmen, denn Du erstellst doch ein Runnable und in dem greifst Du auf die Queue zu und Du hast die Queue nicht in dem Runnable initialisiert. Und mit dem Runnable erstellst Du einen Thread und startest diesen...

Daher hast Du mindestens 2 Threads in denen Du auf die Queue zugreifst.

 

[WIng2005]

Hi Konrad,
stimmt... du hast recht. Jetzt verstehe ich auch die Posts vorher. Mmmh...da werde ich wohl kaum um eine neuer Version der Klasse mit LinkedBlockingQueue rumkommen. Ich google schon die ganze Zeit... mein Buch von der Insel kennt das leider noch nicht. Spannend ist aber immer noch die Frage, warum das unter Windows lief....und, warum nicht einfach nur genau 1 Zugriff daneben geht, sondern die ganze App in einen komischen Leerlauf springt (unendliche Verarbeitung von leeren CMD. Ich lasse gerade mal die Version von oben laufen... so hätte ich zumindest Zeit, mich in das andere in Ruhe einzuarbeiten, ohne den Win-Rechner noch einmal akquirieren zu müssen.

Um mal zu verdeutlichen, was ich bastel:

1x pro Minute wird eine DB ausgelesen, die eine Konfiguration ip-> Port -> Typ enthält. Diese wird übersetzt (CMD) und in eine Warteschlange gepackt, die asap abgearbeitet wird. Es werden also eine Reihe Befehle erzeugt, die eine Abfrage von div. Arduinos ansprechen (192.168.1.xx --> getport#24 = Status Port 24 an IP xx). Das rechte Textfeld repräsentiert den Inhalt der Queue, die sich sukzessive abbaut und ins linke Textfeld geschrieben wird (log). Möglich ist auch ein zusätzliches Einbringen in eine laufende Abbarbeitung per Socket, das ganze wird dann einfach an die Queue gehangen... Ein Tab weiter ist ein weiteres Textfeld, in welches Fehlermeldungen gepackt werden. Soweit so gut... lief fehlerfrei unter Windows. Ich habe etwas völlig anderes studiert, also seht mir eine gewisse Grundform von Dilettantismus nach.

VG
Steffen

 

[KonradN]

Also generell machen die Klassen alle nichts so besonderes. Evtl. macht es auch einfach Sinn, sich etwas mehr mit dem Thema Multithreading zu beschäftigen um dann gezielt z.B. mit Synchronized Blocks zu arbeiten. Ggf. macht es auch Sinn, Threads schlafen zu legen und dann gezielt wieder aufzuwecken oder so, wenn ein Element hinzu gefügt wird.

Ich bin kein Fan davon, einfach eine Klasse zu wechseln um dann zu hoffen, dass da auf magische Weise dann alles läuft. Und das sieht ja nach einem typischen Problem aus, dass Du einen Thread hast, der irgend welche Daten liefert und einen anderen Thread, der diese dann entgegen nehmen kann und verarbeitet.

Und wichtig ist auch: Da gibt es ggf. auch andere Ideen und Lösungen, die da passen könnten. Wenn eine parallele Abarbeitung denkbar ist, dann kann man ggf. auch einfach jeden Datensatz direkt beauftragen. Die Queue verwaltet man dann nicht selbst sondern man hat dann einen ThreadPool und das System ist dann selbst für die Abarbeitung verantwortlich und so ....

Also das Thema schreit direkt danach, dass man sich da deutlich intensiver in einer deutlich grundlegenderen Art und Weise mit beschäftigt.

 

[WIng2005]

Hallo Konrad,

da hast du natürlich Recht, wie gesagt... ich lerne noch und bin mit den aktuellen Ergebnissen trotz aller (berechtigter) Kritik zufrieden, wenn ich bedenke, dass ich das alles "nebenbei" lerne. Der Vollständigkeit halber noch eine Fehlermeldung, die nach CMD.remove:

try {
          cmd = cmdQueue.remove();


    }catch (Exception e) {
        Main.mainFrame.appendTextAreaErrorMessages("->CmdQH.sendSToC.remove(): "+e);
        Main.mainFrame.appendTextAreaErrorMessages("--size: "+cmdQueue.size());
        Main.mainFrame.appendTextAreaErrorMessages("--CMD: "+cmd);
        break;
     }

geworfen wird (siehe Anhang). Ich vermute, das der Eintrag in CMD noch den Stand VOR remove hat. getwin#0 ist der letzte Befehl der Kette und wenn dieser bereits gelesen wurde, wäre die Queue tatsächlich leer und sizeof gibt einen falschen Wert aus.

Die Schlagworte aus deinem letzten Beitrag schaue ich mir mal an.

VG
Steffen

 

[mrBrown]

Spannend ist aber immer noch die Frage, warum das unter Windows lief

Andere JVM-Version mit anderen Optimierungen, anderes Threading-Verhalten, andere Hardware - bei sowas kann das schnell mal alles sein.

Fehler kann sein, dass die Größe der Liste je Thread gecached wird, der lesende Thread bekommt daher irgendwann einen festen Wert, der nicht dem echten entspricht.

 

[Oneixee5]

Ich habe hier mal versucht, wie es aussehen kann, etwas asynchron zu verarbeiten. Es wird alle 3 Sekunden ein zufälliger String in die Queue eingefügt. Ein anderer Thread verarbeitet den Inhalt der Queue sobald er dazu in der Lage ist. Im Prinzip könnte das auch mit mehreren Threads durchgeführt werden. Die Verarbeitung endet wenn die Queue leer ist (nur für mein Beispiel) - das muss man nicht so machen. Nach einer Minute wird die Queue nicht mehr weiter befüllt, damit das Programm irgendwann zu Ende kommt.

import java.time.Duration;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

public class Main {

    // Start writing to the queue immediately, every 3 seconds
    private static final int DELAY = 3;

    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();

    private String createRandomString() {
        final int leftLimit = 97; // letter 'a'
        final int rightLimit = 122; // letter 'z'
        final int targetStringLength = 10;
        final Random random = ThreadLocalRandom.current();
        final StringBuilder buffer = new StringBuilder(targetStringLength);
        for (int i = 0; i < targetStringLength; i++) {
            final int randomLimitedInt = leftLimit + (int) (random.nextFloat() * (rightLimit - leftLimit + 1));
            buffer.append((char) randomLimitedInt);
        }
        return buffer.toString();
    }

    private void enqueue(final long shutdownAfter) {
        final ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
        // executes method #generate every #DELAY seconds
        final Future<?> future = pool.scheduleAtFixedRate(this::generate , DELAY, DELAY, TimeUnit.SECONDS);
        // stopping after #shutdownAfter minutes else run forever
        if (shutdownAfter > 0L) {
            pool.schedule(
                    () -> {
                        future.cancel(true);
                        pool.shutdown();
                    },
                    shutdownAfter, TimeUnit.MINUTES);
        }
    }

    // generate random string an add to queue
    private void generate() {
        final String generatedString = createRandomString();
        System.out.format("🚩 create random string and put to queue: %s%n", generatedString);
        queue.add(generatedString);
    }

    // new thread processing queue's content
    private void dequeue() {
        Executors.newSingleThreadExecutor().execute(this::process);
    }

    private void process() {
        do {
            try {
                final String element = queue.take();
                System.out.format("🏁 get string from queue if available and process in long running task: %s%n", element);
                doLongRunningStuff(element);
            } catch (final InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        } while (!queue.isEmpty()); // stopping the task when #queue is empty
    }

    private void doLongRunningStuff(final String element) throws InterruptedException {
        Objects.requireNonNull(element);
        Thread.sleep(Duration.ofSeconds(DELAY * DELAY)); // simulate long running task
    }

    public static void main(final String[] args) throws InterruptedException {
        final Main main = new Main();
        main.enqueue(1); // stopping after 1 minute but program ends, when queue is empty
        main.dequeue();

        System.out.println("Wait ...");
    }

}

 

[WIng2005]

Wow... stark. Das schaue ich mir mal an.
Ich habe heute noch einmal ein wenig "gebastelt" und die Zugriffe außerhalb von sendSocketToClient() unterbunden. In der Methode writeTextAreaOpen() hatte ich die Queue nach jedem Durchlauf von sendSocket einmal ausgelesen und in die Textarea gepackt. Ich übergebe das gante jetzt als Kopie, und greife nicht mehr direkt auf die Warteschlage zu. Zusätzlich blockiere ich den Zugriff durch addQueue auf die Liste, solange ich das remove durchführe...... zack: seit heute morgen keine Fehlermeldung mehr. Ist bestimmt alles Russisch, aber die Anregungen und Beispiele hier muss ich erst einmal verdauen

static void addCmdQueue(String cmd){
        try {
            
            if (cmd.indexOf(".")<1) {
                throw new Exception("->CmdQH.addCmdQ: illegal cmd: "+cmd);
            }
            do {
                //wait to add
            }while (queueBlocked==true);

            cmdQueue.add(cmd);

            if (!threadRunning) {
                sendSocketToClient();
            }
            
        } catch (Exception e) {
            Main.mainFrame.appendTextAreaErrorMessages("->CmdQH.addCmdQueue: "+e);

        }
    }

private static void writeTextAreaOpen(Object[] cmdList){
    ...
}

 private static void sendSocketToClient() {
        Runnable runnable = new Runnable() {

            @Override

            public void run() {
                try {
                    threadRunning=true;
                    String resultStr="";

                    while (cmdQueue.size()>0) {

                       try {
                            queueBlocked=true;
                                cmd = cmdQueue.remove();
                            queueBlocked=false;

                        }catch (Exception e) {                     
                            Main.mainFrame.appendTextAreaErrorMessages("->CmdQH.sendSToC.remove(): "+e);                         
                            break;
                        }
                       
                       if (queueBlocked) {  
                          break;
                        }
 
                        ...
                            ...
                        try {
                            writeTextAreaOpen(cmdQueue.toArray());
                        } catch (Exception e) {
                            Main.mainFrame.appendTextAreaErrorMessages(getNow());
                            Main.mainFrame.appendTextAreaErrorMessages("->CmdQH.sendSToC.QueueToArray: "+e);
                            Main.mainFrame.appendTextAreaErrorMessages("--size: "+cmdQueue.size());
                        }
                        
                        
                        ...

                     } finally {

                        socketRunning=false;
                        cmd="";
                        cmdQueue.clear();
                        //writeTextAreaOpen(cmdQueue.toArray());
                        Main.mainFrame.clearTextAreaCmdQueue();
                        threadRunning=false;
                        queueBlocked=false;
                        System.gc();
}

Vielen Dank für euren Support!!!

VG
Steffen

 

[Oneixee5]

Du hast Recht: es ist russisch und funktioniert nicht wie du vielleicht denkst. Es gibt Möglichkeiten so eine Zugriffsregel richtig zu bauen:

    ReentrantLock lock = new ReentrantLock();

    public void doInLook() {
        lock.lock();
        try {
            // Critical section here
        } finally {
            lock.unlock();
        }
    }

    public void tryLock(){
        boolean isLockAcquired = lock.tryLock(1, TimeUnit.SECONDS);
        if(isLockAcquired) {
            try {
                //Critical section here
            } finally {
                lock.unlock();
            }
        }
    }

Mit einer BlockingQueue ist das aber nicht notwendig, weil das dort schon intern so gemacht wird.

 

[WIng2005]

Hi,

ich interpretiere das so, dass
tryLock():
-> mein addCmdQueue
- Bei Aufruf wird 1 Sekunde versucht, zu sperren und auf den Code in "tryLock() //Critical section" zuzugreifen, danach Sperre aufgehoben

doInLook()
-> mein sendSocketToClient()
-> Sperre wir gesetzt und verhindert einen Zugriff auf CodeBlöcke, die diese Sperre vorher abfragen.

Korrekt? Gestern mal probiert... Wirre Meldung erhalten, muss ich mir mal genauer anschauen.

 

[Oneixee5]

Siehst du hier:

Einführung in die nebenläufige Programmierung

openbook.rheinwerk-verlag.de

Rheinwerk Computing :: Java SE 8 Standard-Bibliothek - Threads und nebenläufige Programmierung

openbook.rheinwerk-verlag.de

und/oder hier:

Concurrency (Multithreading)

Concurrency (Multithreading)

www.torsten-horn.de

 

[WIng2005]

OK, vielen Dank...das wird ein wenig dauern

 

[WIng2005]

Hallo,

also: Gestern (bis heute nach) noch gelesen, getestet und hoffentlich RICHTIG umgesetzt:

public class CmdQueueHandler {
    static Queue<String> cmdQueue = new LinkedList<>();

    static boolean threadRunning=false;

    static String cmd;

    static Lock QLock = new ReentrantLock();
    static Lock TLock = new ReentrantLock();


    static void addCmdQueue(String cmd) {

        Runnable runnable = new Runnable() {

            @Override

            public void run() {

                try {

                    if (cmd.indexOf(".") < 1) {
                        throw new Exception("->CmdQH.addCmdQ: illegal cmd: " + cmd);
                    }

                        try {
                            QLock.lock();
                                cmdQueue.add(cmd);
                        } finally {
                            QLock.unlock();
                        }

                        try {                                                     //Ohne das Lock hier, startet der Thread trotz Abfrage mehrfach
                            TLock.lock();
                            if (!threadRunning) {
                                threadRunning = true;
                                sendSocketToClient();
                            }
                        }finally{
                            TLock.unlock();
                        }

                    //writeTextAreaOpen();
                } catch (Exception e) {
                    Main.mainFrame.appendTextAreaErrorMessages("->CmdQH.addCmdQueue: " + e);
                }
            }
        };

        Thread addCmdQ = new Thread(runnable);      // addCMDQueue als Thread, da zumindest die Chance besteht, dass hier paralele Aufrufe erfolgen
        addCmdQ.start();
    }


    private static void sendSocketToClient() {
        Runnable runnable = new Runnable() {

            @Override

            public void run() {
                Main.mainFrame.appendTextAreaLog(getNow() +"--> Start" );
                try {
                    threadRunning = true;
                    String resultStr = "";

                    while (!cmdQueue.isEmpty()) {
                        try {

                            QLock.lock();
                            cmd = cmdQueue.remove();

                            //queueBlocked=false;

                        } catch (Exception e) {
                            ...
                            break;
                        } finally {
                            QLock.unlock();
                        }

                        ...
                        ...
                        ...
                                            
                        try {
                            Thread.sleep(400);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }

                        ...
                        ...

                    }//while

                } catch (Exception e) {
                    ...
                    ...

                } finally {
                    Main.mainFrame.appendTextAreaLog(getNow() +"--> Stop" );
                    cmd = "";
                    cmdQueue.clear();
                    Main.mainFrame.clearTextAreaCmdQueue();
                    threadRunning = false;
                    System.gc();
                }
            }
        };

            Thread thread = new Thread(runnable);
            thread.start();

    }

}//Class

AddQueue wir nun als Thread gestartet, da die Chance von Parallelität besteht. Somit wird aber auch der normal getimete (gibt es das Wort?) Aufruf per Default 10x parallel gestartet. Diese starten dann genau 1x das Sendsocket. Die restlichen Threads afügen nur die Kommandos an die Queue. Ist zwar alles noch nicht über Blocking Queues, aber lief zumindest seither fehlerfrei.

VG
Steffen