public synchronized boolean distributeFile(String in, int replica, Job job) {
		System.out.println("Distributing files on DFS...");
		if (slaveList.size() <= 0) {
			System.out.println("No slave available. Cannot start job.");
			return false;
		}
		File inputDir = new File(in);
		if (!inputDir.exists()) {
			System.out.println("Input Directory doesn't exist");
			return false;
		}
		int jobID = job.getId();
		HashMap<Integer, HashSet<FileSplit>> slaveToFile = jobToSlaveFile
				.get(jobID);
		if (slaveToFile == null) {
			slaveToFile = new HashMap<Integer, HashSet<FileSplit>>();
		}
		int taskID = 0;
		// iterate through the input files to generate file split
		for (String filename : inputDir.list()) {
			String path = in + filename;
			// create reader and writer for the file
			LineRecordReader reader = new LineRecordReader(path, job);
			int recordNum = reader.getRecordNum();
			LineRecordWriter writer = new LineRecordWriter(path, job);
			// generate file splits
			for (int i = 0; i < recordNum; i++) {
				FileSplit split = writer.createSplit(i, filename);
				String fn = split.getFilename();
				// write the data into a file chunk
				FileChunk chunk = writer.write(fn);
				// create a new map task accordingly
				MapTask newTask = new MapTask();
				newTask.setInput(split);
				newTask.setJob(job);
				newTask.setStatus(Status.PENDING);
				newTask.setTaskID(taskID);
				// choose a node with least files and replicate the file chunk
				for (int j = 0; j < replica; j++) {
					Pair node = dfsNode.poll();
					if (node == null) {
						break;
					}

					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;
						}
					}
					DFSMessage msg = new DFSMessage("distribute", chunk,
							slaveList.get(node.id), Configuration.SERVER_PORT);
					try {
						commModule.send(msg);
					} catch (RemoteException e) {
						break;
					}
					newTask.setSlaveID(node.id);
					// let tracker know the node can work on a new task
					tracker.addQueuedMapTask(node.id, newTask);
				}
				taskID++;
			}
		}
		jobToSlaveFile.put(jobID, slaveToFile);
		jobTaskCount.put(jobID, taskID);
		return true;
	}