Thrift In a Nutshell

Generating the Lucene Index

public class IndexBuilder {

    public static void main(String[] args) throws Exception {
        String namesFile = "names.csv";
        Document doc = new Document();
        Field[] fields = new Field[]{new Field("firstName", "", Field.Store.YES, Field.Index.ANALYZED_NO_NORMS),
                new Field("lastName", "", Field.Store.YES, Field.Index.ANALYZED_NO_NORMS),
                new Field("address", "", Field.Store.YES, Field.Index.ANALYZED_NO_NORMS),
                new Field("email", "", Field.Store.YES, Field.Index.ANALYZED_NO_NORMS)};
        addFieldsToDocument(doc, fields);

        BufferedReader reader = new BufferedReader(new FileReader(namesFile));

        IndexWriter indexWriter = new IndexWriter(FSDirectory.open(new File("blog-index")),new IndexWriterConfig(Version.LUCENE_31, new StandardAnalyzer(Version.LUCENE_31)));

        String line;
        while ((line = reader.readLine()) != null) {
            String[] personData = getPersonData(line);
            setFieldData(personData, fields);
            indexWriter.addDocument(doc);
        }
        indexWriter.optimize();
        indexWriter.close();
    }

    private static String[] getPersonData(String line) {
        return line.split(",");
    }

    private static void setFieldData(String[] data, Field[] fields) {
        int index = 0;
        for (Field field : fields) {
            field.setValue(data[index++]);
        }
    }

    private static void addFieldsToDocument(Document doc, Field[] fields) {
        for (Field field : fields) {
            doc.add(field);
        }
    }
}

Creating the .thrift File

//all generated java code will have the following for package name
namespace java bbejeck.thrift.gen

//this is the person object 
struct Person {
  1: string firstName,
  2: string lastName,
  3: string address,
  4: string email
}

//exception used to send meaningful error messages back to user
exception LuceneSearchException {
  1: string message
}

//service definition used by client and server
service LuceneSearch { 
    list<Person> search(1: string query) throws (1:LuceneSearchException error) 
}

• $ thrift --gen java lucene_search.thrift
• $ thrift --gen rb lucene_search.thrift

Thrift Server - Java

public class LuceneThriftServer {
    private static final int PORT = 9090;
    private static int numberThreads = 5;

    public static void main(String[] args) throws Exception {
        TServerSocket serverSocket = new TServerSocket(PORT, 100000);
        LuceneSearch.Processor searchProcessor = new LuceneSearch.Processor(new SearchHandler(args[0]));
        if (args.length > 1) {
            numberThreads = Integer.parseInt(args[1]);
        }
        TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverSocket);
        serverArgs.maxWorkerThreads(numberThreads);
        TServer thriftServer = new TThreadPoolServer(serverArgs.processor(searchProcessor).protocolFactory(new TBinaryProtocol.Factory()));
        thriftServer.serve();
    }

Iface Implementation

public class SearchHandler implements LuceneSearch.Iface {
    private IndexSearcher searcher;
    private QueryParser queryParser;
    private static final int MAX_RESULTS = 1000;

    public SearchHandler(String indexPath) {
        try {
            searcher = new IndexSearcher(FSDirectory.open(new File(indexPath)), true);
            queryParser = new QueryParser(Version.LUCENE_31, null, new StandardAnalyzer(Version.LUCENE_31));
            queryParser.setAllowLeadingWildcard(true);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public List<Person> search(String query) throws LuceneSearchException {
        List<Person> results = new ArrayList<Person>();
        try {
            Query q = queryParser.parse(query);
            TopDocs topDocs = searcher.search(q, MAX_RESULTS);
            for (ScoreDoc sd : topDocs.scoreDocs) {
                Document document = searcher.doc(sd.doc);
                results.add(getPersonFromDocument(document));
            }
        } catch (Exception e) {
            throw new LuceneSearchException(e.getMessage());
        }
        return results;
    }

    private Person getPersonFromDocument(Document document) {
        Person p = new Person();
        p.firstName = document.get("firstName");
        p.lastName = document.get("lastName");
        p.address = document.get("address");
        p.email = document.get("email");

        return p;
    }
}

Thrift Client - Ruby

module ThriftConnection

  class LuceneClient

    def initialize(host='localhost', port=9090)
      socket = Thrift::Socket.new(host, port)
      @transport = Thrift::BufferedTransport.new(socket)
      protocol_factory = ::Thrift::BinaryProtocolFactory.new
      protocol = protocol_factory.get_protocol(@transport)
      @transport.open
      @client = LuceneSearch::Client.new(protocol)
    end

    def search(query)
      @client.search(query)
    end

    def close
      @transport.close
    end

  end
end

Running With Scissors

#added to translate from symbol to expected search format
  SEARCH_KEYS_MAPPING = {:first_name => 'firstName',
                                            :last_name => 'lastName',
                                            :email => 'email',
                                            :address => 'address'}


  def self.method_missing(method_name, *args)
    lucene_client = ThriftConnection::LuceneClient.new
    query = ""
        #handles find_by_first_name etc
    if method_name.to_s =~ /^find_by_([a-z]+_?[a-z]*)$/
      query = "#{SEARCH_KEYS_MAPPING[$1.to_sym]}:#{args[0]}"
        #handles find_by_first_name_or_last_name, find_by_first_name_and_email 
    elsif method_name.to_s =~/^find_by_([a-z]+_[a-z]+)_([a-z]+)_([a-z]+_?[a-z]*)$/
      query ="#{SEARCH_KEYS_MAPPING[$1.to_sym]}:#{args[0]} #{$2.upcase} #{SEARCH_KEYS_MAPPING[$3.to_sym]}:#{args[1]}"
    else
       raise ArgumentError.new("search method pattern #{method_name} not recognized")
    end

    results = lucene_client.search(query)
    lucene_client.close
    results
  end

Testing

head names.csv

cat names.csv | grep -iE 'elizabeth|krause' | wc -l 

class SearchTest < Test::Unit::TestCase

  def setup
    @lucene_client = ThriftConnection::LuceneClient.new
  end


  def teardown
    @lucene_client.close
  end

  def test_search_client_first_name
    persons = @lucene_client.search("firstName:Tia")
    assert_equal(5, persons.length)

    persons.each do |person|
      assert_equal("Tia", person.firstName)
    end
  end

  def test_search_person_class_first_name
    persons = Person.find_by_first_name("Tia")
    assert_equal(5, persons.length)

    persons.each do |person|
      assert_equal("Tia", person.firstName)
    end
  end

  def test_search_client_first_name_email_domain
    persons = @lucene_client.search("+firstName:Elizabeth +email:*pookmail.com")
    assert_equal(59, persons.length)
  end

  def test_search_person_class_first_name_email_domain
    persons = Person.find_by_first_name_and_email("elizabeth", "*pookmail.com")
    assert_equal(59, persons.length)
  end

  def test_search_client_first_name_and_last_name
    persons = @lucene_client.search("+firstName:Elizabeth +lastName:Krause")
    assert_equal(1, persons.length)
    person = persons[0]

    assert_equal("Elizabeth", person.firstName)
    assert_equal("Krause", person.lastName)
  end

  def test_search_person_class_first_name_and_last_name
    persons = Person.find_by_first_name_and_last_name("elizabeth", "krause")
    assert_equal(1, persons.length)
    person = persons[0]

    assert_equal("Elizabeth", person.firstName)
    assert_equal("Krause", person.lastName)
  end

  def test_search_person_class_first_name_or_last_name
    persons = Person.find_by_first_name_or_last_name("elizabeth", "krause")
    assert_equal(289, persons.length)
  end

  def test_invalid_search
    assert_raises ArgumentError do
      Person.find_person_by_name("tia")
    end
  end

end

Conclusion

Resources

• For more information on thrift the thrift wiki is a great start
• More information on Lucene can be found here

Background

public class SortingBenchmarks extends SimpleBenchmark {
	
	private static final int SIZE = 100000;
	private static final int MAX_VALUE = 80000;
	private int[] values;

	@Override
	protected void setUp() throws Exception {
		values = new int[SIZE];
		Random generator = new Random();
		for (int i = 0; i < values.length; i++) {
			values[i] = generator.nextInt(MAX_VALUE);
		}
	}

	public void timeHeapSort(int reps) {
		for (int i = 0; i < reps; i++) {
			HeapSort.sort(values);
		}
	}

	public void timeMergeSort(int reps) {
		for (int i = 0; i < reps; i++) {
			MergeSort.sort(values);
		}
	}
	
	public void timeQuickSort(int reps) {
		for (int i = 0; i < reps; i++) {
			QuickSort.sort(values);
		}
	}

	public void timeArraysSort(int reps) {
		for (int i = 0; i < reps; i++) {
			Arrays.sort(values);
		}
	}
}

Getting Started

1. Extend the class SimpleBenchmark
2. Do any test setup/clean up in respective setUp or tearDown methods (Similar to JUnit setUp/tearDown)
3. Write the methods that will execute the code to benchmark starting with the word "time" (Again similar to JUnit, just "time" instead of "test")
4. Place the code you want to benchmark inside your timeSomeOperation methods

Getting and Running Caliper

1. Go here to get a read-only svn link to the source code
2. Check out the code then cd into <CALIPER_INSTALL_DIR> and run ant (obviously you need ant installed and on your path)

1. In <CALIPER_INSTALL_DIR>/build/caliper-0.0/lib/ there are two jar files, allocation.jar and caliper-0.0.jar. These will need to be on the classpath of your project. I chose to create a user library in Eclipse
2. JUnit will also need to be on the project classpath. Again for me it's referenced in a user library
3. I created a driver class that simply executes the main method of the com.google.caliper.Runner class. I pass in the name of my benchmark class by setting up a run configuration in Eclipse for my driver class

package bbejeck.caliper;

public class CaliperRunner {

    public static void main(String[] args) {
	com.google.caliper.Runner.main(args[0]);
    }

}

0% Scenario{vm=java, trial=0, benchmark=HeapSort} 10889428.57 ns; ?=69810.62 ns @ 3 trials
25% Scenario{vm=java, trial=0, benchmark=MergeSort} 9066618.18 ns; ?=44341.70 ns @ 3 trials
50% Scenario{vm=java, trial=0, benchmark=QuickSort} 3312312.93 ns; ?=21028.91 ns @ 3 trials
75% Scenario{vm=java, trial=0, benchmark=ArraysSort} 3104668.79 ns; ?=23965.54 ns @ 3 trials

 benchmark    ms logarithmic runtime
  HeapSort 10.89 ==============================
 MergeSort  9.07 =========================
 QuickSort  3.31 ==
ArraysSort  3.10 =

vm: java
trial:

Conclusion

Environment Used

• a T61 Thinkpad running Ubuntu 10.10(64Bit) and 4G ram
• Java JDK 1.6.0_21
• Glassfish Version 3. (I tried Tomcat 7 and Jetty 8, but had the best luck at this point with Glassfish)

DispatcherServlet Code

@WebServlet(urlPatterns = {"/async/*"}, asyncSupported = true, name = "async")
public class AsyncDispatcherServlet extends DispatcherServlet {
    private ExecutorService executor;
    private static final int NUM_ASYNC_TASKS = 15;
    private static final long TIME_OUT = 10 * 1000;
    private final Log log = LogFactory.getLog(AsyncDispatcherServlet.class);
@Override
    public void init(ServletConfig config) throws ServletException {
        super.init(config);
        executor = Executors.newFixedThreadPool(NUM_ASYNC_TASKS);
    }

    @Override
    public void destroy() {
        executor.shutdownNow();
        super.destroy();
    }

    @Override
    protected void doDispatch(final HttpServletRequest request, final HttpServletResponse response) throws Exception {
        final AsyncContext ac = request.startAsync(request, response);
        ac.setTimeout(TIME_OUT);
        FutureTask task = new FutureTask(new Runnable() {

            @Override
            public void run() {
                try {
                    log.debug("Dispatching request " + request);
                    AsyncDispatcherServlet.super.doDispatch(request,response );
                    log.debug("doDispatch returned from processing request " + request);
                    ac.complete();
                } catch (Exception ex) {
                    log.error("Error in async request", ex);
                }
            }
        }, null);

        ac.addListener(new AsyncDispatcherServletListener(task));
        executor.execute(task);
    }

1. The @WebServlet annotation at the class definition level. This is part of Servlet 3.0 specification, you can now declare servlets, filters etc via annotations, although you can still use web.xml. To enable asynchronous support set the 'asyncSupported' attribute to true.
2. On line 22 the setting of a timeout for the asyncContext object. In this case the timeout is 10 seconds
3. On line 38 setting an AsyncContextEventListener.
4. The application server thread returns almost immediately.

Listener Code

private class AsyncDispatcherServletListener implements AsyncListener {

        private FutureTask futureTask;

        public AsyncDispatcherServletListener(FutureTask futureTask) {
            this.futureTask = futureTask;
        }

        @Override
        public void onTimeout(AsyncEvent event) throws IOException {
            log.warn("Async request did not complete timeout occured");
            handleTimeoutOrError(event, "Request timed out");
        }

        @Override
        public void onComplete(AsyncEvent event) throws IOException {
            log.debug("Completed async request");
        }

        @Override
        public void onError(AsyncEvent event) throws IOException {
            log.error("Error in async request", event.getThrowable());
            handleTimeoutOrError(event, "Error processing " + event.getThrowable().getMessage());
        }

        @Override
        public void onStartAsync(AsyncEvent event) throws IOException {
            log.debug("Async Event started..");
        }

        private void handleTimeoutOrError(AsyncEvent event, String message) {
            PrintWriter writer = null;
            try {
                future.cancel(true);
                HttpServletResponse response = (HttpServletResponse) event.getAsyncContext().getResponse();
                //HttpServletRequest request = (HttpServletRequest) event.getAsyncContext().getRequest();
                //request.getRequestDispatcher("/app/error.htm").forward(request, response);
                writer = response.getWriter();
                writer.print(message);
                writer.flush();
            } catch (IOException ex) {
                log.error(ex);
            } finally {
                event.getAsyncContext().complete();
                if (writer != null) {
                    writer.close();
                }
            }
        }
    }

Web Application Structure

This web application is very simple and has only two controllers - SimpleViewControler and SearchController.

@Controller
public class SimpleViewController {
    
    @RequestMapping({"/","/index.htm"})
    public String showHome(){
        return "index";
    }

    @RequestMapping({"/error.htm"})
    public String error(){
        return "error";
    }

@Controller
public class SearchController {

    @RequestMapping("/search.htm")
    public String doSearch(@RequestParam(value = "latency", defaultValue = "2000") long latency,
                                        @RequestParam(value = "blowup", defaultValue = "false") boolean blowUp,
                                        Model model) throws Exception {

        String searchResult = getSearchResult(latency, blowUp);

        model.addAttribute("result", searchResult);
        return "searchResult";
    }

    @RequestMapping("/search.ajax")
    public void doSearchAjax(@RequestParam(value = "latency", defaultValue = "2000") long latency,
            @RequestParam(value = "blowup", defaultValue = "false") boolean blowUp,
            HttpServletResponse response) throws Exception {

        String searchResult = getSearchResult(latency, blowUp);
        
        PrintWriter writer = null;
        try {
            writer = response.getWriter();
            writer.print(searchResult);
            writer.flush();
        } finally {
            if (writer != null) {
                writer.close();
            }
        }
    }

    private String getSearchResult(long latency, boolean blowUp) throws Exception {

        if (blowUp) {
            throw new RuntimeException("Bad error happened in controller");
        }

        Thread.sleep(latency);

        StringBuilder builder = new StringBuilder("Some search/whatever results being returned");
        Date now = new Date();
        builder.append(" @").append(now);
        
        return builder.toString();
    }

Testing

1. A "control" test - There are two thread-groups consisting of 50 threads each and will ramp up to run all threads in 3 seconds. One thread group will make requests to /app/index.htm and the other thread group will make requests to /app/search.htm. The thread-groups will execute simultaneously and loop 3 times for a total of 300 requests. Each thread-group has a "listener" attached to it to measure throughput, and there is a listener attached to the test to measure overall throughput. This test will give us our baseline. The requests to /app/search.htm will not set any parameters, so each request will have the default value of 2 seconds for latency.
2. The "asynchronous" test - This test will measure the effect of using asynchronous servlets in the application. Setup is identical to the control test above with one exception - the search requests will go to /async/search.htm and hit the AsynchronousDispatcherServlet.
3. An error condition test - This test will be structured a little differently. The thread-group for /app/index.htm has a longer ramp up time, but will remain the same otherwise. The thread group for /async/search.htm will add a JMeter option known as a 'RandomController'. There will be 3 possible search requests sent, a valid request, a request with the latency parameter set to 12 seconds causing a timeout and a request with the blowup parameter set to true, so a RuntimeException will be thrown.

Test Results

As we can see from the test results, sending the search requests through the AsyncronousRequestDispatcher increased application throughput. The request per minute numbers don't mean that much though, given that the web application was so simple and the test was very contrived. What matters more is that the index requests had roughly the same response time and were seemingly unaffected when asynchronous support was used for the search requests

Summary

For me there were two main takeaways from this experiment:

• Even though asynchronous support seemed help with throughput, it is still using a thread pool which consumes server resources, so it should be only be applied to very select parts of an application.
• By setting timeouts and getting a chance to handle them gracefully via the event listener, asynchronous support acts a "circuit breaker" of sorts. This could be valuable when your application makes requests to outside resources that may be down or otherwise unresponsive.

Resources

Source for everything is available on github.
To run the JMeter tests

1. Download JMeter and extract the tar/zip file to some directory
2. Copy all of the *.jmx files in the jmeter directory from the github site for the code into <JMeter install>/bin. From the bin directory run jmeter or jmeter.bat depending on your platform. Once JMeter is up and running select File and you should see AsyncWebTestControl.jmx, AsyncWebTestErrors.jmx, AsyncWebTest.jmx in the File menu. Just click on one of those to open then Ctrl+r to run a test
3. Download the war file and deploy to glassfish. I placed the war file in the autodeploy directory in glassfish. On my laptop it's in /usr/local/servers/glassfishv3/glassfish/domains/domain1/autodeploy.