public synchronized void markDone(ArrayDeque<Task> finishedTask) {
		for (Task task : finishedTask) {
			int jobID = task.getJob().getId();
			// if the job is in mapping process
			if (task.getType() == 'M') {
				int slaveID = task.getSlaveID();
				// get the running tasks and remove the finished ones
				HashSet<MapTask> running = slaveRunningTask.get(slaveID);
				running.remove(task);
				slaveRunningTask.put(slaveID, running);
				// remove the task from job task list
				HashSet<MapTask> hs = jobMapTask.get(jobID);
				hs.remove(task);
				jobMapTask.put(jobID, hs);
				HashSet<Task> ft = jobFinishedTask.get(jobID);
				ft.add(task);
				jobFinishedTask.put(jobID, ft);
				// check if there is remaining task
				if (hs.isEmpty()) {
					// start reduce for this job
					jobMapTask.remove(jobID);
					System.out.println("Finished mapping job #" + jobID);
					submitReduceJob(task.getJob());
				}
			}
			// if the job is in reducing process
			else if (task.getType() == 'R') {
				HashSet<ReduceTask> hs = jobReduceTask.get(jobID);
				hs.remove(task);
				HashSet<Task> ft = jobFinishedTask.get(jobID);
				ft.add(task);
				jobFinishedTask.put(jobID, ft);
				String path = ((ReduceTask) task).getOutput();
				ArrayDeque<String> curOutput = jobOutput.get(jobID);
				if (curOutput == null) {
					curOutput = new ArrayDeque<String>();
				}
				// add the output path to the list
				curOutput.add(path);
				jobOutput.put(jobID, curOutput);
				System.out.println("Reducing output generated at " + path);
				// if all the reduce jobs finished, report to the client
				if (hs.isEmpty()) {
					jobReduceTask.remove(jobID);
					System.out.println("Job #" + jobID + " finished.");
					String toHost = jobClient.get(jobID);
					StringBuilder sb = new StringBuilder();
					ArrayDeque<String> outputs = jobOutput.get(jobID);
					for (String s : outputs) {
						sb.append(s);
						sb.append("\n");
					}
					// send finish message back
					Message msg = new Message(sb.toString());
					CommModule.send(msg, toHost, Configuration.CLIENT_PORT);
				}
			} else {
				System.out.println("Invalid task type");
			}
		}
	}