jobmanager.patch

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<Job> _queuedJobSet;
 
+	private Set<Job> _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<Job>(new JobComparator());
+		_overflowJobSet = new TreeSet<Job>(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<Job> getAllJobsFromQueue() {
-		return Collections.unmodifiableSet(_queuedJobSet);
+        Set<Job> retSet = 
+            new TreeSet(Collections.unmodifiableSet(_queuedJobSet));
+        retSet.addAll(Collections.unmodifiableSet(_overflowJobSet));
+		return retSet;
 	}
 
 	public synchronized Job getNextJob(Set<RemoteSlave> 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() {