Use a POSIX semaphore for synchronising terminal access during indexing

Nice side-effect: indexamajig -j and --int-diag can now be used together

4 files changed, 57 insertions, 39 deletions

diff --git a/src/im-sandbox.c b/src/im-sandbox.c
index ba709f6f..b800f5b4 100644
--- a/src/im-sandbox.c
+++ b/src/im-sandbox.c
@@ -48,6 +48,8 @@
 #include <signal.h>
 #include <sys/stat.h>
 #include <assert.h>
+#include <sys/mman.h>
+#include <semaphore.h>
 
 #ifdef HAVE_CLOCK_GETTIME
 #include <time.h>
@@ -85,6 +87,12 @@ struct sb_reader
 };
 
 
+struct sb_shm
+{
+	sem_t term_sem;
+};
+
+
 struct sandbox
 {
 	pthread_mutex_t lock;
@@ -96,7 +104,6 @@ struct sandbox
 	int n_hadcrystals_last_stats;
 	int n_crystals_last_stats;
 	int t_last_stats;
-	int suspend_stats;
 
 	struct index_args *iargs;
 
@@ -110,6 +117,8 @@ struct sandbox
 	struct filename_plus_event **last_filename;
 	int serial;
 
+	struct sb_shm *shared;
+
 	char *tmpdir;
 
 	struct sb_reader *reader;
@@ -359,7 +368,7 @@ static int read_fpe_data(struct buffer_data *bd)
 
 static void run_work(const struct index_args *iargs,
                      int filename_pipe, int results_pipe, Stream *st,
-                     int cookie, const char *tmpdir)
+                     int cookie, const char *tmpdir, sem_t *term_sem)
 {
 	FILE *fh;
 	int allDone = 0;
@@ -494,7 +503,7 @@ static void run_work(const struct index_args *iargs,
 
 			pargs.n_crystals = 0;
 			process_image(iargs, &pargs, st, cookie, tmpdir,
-			              results_pipe, ser);
+			              results_pipe, ser, term_sem);
 
 			/* Request another image */
 			c = sprintf(buf, "%i\n", pargs.n_crystals);
@@ -815,11 +824,12 @@ static void start_worker_process(struct sandbox *sb, int slot)
 
 		st = open_stream_fd_for_write(stream_pipe[1]);
 		run_work(sb->iargs, filename_pipe[0], result_pipe[1],
-		         st, slot, tmp);
+		         st, slot, tmp, &sb->shared->term_sem);
 		close_stream(st);
 
 		//close(filename_pipe[0]);
 		close(result_pipe[1]);
+		munmap(sb->shared, sizeof(struct sb_shm));
 
 		free(sb);
 
@@ -894,6 +904,25 @@ static void handle_zombie(struct sandbox *sb)
 }
 
 
+static int setup_shm(struct sandbox *sb)
+{
+	sb->shared = mmap(NULL, sizeof(struct sb_shm), PROT_READ | PROT_WRITE,
+	                  MAP_SHARED | MAP_ANONYMOUS, -1, 0);
+
+	if ( sb->shared == MAP_FAILED ) {
+		ERROR("SHM setup failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	if ( sem_init(&sb->shared->term_sem, 1, 1) ) {
+		ERROR("Terminal semaphore setup failed: %s\n", strerror(errno));
+		return 1;
+	}
+
+	return 0;
+}
+
+
 void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
                     int config_basename, FILE *fh,
                     Stream *stream, const char *tempdir)
@@ -930,7 +959,6 @@ void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
 	sb->n_hadcrystals_last_stats = 0;
 	sb->n_crystals_last_stats = 0;
 	sb->t_last_stats = get_monotonic_seconds();
-	sb->suspend_stats = 0;
 	sb->n_proc = n_proc;
 	sb->iargs = iargs;
 	sb->serial = 1;
@@ -939,6 +967,12 @@ void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
 	sb->reader->fhs = NULL;
 	sb->reader->stream = stream;
 
+	if ( setup_shm(sb) ) {
+		ERROR("Failed to set up SHM.\n");
+		free(sb);
+		return;
+	}
+
 	sb->stream_pipe_write = calloc(n_proc, sizeof(int));
 	if ( sb->stream_pipe_write == NULL ) {
 		ERROR("Couldn't allocate memory for pipes.\n");
@@ -1106,34 +1140,20 @@ void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
 
 			chomp(results);
 
-			if ( strcmp(results, "SUSPEND") == 0 ) {
-				sb->suspend_stats++;
-				continue;  /* Do not send next filename */
-			} else if ( strcmp(results, "RELEASE") == 0 ) {
-				if ( sb->suspend_stats > 0 ) {
-					sb->suspend_stats--;
-				} else {
-					ERROR("RELEASE before SUSPEND.\n");
+			strtol(results, &eptr, 10);
+			if ( eptr == results ) {
+				if ( strlen(results) > 0 ) {
+					ERROR("Invalid result '%s'\n",
+					      results);
 				}
-				continue;  /* Do not send next filename */
 			} else {
 
-				strtol(results, &eptr, 10);
-				if ( eptr == results ) {
-					if ( strlen(results) > 0 ) {
-						ERROR("Invalid result '%s'\n",
-						      results);
-					}
-				} else {
-
-					int nc = atoi(results);
-					sb->n_crystals += nc;
-					if ( nc > 0 ) {
-						sb->n_hadcrystals++;
-					}
-					sb->n_processed++;
-
+				int nc = atoi(results);
+				sb->n_crystals += nc;
+				if ( nc > 0 ) {
+					sb->n_hadcrystals++;
 				}
+				sb->n_processed++;
 
 			}
 
@@ -1210,8 +1230,8 @@ void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
 		/* Update progress */
 		lock_sandbox(sb);
 		tNow = get_monotonic_seconds();
-		if ( !sb->suspend_stats
-		    && (tNow >= sb->t_last_stats+STATS_EVERY_N_SECONDS) )
+		r = sem_trywait(&sb->shared->term_sem);
+		if ((r==0) && (tNow >= sb->t_last_stats+STATS_EVERY_N_SECONDS))
 		{
 
 			STATUS("%4i indexable out of %4i processed (%4.1f%%), "
@@ -1228,7 +1248,9 @@ void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
 			sb->n_crystals_last_stats = sb->n_crystals;
 			sb->t_last_stats = tNow;
 
+
 		}
+		if ( r == 0 ) sem_post(&sb->shared->term_sem);
 		unlock_sandbox(sb);
 
 		allDone = 1;
@@ -1264,6 +1286,7 @@ void create_sandbox(struct index_args *iargs, int n_proc, char *prefix,
 	free(sb->result_fhs);
 	free(sb->pids);
 	free(sb->tmpdir);
+	munmap(sb->shared, sizeof(struct sb_shm));
 
 	pthread_mutex_destroy(&sb->lock);
 
diff --git a/src/indexamajig.c b/src/indexamajig.c
index eb619a87..c7e4a270 100644
--- a/src/indexamajig.c
+++ b/src/indexamajig.c
@@ -695,11 +695,6 @@ int main(int argc, char *argv[])
 
 		free(int_diag);
 
-		if ( (n_proc > 1) && (iargs.int_diag != INTDIAG_NONE) ) {
-			n_proc = 1;
-			STATUS("Ignored \"-j\" because you used --int-diag.\n");
-		}
-
 	}
 
 	st = open_stream_for_write_2(outfile, geom_filename, argc, argv);
diff --git a/src/process_image.c b/src/process_image.c
index 1a5a3477..5ead2c61 100644
--- a/src/process_image.c
+++ b/src/process_image.c
@@ -78,7 +78,7 @@ static void try_refine_autoR(struct image *image, Crystal *cr)
 
 void process_image(const struct index_args *iargs, struct pattern_args *pargs,
                    Stream *st, int cookie, const char *tmpdir, int results_pipe,
-                   int serial)
+                   int serial, sem_t *term_sem)
 {
 	float *data_for_measurement;
 	size_t data_size;
@@ -260,7 +260,7 @@ void process_image(const struct index_args *iargs, struct pattern_args *pargs,
 	                iargs->ir_inn, iargs->ir_mid, iargs->ir_out,
 	                iargs->int_diag, iargs->int_diag_h,
 	                iargs->int_diag_k, iargs->int_diag_l,
-	                results_pipe);
+	                term_sem);
 
 	ret = write_chunk(st, &image, hdfile,
 	                  iargs->stream_peaks, iargs->stream_refls,
diff --git a/src/process_image.h b/src/process_image.h
index de364772..6e44c173 100644
--- a/src/process_image.h
+++ b/src/process_image.h
@@ -104,7 +104,7 @@ struct pattern_args
 extern void process_image(const struct index_args *iargs,
                           struct pattern_args *pargs, Stream *st,
                           int cookie, const char *tmpdir, int results_pipe,
-                          int serial);
+                          int serial, sem_t *term_sem);
 
 
 #endif	/* PROCESS_IMAGEs_H */