public void submitMapJob(String host, Job job) {
		job.setID(getJobID());
		synchronized (jobList) {
			jobList.put(job.getId(), job);
		}
		synchronized (jobClient) {
			jobClient.put(job.getId(), host);
		}
		synchronized (jobAssignedTask) {
			HashSet<Task> hs = new HashSet<Task>();
			jobAssignedTask.put(job.getId(), hs);
		}
		synchronized (jobFinishedTask) {
			HashSet<Task> hs = new HashSet<Task>();
			jobFinishedTask.put(job.getId(), hs);
		}
		System.out.println("Start job #" + job.getId());
		// create and distribute splits
		if (!dfs.distributeFile(job.conf.INPUT_DIR, job.conf.REPLICA, job)) {
			String toHost = jobClient.get(jobID);
			Message msg = new Message("fail");
			CommModule.send(msg, toHost, Configuration.CLIENT_PORT);
		}
	}