synchronized void shuffle() {
		boolean hasException = false;
		Exception exception = null;
		long beginTime = System.currentTimeMillis();
		try {
			// check if we really have reducers
			if (hasReduceClass) {
				// try to randomize the sending order
				List<Integer> keyList = new ArrayList<Integer>(
						this.reduceInputs.keySet());
				Random randomGenerator = new Random(this.daemonNo);
				while (!keyList.isEmpty()) {
					int reduceNo = keyList.remove(randomGenerator
							.nextInt(keyList.size()));
					ReduceInput input = this.reduceInputs.get(reduceNo);
					// One map output goes to one reducer
					TwisterMessage newInputMsg = copyDataToCacheIfLargeAndGetReduceInput(
							input, 1);
					this.pubsubService
							.send(input.getReduceTopic(), newInputMsg);
				}
				/*
				for (ReduceInput input : this.reduceInputs.values()) {
					// One map output goes to one reducer
					TwisterMessage newInputMsg = copyDataToCacheIfLargeAndGetReduceInput(
							input, 1);
					this.pubsubService
							.send(input.getReduceTopic(), newInputMsg);
				}
				*/
				if (this.bcastSupported) {
					for (ReduceInput input : this.rowBCastValues.values()) {
						// sqrtReducers is the number of receiver of the data
						TwisterMessage newInputMsg = copyDataToCacheIfLargeAndGetReduceInput(
								input, this.sqrtReducers);
						this.pubsubService.send(input.getReduceTopic(),
								newInputMsg);
					}
				}
				long endTime = System.currentTimeMillis();
				// send a response to driver here 
				TaskStatus status = new TaskStatus(
						TwisterConstants.SHUFFLE_TASK,
						TwisterConstants.SUCCESS, this.daemonNo, this.daemonNo,
						(endTime - beginTime), iteration);
				status.setReduceInputMap(this.reduceInputMap);
				this.pubsubService.send(TwisterConstants.RESPONSE_TOPIC_BASE
						+ "/" + this.jobId, status);
				// then do clean
				this.reduceInputs.clear();
				this.reduceInputMap.clear();
				if (this.bcastSupported) {
					this.rowBCastValues.clear();
				}
			}
			hasException = false;
		} catch (Exception e) {
			e.printStackTrace();
			hasException = true;
			exception = e;
		}
		// if there is exception
		if (hasException) {
			TaskStatus status = new TaskStatus(TwisterConstants.SHUFFLE_TASK,
					TwisterConstants.FAILED, this.daemonNo, this.daemonNo,
					(System.currentTimeMillis() - beginTime), iteration);
			status.setExceptionString(exception.getMessage());
			try {
				this.pubsubService.send(TwisterConstants.RESPONSE_TOPIC_BASE
						+ "/" + this.jobId, status);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}