Thread Executors in java

With Java 1.4, when we used to write threaded applications, we did not have much options in limiting and reusing the same thread for the available tasks. I had written applications earlier where each process that needed to be executed freely (or a process which can be executed on a thread) used to create its own thread. And once a thread was created and used, it was simply discarded. New threads were created whenever needed.

With java 1.5, there is an executor class which allows you to create controlled thread pools and execute the list of tasks on the same number of threads. If the number of tasks exceeds the number of threads then these extra tasks just wait for a thread to become available and then start their execution.

A simple threaded server using java 1.4 :

1 import java.io.*;
2 import java.net.*;
3 import java.util.*;
4
5 public class testPool implements Runnable
6 {
7   private int port;
8   private int active, total;
9
10   public testPool(int port)
11   {
12     this.port = port;
13     System.out.println(“Thread pool constructor – creating thread”);
14     new Thread(this).start();
15   }
16
17   public void run()
18   {
19     try
20     {
21       System.out.println(“Thread pool – new thread created”);
22       ServerSocket ss = new ServerSocket(port);
23       while(true)
24       {
25         Socket s = ss.accept();
26         total++;
27         new Handler(s);
28       }
29     }catch(Exception ex)
30     {
31       ex.printStackTrace();
32     }
33   }
34
35   public class Handler implements Runnable
36   {
37     private Socket socket;
38     public Handler(Socket s)
39     {
40       this.socket = s;
41       System.out.println(“Handler constructor – creating thread”);
42       new Thread(this).start();
43     }
44     public void run()
45     {
46       System.out.println(Thread.currentThread().getName()+” – Handler – new thread created”);
47       active++;
48       boolean loop = true;
49       try
50       {
51         BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
52         DataOutputStream out = new DataOutputStream(socket.getOutputStream());
53         while(loop)
54         {
55           String cmd = in.readLine();
56           if(cmd.equals(“QUIT”))
57           {
58             loop = false;
59             socket.close();
60           }
61           else if(cmd.equals(“INFO”))
62           {
63             System.out.println(“Active Connections : “+active);
64             System.out.println(“Total Connections : “+total);
65           }
66           else
67           {
68             System.out.println(“Command = “+cmd);
69           }
70         }
71       }catch(Exception ex)
72       {
73         ex.printStackTrace();
74       }
75       active–;
76     }
77   }
78
79   public static void main(String[] args)
80   {
81     System.out.println(“Starting Daemon”);
82     new testPool(Integer.parseInt(args[0]));
83   }
84 }

Compile and run the program

jayant@jayantbox:~/myprogs/java$ java -cp . testPool 5000
Starting Daemon
Thread pool constructor – creating thread
Thread pool – new thread created

Now, as you keep on connecting to localhost 5000 port, you can see that new threads are created. Even when the old connections are closed, the same thread is not used again but new threads are created.

so, if i do

telnet localhost 5000

from 3 consoles and issue the “INFO” command from 4th console, the output is

jayant@jayantbox:~/myprogs/java$ java -cp . testPool 5000
Starting Daemon
Thread pool constructor – creating thread
Thread pool – new thread created
Handler constructor – creating thread
Thread-1 – Handler – new thread created
Handler constructor – creating thread
Thread-2 – Handler – new thread created
Handler constructor – creating thread
Thread-3 – Handler – new thread created
Handler constructor – creating thread
Thread-4 – Handler – new thread created
Active Connections : 1
Total Connections : 4

4 threads are created. In this case, everytime you connect to the server, a new thread would be created.

Now, lets modify the code to work on java 1.5 using thread Executors…

Add the following code snippets at or after the following line numbers

4 import java.util.concurrent.*;
10  private int pool;
12  this.pool=2;
22  ExecutorService threadExecutor = Executors.newFixedThreadPool(pool);
27  threadExecutor.execute(new Handler(s));

And comment the following 2 lines

27  new Handler(s);
42  new Thread(this).start();

Not lets run the program again on port 5000 and try connecting to it. We have created a pool of 2 threads here. Observer that:

  • pool-1-thread-1 & pool-1-thread-2 are the two threads created.
  • After two connections to the server, the connections are not rejected, but are queued.
  • If a thread becomes available, a connection from the queue is assigned to the thread for processing
  • The same two threads are used again and again. No new threads are created.
  • Connections which are in queue are not able to process any requests unless they are assigned to a worker thread

Process for testing

console 1: telnet localhost 5000
console 2: telnet localhost 5000
console 3: telnet localhost 5000
console 1: from 1
console 2: from 2
console 3: from 3
console 1: INFO
console 1: 1 quitting
console 1: QUIT

Output for the testing process

jjayant@jayantbox:~/myprogs/java$ java -cp . testPool 5000
Starting Daemon
Thread pool constructor – creating thread
Thread pool – new thread created
Handler constructor – creating thread
pool-1-thread-1 – Handler – new thread created
Handler constructor – creating thread
pool-1-thread-2 – Handler – new thread created
Handler constructor – creating thread
Command = from 1
Command = from 2
Active Connections : 2
Total Connections : 3
Command = 1 quitting
pool-1-thread-1 – Handler – new thread created
Command = from 3

As seen, processing of connection 3 starts only after connection 1 is closed by the client.

The benefits of using a thread executor instead of a normal thread are

a] Overhead of creating and destroying threads is avoided.
b] A queue of threads is created automatically when the no of tasks exceeds the no of threads in the pool. The tasks in queue are automatically executed when the threads become free.
c] A limit (maximum no of threads) could be imposed thus controlling the available resources for delivering better performance.

Berkeley DB

I had a chance to look in BDB-XML some time back. But i did not find it that good in performance. BDB on the other hand promises fast storage and retrieval of key-value pairs. For the new-comers, BDB is a embedded database API which can be used to create your own database and fire a defined set of queries on it. BDB does not provide you with an SQL interface or even a database server. All you have is a set of API using which you can write programs to create your own database, populate data into the database, define your own cache and indexes and then use the database to retrieve values for a particular key using your own programs.

With BDB, there is no sql/query optimizer/query parser layer between you and the database.

There are certain types of BDB engines available:

1.] BDB -> the original BDB with c/c++ api. You write programs in c/c++ to create and access your database.
2.] BDB-JAVA -> The java api to BDB. The java api uses JNI in the backend to communicate with the BDB library (written in c/c++).
3.] BDB-JE -> The java edition of the BDB engine. It is a pure java implementation of the BDB engine. It does not use JNI for communication with the system.
4.] BDB-XML -> This is a very sophesticated version of BDB – where you can store XML documents and retrieve documents using any of the keys in the XML Document. You have an XQuery interface where you can fire XML based queries and retrieve results.

The original BDB is ofcourse the fastest.

For a startup, we will take a look at the DPL API of BDB-JAVA. DPL stands for Direct Persistence Layer and is used generally for storing and managing java objects in the database. DPL works best with a static database schema and requires java 1.5.

To create a database using DPL, you generally require an entity class and then create/open a database environment and insert the entity object into the entity store in the database environment. Sounds greek right ?? Lets see an example

Entity Class

import com.sleepycat.persist.*;
import com.sleepycat.db.*;
import com.sleepycat.persist.model.*;
import static com.sleepycat.persist.model.Relationship.*;

@Entity
public class SimpleEntityClass {

   // Primary key is pKey
   @PrimaryKey
   private String pKey;

   // Secondary key is the sKey
   @SecondaryKey(relate=MANY_TO_ONE)
   private String sKey;

   public SimpleEntityClass(String pk, String sk)
   {
      this.pKey = pk;
      this.sKey = sk;
   }

   public void setpKey(String data)
   {
      pKey = data;
   }

   public void setsKey(String data)
   {
      sKey = data;
   }

   public String getpKey()
   {
      return pKey;
   }

   public String getsKey()
   {
      return sKey;
   }
}

Then create a Database Access class which can insert and retrieve Entity objects from the database.

import java.io.*;
import com.sleepycat.db.*;
import com.sleepycat.persist.*;

public class SimpleDA
{
   PrimaryIndex pIdx;
   SecondaryIndex sIdx;

   EntityCursor pcursor;
   EntityCursor scursor;

   public SimpleDA(EntityStore store) throws Exception
   {
      pIdx = store.getPrimaryIndex(String.class, SimpleEntityClass.class);
      sIdx = store.getSecondaryIndex(pIdx, String.class, “sKey”);
   }

   public void addEntry(String pk, String sk) throws DatabaseException
   {
      pIdx.put(new SimpleEntityClass(pk, sk));
   }

   public SimpleEntityClass findByPk(String pk) throws DatabaseException
   {
      SimpleEntityClass found = pIdx.get(pk);
      return found;
   }

   public ArrayList findBySk(String sk) throws DatabaseException
   {
      ArrayList ret = new ArrayList();
      scursor = sIdx.subIndex(sk).entities();
      for(SimpleEntityClass sec1 = scursor.first(); sec1!=null; sec1 = scursor.next())
      {
         ret.add(sec1);
      }
      scursor.close();
      return ret;
   }
}

Create/open the environment to put and retrieve records from the database

import java.io.*;
import com.sleepycat.db.*;
import com.sleepycat.persist.*;

public class SimpleStore
{
   private static File envHome = new File(“./bdbjava”);
   private Environment env;
   private EntityStore store;
   private SimpleDA sda;

   public void setup() throws DatabaseException
   {
   // put all config options here.
      EnvironmentConfig envConfig = new EnvironmentConfig();
      envConfig.setAllowCreate(true);
      envConfig.setCacheSize(536870912); //512 MB
      envConfig.setCacheCount(2); // 2 caches of 256 MB each
      envConfig.setTxnWriteNoSync(true);
      envConfig.setInitializeCache(true);
      envConfig.setThreaded(true);
      envConfig.setInitializeLogging(true);
      envConfig.setTransactional(true);

      StoreConfig sConfig = new StoreConfig();
      sConfig.setAllowCreate(true);

      try
      {
         env = new Environment(envHome, envConfig);
         store = new EntityStore(env, “MyDatabaseName”, sConfig);
      }catch(Exception ex)
      {
         ex.printStackTrace();
         System.exit(-1);
      }
   }

   public SimpleStore()
   {
      setup();
   }

   public void putData(String pk, String sk) throws Exception
   {
      sda = new SimpleDA(store);
      sda.addEntry(pk, sk);
   }

   public void getDataPk(String pk) throws Exception
   {
      sda = new SimpleDA(store);
      SimpleEntityClass sec = sda.findByPk(pk);
      System.out.println(“pk = “+sec.getpKey()+”, sk = “+sec.getsKey());
   }

   public void getDataSk(String sk) throws Exception
   {
      sda = new SimpleDA(store);
      ArrayList data = sda.findBySk(sk);
      for(int x=0; x<data.size(); x++)
      {
         SimpleEntityClass sec = data.get(x);
         System.out.println(“pk = “+sec.getpKey()+”, sk = “+sec.getsKey());
      }
   }

   public void closeAll() throws Exception
   {
      store.close();
      env.close();
   }

   public static void main(String[] args)
   {
      SimpleStore ss = new SimpleStore();
      ss.putData(“pk1″,”sk1”);
      ss.putData(“pk2″,”sk2”);
      ss.putData(“pk3″,”sk1”);
      ss.putData(“pk4″,”sk3”);

      ss.getDataPk(“pk1”);

      ss.getDataSk(“sk1”);

      ss.closeAll();
   }
}

So, now you have your own program for creating entities of the type SimpleEntityClass and store them in the database in serialized form. These objects cab be retrieved using primary or secondary keys.

For the relationship between primary and secondary keys please refer to http://www.oracle.com/technology/documentation/berkeley-db/db/java/com/sleepycat/persist/model/Relationship.html

Since you would be storing complete objects instead of just the required data sets, the database size would be relatively high and that would slow down things a bit.

The Other World…

There is this world where you buy your tickets (be it movie tickets or travel tickets) online. Where you go to the big baazaar an swipe your credit card to make the payment. Where you call up for your car service and the service guys take the car from your place, service it and give it back to you with a bill. You use your credit card for most of the transactions and you rarely deal with cash. You sit in an air-conditioned office and punch a few keys on your computer and you get paid heavily for it. The world of a software related person.

And then there is this other world where everything is done through cash. You rely on the travel agent to get your rail and air tickets. And you get your movie tickets from the movie hall ticket window. You take your car to the service station yourself and stand there and get the service done in front of your eyes. You pay cash all the time and try to avoid getting bills to avoid paying the 12.XX% tax on the total amount. You are not eligible for a credit card cause you dont have any asset to show to the banks.

I had an encounter with the “other world” when i got wood work and painting done in my new flat. I came to know that the margin of profit could be increased for 10% to 60% by simply twisting the quality of materials. I also came to know the fact that the MRP quoted on most materials includes a huge profit margin and that you can actually bargain on the MRP.

When i went to get the first lot of wood i came to know that wood comes from 30Rs/sq ft to more than 100Rs/sq ft. And a increase of 5 Rs/sq ft can make a huge difference in the final cost. Dealing with the workers is a pain. For them you are a fresh,delicious and juicy piece of chicken almost ready for being chopped. And they would love to chop you irrespective of what you think. If you leave them to their will, they would chop you and fight over your pieces. For them even 5 Rs has a meaning. And then there are these people in the local market who only understand cash. If you look at their shop, you wont believe that the shop can have lakhs of rs in cash within itself. There are no guards to guard the shop. And they still have lakhs of cash with them and crores of raw materials just lying there. They dont have a credit card machine cause the bank wont give them any. They dont have credit cards themselves cause they dont have any proof of their huge income. They make lakhs in a month and dont file any income tax returns. They hardly pay any tax. And the fact may be that though they may be serving you, but they may be having more wealth and property than you do.

When i made my first payment of 46000 (cash) to the wood supplier, i filled up my office bag with rupees. I was scared of driving on the road and my hands were literally trembling when i was counting the amount i had to pay. But the shop owner was calm and composed as if it was his daily job. These guys dont know what the internet is or how to use it. All they know is how to make money. It is the other world – the actual world. For us this world is hidden behind a layer of service guys and internet.