package aj.hadoop.monitor.examples;

import java.io.*;
import java.net.*;
import java.util.*;
import java.text.*;
import java.lang.*;
import java.lang.management.*;
import java.util.StringTokenizer;
import java.util.logging.FileHandler;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.Signature;

import aj.hadoop.monitor.util.PickerClient;

@Aspect
public class WordCountMonitor {

	private String HOST = "127.0.0.1";
	private int PORT = 55000;

	/** log file */
	public static final String LOGFILE = "Hanoi-MethodTraceMonitor.log";

	/**
	 * pointcut
	 * @param word
	 * @param one
	 */
	@Pointcut ("call(void org.apache.hadoop.mapreduce.Mapper.Context.write" +
			"(" +
			"java.lang.Object, " +
			"java.lang.Object" +
			"))" +
			"&& cflow(execution(public void org.apache.hadoop.examples.WordCount$TokenizerMapper.map(..)))" +
			"&& args(word, one)")
	public void catch_map_method(Object word, Object one){}

	@Before ( value = "catch_map_method( key, one )")
	public void logging_current_method( JoinPoint thisJoinPoint,
			Object key,
			Object one) {

		try { 
			PickerClient client = new PickerClient();
			client.send(key);
			client.socketClose();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

@@ -188,24 +188,60 @@
 gTokenizer;%0A
+import java.util.concurrent.Phaser;%0A
 import java.
@@ -995,16 +995,31 @@
 pointcut
+ for map method
 %0A%09 * @pa
@@ -1375,16 +1375,518 @@
 one)%7B%7D%0A%0A
+%09/**%0A%09 * pointcut for reduce method%0A%09 * @param word%0A%09 * @param values%0A%09 * @param cntext%0A%09 */%0A%09@Pointcut (%22call(public void org.apache.hadoop.examples.WordCount$IntSumReducer.reduce(%22 +%0A%09%09%09%22(%22 +%0A%09%09%09%22java.lang.Object, %22 +%0A%09%09%09%22java.lang.Object, %22 +%0A%09%09%09%22java.lang.Object%22 +%0A%09%09%09%22))%22 +%0A%09%09%09%22&& args(key, values, context)%22)%0A%09public void catch_reduce_method(Object key, Object values, Object context)%7B%7D%0A%0A%0A%09/**%0A%09 * logging map method using HanoiPicker%0A%09 * @param thisJoinPoint%0A%09 * @param key%0A%09 * @param one%0A%09 */%0A
 %09@Before
@@ -1949,23 +1949,19 @@
 logging_
-current
+map
 _method(
@@ -2019,16 +2019,40 @@
 one) %7B%0A%0A
+%09%09String phase = %22MAP%22;%0A
 %09%09try %7B 
@@ -2112,16 +2112,522 @@
 nt.send(
+phase + %22,%22 + key);%0A%09%09%09client.socketClose();%0A%09%09%7D catch (Exception e) %7B%0A%09%09%09e.printStackTrace();%0A%09%09%7D%0A%09%7D%0A%0A%09/**%0A%09 * logging reduce method using HanoiPicker%0A%09 * @param thisJoinPoint%0A%09 * @param key%0A%09 * @param one%0A%09 */%0A%09@Before ( value = %22catch_reduce_method( key, values, context )%22)%0A%09public void logging_reduce_method( JoinPoint thisJoinPoint,%0A%09%09%09Object key,%0A%09%09%09Object values,%0A%09%09%09Object context) %7B%0A%0A%09%09String phase = %22SHUFFLE%22;%0A%09%09try %7B %0A%09%09%09PickerClient client = new PickerClient();%0A%09%09%09client.send(phaser + %22,%22 + 
 key);%0A%09%09

send message to HanoipPicker with map/shuffle/reduce phase.