Replaced daemon threads with executor service.

The certificate status update and retrieving modifications tasks
have been modified to use the executor service. Unlike daemon
threads, the service will allow existing task to exit gracefully
before shutting down. An abandon operation is used terminate the
persistent search used for retrieving modifications. Some methods
have been moved to CertificateRepository class to simplify
synchronizations.

Ticket #73

6 files changed, 193 insertions, 145 deletions

diff --git a/pki/base/ca/src/com/netscape/ca/CRLIssuingPoint.java b/pki/base/ca/src/com/netscape/ca/CRLIssuingPoint.java
index 1c89a29c8..3de6128b6 100644
--- a/pki/base/ca/src/com/netscape/ca/CRLIssuingPoint.java
+++ b/pki/base/ca/src/com/netscape/ca/CRLIssuingPoint.java
@@ -62,8 +62,6 @@ import com.netscape.certsrv.common.NameValuePair;
 import com.netscape.certsrv.common.NameValuePairs;
 import com.netscape.certsrv.dbs.EDBNotAvailException;
 import com.netscape.certsrv.dbs.IElementProcessor;
-import com.netscape.certsrv.dbs.certdb.ICertRecord;
-import com.netscape.certsrv.dbs.certdb.ICertRecordList;
 import com.netscape.certsrv.dbs.certdb.ICertificateRepository;
 import com.netscape.certsrv.dbs.certdb.IRevocationInfo;
 import com.netscape.certsrv.dbs.crldb.ICRLIssuingPointRecord;
@@ -1863,26 +1861,7 @@ public class CRLIssuingPoint implements ICRLIssuingPoint, Runnable {
      */
     public void processRevokedCerts(IElementProcessor p)
             throws EBaseException {
-        CertRecProcessor cp = (CertRecProcessor) p;
-        String filter = getFilter();
-
-        // NOTE: dangerous cast. 
-        // correct way would be to modify interface and add
-        // accessor but we don't want to touch the interface
-        CertificateRepository cr = (CertificateRepository) mCertRepository;
-
-        synchronized (cr.mCertStatusUpdateThread) {
-            CMS.debug("Starting processRevokedCerts (entered lock)");
-            ICertRecordList list = mCertRepository.findCertRecordsInList(filter,
-                    new String[] { ICertRecord.ATTR_ID, ICertRecord.ATTR_REVO_INFO, "objectclass" },
-                    "serialno",
-                    mPageSize);
-
-            int totalSize = list.getSize();
-
-            list.processCertRecords(0, totalSize - 1, cp);
-            CMS.debug("processRevokedCerts done");
-        }
+        mCertRepository.processRevokedCerts(p, getFilter(), mPageSize);
     }
 
     /**
diff --git a/pki/base/common/src/com/netscape/certsrv/dbs/IDBSSession.java b/pki/base/common/src/com/netscape/certsrv/dbs/IDBSSession.java
index ec019423c..c186d1145 100644
--- a/pki/base/common/src/com/netscape/certsrv/dbs/IDBSSession.java
+++ b/pki/base/common/src/com/netscape/certsrv/dbs/IDBSSession.java
@@ -177,6 +177,8 @@ public interface IDBSSession {
     public LDAPSearchResults persistentSearch(String base, String filter,
             String attrs[]) throws EBaseException;
 
+    public void abandon(LDAPSearchResults results) throws EBaseException;
+
     /**
      * Retrieves a list of objects.
      * 
diff --git a/pki/base/common/src/com/netscape/certsrv/dbs/certdb/ICertificateRepository.java b/pki/base/common/src/com/netscape/certsrv/dbs/certdb/ICertificateRepository.java
index 2e9e00f60..a8505c2a2 100644
--- a/pki/base/common/src/com/netscape/certsrv/dbs/certdb/ICertificateRepository.java
+++ b/pki/base/common/src/com/netscape/certsrv/dbs/certdb/ICertificateRepository.java
@@ -28,6 +28,7 @@ import netscape.security.x509.X509CertImpl;
 
 import com.netscape.certsrv.base.EBaseException;
 import com.netscape.certsrv.base.MetaInfo;
+import com.netscape.certsrv.dbs.IElementProcessor;
 import com.netscape.certsrv.dbs.ModificationSet;
 import com.netscape.certsrv.dbs.repository.IRepository;
 import com.netscape.cmscore.dbs.CertificateRepository.RenewableCertificateCollection;
@@ -512,5 +513,16 @@ public interface ICertificateRepository extends IRepository {
      */
     public void removeCertRecords(BigInteger beginS, BigInteger endS) throws EBaseException;
 
+    /**
+     * Builds a list of revoked certificates to put them into CRL.
+     * Calls certificate record processor to get necessary data
+     * from certificate records.
+     * This also regenerates CRL cache.
+     *
+     * @param cp certificate record processor
+     * @exception EBaseException if an error occurred in the database.
+     */
+    public void processRevokedCerts(IElementProcessor cp, String filter, int pageSize) throws EBaseException;
+
     public void shutdown();
 }
diff --git a/pki/base/common/src/com/netscape/cmscore/dbs/CertificateRepository.java b/pki/base/common/src/com/netscape/cmscore/dbs/CertificateRepository.java
index c563a3a45..ae77330fc 100644
--- a/pki/base/common/src/com/netscape/cmscore/dbs/CertificateRepository.java
+++ b/pki/base/common/src/com/netscape/cmscore/dbs/CertificateRepository.java
@@ -25,10 +25,13 @@ import java.util.Date;
 import java.util.Enumeration;
 import java.util.Hashtable;
 import java.util.Vector;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 import netscape.ldap.LDAPAttributeSet;
 import netscape.ldap.LDAPEntry;
-import netscape.ldap.LDAPException;
 import netscape.ldap.LDAPSearchResults;
 import netscape.security.x509.CertificateValidity;
 import netscape.security.x509.RevokedCertImpl;
@@ -46,6 +49,7 @@ import com.netscape.certsrv.dbs.IDBSSession;
 import com.netscape.certsrv.dbs.IDBSearchResults;
 import com.netscape.certsrv.dbs.IDBSubsystem;
 import com.netscape.certsrv.dbs.IDBVirtualList;
+import com.netscape.certsrv.dbs.IElementProcessor;
 import com.netscape.certsrv.dbs.Modification;
 import com.netscape.certsrv.dbs.ModificationSet;
 import com.netscape.certsrv.dbs.certdb.ICertRecord;
@@ -81,6 +85,11 @@ public class CertificateRepository extends Repository
     private int mTransitMaxRecords = 1000000;
     private int mTransitRecordPageSize = 200;
 
+    CertStatusUpdateTask certStatusUpdateTask;
+    RetrieveModificationsTask retrieveModificationsTask;
+
+    IRepository requestRepository;
+
     /**
      * Constructs a certificate repository.
      */
@@ -216,76 +225,93 @@ public class CertificateRepository extends Repository
      * 0 - disable
      * >0 - enable
      */
-    public CertStatusUpdateThread mCertStatusUpdateThread = null;
-    public RetrieveModificationsThread mRetrieveModificationsThread = null;
+    public void setCertStatusUpdateInterval(IRepository requestRepository, int interval, boolean listenToCloneModifications) {
 
-    public void setCertStatusUpdateInterval(IRepository requestRepo, int interval, boolean listenToCloneModifications) {
         CMS.debug("In setCertStatusUpdateInterval " + interval);
+
+        this.requestRepository = requestRepository;
+
         if (interval == 0) {
-            CMS.debug("In setCertStatusUpdateInterval interval = 0" + interval);
-            if (mCertStatusUpdateThread != null) {
-                mCertStatusUpdateThread.stop();
+            CMS.debug("In setCertStatusUpdateInterval interval = 0");
+            if (certStatusUpdateTask != null) {
+                certStatusUpdateTask.stop();
             }
-            if (mRetrieveModificationsThread != null) {
-                mRetrieveModificationsThread.stop();
+            if (retrieveModificationsTask != null) {
+                retrieveModificationsTask.stop();
             }
             return;
         }
 
-        CMS.debug("In setCertStatusUpdateInterval  listenToCloneModifications=" + listenToCloneModifications +
-                  "  mRetrieveModificationsThread=" + mRetrieveModificationsThread);
-        if (listenToCloneModifications && mRetrieveModificationsThread == null) {
-            CMS.debug("In setCertStatusUpdateInterval about to create RetrieveModificationsThread");
-            mRetrieveModificationsThread = new RetrieveModificationsThread(this, "RetrieveModificationsThread");
-            LDAPSearchResults mResults = null;
+        CMS.debug("In setCertStatusUpdateInterval  listenToCloneModifications=" + listenToCloneModifications);
+
+        if (listenToCloneModifications) {
+            CMS.debug("In setCertStatusUpdateInterval listening to modifications");
             try {
-                mResults = startSearchForModifiedCertificateRecords();
-            } catch (Exception e) {
-                mResults = null;
-            }
-            if (mResults != null) {
-                mRetrieveModificationsThread.setResults(mResults);
-                mRetrieveModificationsThread.setDaemon(true);
-                mRetrieveModificationsThread.start();
+                retrieveModificationsTask = new RetrieveModificationsTask(this);
+                retrieveModificationsTask.start();
+            } catch (EBaseException e) {
+                retrieveModificationsTask = null;
+                e.printStackTrace();
             }
         }
 
-        CMS.debug("In setCertStatusUpdateInterval  mCertStatusUpdateThread " + mCertStatusUpdateThread);
-        if (mCertStatusUpdateThread == null) {
-            CMS.debug("In setCertStatusUpdateInterval about to create CertStatusUpdateThread ");
-            mCertStatusUpdateThread = new CertStatusUpdateThread(this, requestRepo, "CertStatusUpdateThread");
-            mCertStatusUpdateThread.setInterval(interval);
-            mCertStatusUpdateThread.setDaemon(true);
-            mCertStatusUpdateThread.start();
-        } else {
-            CMS.debug("In setCertStatusUpdateInterval it thinks the thread is up already ");
-            mCertStatusUpdateThread.setInterval(interval);
-            // dont do anything if we have a thread running already
-        }
+        CMS.debug("In setCertStatusUpdateInterval scheduling cert status update every " + interval + " seconds.");
+        certStatusUpdateTask = new CertStatusUpdateTask(this, interval);
+        certStatusUpdateTask.start();
     }
 
     /**
-     * Blocking method.
+     * This method blocks when another thread (such as the CRL Update) is running
      */
-    public void updateCertStatus() throws EBaseException {
+    public synchronized void updateCertStatus() {
 
         CMS.debug("In updateCertStatus()");
 
-        CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
-                CMS.getLogMessage("CMSCORE_DBS_START_VALID_SEARCH"));
-        transitInvalidCertificates();
-        CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
-                CMS.getLogMessage("CMSCORE_DBS_FINISH_VALID_SEARCH"));
-        CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
-                CMS.getLogMessage("CMSCORE_DBS_START_EXPIRED_SEARCH"));
-        transitValidCertificates();
-        CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
-                CMS.getLogMessage("CMSCORE_DBS_FINISH_EXPIRED_SEARCH"));
-        CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
-                CMS.getLogMessage("CMSCORE_DBS_START_REVOKED_EXPIRED_SEARCH"));
-        transitRevokedExpiredCertificates();
-        CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
-                CMS.getLogMessage("CMSCORE_DBS_FINISH_REVOKED_EXPIRED_SEARCH"));
+        try {
+            CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
+                    CMS.getLogMessage("CMSCORE_DBS_START_VALID_SEARCH"));
+            transitInvalidCertificates();
+            CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
+                    CMS.getLogMessage("CMSCORE_DBS_FINISH_VALID_SEARCH"));
+
+            CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
+                    CMS.getLogMessage("CMSCORE_DBS_START_EXPIRED_SEARCH"));
+            transitValidCertificates();
+            CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
+                    CMS.getLogMessage("CMSCORE_DBS_FINISH_EXPIRED_SEARCH"));
+
+            CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
+                    CMS.getLogMessage("CMSCORE_DBS_START_REVOKED_EXPIRED_SEARCH"));
+            transitRevokedExpiredCertificates();
+            CMS.getLogger().log(ILogger.EV_SYSTEM, ILogger.S_OTHER,
+                    CMS.getLogMessage("CMSCORE_DBS_FINISH_REVOKED_EXPIRED_SEARCH"));
+
+            CMS.debug("Starting cert checkRanges");
+            checkRanges();
+            CMS.debug("cert checkRanges done");
+
+            CMS.debug("Starting request checkRanges");
+            requestRepository.checkRanges();
+            CMS.debug("request checkRanges done");
+
+        } catch (Exception e) {
+            CMS.debug("updateCertStatus done: " + e.toString());
+        }
+    }
+
+    public synchronized void processRevokedCerts(IElementProcessor p, String filter, int pageSize)
+            throws EBaseException {
+
+        CMS.debug("Starting processRevokedCerts (entered lock)");
+        ICertRecordList list = findCertRecordsInList(filter,
+                new String[] { ICertRecord.ATTR_ID, ICertRecord.ATTR_REVO_INFO, "objectclass" },
+                "serialno",
+                pageSize);
+
+        int totalSize = list.getSize();
+
+        list.processCertRecords(0, totalSize - 1, p);
+        CMS.debug("processRevokedCerts done");
     }
 
     /**
@@ -1789,23 +1815,19 @@ public class CertificateRepository extends Repository
         return e;
     }
 
-    private LDAPSearchResults startSearchForModifiedCertificateRecords()
-            throws EBaseException {
-        CMS.debug("startSearchForModifiedCertificateRecords");
-        LDAPSearchResults r = null;
-        IDBSSession s = mDBService.createSession();
+    LDAPSearchResults searchForModifiedCertificateRecords(IDBSSession session) {
+        CMS.debug("searchForModifiedCertificateRecords");
+        LDAPSearchResults results = null;
 
         String filter = "(" + CertRecord.ATTR_CERT_STATUS + "=*)";
         try {
-            r = s.persistentSearch(getDN(), filter, null);
-            CMS.debug("startSearchForModifiedCertificateRecords  persistentSearch started");
+            results = session.persistentSearch(getDN(), filter, null);
+
         } catch (Exception e) {
             CMS.debug("startSearchForModifiedCertificateRecords  persistentSearch Exception=" + e);
-            r = null;
-            if (s != null)
-                s.close();
         }
-        return r;
+
+        return results;
     }
 
     public void getModifications(LDAPEntry entry) {
@@ -1905,91 +1927,106 @@ public class CertificateRepository extends Repository
     }
 
     public void shutdown() {
+        if (certStatusUpdateTask != null) {
+            certStatusUpdateTask.stop();
+        }
+
+        if (retrieveModificationsTask != null) {
+            retrieveModificationsTask.stop();
+        }
     }
 }
 
-class CertStatusUpdateThread extends Thread {
-    CertificateRepository _cr = null;
-    IRepository _rr = null;
-    int _interval;
+class CertStatusUpdateTask implements Runnable {
+
+    CertificateRepository repository;
+    int interval;
 
-    CertStatusUpdateThread(CertificateRepository cr, IRepository rr, String name) {
-        super(name);
-        CMS.debug("new CertStatusUpdateThread");
-        //setName(name);
+    ScheduledExecutorService executorService;
 
-        _cr = cr;
-        _rr = rr;
+    public CertStatusUpdateTask(CertificateRepository repository, int interval) {
+        this.repository = repository;
+        this.interval = interval;
     }
 
-    public void setInterval(int interval) {
-        _interval = interval;
+    public void start() {
+        // schedule task to run immediately and repeat after specified interval
+        executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+            public Thread newThread(Runnable r) {
+                return new Thread(r, "CertStatusUpdateTask");
+            }
+        });
+        executorService.scheduleWithFixedDelay(this, 0, interval, TimeUnit.SECONDS);
     }
 
     public void run() {
-        CMS.debug("Inside run method of CertStatusUpdateThread");
-
-        while (true) {
-            try {
-                // block the update while another thread
-                // (such as the CRL Update) is running
-                CMS.debug("About to start updateCertStatus");
-                synchronized (_cr.mCertStatusUpdateThread) {
-                    CMS.debug("Starting updateCertStatus (entered lock)");
-                    _cr.updateCertStatus();
-                    CMS.debug("updateCertStatus done");
-
-                    CMS.debug("Starting cert checkRanges");
-                    _cr.checkRanges();
-                    CMS.debug("cert checkRanges done");
-
-                    CMS.debug("Starting request checkRanges");
-                    _rr.checkRanges();
-                    CMS.debug("request checkRanges done");
-                }
+        repository.updateCertStatus();
+    }
 
-            } catch (Exception e) {
-                CMS.debug("updateCertStatus done: " + e.toString());
-            }
-            try {
-                sleep(_interval * 1000);
-            } catch (InterruptedException e) {
-            }
-        }
+    public void stop() {
+        // shutdown executorService without interrupting running task
+        if (executorService != null) executorService.shutdown();
     }
 }
 
-class RetrieveModificationsThread extends Thread {
-    CertificateRepository _cr = null;
-    LDAPSearchResults _results = null;
+class RetrieveModificationsTask implements Runnable {
 
-    RetrieveModificationsThread(CertificateRepository cr, String name) {
-        super(name);
-        CMS.debug("new RetrieveModificationsThread");
-        //setName(name);
+    CertificateRepository repository;
 
-        _cr = cr;
+    IDBSSession session;
+    LDAPSearchResults results;
+
+    ScheduledExecutorService executorService;
+
+    public RetrieveModificationsTask(CertificateRepository repository) {
+        this.repository = repository;
     }
 
-    public void setResults(LDAPSearchResults results) {
-        _results = results;
+    public void start() throws EBaseException {
+        // start persistent search
+        try {
+            session = repository.getDBSubsystem().createSession();
+            results = repository.searchForModifiedCertificateRecords(session);
+        } catch (EBaseException e) {
+            stop(); // avoid leaks
+            throw e;
+        }
+
+        // schedule task to run immediately and repeat without delay
+        executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+            public Thread newThread(Runnable r) {
+                return new Thread(r, "RetrieveModificationsTask");
+            }
+        });
+        executorService.scheduleWithFixedDelay(this, 0, 1, TimeUnit.MICROSECONDS);
     }
 
     public void run() {
         CMS.debug("Inside run method of RetrieveModificationsThread");
 
-        if (_results != null) {
-            try {
-                while (_results.hasMoreElements()) {
-                    LDAPEntry entry = _results.next();
-                    _cr.getModifications(entry);
-                }
-            } catch (LDAPException e) {
-                CMS.debug("LDAPException: " + e.toString());
+        try {
+            // results.hasMoreElements() will block until next element becomes available
+            // or return false if the search is abandoned or connection is closed
+            if (results.hasMoreElements()) {
+                LDAPEntry entry = results.next();
+                repository.getModifications(entry);
             }
-        } else {
-            CMS.debug("_results are null");
+        } catch (Exception e) {
+            CMS.debug("Exception: " + e.toString());
         }
         CMS.debug("Done with run method of RetrieveModificationsThread");
     }
+
+    public void stop() {
+        if (executorService != null) executorService.shutdown();
+
+        if (session != null) {
+            // closing the session doesn't actually close the connection,
+            // so the search needs to be abandoned explicitly
+            if (results != null) try { session.abandon(results); } catch (Exception e) { e.printStackTrace(); }
+
+            // close session
+            try { session.close(); } catch (Exception e) { e.printStackTrace(); }
+        }
+    }
 }
diff --git a/pki/base/common/src/com/netscape/cmscore/dbs/DBSSession.java b/pki/base/common/src/com/netscape/cmscore/dbs/DBSSession.java
index db2f3c4e7..7201f61e7 100644
--- a/pki/base/common/src/com/netscape/cmscore/dbs/DBSSession.java
+++ b/pki/base/common/src/com/netscape/cmscore/dbs/DBSSession.java
@@ -404,6 +404,21 @@ public class DBSSession implements IDBSSession {
         }
     }
 
+    public void abandon(LDAPSearchResults results) throws EBaseException {
+        try {
+            mConn.abandon(results);
+
+        } catch (LDAPException e) {
+            if (e.getLDAPResultCode() == LDAPException.UNAVAILABLE)
+                throw new EDBNotAvailException(
+                        CMS.getUserMessage("CMS_DBS_INTERNAL_DIR_UNAVAILABLE"));
+            // XXX error handling, should not raise exception if
+            // entry not found
+            throw new EDBException(CMS.getUserMessage("CMS_DBS_LDAP_OP_FAILURE",
+                        e.toString()));
+        }
+    }
+
     /**
      * Retrieves a list of objects.
      */
diff --git a/pki/base/common/test/com/netscape/cmscore/dbs/DBSSessionDefaultStub.java b/pki/base/common/test/com/netscape/cmscore/dbs/DBSSessionDefaultStub.java
index c00be1f45..09a2e1498 100644
--- a/pki/base/common/test/com/netscape/cmscore/dbs/DBSSessionDefaultStub.java
+++ b/pki/base/common/test/com/netscape/cmscore/dbs/DBSSessionDefaultStub.java
@@ -64,6 +64,9 @@ public class DBSSessionDefaultStub implements IDBSSession {
         return null;
     }
 
+    public void abandon(LDAPSearchResults results) throws EBaseException {
+    }
+
     public <T> IDBVirtualList<T> createVirtualList(String base, String filter, String attrs[], String sortKey, int pageSize)
             throws EBaseException {
         return null;