Thread Pool Manager and Timer

This document describes the Thread Pool Manager and Timer services, implemented in the ProSyst Util Bundle.

Contents:

• Thread Pool Manager
  • The Concept of Thread Pools
  • Autoshrinking
  • Ways of Threads Allocation
  • Executing a Runnable Job
  • Accessing Runnable Instances
  • System Properties
• Timer Service
  • Overview
  • Notification Types
  • No-Delay Timers
  • Registering for Notification
  • System Properties
• References

Thread Pool Manager

The Concept of Thread Pools

To be able to operate with multiple clients in a system with scanty resources, the OSGi framework must highly optimize the speed of the requests' execution. The speed of the requests' processing depends on how the framework manages its memory buffers, thread pools, sockets, and database connections. Frequently, services in a bundle require creating their own threads when activated, but creating a new thread is a time-consuming operation, which slows down the performance of the framework. In addition, there are server services which create multiple threads to serve concurrent client requests. For optimization of resource usage for creating threads developers can benefit from the thread pooling mechanism of the Thread Pool Manager.

A thread pool is a collection of created threads, put in waiting state. The Thread Pool Manager maintains a default thread pool whose threads are shared by all bundles in the OSGi framework.

The Thread Pool Manager uses two pools for thread pool management:

• A pool with threads that are waiting for execution - idle threads. If there are threads in this pool, the Thread Pool Manager notifies them and attaches a job (an implementation of the java.lang.Runnable interface) for execution.
• A pool containing jobs that are waiting for execution because there are no idle threads - waiting jobs.

At startup, the Thread Pool Manager generates a certain number of threads according to a minimum value defined as a VM system property and pushes them in the idle thread pool. Jobs that request execution are attached to threads from this pool and are executed. The number of existing threads can be extended up to a maximum value assigned again as a system property. If the count of requesting jobs exceeds this value, they are temporarily kept in the waiting jobs pool in which they wait for threads to free. If a thread completes its job, the waiting jobs pool is examined and the job that is first in the wait queue is released and attached to the thread without pooling it.

Note that such a mechanism is suitable mainly for short-lived jobs. Otherwise, the waiting jobs pool will extremely grow and the operation of the Thread Pool Manager will slow down. Another problem appears when processing bad-written Runnable objects (never exiting their run method). In this case, the number of free threads decreases and the number of already running threads will reach the limit, so it may occur that no more threads can be processed.

Autoshrinking

After some time of work, idle threads in the pool may increase their number much above the minimum. That’s why an autoshrinking mechanism is implemented - every 30 seconds (or the value of the mbs.util.threadpool.inactiveTime system property) the Thread Pool Manager checks the threads in the pool and if needed reduces the idle threads number to the number of accessed threads within this time period. If accessed threads are less than the minimum, idle threads count is shrunk to the defined minimum.

Ways of Threads Allocation

The Thread Pool Manager supports two approaches for allocating threads from the thread pool:

• Allocating a specified portion of threads in the thread pool, needed for the correct operation of the bundle.

  This operation can be done by using the Thread Pool Factory, available as a com.prosyst.util.threadpool.ThreadPoolFactory service object in the OSGi framework. The factory's only method getThreadPool provides restricted access to the thread pool - the caller bundle can use no more threads than the number or percent specified when the getThreadPool method has been invoked.

• Allocating predefined percent of threads in the thread pool.

  Bundles can refer directly to the Thread Pool Manager, available as a com.prosyst.util.threadpool.ThreadPoolManager service object in the OSGi framework, to use threads from its thread pool. In such case, the Thread Pool Manager allows the caller bundle to use up to 30% (or the percent specified through the mbs.util.threadpool.percent system property) of the threads in the pool.

In both ways it becomes possible to avoid having all pool's threads allocated by a single bundle. The bundle is not allowed to take concurrently more threads than the number specified earlier to the Thread Pool Manager. If the bundle requests a thread over the indicated limit, the request will wait until the bundle reduces consumed threads. Then, the job will be queued up in the pool of waiting jobs.

Executing a Runnable Job

A Runnable job can be executed within the thread pool by passing it in the corresponding execute method of the ThreadPoolManager instance either returned by the Thread Pool Factory or got as a service from the OSGi framework. The Thread Pool Manager allows setting different priorities of its subordinate threads as well as, depending on the mbs.util.threadpool.useNames system property, naming each executed job.

The code example below contains two components: a bundle activator that refers to the ThreadPoolManager service, and a simple implementation of the java.lang.Runnable interface representing a job. The job is simple and common - to process an operation, it waits for an event to occur; if the event does not occur within a certain time period (1 second in the example), another operation is loaded. In the example below, the event is the turning on of the check flag and all operations are related to printing messages in the system output.

  public class RunnableTest implements Runnable {
         
    private Object synch; 
    private boolean check;
          
    //Constructs a test job
    public RunnableTest() {
      check = false;
      synch = new Object();
    } 
         
    //This method is executed by the Thread Pool Manager service
    public void run() {
      synchronized(synch) {     
        System.out.println("Job  " 
                           + Thread.currentThread().getName() 
                           + " starts running!"); 
        if (!check) {       
          try {         
            synch.wait(1000);       
           } catch (InterruptedException e) { }     
        }
        if (check) {       
          System.out.println("OK: CHECKED");     
        } else {       
          System.out.println("NOT OK: NOT CHECKED");     
        }
      }
    }
    
    //Turns the check flag on 
    public void checkIt() { 
      synchronized (synch) { 
        check = true; 
        synch.notify(); 
      } 
    }
 
    //Wakes up this Runnable object causing exiting its run method
    public void stopIt() { 
      synchronized (synch) { 
        synch.notify(); 
      } 
    } 
  }

  import org.osgi.framework.BundleContext;
  import org.osgi.framework.BundleActivator; 
  import org.osgi.framework.BundleException;
  import org.osgi.framework.ServiceReference;  
  import com.prosyst.util.threadpool.ThreadPoolManager;        
 
  public class TestActivator implements BundleActivator {

    ServiceReference tpoolManRef;
    ThreadPoolManager tpoolMan;  
    private static String servName = ThreadPoolManager.class.getName();
 
    public void start(BundleContext bc) throws BundleException { 
 
      // Retrieving the Thread Pool Manager from the OSGi framework
      tpoolManRef = bc.getServiceReference(servName);   
      if(tpoolManRef != null) {
        tpoolMan = (ThreadPoolManager) bc.getService(tpoolManRef);  
 
        // Starting the execution of the RunnableTest job
        tpoolMan.execute(new RunnableTest(), "test"); 
      } else {
        throw new BundleException("No service reference is available!");
      }
    }   

    public void stop(BundleContext bc) {
      bc.ungetService(tpoolManRef);
    }
  }        

Accessing Runnable Instances

If the thread that an entity operates in is taken from the pool of the Thread Pool Manager, the entity can retrieve the associated Runnable job and make use of some additional functionality, offered by the job. Each thread in the thread pool is a com.prosyst.util.threadpool.ThreadContext instance whose getRunnable method returns the job presently attached to this thread.

                         . . .
    Thread current = Thread.currentThread();
    if (current instanceof ThreadContext) {
      Runnable runnable = ((ThreadContext) current).getRunnable();
    }
                         . . .

System Properties

The Thread Pool Manager's operation can be configured through the following system properties:

Note: If minThreads is set to 0, then the Thread Pool Manager will not act as a thread pooling service. When a job comes for execution, it simply creates a thread, passes the job to it, and starts the thread. The manager does not keep a fixed number of threads, or return the threads finished their job back in the pool. This feature is added to avoid duplication of the thread pooling functionality in case it is supported by the current JVM.

Note: The framework administrator should precisely estimate the number of threads that are necessary to bundles using the Thread Pool Manager service.

Timer Service

Overview

Many operations are executed at a defined timeout, e.g. sessions are invalidated when users are inactive for a long time, information blocks are persistently stored after they have been kept in memory for some time, and so on. To handle such cases, a bundle should manage its own thread that should take care to inform of time period expiration, or to realize a scenario based on the wait/notify model.

The Timer service saves bundles repeated implementation of the timer model. The service provides an asynchronous mechanism for time notification to each of its subscribers using the pooling mechanisms of the Thread Pool Manager.

Notification Types

There are two types of notifications offered by the Timer service:

• Periodical - An event is asynchronously fired on equal amounts of time.
• One-shot - An event is fired only once.

The Timer service schedules events in a separate thread. When the time to notify a timer listener comes, the Timer service gets a thread from the pool and performs the notification.

No-Delay Timers

As Timer service uses the Thread Pool Manager for execution of timer notifications, it can allocate no more than 30 percent of the thread pool's capacity. This suggests that some notifications may be done later than requested. Mostly, it is considered that it is important to be notified in some time and accuracy is not an issue. But there are cases when it is critical to be notified on time. To answer the needs of critical notifications, the Timer service allows bundles to specify that notification must be as close to the time of the event as possible by adding timers of type no-delay. Such timers are executed in newly created threads if the thread pool does not provide enough free ones.

Registering for Notification

The Timer service interface is com.prosyst.util.timer.Timer. To register for notification, a bundle must supply a com.prosyst.util.timer.TimerListener object to the Timer by invoking Timer's addNotifyListener method. Each timer listener should specify timeout after which it will be notified with an event, notification type (with or without delay) and priority of notification thread.

Note: If, on registering a timer listener for an event, such a listener for the same event already exists in the Timer service, the old one is overwritten with the new one no matter if their notification types are the same or are different.

You can benefit from the timer reference utility, whose class is com.prosyst.util.ref.TimerRef, to register one-shot timers saving the efforts for executing common operations such as getting, listening for and releasing the Timer service.

Listing 2 shows a timer listener, which subscribes for both one-shot and periodical time events.

import org.osgi.framework.BundleActivator;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;
import com.prosyst.util.timer.Timer;
import com.prosyst.util.timer.TimerListener;

public class MyTimerListener implements BundleActivator, TimerListener{ 
  private ServiceReference refTimer = null;
  private Timer timer = null; 
  final static String SERVICE_TIMER = Timer.class.getName();
  static final int EVENT_TYPE_1 = 1;
  static final int EVENT_TYPE_2 = 2;

  public void start(BundleContext bc) throws Exception {
    try {
      refTimer = bc.getServiceReference(SERVICE_TIMER);
      if (refTimer != null) {
        timer = (Timer)bc.getService(refTimer);
        timer.addNotifyListener(this, 
                                Thread.MIN_PRIORITY, 
                                Timer.ONE_SHOT_TIMER,  
                                (long)100,
                                EVENT_TYPE_1);
        timer.addNotifyListener(this,
                                Thread.NORM_PRIORITY,
                                Timer.PERIODICAL_TIMER,
                                (long)1000,
                                EVENT_TYPE_2);
      }
    } catch (Exception e) {
      e.printStackTrace();
    } 
  }

  public void stop(BundleContext bc) throws Exception {
    if (refTimer != null) {
      timer.removeListener(this, EVENT_TYPE_1);
      timer.removeListener(this, EVENT_TYPE_2);
      bc.ungetService(refTimer);
    }
  }

  public void timer(int event) { 
    int type2Counter = 0;
    if(event == EVENT_TYPE_1){
      System.out.println("Event of type 1 received!");
    } else if( event == EVENT_TYPE_2) {
      if(type2Counter < 10) {
        System.out.println("Event of type 2 received!");
        type2Counter++;
      }
      timer.removeListener(this, EVENT_TYPE_2);
    }
  }
}       

System Properties

Use the mbs.timer.priority system property to specify the priority of the thread scheduling timer events. By default, the thread's priority is 5 (Thread.NORM_PRIORITY).

References

• Thread Pool Manager API Documentation.
• Timer Service API Documentation.