For review: coroutining changes

Thomas Charles CONWAY conway at cs.mu.OZ.AU
Wed Nov 25 10:03:30 AEDT 1998


Hi

Fergus or Tyson, can you please look at these changes?

Thomas
-- 
Thomas Conway <conway at cs.mu.oz.au> )O+
To a killer whale, otters are like hairy popcorn -- Paul Dayton

Improvements to coroutining support. These changes allow us to do
provide io primatives that cause the Mercury context to suspend rather
than causing the engine to block.

compiler/pragma_c_gen.m:
	Include the predicate name in the calls to MR_OBTAIN_GLOBAL_C_LOCK
	and MR_RELEASE_GLOBAL_C_LOCK for improved debugging.

runtime/mercury_thread.h:
	Change the global lock macros to include the message generated
	by the changes to pragma_c_gen.

library/char.m:
library/std_util.m:
	include `thread_safe' in the flags for a couple of pragma c
	definitions that seem to have missed out.

runtime/mercury_context.{c,h}:
	Add a list of "pending" contexts that are blocked on a
	file descriptor. When the runqueue becomes empty, we call
	select on all the pending contexts.

	TODO: add a nonblocking call to select so that we can poll
	for io from time to time rather than waiting till there is
	nothing else to do.

runtime/mercury_select.{c,h}:
	Make functions that forward to the FD_* macros, which on Linux
	attempt to use registers that we've already grabbed. Aarrggh!

runtime/mercury_thread.c:
	Tidy up some of the conditional compilation.

cvs diff: Diffing .
cvs diff: Diffing bindist
cvs diff: Diffing boehm_gc
cvs diff: Diffing boehm_gc/Mac_files
cvs diff: Diffing boehm_gc/cord
cvs diff: Diffing boehm_gc/cord/private
cvs diff: Diffing boehm_gc/include
cvs diff: Diffing boehm_gc/include/private
cvs diff: Diffing browser
cvs diff: Diffing bytecode
cvs diff: Diffing bytecode/test
cvs diff: Diffing compiler
Index: compiler/pragma_c_gen.m
===================================================================
RCS file: /home/staff/zs/imp/mercury/compiler/pragma_c_gen.m,v
retrieving revision 1.23
diff -u -r1.23 pragma_c_gen.m
--- pragma_c_gen.m	1998/11/20 04:08:53	1.23
+++ pragma_c_gen.m	1998/11/22 22:45:41
@@ -382,14 +382,20 @@
 	%
 	% Code fragments to obtain and release the global lock
 	%
+	code_info__get_module_info(ModuleInfo),
 	{ ThreadSafe = thread_safe ->
 		ObtainLock = pragma_c_raw_code(""),
 		ReleaseLock = pragma_c_raw_code("")
 	;
-		ObtainLock =
-			pragma_c_raw_code("\tMR_OBTAIN_GLOBAL_C_LOCK();\n"),
-		ReleaseLock =
-			pragma_c_raw_code("\tMR_RELEASE_GLOBAL_C_LOCK();\n")
+		module_info_pred_proc_info(ModuleInfo, PredId, ProcId,
+			PredInfo, _),
+		pred_info_name(PredInfo, Name),
+		string__append_list(["\tMR_OBTAIN_GLOBAL_C_LOCK(""",
+			Name, """);\n"], ObtainLockStr),
+		ObtainLock = pragma_c_raw_code(ObtainLockStr),
+		string__append_list(["\tMR_RELEASE_GLOBAL_C_LOCK(""",
+			Name, """);\n"], ReleaseLockStr),
+		ReleaseLock = pragma_c_raw_code(ReleaseLockStr)
 	},
 
 	%
@@ -447,7 +453,7 @@
 	% join all the components of the pragma_c together
 	%
 	{ Components = [InputComp, SaveRegsComp, ObtainLock, C_Code_Comp,
-			CheckR1_Comp, ReleaseLock, RestoreRegsComp,
+			ReleaseLock, CheckR1_Comp, RestoreRegsComp,
 			OutputComp] },
 	{ PragmaCCode = node([
 		pragma_c(Decls, Components, MayCallMercury, MaybeFailLabel, no)
cvs diff: Diffing compiler/notes
cvs diff: Diffing doc
cvs diff: Diffing extras
cvs diff: Diffing extras/cgi
cvs diff: Diffing extras/complex_numbers
cvs diff: Diffing extras/complex_numbers/samples
cvs diff: Diffing extras/complex_numbers/tests
cvs diff: Diffing extras/exceptions
cvs diff: Diffing extras/graphics
cvs diff: Diffing extras/graphics/mercury_opengl
cvs diff: Diffing extras/graphics/mercury_tcltk
cvs diff: Diffing extras/graphics/samples
cvs diff: Diffing extras/graphics/samples/calc
cvs diff: Diffing extras/graphics/samples/maze
cvs diff: Diffing extras/graphics/samples/pent
cvs diff: Diffing extras/odbc
cvs diff: Diffing extras/references
cvs diff: Diffing extras/references/samples
cvs diff: Diffing extras/references/tests
cvs diff: Diffing extras/trailed_update
cvs diff: Diffing extras/trailed_update/samples
cvs diff: Diffing extras/trailed_update/tests
cvs diff: Diffing library
Index: library/char.m
===================================================================
RCS file: /home/staff/zs/imp/mercury/library/char.m,v
retrieving revision 1.25
diff -u -r1.25 char.m
--- char.m	1998/05/11 20:12:36	1.25
+++ char.m	1998/10/14 23:52:12
@@ -401,16 +401,18 @@
 
 %-----------------------------------------------------------------------------%
 
-:- pragma c_code(char__to_int(Character::in, Int::out), will_not_call_mercury, "
+:- pragma c_code(char__to_int(Character::in, Int::out),
+               [will_not_call_mercury, thread_safe] , "
 	Int = (UnsignedChar) Character;
 ").
 
-:- pragma c_code(char__to_int(Character::in, Int::in), will_not_call_mercury, "
+:- pragma c_code(char__to_int(Character::in, Int::in),
+               [will_not_call_mercury, thread_safe] , "
 	SUCCESS_INDICATOR = ((UnsignedChar) Character == Int);
 ").
 
-:- pragma c_code(char__to_int(Character::out, Int::in), will_not_call_mercury,
-"
+:- pragma c_code(char__to_int(Character::out, Int::in),
+               [will_not_call_mercury, thread_safe] , "
 	/*
 	** If the integer doesn't fit into a char, then
 	** the assignment `Character = Int' below will truncate it.
@@ -427,7 +429,8 @@
 char__min_char_value(0).
 
 :- pragma c_header_code("#include <limits.h>").
-:- pragma c_code(char__max_char_value(Max::out), will_not_call_mercury, "
+:- pragma c_code(char__max_char_value(Max::out),
+               [will_not_call_mercury, thread_safe], "
 	Max = UCHAR_MAX;
 ").
 	
Index: library/std_util.m
===================================================================
RCS file: /home/staff/zs/imp/mercury/library/std_util.m,v
retrieving revision 1.137
diff -u -r1.137 std_util.m
--- std_util.m	1998/11/19 06:11:00	1.137
+++ std_util.m	1998/11/22 22:46:20
@@ -959,14 +959,17 @@
 % to make sure that the compiler doesn't issue any determinism warnings
 % for them.
 
-:- pragma c_code(semidet_succeed, will_not_call_mercury,
+:- pragma c_code(semidet_succeed, [will_not_call_mercury, thread_safe],
 		"SUCCESS_INDICATOR = TRUE;").
-:- pragma c_code(semidet_fail, will_not_call_mercury,
+:- pragma c_code(semidet_fail, [will_not_call_mercury, thread_safe],
 		"SUCCESS_INDICATOR = FALSE;").
-:- pragma c_code(cc_multi_equal(X::in, Y::out), will_not_call_mercury,
+:- pragma c_code(cc_multi_equal(X::in, Y::out),
+               [will_not_call_mercury, thread_safe],
 		"Y = X;").
-:- pragma c_code(cc_multi_equal(X::di, Y::uo), will_not_call_mercury,
+:- pragma c_code(cc_multi_equal(X::di, Y::uo),
+               [will_not_call_mercury, thread_safe],
 		"Y = X;").
+
 %-----------------------------------------------------------------------------%
 
 	% The type `std_util:type_info/0' happens to use much the same
cvs diff: Diffing lp_solve
cvs diff: Diffing lp_solve/lp_examples
cvs diff: Diffing profiler
cvs diff: Diffing runtime
Index: runtime/Mmakefile
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/Mmakefile,v
retrieving revision 1.42
diff -u -r1.42 Mmakefile
--- Mmakefile	1998/11/09 10:24:41	1.42
+++ Mmakefile	1998/11/22 22:46:27
@@ -58,6 +58,7 @@
 			mercury_prof_mem.h	\
 			mercury_regorder.h	\
 			mercury_regs.h		\
+			mercury_select.h	\
 			mercury_signal.h	\
 			mercury_std.h		\
 			mercury_stacks.h	\
@@ -114,6 +115,7 @@
 			mercury_prof.c		\
 			mercury_prof_mem.c	\
 			mercury_regs.c		\
+			mercury_select.c	\
 			mercury_signal.c	\
 			mercury_stack_trace.c	\
 			mercury_tabling.c	\
Index: runtime/mercury_context.c
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/mercury_context.c,v
retrieving revision 1.16
diff -u -r1.16 mercury_context.c
--- mercury_context.c	1998/11/11 02:14:16	1.16
+++ mercury_context.c	1998/11/24 22:41:15
@@ -21,6 +21,7 @@
 #include "mercury_memory_handlers.h"
 #include "mercury_context.h"
 #include "mercury_engine.h"	/* for `MR_memdebug' */
+#include "mercury_select.h"	/* for `MR_fd* stuff' */
 
 MR_Context	*MR_runqueue_head;
 MR_Context	*MR_runqueue_tail;
@@ -29,6 +30,11 @@
   MercuryCond	*MR_runqueue_cond;
 #endif
 
+MR_PendingContext	*MR_pending_contexts;
+#ifdef	MR_THREAD_SAFE
+  MercuryLock		*MR_pending_contexts_lock;
+#endif
+
 /*
 ** free_context_list is a global linked list of unused context
 ** structures. If the MemoryZone pointers are not NULL,
@@ -57,6 +63,9 @@
 
 	pthread_mutex_init(&MR_global_lock, MR_MUTEX_ATTR);
 
+	MR_pending_contexts_lock = make(MercuryLock);
+	pthread_mutex_init(MR_pending_contexts_lock, MR_MUTEX_ATTR);
+
 	MR_KEY_CREATE(&MR_engine_base_key, NULL);
 
 #endif
@@ -160,6 +169,82 @@
 	fatal_error("computation floundered");
 }
 
+	/*
+	** Check to see if any contexts that blocked on IO have become
+	** runnable. Return the number of contexts that are still blocked.
+	** The parameter specifies whether or not the call to select should
+	** block or not.
+	*/
+static int check_pending_contexts(Bool block);
+static int
+check_pending_contexts(Bool block)
+{
+	int	err, n;
+	fd_set	rd, wr, ex;
+	struct timeval timeout;
+	MR_PendingContext *tmp;
+
+	if (MR_pending_contexts == NULL)
+		return 0;
+
+	MR_fd_zero(&rd); MR_fd_zero(&wr); MR_fd_zero(&ex);
+	n = -1;
+	for (tmp = MR_pending_contexts ; tmp ; tmp = tmp -> next) {
+		if (tmp->waiting_mode & MR_PENDING_READ) {
+			n = ((n > tmp->fd) ? n : tmp->fd);
+			FD_SET(tmp->fd, &rd);
+		}
+		if (tmp->waiting_mode & MR_PENDING_WRITE) {
+			n = ((n > tmp->fd) ? n : tmp->fd);
+			FD_SET(tmp->fd, &wr);
+		}
+		if (tmp->waiting_mode & MR_PENDING_EXEC) {
+			n = ((n > tmp->fd) ? n : tmp->fd);
+			FD_SET(tmp->fd, &ex);
+		}
+	}
+	n++;
+
+	if (n == 0)
+		fatal_error("no fd's set!");
+
+	if (block) {
+		err = select(n, &rd, &wr, &ex, NULL);
+	} else {
+		timeout.tv_sec = 0;
+		timeout.tv_usec = 0;
+		err = select(n, &rd, &wr, &ex, &timeout);
+	}
+
+	if (err < 0)
+		fatal_error("select failed!");
+
+	n = 0;
+	for (tmp = MR_pending_contexts ; tmp ; tmp = tmp -> next) {
+		n++;
+		if (tmp->waiting_mode & MR_PENDING_READ) {
+			if(FD_ISSET(tmp->fd, &rd)) {
+				schedule(tmp->context);
+				continue;
+			}
+		}
+		if (tmp->waiting_mode & MR_PENDING_WRITE) {
+			if(FD_ISSET(tmp->fd, &wr)) {
+				schedule(tmp->context);
+				continue;
+			}
+		}
+		if (tmp->waiting_mode & MR_PENDING_EXEC) {
+			if(FD_ISSET(tmp->fd, &ex)) {
+				schedule(tmp->context);
+				continue;
+			}
+		}
+	}
+
+	return n;
+}
+
 Define_extern_entry(do_runnext);
 
 BEGIN_MODULE(scheduler_module)
@@ -179,15 +264,16 @@
 	MR_LOCK(MR_runqueue_lock, "do_runnext (i)");
 
 	while (1) {
-		if (MR_exit_now = TRUE) {
+		if (MR_exit_now == TRUE) {
 			MR_UNLOCK(MR_runqueue_lock, "do_runnext (ii)");
 			destroy_thread(MR_engine_base);
 		}
 		tmp = MR_runqueue_head;
+		/* XXX check pending io */
 		prev = NULL;
 		while(tmp != NULL) {
-			if (depth > 0 && tmp->owner_thread == thd
-					|| tmp->owner_thread == NULL)
+			if ((depth > 0 && tmp->owner_thread == thd)
+				|| (tmp->owner_thread == (MercuryThread) NULL))
 				break;
 			prev = tmp;
 			tmp = tmp->next;
@@ -209,8 +295,11 @@
 }
 #else /* !MR_THREAD_SAFE */
 {
-	if (MR_runqueue_head == NULL)
+	if (MR_runqueue_head == NULL && MR_pending_contexts == NULL)
 		fatal_error("empty runqueue!");
+
+	while (MR_runqueue_head == NULL)
+		check_pending_contexts(TRUE); /* block */
 
 	MR_ENGINE(this_context) = MR_runqueue_head;
 	MR_runqueue_head = MR_runqueue_head->next;
Index: runtime/mercury_context.h
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/mercury_context.h,v
retrieving revision 1.8
diff -u -r1.8 mercury_context.h
--- mercury_context.h	1998/11/11 02:14:16	1.8
+++ mercury_context.h	1998/11/22 22:46:32
@@ -129,8 +129,7 @@
 typedef MR_Context Context;	/* for backwards compatibility */
 
 /*
-** the runqueue is a linked list of contexts that are
-** runnable.
+** The runqueue is a linked list of contexts that are runnable.
 */
 extern		MR_Context	*MR_runqueue_head;
 extern		MR_Context	*MR_runqueue_tail;
@@ -139,6 +138,37 @@
   extern	MercuryCond	*MR_runqueue_cond;
 #endif
 
+/*
+** As well as the runqueue, we maintain a linked list of contexts
+** and associated file descriptors that are suspended blocked for
+** reads/writes/exceptions. When the runqueue becomes empty, if
+** this list is not empty then we call select and block until one
+** or more of the file descriptors become ready for I/O, then
+** wake the appropriate context.
+** In addition, we periodically check to see if the list of blocked
+** contexts is non-empty and if so, poll to wake any contexts that
+** can unblock. This, while not yielding true fairness (since this
+** requires the current context to perform some yield-like action),
+** ensures that it is possible for programmers to write concurrent
+** programs with continuous computation and interleaved I/O dependent
+** computation in a straight-forward manner.
+*/
+
+#define	MR_PENDING_READ		0x01
+#define	MR_PENDING_WRITE	0x02
+#define	MR_PENDING_EXEC		0x04
+
+typedef struct MR_PENDING_CONTEXT {
+	struct MR_PENDING_CONTEXT	*next;
+	MR_Context			*context;
+	int				fd;
+	char				waiting_mode;
+} MR_PendingContext;
+
+extern	MR_PendingContext	*MR_pending_contexts;
+#ifdef	MR_THREAD_SAFE
+  extern	MercuryLock	*MR_pending_contexts_lock;
+#endif
 
 /*
 ** Initializes a context structure.
Index: runtime/mercury_library_types.h
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/mercury_library_types.h,v
retrieving revision 1.2
diff -u -r1.2 mercury_library_types.h
--- mercury_library_types.h	1998/11/09 10:24:37	1.2
+++ mercury_library_types.h	1998/11/22 22:46:37
@@ -15,6 +15,8 @@
 #include <stdio.h>		/* for `FILE' */
 #include "mercury_types.h"	/* for `Word' and `Integer' */
 
+#include "mercury_thread.h"
+
 /*
 ** The C `MercuryFile' type is used for the Mercury `io__stream' type
 ** in library/io.m.
@@ -25,6 +27,9 @@
 typedef struct mercury_file {
 	FILE *file;
 	int line_number;
+#ifdef MR_THREAD_SAFE
+	MercuryLock	lock;
+#endif
 } MercuryFile;
 
 /*
Index: runtime/mercury_thread.c
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/mercury_thread.c,v
retrieving revision 1.7
diff -u -r1.7 mercury_thread.c
--- mercury_thread.c	1998/09/21 10:21:40	1.7
+++ mercury_thread.c	1998/10/07 01:54:53
@@ -125,15 +125,13 @@
 }
 #endif
 
-#ifdef	MR_THREAD_SAFE
+#if defined(MR_THREAD_SAFE) && defined(MR_DEBUG_THREADS)
 void
 MR_mutex_lock(MercuryLock *lock, const char *from)
 {
 	int err;
 
-#ifdef MR_DEBUG_THREADS
 	fprintf(stderr, "%d locking on %p (%s)\n", pthread_self(), lock, from);
-#endif
 	err = pthread_mutex_lock(lock);
 	assert(err == 0);
 }
@@ -143,10 +141,8 @@
 {
 	int err;
 
-#ifdef MR_DEBUG_THREADS
 	fprintf(stderr, "%d unlocking on %p (%s)\n",
 		pthread_self(), lock, from);
-#endif
 	err = pthread_mutex_unlock(lock);
 	assert(err == 0);
 }
@@ -156,9 +152,7 @@
 {
 	int err;
 
-#ifdef MR_DEBUG_THREADS
 	fprintf(stderr, "%d signaling %p\n", pthread_self(), cond);
-#endif
 	err = pthread_cond_broadcast(cond);
 	assert(err == 0);
 }
@@ -168,9 +162,7 @@
 {
 	int err;
 
-#ifdef MR_DEBUG_THREADS
 	fprintf(stderr, "%d waiting on %p (%p)\n", pthread_self(), cond, lock);
-#endif
 	err = pthread_cond_wait(cond, lock);
 	assert(err == 0);
 }
Index: runtime/mercury_thread.h
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/mercury_thread.h,v
retrieving revision 1.3
diff -u -r1.3 mercury_thread.h
--- mercury_thread.h	1998/08/07 00:50:30	1.3
+++ mercury_thread.h	1998/09/10 02:43:13
@@ -15,7 +15,7 @@
   #if defined(MR_DIGITAL_UNIX_PTHREADS)
     #define MR_MUTEX_ATTR		pthread_mutexattr_default
     #define MR_COND_ATTR		pthread_condattr_default
-    #define MR_THREAD_ATTR	pthread_attr_default
+    #define MR_THREAD_ATTR		pthread_attr_default
   #else
     #define MR_MUTEX_ATTR		NULL
     #define MR_COND_ATTR		NULL
@@ -27,7 +27,7 @@
   typedef pthread_mutex_t	MercuryLock;
   typedef pthread_cond_t	MercuryCond;
 
-  #if 0
+  #ifndef MR_DEBUG_THREADS
 	/*
 	** The following macros should be used once the
 	** use of locking in the generated code is considered
@@ -57,11 +57,11 @@
 	** predicates which are not thread-safe.
 	** See the comments below.
 	*/
-  #define MR_OBTAIN_GLOBAL_C_LOCK()	MR_mutex_lock(&MR_global_lock, \
-						"pragma c code");
+  #define MR_OBTAIN_GLOBAL_C_LOCK(where)	MR_LOCK(&MR_global_lock, \
+							where);
 
-  #define MR_RELEASE_GLOBAL_C_LOCK()	MR_mutex_unlock(&MR_global_lock, \
-						"pragma c code");
+  #define MR_RELEASE_GLOBAL_C_LOCK(where)	MR_UNLOCK(&MR_global_lock, \
+							where);
 
   #if defined(MR_DIGITAL_UNIX_PTHREADS)
     #define MR_GETSPECIFIC(key) 	({		\
Index: runtime/mercury_types.h
===================================================================
RCS file: /home/staff/zs/imp/mercury/runtime/mercury_types.h,v
retrieving revision 1.13
diff -u -r1.13 mercury_types.h
--- mercury_types.h	1998/07/28 03:10:01	1.13
+++ mercury_types.h	1998/08/24 04:49:19
@@ -65,9 +65,6 @@
 /* continuation function type, for --high-level-C option */
 typedef void (*Cont) (void);
 
-/* spinlocks -- see mercury_spinlock.h */
-typedef Word	SpinLock;
-
 /*
 ** semidet predicates indicate success or failure by leaving nonzero or zero
 ** respectively in register r1
cvs diff: Diffing runtime/GETOPT
cvs diff: Diffing runtime/machdeps
cvs diff: Diffing samples
cvs diff: Diffing samples/c_interface
cvs diff: Diffing samples/c_interface/c_calls_mercury
cvs diff: Diffing samples/c_interface/cplusplus_calls_mercury
cvs diff: Diffing samples/c_interface/mercury_calls_c
cvs diff: Diffing samples/c_interface/mercury_calls_cplusplus
cvs diff: Diffing samples/c_interface/mercury_calls_fortran
cvs diff: Diffing samples/c_interface/simpler_c_calls_mercury
cvs diff: Diffing samples/c_interface/simpler_cplusplus_calls_mercury
cvs diff: Diffing samples/diff
cvs diff: Diffing samples/muz
cvs diff: Diffing samples/rot13
cvs diff: Diffing scripts
cvs diff: Diffing tools
cvs diff: Diffing trace
cvs diff: Diffing trial
cvs diff: Diffing util

-begin-mercury_select.c----------------------------------------------------
#include "mercury_select.h"

void
MR_fd_zero(fd_set *fdset)
{
	FD_ZERO(fdset);
}

-end-mercury_select.c----------------------------------------------------
-begin-mercury_select.h----------------------------------------------------

#ifndef	MR_MERCURY_SELECT_H
#define MR_MERCURY_SELECT_H

#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

void MR_fd_zero(fd_set *fdset);

#endif
-end-mercury_select.h----------------------------------------------------



More information about the developers mailing list