Use job scheduler when processing all spools.alpha-3.1.90

* src/job.c (job_printable): Improve job identification.
(wydawca_scanner): Don't log statistics in single process mode.
(job_queue_wait): New function.
* src/net.c (wydawca_listener): Move the call to job_init to main.
* src/process.c (scan_spool_unlocked): Enqueue the triplet.
Don't call spool_commit_triplets
* src/triplet.c (remove_triplet): Omit checks in dry-run mode.
(triplet_processor,spool_commit_triplets)
(count_collected_triplets): Remove.
* src/wydawca.c (main): Call job_queue_wait if not daemon.
* tests/inotify-unatt01.at: Update expected output.
* tests/upload-dry.at: Likewise.

1 files changed, 54 insertions, 14 deletions

diff --git a/src/job.c b/src/job.c
index 48f1509..84d3bb5 100644
--- a/src/job.c
+++ b/src/job.c
@@ -69,21 +69,23 @@ char const *
 job_printable(struct job *job)
 {
     if (!job->printable) {
+	size_t s = 0;
+
 	switch (job->type) {
 	case JOB_TRIPLET:
-	    job->printable = grecs_strdup("triplet");
+	    if (grecs_asprintf(&job->printable, &s,
+			       "triplet(%s,%s)",
+			       job->v.triplet->spool->tag,
+			       job->v.triplet->name))
+		grecs_alloc_die();
 	    break;
 
-	case JOB_SPOOL:{
-		size_t s = 0;
-		int rc;
-		rc = grecs_asprintf(&job->printable, &s,
-				    "spool(%s)",
-				    job->v.spool->tag);
-		if (rc)
-		    grecs_alloc_die();
-		break;
-	    }
+	case JOB_SPOOL:
+	    if (grecs_asprintf(&job->printable, &s,
+			       "spool(%s)",
+			       job->v.spool->tag))
+		grecs_alloc_die();
+	    break;
 
 	case JOB_ALL_SPOOLS:
 	    job->printable = grecs_strdup("all spools");
@@ -149,7 +151,8 @@ wydawca_scanner(struct job *job)
 	notify_flush(job->v.spool);
     }
     timer_stop("wydawca");
-    logstats();
+    if (!single_process)
+	logstats();
     return rc;
 }
 
@@ -329,7 +332,6 @@ print_status(struct job *job, int expect_term)
 void
 job_queue_runner(time_t min_interval)
 {
-    int status;
     struct job *job;
     time_t now = time(NULL);
 
@@ -337,6 +339,8 @@ job_queue_runner(time_t min_interval)
 	wakeup = 0;
 
 	for (;;) {
+	    int status;
+	    
 	    pid_t pid = waitpid((pid_t) - 1, &status, WNOHANG);
 	    if (pid <= 0)
 		break;
@@ -394,7 +398,43 @@ job_queue_runner(time_t min_interval)
 }
 
 void
-job_init()
+job_queue_wait(void)
+{
+    while (queue) {
+	int status;
+	struct job *job;
+	
+	pid_t pid = waitpid((pid_t) - 1, &status, 0);
+	if (pid <= 0)
+	    break;
+	for (job = queue; job; job = job->next) {
+	    if (job->state == STATE_ACTIVE && job->pid == pid) {
+		remove_triplet(job->v.triplet, 1);
+		job->v.triplet = NULL;
+		job->exit_status = status;
+		job->state = STATE_FINISHED;
+		jobcnt--;
+	    }
+	}
+
+	for (job = queue; job;) {
+	    struct job *next = job->next;
+	    if (job->state == STATE_FINISHED) {
+		print_status(job, 0);
+		job_remove(job);
+		job_destroy(job);
+	    } else if (job->state == STATE_QUEUED) {
+		job_remove(job);
+		job_destroy(job);
+	    }		
+	    job = next;
+	}
+    }
+    logstats();
+}
+
+void
+job_init(void)
 {
     signal(SIGCHLD, queue_signal);
     signal(SIGALRM, queue_signal);