Deliverable of code associated with assignment 2 of Software Reengineering (dlim1 and sbijl)

.gitignore

@@ -17,3 +17,7 @@ bin/
 *.ipr
 *.iws
 *.db
+alitheia/core/nb-configuration.xml
+alitheia/pom.xml
+alitheia/pom.xml
+alitheia/pom.xml

alitheia/core/src/main/java/eu/sqooss/impl/service/scheduler/SchedulerServiceImpl.java

@@ -38,9 +38,7 @@
 import java.util.List;
 import java.util.PriorityQueue;
 import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.*;
 
 import org.osgi.framework.BundleContext;
 
@@ -65,13 +63,17 @@ public class SchedulerServiceImpl implements Scheduler {
     // thread safe job queue
     private PriorityQueue<Job> blockedQueue = new PriorityQueue<Job>(1,
             new JobPriorityComparator());
-    private BlockingQueue<Job> workQueue = new PriorityBlockingQueue<Job>(1,
-            new JobPriorityComparator());
 
     private BlockingQueue<Job> failedQueue = new ArrayBlockingQueue<Job>(1000);
 
+	private ExecutorService executorService;
+
+	private List<Runnable> frozenJobs = new LinkedList<Runnable>();
+
     private List<WorkerThread> myWorkerThreads = null;
 
+    private boolean isExecuting = false;
+
     public SchedulerServiceImpl() { }
 
     public void enqueue(Job job) throws SchedulerException {
@@ -89,10 +91,14 @@ public void enqueue(Job job) throws SchedulerException {
     public void enqueueNoDependencies(Set<Job> jobs) throws SchedulerException {
         synchronized (this) {
             for (Job job : jobs) {
-                logger.debug("Scheduler ServiceImpl: queuing job "
+                if (logger != null)
+                    logger.debug("Scheduler ServiceImpl: queuing job "
                         + job.toString());
                 job.callAboutToBeEnqueued(this);
-                workQueue.add(job);
+
+                Future<Void> future = executorService.submit(job); 
+                job.future = future;
+
                 stats.addWaitingJob(job.getClass().toString());
                 stats.incTotalJobs();
             }
@@ -102,7 +108,8 @@ public void enqueueNoDependencies(Set<Job> jobs) throws SchedulerException {
     public void enqueueBlock(List<Job> jobs) throws SchedulerException {
         synchronized (this) {
             for (Job job : jobs) {
-                logger.debug("SchedulerServiceImpl: queuing job " + job.toString());
+                if (logger != null)
+                    logger.debug("SchedulerServiceImpl: queuing job " + job.toString());
                 job.callAboutToBeEnqueued(this);
                 blockedQueue.add(job);
                 stats.addWaitingJob(job.getClass().toString());
@@ -115,7 +122,7 @@ public void enqueueBlock(List<Job> jobs) throws SchedulerException {
 
     public void dequeue(Job job) {
         synchronized (this) {
-            if (!blockedQueue.contains(job) && !workQueue.contains(job)) {
+            if (!blockedQueue.contains(job) && job.future == null) {
                 if (logger != null) {
                     logger.info("SchedulerServiceImpl: job " + job.toString()
                             + " not found in the queue.");
@@ -124,7 +131,12 @@ public void dequeue(Job job) {
             }
             job.callAboutToBeDequeued(this);
             blockedQueue.remove(job);
-            workQueue.remove(job);
+            if (job.future != null) {
+            	job.future.cancel(false);
+            }
+
+            stats.removeWaitingJob(job.getClass().toString());
+            stats.decTotalJobs();
         }
         if (logger != null) {
             logger.warn("SchedulerServiceImpl: job " + job.toString()
@@ -133,23 +145,11 @@ public void dequeue(Job job) {
     }
 
     public Job takeJob() throws java.lang.InterruptedException {
-        /*
-         * no synchronize needed here, the queue is doing that adding
-         * synchronize here would actually dead-lock this, since no new items
-         * can be added as long someone is waiting for items
-         */
-        return workQueue.take();
+    	return null;
     }
 
     public Job takeJob(Job job) throws SchedulerException {
-        synchronized (workQueue) {
-            if (!workQueue.contains(job)) {
-                throw new SchedulerException("Can't take job " + job
-                        + ": It is not in the scheduler's queue right now.");
-            }
-            workQueue.remove(job);
-            return job;
-        }
+    	return null;
     }
 
     public void jobStateChanged(Job job, Job.State state) {
@@ -179,56 +179,46 @@ public void jobStateChanged(Job job, Job.State state) {
 
     public void jobDependenciesChanged(Job job) {
         synchronized (this) {
-            if (workQueue.contains(job) && !job.canExecute()) {
-                workQueue.remove(job);
-                blockedQueue.add(job);
+            if (jobIsQueuedForWork(job) && !job.canExecute()) {
+                if (job.future.cancel(false)) {
+                	job.future = null;
+                	blockedQueue.add(job);                	
+                }                
             } else if (job.canExecute()) {
                 blockedQueue.remove(job);
-                workQueue.add(job);
+                Future<Void> future = executorService.submit(job);
+                job.future = future;
             }
         }
     }
 
+    /**
+     * Returns whether the job meets its dependencies and is
+     * therefore not blocked but queued for work
+     */
+    public boolean jobIsQueuedForWork(Job job)
+    {
+    	return (job.future != null && !job.future.isDone());
+    }
+
     public void startExecute(int n) {
-        if (logger != null)
-            logger.info("Starting " + n + " worker threads");
-        synchronized (this) {
-            if (myWorkerThreads == null) {
-                myWorkerThreads = new LinkedList<WorkerThread>();
-            }
-
-            for (int i = 0; i < n; ++i) {
-                WorkerThread t = new WorkerThreadImpl(this, i);
-                t.start();
-                myWorkerThreads.add(t);
-                stats.incWorkerThreads();
-            }
+        executorService = Executors.newCachedThreadPool();
+        for(Runnable runnable : frozenJobs)
+        {
+        	Job job = (Job)runnable;
+            Future<Void> future = executorService.submit(job); 
+            job.future = future;
         }
+        isExecuting = true;
     }
 
-    public void stopExecute() {
-        synchronized (this) {
-            if (myWorkerThreads == null) {
-                return;
-            }
-
-            for (WorkerThread t : myWorkerThreads) {
-                t.stopProcessing();
-                stats.decWorkerThreads();
-            }
-
-            myWorkerThreads.clear();
-        }
+    public void stopExecute() {    	
+    	frozenJobs = executorService.shutdownNow();
+        isExecuting = false;
     }
 
     synchronized public boolean isExecuting() {
-        synchronized (this) {
-            if (myWorkerThreads == null) {
-                return false;
-            } else {
-                return !myWorkerThreads.isEmpty();
-            }
-        }
+        return isExecuting; 
     }
 
     public SchedulerStats getSchedulerStats() {
@@ -245,8 +235,6 @@ public WorkerThread[] getWorkerThreads() {
     }
 
     public void startOneShotWorkerThread() {
-        WorkerThread t = new WorkerThreadImpl(this, true);
-        t.start();
     }
 
 	@Override
@@ -304,8 +292,12 @@ public synchronized void yield(Job j, ResumePoint p) throws SchedulerException {
 
         if (j.state() != Job.State.Yielded)
             j.yield(p);
-        workQueue.remove(j);
-        blockedQueue.add(j);
+
+        if (jobIsQueuedForWork(j) && j.future.cancel(false)) {
+        	j.future = null;
+        	blockedQueue.add(j);                	
+        }
+
     }
 }