private synchronized void submitReduceJob(Job job) {
		System.out.println("Start reducing job #" + job.getId());
		int reducerNum = job.conf.REDUCER_NUM;
		if (dfs.getSlaveList().size() < reducerNum) {
			System.out.println("Not enough reducer. Cannot reduce. Job terminated.");
			String toHost = jobClient.get(jobID);
			Message msg = new Message("fail");
			CommModule.send(msg, toHost, Configuration.CLIENT_PORT);
			return;
		}
		int taskID = 0;
		int slaveID = 1;
		for (int i = 1; i <= reducerNum; i++) {
			ReduceTask task = new ReduceTask();
			task.setJob(job);
			while (!dfs.getSlaveList().containsKey(slaveID)) {
				slaveID ++;
			}
			task.setSlaveID(slaveID);
			task.setTaskID(taskID);
			ArrayDeque<ReduceTask> tasks = new ArrayDeque<ReduceTask>();
			tasks.push(task);
			slaveReduceTaskList.put(slaveID, tasks);
			HashSet<ReduceTask> hs = jobReduceTask.get(job.getId());
			if (hs == null) {
				hs = new HashSet<ReduceTask>();
			}
			hs.add(task);
			jobReduceTask.put(job.getId(), hs);
			taskID++;
			slaveID++;
		}
	}