Index: conf/jobmanager.conf.dist =================================================================== --- conf/jobmanager.conf.dist (revision 1738) +++ conf/jobmanager.conf.dist (working copy) @@ -9,3 +9,6 @@ # use SSL transfer useSSLTransfers=true +# Max threads for concurrent JobTransfers +maxThreads=5 + Index: src/net/sf/drftpd/mirroring/JobManager.java =================================================================== --- src/net/sf/drftpd/mirroring/JobManager.java (revision 1738) +++ src/net/sf/drftpd/mirroring/JobManager.java (working copy) @@ -47,12 +47,16 @@ private Set _queuedJobSet; + private Set _overflowJobSet; // Jobs not yet queued + private boolean _useCRC; private boolean _useSSL; private long _sleepSeconds; + private int _threads; // Max threads for concurrent JobTransfers + private GlobalContext _gctx; private TimerTask _runJob = null; @@ -63,6 +67,7 @@ public JobManager(GlobalContext gctx) { _gctx = gctx; _queuedJobSet = new TreeSet(new JobComparator()); + _overflowJobSet = new TreeSet(new JobComparator()); reload(); } @@ -83,7 +88,18 @@ jobiter.remove(); } } - _queuedJobSet.addAll(jobs2); + + /* Add jobs to the real Set until the size limit + * is reached, then add to the overflow Set + */ + for(Job j : jobs2){ + if(_queuedJobSet.size() <= _threads){ + _queuedJobSet.add(j); + } + else{ + _overflowJobSet.add(j); + } + } } public synchronized void addJobToQueue(Job job) { @@ -91,10 +107,13 @@ } /** - * Gets all jobs. + * Gets all jobs from the queuedJobSet and overflowJobSet. */ public synchronized Set getAllJobsFromQueue() { - return Collections.unmodifiableSet(_queuedJobSet); + Set retSet = + new TreeSet(Collections.unmodifiableSet(_queuedJobSet)); + retSet.addAll(Collections.unmodifiableSet(_overflowJobSet)); + return retSet; } public synchronized Job getNextJob(Set busySlaves, Set skipJobs) { @@ -228,6 +247,7 @@ _useCRC = true; _useSSL = false; _sleepSeconds = 10000; // 10 seconds + _threads = 5; // 5 threads return; } finally { if (fis != null) { @@ -243,6 +263,7 @@ _useCRC = p.getProperty("useCRC", "true").equals("true"); _useSSL = p.getProperty("useSSLTransfers", "true").equals("true"); _sleepSeconds = 1000 * Integer.parseInt(PropertyHelper.getProperty(p,"sleepSeconds")); + _threads = Integer.parseInt(PropertyHelper.getProperty(p,"maxThreads")); if (_runJob != null) { _runJob.cancel(); } @@ -259,6 +280,13 @@ public synchronized void removeJobFromQueue(Job job) { _queuedJobSet.remove(job); + + if(!_overflowJobSet.isEmpty()){ + Job firstOverflowJob = _overflowJobSet.iterator().next(); + addJobToQueue(firstOverflowJob); + _overflowJobSet.remove(firstOverflowJob); + } + } public void startJobs() {