public synchronized void enforceReplication(int id,
			ArrayDeque<MapTask> tasks) {
		if (tasks == null) {
			tasks = new ArrayDeque<MapTask>();
		}
		// get running task for that job id
		ArrayDeque<MapTask> runningTask = tracker.getRunningMapList(id);
		tasks.addAll(runningTask);
		for (MapTask task : tasks) {
			int jobID = task.getJob().getId();
			FileSplit split = task.getInput();
			// regenerate file split accordingly
			LineRecordWriter writer = new LineRecordWriter(
					task.getJob().conf.INPUT_DIR + split.getOriginalName(),
					split.getStart(), task.getJob());
			Pair node = dfsNode.poll();
			if (node != null && node.id == id) {
				node = dfsNode.poll();
			}
			if (node == null) {
				return;
			}
			HashMap<Integer, HashSet<FileSplit>> slaveToFile = jobToSlaveFile
					.get(jobID);
			HashSet<FileSplit> files = slaveToFile.get(node.id);
			// get the node with least files
			if (files == null) {
				files = new HashSet<FileSplit>();
				files.add(split);
				slaveToFile.put(node.id, files);
				node.fileCount++;
				dfsNode.add(node);
			} else if (!files.contains(split)) {
				files.add(split);
				slaveToFile.put(node.id, files);
				node.fileCount++;
				dfsNode.add(node);
			} else {
				// if the node already contains that file chunk
				// randomly select another node to put the chunk
				dfsNode.add(node);
				Iterator<Pair> itr = dfsNode.iterator();
				boolean found = false;
				while (itr.hasNext()) {
					node = itr.next();
					files = slaveToFile.get(node.id);
					if (files == null) {
						files = new HashSet<FileSplit>();
						files.add(split);
						slaveToFile.put(node.id, files);
						node.fileCount++;
						found = true;
						break;
					} else if (!files.contains(split)) {
						files.add(split);
						slaveToFile.put(node.id, files);
						node.fileCount++;
						found = true;
						break;
					}
				}
				if (!found) {
					System.out.println("Failed to enforce replication for "
							+ split.getFilename());
					continue;
				}
			}
			FileChunk chunk = writer.write(split.getFilename());
			DFSMessage msg = new DFSMessage("distribute", chunk,
					slaveList.get(node.id), Configuration.SERVER_PORT);
			try {
				commModule.send(msg);
			} catch (RemoteException e) {
				break;
			}
			HashSet<Task> ft = new HashSet<Task>();
			ft = tracker.getFinishedTaskList(jobID);
			if (ft.contains(task)) {
				continue;
			}
			task.setSlaveID(node.id);
			tracker.addQueuedMapTask(node.id, task);
		}
	}