more thread updates.

git-svn-id: https://yap.svn.sf.net/svnroot/yap/trunk@1001 b08c6af1-5177-4d33-ba66-4b1c6b8b522a

C/absmi.c

@@ -6071,6 +6071,7 @@ Yap_absmi(int inp)
 \************************************************************************/
 
       BOp(call_cpred, sla);
+     
 #ifdef FROZEN_STACKS
       { 
 	choiceptr top_b = PROTECT_FROZEN_B(B);
@@ -6376,6 +6377,10 @@ Yap_absmi(int inp)
 #endif
  	saveregs();
 	pt0 = Yap_ExpandIndex(pe);
+	if (PP == NULL) {
+	  READ_UNLOCK(pe->PRWLock);
+	  PP = pe;
+	}
 	/* restart index */
 	setregs();
 	UNLOCK(pe->PELock);

C/dbase.c

@@ -3806,6 +3806,7 @@ complete_lu_erase(LogUpdClause *clau)
     return;
   }
 #ifdef DEBUG
+#ifndef THREADS
   if (clau->ClNext)
     clau->ClNext->ClPrev = clau->ClPrev;
   if (clau->ClPrev) {
@@ -3813,6 +3814,7 @@ complete_lu_erase(LogUpdClause *clau)
   } else {
     DBErasedList = clau->ClNext;
   }
+#endif
 #endif
   if (cp != NULL) {
     DBRef ref;
@@ -3888,6 +3890,7 @@ EraseLogUpdCl(LogUpdClause *clau)
     ap->cs.p_code.NOfClauses--;
     clau->ClFlags |= ErasedMask;
 #ifdef DEBUG
+#ifndef THREADS
     {
       LogUpdClause *er_head = DBErasedList;
       if (er_head == NULL) {
@@ -3899,6 +3902,7 @@ EraseLogUpdCl(LogUpdClause *clau)
       }
       DBErasedList = clau;
     }
+#endif
 #endif
     /* we are holding a reference to the clause */
     clau->ClRefCount++;

C/index.c

@@ -4129,7 +4129,7 @@ expand_index(struct intermediates *cint) {
       lab = do_index(cls, max, cint, argno+1, fail_l, isfirstcl, clleft, top);
     }
   }
-  if (labp)
+  if (labp && !(lab & 1))
     *labp = (yamop *)lab; /* in case we have a single clause */
   return labp;
 }

C/init.c

@@ -781,6 +781,8 @@ InitCodes(void)
   heap_regs->thread_handle[0].default_yaam_regs = 
     &Yap_standard_regs;
   heap_regs->thread_handle[0].handle = pthread_self();
+  heap_regs->thread_handle[0].handle = pthread_self();
+  pthread_mutex_init(&ThreadHandle[0].tlock, NULL);
 #endif
 #if defined(YAPOR) || defined(THREADS)
   INIT_LOCK(heap_regs->bgl);

C/threads.c

@@ -71,12 +71,13 @@ store_specs(int new_worker_id, UInt ssize, UInt tsize, Term tgoal, Term tdetach)
 
 
 static void
-thread_die(int wid)
+thread_die(int wid, int always_die)
 {
   Prop p0;
 
   LOCK(ThreadHandlesLock);
-  if (ThreadHandle[wid].tdetach == MkAtomTerm(AtomTrue)) {
+  if (ThreadHandle[wid].tdetach == MkAtomTerm(AtomTrue) ||
+      always_die) {
     p0 = AbsPredProp(heap_regs->thread_handle[wid].local_preds);
     /* kill all thread local preds */
     while(p0) {
@@ -117,9 +118,8 @@ thread_run(void *widp)
   tgs[0] = Yap_FetchTermFromDB(ThreadHandle[worker_id].tgoal);
   tgs[1] = ThreadHandle[worker_id].tdetach;
   tgoal = Yap_MkApplTerm(FunctorThreadRun, 2, tgs);
-  pthread_mutex_unlock(&(ThreadHandle[worker_id].tlock));
   out = Yap_RunTopGoal(tgoal);
-  thread_die(worker_id);
+  thread_die(worker_id, FALSE);
   return NULL;
 }
 
@@ -138,19 +138,15 @@ p_create_thread(void)
   Term tgoal = Deref(ARG1);
   Term tdetach = Deref(ARG5);
   int new_worker_id = IntegerOfTerm(Deref(ARG6));
-  pthread_attr_t at;
   
   if (new_worker_id == -1) {
     /* YAP ERROR */
     return FALSE;
   }
   ThreadHandle[new_worker_id].id = new_worker_id;
-  pthread_mutex_init(&ThreadHandle[new_worker_id].tlock, NULL);
-  pthread_mutex_lock(&(ThreadHandle[new_worker_id].tlock));
   store_specs(new_worker_id, ssize, tsize, tgoal, tdetach);
-  pthread_attr_init(&at);
+  pthread_mutex_init(&ThreadHandle[new_worker_id].tlock, NULL);
-  pthread_attr_setstacksize(&at, 32*4096);
+  if ((ThreadHandle[new_worker_id].ret = pthread_create(&ThreadHandle[new_worker_id].handle, NULL, thread_run, (void *)(&(ThreadHandle[new_worker_id].id)))) == 0) {
-  if ((ThreadHandle[new_worker_id].ret = pthread_create(&(ThreadHandle[new_worker_id].handle), NULL, thread_run, (void *)(&(ThreadHandle[new_worker_id].id)))) == 0) {
     return TRUE;
   }
   /* YAP ERROR */
@@ -167,8 +163,6 @@ static Int
 p_thread_join(void)
 {
   Int tid = IntegerOfTerm(Deref(ARG1));
-  pthread_t th;
-  void *retval;
 
   LOCK(ThreadHandlesLock);
   if (!ThreadHandle[tid].in_use) {
@@ -179,13 +173,13 @@ p_thread_join(void)
     UNLOCK(ThreadHandlesLock);
     return FALSE;
   }
-  ThreadHandle[tid].tdetach = MkAtomTerm(AtomTrue);
-  th = ThreadHandle[tid].handle;
   UNLOCK(ThreadHandlesLock);
-  if (pthread_join(th, &retval) < 0) {
+  if (pthread_join(ThreadHandle[tid].handle, NULL) < 0) {
     /* ERROR */
+  fprintf(stderr, "join error %d %d\n", tid, errno);
     return FALSE;
   }
+  fprintf(stderr, "join %d\n", tid);
   return TRUE;
 }
 
@@ -194,15 +188,14 @@ p_thread_destroy(void)
 {
   Int tid = IntegerOfTerm(Deref(ARG1));
 
-  thread_die(tid);
+  thread_die(tid, TRUE);
   return TRUE;
 }
 
 static Int
 p_thread_detach(void)
 {
-  pthread_t th = ThreadHandle[IntegerOfTerm(Deref(ARG1))].handle;
+  if (pthread_detach(ThreadHandle[IntegerOfTerm(Deref(ARG1))].handle) < 0) {
-  if (pthread_detach(th) < 0) {
     /* ERROR */
     return FALSE;
   }
@@ -212,7 +205,8 @@ p_thread_detach(void)
 static Int
 p_thread_exit(void)
 {
-  thread_die(worker_id);
+  fprintf(stderr,"here i go %d %ld\n", worker_id, ThreadHandle[worker_id].handle);
+  thread_die(worker_id, FALSE);
   pthread_exit(NULL);
   return TRUE;
 }

C/tracer.c

@@ -77,6 +77,10 @@ unsigned long long vsc_count;
 unsigned long vsc_count;
 #endif
 
+#if THREADS
+static int thread_trace;
+#endif
+
 /*
 static int
 check_trail_consistency(void) {
@@ -115,6 +119,11 @@ low_level_trace(yap_low_level_port port, PredEntry *pred, CELL *args)
   /*  extern int gc_calls; */
 
   LOCK(heap_regs->low_level_trace_lock);
+  if (thread_trace &&
+      worker_id +1 != thread_trace) {
+    UNLOCK(heap_regs->low_level_trace_lock);
+    return;    
+  }
   vsc_count++;
 #ifdef COMMENTED
   //  if (vsc_count == 218280)
@@ -256,6 +265,15 @@ static Int p_start_low_level_trace(void)
   return(TRUE);
 }
 
+#ifdef THREADS
+static Int p_start_low_level_trace2(void)
+{
+  thread_trace = IntegerOfTerm(Deref(ARG1))+1;
+  Yap_do_low_level_trace = TRUE;
+  return(TRUE);
+}
+#endif
+
 static Int p_stop_low_level_trace(void)
 {
   Yap_do_low_level_trace = FALSE;
@@ -267,6 +285,9 @@ void
 Yap_InitLowLevelTrace(void)
 {
   Yap_InitCPred("start_low_level_trace", 0, p_start_low_level_trace, SafePredFlag);
+#if THREADS
+  Yap_InitCPred("start_low_level_trace", 1, p_start_low_level_trace2, SafePredFlag);
+#endif
   Yap_InitCPred("stop_low_level_trace", 0, p_stop_low_level_trace, SafePredFlag);
 }
 

pl/threads.yap

@@ -103,7 +103,10 @@ thread_create(Goal, Id, Options) :-
 '$thread_ground_stacks'(_).
 
 '$add_thread_aliases'([Alias|Aliases], Id) :-
-	recorda('$thread_alias',[Id0|Alias],_),
+	recorded('$thread_alias',[Id0|Alias],_), !,
+	'$do_error'(permission_error(alias,new,Alias),thread_create_alias(Id,Alias)).
+'$add_thread_aliases'([Alias|Aliases], Id) :-
+	recorda('$thread_alias',[Id|Alias],_),
 	'$add_thread_aliases'(Aliases, Id).
 '$add_thread_aliases'([], _).
 
@@ -178,7 +181,7 @@ current_thread(Tid, Status) :-
 	
 '$thr_status'(Tid, Status) :-
 	recorded('$thread_exit_status', [Tid|Status], _), !.
-'$thr_status'(Tid, running).
+'$thr_status'(_, running).
 
 
 mutex_create(V) :-
@@ -198,7 +201,7 @@ mutex_create(V) :-
 	
 mutex_destroy(V) :-
 	var(V), !,
-	'$do_error'(instantiation_error,mutex_destroy(A)).
+	'$do_error'(instantiation_error,mutex_destroy(V)).
 mutex_destroy(A) :-
 	recorded('$mutex',[A|Id],R), !,
 	'$destroy_mutex'(Id),
@@ -211,7 +214,7 @@ mutex_destroy(V) :-
 	
 mutex_lock(V) :-
 	var(V), !,
-	'$do_error'(instantiation_error,mutex_lock(A)).
+	'$do_error'(instantiation_error,mutex_lock(V)).
 mutex_lock(A) :-
 	recorded('$mutex',[A|Id],_), !,
 	'$lock_mutex'(Id).
@@ -224,7 +227,7 @@ mutex_lock(V) :-
 	
 mutex_trylock(V) :-
 	var(V), !,
-	'$do_error'(instantiation_error,mutex_trylock(A)).
+	'$do_error'(instantiation_error,mutex_trylock(V)).
 mutex_trylock(A) :-
 	recorded('$mutex',[A|Id],_), !,
 	'$trylock_mutex'(Id).
@@ -237,7 +240,7 @@ mutex_trylock(V) :-
 	
 mutex_unlock(V) :-
 	var(V), !,
-	'$do_error'(instantiation_error,mutex_unlock(A)).
+	'$do_error'(instantiation_error,mutex_unlock(V)).
 mutex_unlock(A) :-
 	recorded('$mutex',[A|Id],_), !,
 	( '$unlock_mutex'(Id) ->
@@ -297,12 +300,12 @@ message_queue_destroy(Name) :-
 message_queue_destroy(Queue) :-
 	recorded('$queue',q(Queue,Mutex,Cond),R), !,
 	erase(R),
-	mutex_destroy(Mutex),
 	'$cond_destroy'(Cond),
+	mutex_destroy(Mutex),
 	'$clean_mqueue'(Queue).
 message_queue_destroy(Queue) :-
 	atom(Queue), !,
-	'$do_error'(existence_error(queue,Queue),message_queue_destroy(Name)).
+	'$do_error'(existence_error(queue,Queue),message_queue_destroy(QUeue)).
 message_queue_destroy(Name) :-
 	'$do_error'(type_error(atom,Name),message_queue_destroy(Name)).
 
@@ -331,7 +334,7 @@ thread_get_message(Queue, Term) :-
 	mutex_lock(Mutex),
 	'$thread_get_message_loop'(Queue, Term, Mutex, Cond).
 
-'$thread_get_message_loop'(Queue, Term, Mutex, Cond) :-
+'$thread_get_message_loop'(Queue, Term, Mutex, _) :-
 	recorded('$msg_queue',q(Queue,Term),R), !,
 	erase(R),
 	mutex_unlock(Mutex).
@@ -344,14 +347,14 @@ thread_peek_message(Term) :-
 	thread_peek_message(Id, Term).
 
 thread_peek_message(Queue, Term) :-
-	recorded('$queue',q(Queue,Mutex,Cond),_),
+	recorded('$queue',q(Queue,Mutex,_),_),
 	mutex_lock(Mutex),
 	'$thread_peek_message2'(Queue, Term, Mutex).
 
 '$thread_peek_message2'(Queue, Term, Mutex) :-
 	recorded('$msg_queue',q(Queue,Term),_), !,
 	mutex_unlock(Mutex).
-'$thread_peek_message2'(Queue, Term, Mutex) :-
+'$thread_peek_message2'(_, _, Mutex) :-
 	mutex_unlock(Mutex),
 	fail.