thread "lock" bei zugriff auf objektliste

[alderwaran]

hi zusammen,

ich möchte gerne einen multithreaded downloadmanager schreiben, im moment bin ich dabei den teil code zu implementieren der eine datei über mehrere kanäle, also parallel bröckchenweise, downloaded.

zuerst wird ein dateiobjekt generiert das eine verwaltungsliste der einzelnen downloadchunks erzeugt.
jedem chunkobjekt in dieser liste zugeordnet ist ein downloadthread der diesen chunk lesen und schreiben muss.. und hier liegt das problem: alle chunk-threads werden blockiert bis der laufende thread beendet wird, dann wird der "nächste" aktiv und wieder sind alle restlichen blockiert.

hier haufenweise code :/

main, es wird ein downloadthread erzeugt, kein hokuspokus.

package download;

import java.util.logging.Level;
import java.util.logging.Logger;

public class Main {

    public static void main(String[] args) {
        DownloadThread f = new DownloadThread("http://localhost/testfile", "/home/cam/", "stuhl",4);
        while (f.isRunning()) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                Logger.getLogger(Main.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

der downloadthread, ein paar tests, generierung des zieldateiobjekts, der chunkliste und eine schleife die die chunk-threads erzeugt. zum schluss stümperhaft warten und informationen aus der chunkliste ausgeben.

package download;

import java.io.*;
import java.net.*;
import java.util.logging.Level;
import java.util.logging.Logger;

public class DownloadThread implements Runnable {

    private String downloadURL;
    private String savePath;
    private String saveFileName = null;
    private int maxThreads;
    private int downloadSize;
    private int downloaded;
    private boolean running;
    private String fileName = null;
    private RandomAccessFile raRile = null;
    private URL url;
    private static int MALFORMEDURL = 1;
    private static int OPENINITIALCONNECTION = 2;
    private static int UNKNOWNFILESIZE = 3;
    private static int CANNOTOPENFILE = 4;
    private static int CANNOTEXTENDFILE = 5;

    public void error(int _errorcode) {
        // wait for kill
        // set status
        System.out.println("error: "+_errorcode);
    }

    public void run() {
        this.fileName = this.savePath + this.saveFileName;
        HttpURLConnection connection;

        try {
            this.url = new URL(downloadURL);
        } catch (MalformedURLException ex) {
            error(MALFORMEDURL);
        }

        try {
            connection = (HttpURLConnection) url.openConnection();
            this.downloadSize = connection.getContentLength();
        } catch (IOException ex) {
            error(OPENINITIALCONNECTION);
        }

        if (this.downloadSize < 1) {
            error(UNKNOWNFILESIZE);
        }

        try {
            raRile = new RandomAccessFile(this.fileName, "rw");
        } catch (FileNotFoundException ex) {
            error(CANNOTOPENFILE);
        }
        try {
            raRile.setLength(this.downloadSize);
            raRile.close();
        } catch (IOException ex) {
            error(CANNOTEXTENDFILE);
        }
        url = null;

        int threadcount = 0;
        ChunkMap chunkMap = new ChunkMap(this.downloadURL, this.fileName, this.downloadSize);
        TargetFile targetFile = new TargetFile(this.fileName, this.downloadSize);
        
        while (threadcount < this.maxThreads) {
            
            ChunkThread chunkThread = new ChunkThread(chunkMap,targetFile);
            if (chunkThread.isReadyToRun()) {
                chunkThread.setReadBufferKB(1024);
                chunkThread.start();
                threadcount++;
            }
            try {
                Thread.sleep(1);
            } catch (InterruptedException ex) {
                Logger.getLogger(DownloadThread.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        
        for (int i = 0; i < 1000; i++) {
            chunkMap.printInfo();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                Logger.getLogger(DownloadThread.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    private void setRunning(boolean _running) {
        this.running = _running;
    }

    public boolean isRunning() {
        return this.running;
    }

    public DownloadThread(String _downloadURL, String _savePath, String _saveFileName, int _maxThreads) {
        this.downloadURL = _downloadURL;
        this.savePath = _savePath;
        this.saveFileName = _saveFileName;
        this.maxThreads = _maxThreads;
        this.setRunning(true);
        Thread newThread = new Thread(this);
        newThread.start();
        this.setRunning(false);
    }
}

das zieldateiobjekt. synchronized da immer nur ein thread in die datei schreiben soll wenn sein downloadpuffer voll ist.

package download;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.logging.Level;
import java.util.logging.Logger;

public class TargetFile {

    private RandomAccessFile raFile;
    private String raFileName;
    private int raFileSize;

    public TargetFile(String _raFileName, int _raFileSize) {
        this.raFileName = _raFileName;
        this.raFileSize = _raFileSize;

        try {
            this.raFile = new RandomAccessFile(raFileName, "rw");
        } catch (FileNotFoundException ex) {
            Logger.getLogger(TargetFile.class.getName()).log(Level.SEVERE, null, ex);
        }

        try {
            raFile.seek(raFileSize - 1);
        } catch (IOException ex) {
            Logger.getLogger(TargetFile.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public synchronized void write(byte[] _writebuf, int _offset, int _length) {
        try {
            raFile.seek(_offset);
            raFile.write(_writebuf, 0, _length);
        } catch (IOException ex) {
            Logger.getLogger(TargetFile.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public void close() {
        try {
            this.raFile.close();
        } catch (IOException ex) {
            Logger.getLogger(TargetFile.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}

das chunkobjekt das in der liste verwaltet wird. synchronized ist hier redundant da jeder downloadthread nur zugriff auf seinen eigenen chunk hat.

package download;

public class Chunk {

    private int chunkNumber;
    private int start;
    private int stop;
    private int pointer;
    private String fileName = null;

    public Chunk(String _fileName, int _chunkNumber, int _start, int _stop) {
        this.fileName = _fileName;
        this.chunkNumber = _chunkNumber;
        this.start = _start;
        this.pointer = _start;
        this.stop = _stop;
    }

    public synchronized void addToPointer(int _addition) {
        this.pointer += _addition;
    }

    public synchronized int getPointer() {
        return this.pointer;
    }

    public synchronized void setStart(int _start) {
        this.start = _start;
    }

    public synchronized int getStart() {
        return this.start;
    }

    public synchronized void setStop(int _stop) {
        this.stop = _stop;
    }

    public synchronized int getStop() {
        return this.stop;
    }

    public synchronized void setChunkNumber(int _chunkNumber) {
        this.chunkNumber = _chunkNumber;
    }

    public synchronized int getChunkNumber() {
        return this.chunkNumber;
    }

    public synchronized int getBytesLeft() {
        return Math.round(this.stop - this.pointer);
    }
}

hier könnte das probem liegen. die liste (oder vektor, egal, hab beides probiert) der die chunkobjekte beinhaltet.

package download;

import java.util.Iterator;
import java.util.List;
import java.util.Vector;

public class ChunkMap {

    private int maxSize;
    private String downloadURL;
    private String fileName;
    private List<Chunk> chunkVector;

    public ChunkMap(String _downloadURL, String _fileName, int _maxSize) {
        this.downloadURL = _downloadURL;
        this.fileName = _fileName;
        this.maxSize = _maxSize;
        chunkVector = new Vector<Chunk>();
    }

    public void printInfo() {
        Iterator chunkIterator = chunkVector.iterator();
        while (chunkIterator.hasNext()) {
            Chunk chunk = (Chunk) chunkIterator.next();
            System.out.println(chunk.getChunkNumber()+"----------------------------------");
            System.out.println("from "+chunk.getStart()+" to "+chunk.getStop());
            System.out.println("pointer "+chunk.getPointer());
            System.out.println("left "+chunk.getBytesLeft());
        }
        System.out.println("##################################");
    }

    public String getURL() {
        return this.downloadURL;
    }

    public String getFile() {
        return this.fileName;
    }

    /**
     * erzeugt einen neuen chunk im chunkvektor und gibt die
     * werte des neuen chunks zurück
     * @return
     */
    public Chunk addChunk() {
        Chunk chunk = null;

        if (chunkVector.size() == 0) {
            chunk = new Chunk(this.fileName, chunkVector.size() + 1, 0, this.maxSize-1);
            this.chunkVector.add(chunk);
            return chunk;
        }

        // get chunk with max remaining to download
        Chunk biggestChunk = new Chunk(this.fileName, 0, 0, 0);
        Iterator chunkIterator = chunkVector.iterator();
        while (chunkIterator.hasNext()) {
            chunk = (Chunk) chunkIterator.next();
            if (chunk.getBytesLeft() >= biggestChunk.getBytesLeft()) {
                biggestChunk = chunk; // referenz auf biggest !
            }
        }

        // set biggest chunks maxsize to half remaining
        int oldstop = biggestChunk.getStop();
        int newStop = Math.round(biggestChunk.getBytesLeft()/2+biggestChunk.getPointer());
        biggestChunk.setStop(newStop);

        // create new chunk with start=newstop+1 and stop=oldstop
        chunk = new Chunk(this.fileName, chunkVector.size() + 1, newStop+1, oldstop);
        this.chunkVector.add(chunk);

        return chunk;
    }
}

hier, zeile 36
inputStream.skip(chunk.getPointer());
wenn ich anstatt des zugriffs auf das chunkobjekt dieses threads eine konstante eingebe ("10") laufen alle downloadthreads gleichzeitig (aber natürlich falsch da der quelldateioffset nicht stimmt).
warum halten die threads bis auf einen hier an? ich hab versucht mit jconsole drauf zu kommen... nix.

package download;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ChunkThread implements Runnable {

    private boolean isRunning = false;
    private boolean isReadyToRun = false;
    private int bufferSize = 1024 * 1024;
    private RandomAccessFile raFile = null;
    private InputStream inputStream = null;
    private HttpURLConnection connection = null;
    private ChunkMap fileChunkMap = null;
    private Chunk chunk = null;
    private TargetFile targetFile;

    public void start() {
        this.isRunning = true;
        Thread newThread = new Thread(this);
        newThread.start();
        this.isRunning = false;
    }

    public void run() {
        try {
            // create chunk and seek to sthe startpoint dictated by chunk
            //synchronized (fileChunkMap) {
            chunk = fileChunkMap.addChunk();
            inputStream.skip(chunk.getPointer());
            //}
            int i = chunk.getBytesLeft();
            while (i > 0) {
                byte[] buffer = new byte[bufferSize];
                int read = inputStream.read(buffer);
                if (read == -1) {
                    break;
                }
                this.targetFile.write(buffer, chunk.getPointer(), read);
                chunk.addToPointer(read);
                i = chunk.getBytesLeft();
            }
            raFile.close();
            inputStream.close();
            connection.disconnect();
        } catch (IOException ex) {
            Logger.getLogger(ChunkThread.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public boolean isRunning() {
        return this.isRunning;
    }

    public boolean isReadyToRun() {
        return this.isReadyToRun;
    }

    public void setReadBufferKB(int _kb) {
        this.bufferSize = 1024 * _kb;
    }

    public ChunkThread(ChunkMap _fileChunkMap, TargetFile _targetFile) {

        this.fileChunkMap = _fileChunkMap;
        this.targetFile = _targetFile;

        // test new connection
        try {
            URL url = new URL(_fileChunkMap.getURL());
            connection = (HttpURLConnection) url.openConnection();
            inputStream = connection.getInputStream();
        } catch (IOException ex) {
            Logger.getLogger(ChunkThread.class.getName()).log(Level.SEVERE, null, ex);
        }
        // test open file
        try {
            raFile = new RandomAccessFile(_fileChunkMap.getFile(), "rw");
        } catch (FileNotFoundException ex) {
            Logger.getLogger(ChunkThread.class.getName()).log(Level.SEVERE, null, ex);
        }

        // new connection and file open ok
        this.isReadyToRun = true;
    }
}

sorry für den vielen code aber ich hab heute den ganzen tag an dem problem gesessen und bin auf keinen grünen zweig gekommen....:autsch:

zum schluss noch die ausgabe des status der chunks in der chunkliste. man sieht das nur thread 0 aktiv ist, erst wenn dieser left<=0 ist beginnt der nächste zu arbeiten.

##################################
1----------------------------------
from 0 to 25454530
pointer 23994368
left 1460162
2----------------------------------
from 50303025 to 75151512
pointer 50303025
left 24848488
3----------------------------------
from 25454531 to 50303024
pointer 27938816
left 22364208
4----------------------------------
from 75151513 to 100000002
pointer 75151513
left 24848488
##################################
1----------------------------------
from 0 to 25454530
pointer 23994368
left 1460162
2----------------------------------
from 50303025 to 75151512
pointer 50303025
left 24848488
3----------------------------------
from 25454531 to 50303024
pointer 27938816
left 22364208
4----------------------------------
from 75151513 to 100000002
pointer 75151513
left 24848488
##################################
1----------------------------------
from 0 to 25454530
pointer 23994368
left 1460162
2----------------------------------
from 50303025 to 75151512
pointer 50303025
left 24848488
3----------------------------------
from 25454531 to 50303024
pointer 27938816
left 22364208
4----------------------------------
from 75151513 to 100000002
pointer 75151513
left 24848488
##################################

 

[madboy]

Ich habe das jetzt nur überflogen, aber keine Frage gesehen...
Außerdem wäre es sehr hilfreich, wenn du ein KSKB posten würdest. Vermutlich schauen sich die wenigsten den geposteten Code an (vor allem bei der Menge)

 

[SlaterB]

na klingt schon spannend, echte Herausforderungen im Forum-Alltag,
die Frage ist auch recht klar nach parallelen oder nicht parallelen Threads

mit Download und Datei-Streams ist das aber wirklich schlecht zu testen,
zumal dann unbekannte 'localhost/testfile' verwendet wird,

wie siehts aus, wenn man die Threads testweise auf Dummy-Arbeit oder zumindest Lesen aus programminternen ByteArray-Streams umbaut,
so dass es jeder gut testen kann, tritt dann das Problem nicht mehr auf?

 

[alderwaran]

ich habe mir nun per iptraf angesehen was über die leitung geht und erstaunlicherweise sind da 4 offene connections auf denen recht gleichmäßig traffic läuft obwohl die ausgabe des chunkvektors immer nur variablenänderungen in *einem* chunk zeigen.

nach programmende stimmen dann auch noch quell- und zieldatei überein (sollte mich freuen, aber ich beiß grad in die tischkante)

brett ~ # md5sum /var/www/localhost/htdocs/testfile /home/cam/stuhl 
ef8e5c2ba6fbdc99c471d2e41dd2dcc7  /var/www/localhost/htdocs/testfile
ef8e5c2ba6fbdc99c471d2e41dd2dcc7  /home/cam/stuhl

diese übereinstimmung geht nur wenn jeder der threads, die ja gleichzeitig traffic machen, qed, seinen im zugeordneten chunk updaten kann um dann mit den richtigen parametern die methode aufzurufen die synchronized in die zieldatei schreibt.
stellt sich mir die frage ob objektreferenzen irgendwo zwischengespeichert werden um für einen tiefer in der hierarchie liegenden thread ein "konsistentes" datenmodell aufrecht zu erhalten... näääähueh:


zum thema kskb: das würde im endeffekt nicht viel kürzer oder übersichtlicher werden als die codebrocken die ich zuerst gepostet habe - eine main-class, eine thread-class die weitere klassen und weitere threads erzeugt die auf das datenmodell des ersten threads zugreifen, dazu noch ein pseudo-webserver... alles in einem langen listing. da stecke ich die zeit lieber daran das ganze nochmal (mit observables?) zu reimplementiren und hier die ergebnisse und hoffentlich auch eine lösung zu posten.

hätte ja sein können das jemand direkt aufspringt, mir 5 finger ins gesicht drückt mit den worten "das kann man so nicht machen!"

 

[Gelöschtes Mitglied 5909]

wie wäre es denn, wenn jeder chunkthread in seine eigene datei schreibt und am ende werden sie dann gemerged?
Ansonsten könntest du noch den Buffer größer machen, denn das schreiben in die Datei ist "langsam" wenn der buffer klein ist, kann es sein dass du dann dadrauf warten musst.

 

[alderwaran]

wie wäre es denn, wenn jeder chunkthread in seine eigene datei schreibt und am ende werden sie dann gemerged?

habe ich auch überlegt - die entscheidung dagegen war das ich auf platte den doppelten speicherplatz brauche.
das problem ist auch nicht das schreiben in die datei. der fehler, das unerwünschte verhalten , tritt auch auf wenn man die zeile mit der zugriffsmethode auf die zieldatei auskommentiert, also garnichts geschrieben wird.

ich hab nun doch madboy's vorschlag aufgegriffen und kein kleines kskb gemacht das den zugriff mehrerer threads auf ein datenobjekt "simuliert".

damit mir das spass macht habe ich die namen vergnubbelt - man möge mir das verzeihen :lol:

main erzeugt zuerst ein vektorobjekt das dafür da ist gnubbel-datenobjekte zu speichern, dann werden 4 threads erzeugt deren konstruktor eine referenz ihres gnubbels übergeben wird der auch gleich miterzeugt und in den vektor eingefügt wird.
die threads machen nichts anderes als ein datenfeld in ihrem gnubbel zu verändern.
die frage ist nun ob die änderungen die jeder thread macht auch von aussen beim zugriff auf die gnubbel des vektorobjekts sichtbar sind.
dazu startet main eine schleife die jede sekunde die werte aus allen gnubbels über eine verkormethode auf der console ausgibt.

package thread_test02;

import java.util.Iterator;
import java.util.Random;
import java.util.Vector;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Main {

    public static void main(String[] args) {
        GnubbelVektor gnubbelVektor = new GnubbelVektor();
        for (int i = 0; i < 4; i++) {
            GnubbelThread gnubbelThread = new GnubbelThread(gnubbelVektor.addGnubbel(i));
        }
        for (int i = 0; i < 1000; i++) {
            gnubbelVektor.printInfo();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                Logger.getLogger(Main.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

class GnubbelThread implements Runnable {

    private Gnubbel gnubbel = null;

    public void run() {
        Random r = new Random();
        for (int i = 2; i < 1000; i++) {
            this.gnubbel.setGnubbelWert(gnubbel.getGnubbelNummer() * i + i);
            try {
                Thread.sleep(r.nextInt(1000) + 500);
            } catch (InterruptedException ex) {
                Logger.getLogger(GnubbelThread.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    public GnubbelThread(Gnubbel _gnubbel) {
        this.gnubbel = _gnubbel;
        Thread gnubbelThread = new Thread(this);
        gnubbelThread.start();
    }
}

class GnubbelVektor {

    private Vector<Gnubbel> gnubbelVektor;

    public GnubbelVektor() {
        this.gnubbelVektor = new Vector<Gnubbel>();
    }

    public Gnubbel addGnubbel(int _gnubbelWert) {
        Gnubbel gnubbel = new Gnubbel(this.gnubbelVektor.size(), _gnubbelWert);
        this.gnubbelVektor.add(gnubbel);
        return gnubbel;
    }

    public void printInfo() {
        Iterator i = gnubbelVektor.iterator();
        System.out.println("#######################################");
        while (i.hasNext()) {
            Gnubbel gnubbel = (Gnubbel) i.next();
            System.out.println("gnubbel: " + gnubbel.getGnubbelNummer());
            System.out.println("gnubbelwert: " + gnubbel.getGnubbelWert());
            System.out.println("--------------------------------------");
        }
    }
}

class Gnubbel {

    private int gnubbelNummer;
    private int gnubbelWert;

    public Gnubbel(int _gnubbelNummer, int _gnubbelWert) {
        this.gnubbelNummer = _gnubbelNummer;
        this.gnubbelWert = _gnubbelWert;
    }

    public void setGnubbelWert(int _gnubbelWert) {
        this.gnubbelWert = _gnubbelWert;
    }

    public int getGnubbelWert() {
        return this.gnubbelWert;
    }

    public int getGnubbelNummer() {
        return this.gnubbelNummer;
    }
}

bis hier funktioniert alles wie ichs mir erhofft habe. nun werde ich den krempel erweitern und weiterposten.