Bug 667: fix job blocking handling, especially dealing with batches

         keeps low priority jobs from blocking higher-priority jobs
         removes blocking when a job to the destination is not in process
         keeps batching from ignoring MaxConcurrentCalls on retries

diff --git a/CHANGES b/CHANGES
index b9edb0a969a08a75d4ceca1783fdab7710657b69..58d99ece29c4d82c3701a699e5f811249d38e32f 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -2,6 +2,8 @@
 
 Changelog for HylaFAX 4.2.2
 
+* fix job blocking (concurrent call) handling, especially
+  as dealing with batches (13 Jul 2003)
 * fix job modifications on batched, running jobs (13 Jul 2005)
 * throttle faxq from running the scheduler more than once per
   second in order to keep faxq from eating up CPU (11 Jul 2005)

diff --git a/faxd/DestInfo.c++ b/faxd/DestInfo.c++
index b7a199807af6feb134835b840d4b20efd9946abb..ee395fa3f0545398d993301dfc882b03dd5120e1 100644 (file)
--- a/faxd/DestInfo.c++
+++ b/faxd/DestInfo.c++
@@ -30,6 +30,7 @@
 DestInfo::DestInfo()
 {
     activeCount = 0;
+    callCount = 0;
     blockedCount = 0;
     running = NULL;
 }
@@ -39,6 +40,7 @@ DestInfo::DestInfo(const DestInfo& other)
     , info(other.info)
 {
     activeCount = other.activeCount;
+    callCount = other.callCount;
     blockedCount = other.blockedCount;
     running = other.running;
 }

diff --git a/faxd/DestInfo.h b/faxd/DestInfo.h
index 1e2cbcef9c2ecb53957d1e7f40a19fa4215bc1e0..f6fe4779d088ef8a8819042d263e4e68b3596bd7 100644 (file)
--- a/faxd/DestInfo.h
+++ b/faxd/DestInfo.h
@@ -51,6 +51,7 @@ class DestInfo : public QLink {
 private:
     u_short            activeCount;    // count of active jobs to destination
     u_short            blockedCount;   // count of blocked jobs
+    u_short            callCount;      // count of active calls to destination
     FaxMachineInfo     info;           // remote machine capabilities and such
     Job*               running;        // jobs to dest being processed
 public:
@@ -61,6 +62,9 @@ public:
     u_int getActive() const;           // return count of active jobs
     u_int getCount() const;            // return count of active+blocked jobs
     bool isEmpty() const;              // true if any jobs referenced
+    u_int getCalls() const;            // return count of active calls
+    void call();                       // initiate call to destination
+    void hangup();                     // terminate call to destination
 
     bool isActive(Job&) const; // true if job is considered active
     bool supportsBatching();           // if remote supports batch protocol
@@ -78,6 +82,9 @@ inline u_int DestInfo::getActive() const      { return activeCount; }
 inline u_int DestInfo::getCount() const
     { return activeCount + blockedCount; }
 inline bool DestInfo::isEmpty() const          { return getCount() == 0; }
+inline u_int DestInfo::getCalls() const                { return callCount; }
+inline void DestInfo::call()                   { callCount++; }
+inline void DestInfo::hangup()                 { callCount--; }
 
 fxDECLARE_StrKeyDictionary(DestInfoDict, DestInfo)
 #endif /* _DestInfo_ */

diff --git a/faxd/faxQueueApp.c++ b/faxd/faxQueueApp.c++
index 216977caf886e134729a0db36474de3a96b4a355..5877d59415e272c7d96c6580d3344c71c5549d6c 100644 (file)
--- a/faxd/faxQueueApp.c++
+++ b/faxd/faxQueueApp.c++
@@ -412,6 +412,15 @@ faxQueueApp::prepareJobDone(Job& job, int status)
            }
            if (processnext) processJob(*targetjob, targetjob->breq, destJobs[targetjob->dest], destCtrls[targetjob->dest]);
            else if (startsendjob) sendJobStart(*targetjob->bfirst(), targetjob->bfirst()->breq, destCtrls[targetjob->dest]);
+           else {
+               /*
+                * This destination was marked as called, but all jobs to this
+                * destination failed preparation, so we must undo the call marking.
+                */
+               DestInfo& di = destJobs[job.dest];
+               di.hangup();
+               unblockDestJobs(job, di);       // release any blocked jobs
+           }
        }
     }
 }
@@ -1366,6 +1375,10 @@ faxQueueApp::sendJobDone(Job& job, int status)
     int cstatus = status;
     bool seenabort = false;
 
+    DestInfo& di = destJobs[job.dest];
+    di.hangup();
+    unblockDestJobs(job, di);
+
     for (cjob = &job; cjob != NULL; cjob = njob) {
        njob = cjob->bnext;
        req = readRequest(*cjob);               // reread the qfile
@@ -2073,35 +2086,41 @@ faxQueueApp::runJob(Job& job)
  * for this job.  If the job is active and blocking other
  * jobs, we need to unblock...
  */
-#define        isOKToStartJobs(di, dci, n) \
-    (di.getActive()+n <= dci.getMaxConcurrentCalls())
+#define        isOKToCall(di, dci, n) \
+    (di.getCalls()+n <= dci.getMaxConcurrentCalls())
 
 void
-faxQueueApp::removeDestInfoJob (Job& job)
+faxQueueApp::unblockDestJobs(Job& job, DestInfo& di)
+{
+    /*
+     * Check if there are blocked jobs waiting to run
+     * and that there is now room to run one.  If so,
+     * take jobs off the blocked queue and make them
+     * ready for processing.
+     */
+    Job* jb;
+    const DestControlInfo& dci = destCtrls[job.dest];
+    u_int n = 1;
+    while (isOKToCall(di, dci, n) && (jb = di.nextBlocked())) {
+       setReadyToRun(*jb);
+       if (!di.supportsBatching()) n++;
+       FaxRequest* req = readRequest(*jb);
+       if (req) {
+           req->notice = "";
+           updateRequest(*req, *jb);
+           delete req;
+       }
+    }
+}
+
+void
+faxQueueApp::removeDestInfoJob(Job& job)
 {
     DestInfo& di = destJobs[job.dest];
     di.done(job);                      // remove from active destination list
     di.updateConfig();                 // update file if something changed
     if (!di.isEmpty()) {
-       /*
-        * Check if there are blocked jobs waiting to run
-        * and that there is now room to run one.  If so,
-        * take jobs off the blocked queue and make them
-        * ready for processing.
-        */
-       Job* jb;
-       const DestControlInfo& dci = destCtrls[job.dest];
-       u_int n = 1;
-       while ((isOKToStartJobs(di, dci, n) || di.supportsBatching()) && (jb = di.nextBlocked())) {
-           setReadyToRun(*jb);
-           n++;
-           FaxRequest* req = readRequest(*jb);
-           if (req) {
-               req->notice = "";
-               updateRequest(*req, *jb);
-               delete req;
-           }
-        }
+       unblockDestJobs(job, di);
     } else {
        /*
         * This is the last job to the destination; purge
@@ -2225,7 +2244,7 @@ faxQueueApp::runScheduler()
                }
                time_t now = Sys::now();
                time_t tts;
-               if (!di.isActive(job) && !isOKToStartJobs(di, dci, 1)) {
+               if (!isOKToCall(di, dci, 1)) {
                    /*
                     * This job would exceed the max number of concurrent
                     * calls that may be made to this destination.  Put it
@@ -2266,6 +2285,7 @@ faxQueueApp::runScheduler()
                         * allowed on this modem which are not of a lesser priority than
                         * jobs to other destinations.
                         */
+                       unblockDestJobs(job, di);
 
                        JobIter joblist = iter;
                        Job* bjob = &job;       // Last batched Job
@@ -2312,6 +2332,7 @@ faxQueueApp::runScheduler()
                        bjob->bnext = NULL;
                    } else
                        job.bnext = NULL;
+                   di.call();                  // mark as called to correctly block other jobs
                    processJob(job, req, di, dci);
                } else                          // leave job on run queue
                    delete req;

diff --git a/faxd/faxQueueApp.h b/faxd/faxQueueApp.h
index a2f7b4d09c9e121fcf650560e3554c0353acf402..a21ba7da73beefe2da050086ab151195eb29c231 100644 (file)
--- a/faxd/faxQueueApp.h
+++ b/faxd/faxQueueApp.h
@@ -220,7 +220,8 @@ private:
     void       timeoutJob(Job& job, FaxRequest&);
     void       runJob(Job& job);
 
-    void       removeDestInfoJob (Job& job);
+    void       removeDestInfoJob(Job& job);
+    void        unblockDestJobs(Job& job, DestInfo& di);
 
     void       runScheduler();
     void       pokeScheduler();