@Override
	public void run() {
		while (true) {
			try {
				Message ret = null;
				// produce partitions it generated from map tasks
				HashMap<Integer, HashMap<Integer, ArrayDeque<String[]>>> partitions = tracker
						.getPartition();
				if (partitions.size() > 0) {
					// get current working slaves information
					WorkMessage msg = new WorkMessage("slave", toHost, toPort);
					msg.setSlaveID(slaveID);
					ret = commModule.send(msg);
					dispatch(ret);
					HashMap<Integer, String> slaveList = tracker.getSlaveList();
					
					// send the partition information to corresponding slaves
					for (Entry<Integer, HashMap<Integer, ArrayDeque<String[]>>> item : partitions
							.entrySet()) {
						int jobID = item.getKey();
						HashMap<Integer, ArrayDeque<String[]>> slave = item
								.getValue();
						for (Entry<Integer, ArrayDeque<String[]>> par : slave
								.entrySet()) {
							// get the slave ID and send the partitions
							int slaveID = par.getKey();
							ArrayDeque<String[]> partition = par.getValue();
							String host = slaveList.get(slaveID);
							PartitionMessage pm = new PartitionMessage(
									"partition", host, toPort);
							pm.setPartition(jobID, partition);
							ret = commModule.send(pm);
						}
					}
				}
				int idleMapSlot = tracker.getIdleMapSlot();
				int idleReduceSlot = tracker.getIdleReduceSlot();
				// if there are slots for new tasks, ask the master for new task
				if (idleMapSlot > 0 || idleReduceSlot > 0) {
					WorkMessage msg = new WorkMessage("idle", toHost, toPort);
					msg.setFinishedTask(tracker.getFinishedTasks());
					msg.setMapSlot(idleMapSlot);
					msg.setReduceSlot(idleReduceSlot);
					msg.setSlaveID(slaveID);
					ret = commModule.send(msg);
				} else {
					// if there is no available slots, tell the master too
					WorkMessage msg = new WorkMessage("busy", toHost, toPort);
					msg.setSlaveID(slaveID);
					ret = commModule.send(msg);
				}
				dispatch(ret);
				// wait for some interval until the next heart beat
				Thread.sleep(sleepInterval);
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (RemoteException e) {
				retryCount++;
				// recognizing the master is not there, it the slave aborts
				if (retryCount > retryNum) {
					System.out
							.println("Coordinator not responding. Job termitated");
					System.exit(-1);
				}
				System.out.println("Coordinator not responding. Retry in "
						+ sleepInterval / 1000 + " seconds... (attempt "
						+ retryCount + "/" + retryNum + ")");
			}
		}

	}