From 509337b825fc73e026ff4baba28df033e11789b9 Mon Sep 17 00:00:00 2001
From: Endi Sukma Dewata <edewata@redhat.com>
Date: Tue, 6 Mar 2012 20:06:44 -0600
Subject: [PATCH] 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
---
 .../ca/src/com/netscape/ca/CRLIssuingPoint.java    |   23 +--
 .../src/com/netscape/certsrv/dbs/IDBSSession.java  |    2 +
 .../certsrv/dbs/certdb/ICertificateRepository.java |   12 +
 .../cmscore/dbs/CertificateRepository.java         |  283 +++++++++++---------
 .../src/com/netscape/cmscore/dbs/DBSSession.java   |   15 +
 .../cmscore/dbs/DBSSessionDefaultStub.java         |    3 +
 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 1c89a29c8694847ee03ecb1f2964ad11578f04ff..3de6128b6765e44590838915c4f32dc029bc0aeb 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 ec019423ccbe52e4cb854c161e3b776e94268be4..c186d1145fda1b19ec269462a0db0ad8034dc233 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 2e9e00f60e290b24c378a253bf0b293afaaf5f4a..a8505c2a207f61b42b455af7c09e7630b2a7379b 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 c563a3a4504f8dfae892df0ebf32ac6c66afdb08..ae77330fc5f6c4c0e365b19d22dbb97777dda52c 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 {
 
-    CertStatusUpdateThread(CertificateRepository cr, IRepository rr, String name) {
-        super(name);
-        CMS.debug("new CertStatusUpdateThread");
-        //setName(name);
+    CertificateRepository repository;
+    int interval;
 
-        _cr = cr;
-        _rr = rr;
+    ScheduledExecutorService executorService;
+
+    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");
+        repository.updateCertStatus();
+    }
 
-        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");
-                }
-
-            } 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 db2f3c4e70d8d3912108a68e15b6d2ca43fc02d9..7201f61e7f72f4fac9bb3fd53b2124fbb35af060 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 c00be1f4537ba3d13cdcdf6e47e656634c810d64..09a2e1498addd496b8171c1979b2346e2539fb8c 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;
-- 
1.7.7.1.msysgit.0

