pyxes use futexes; faster. remove timeout-0 pun

Author: moon-chilled

jsrc/cd.c

@@ -8,7 +8,8 @@
 #include <Windows.h>
 #include <stdint.h> // portable: uint64_t   MSVC: __int64 
 
-struct jtimeval { long tv_sec, tv_usec; };
+struct jtimespec { long long tv_sec, tv_nsec; };
+struct jtimeval { long long tv_sec, tv_usec; };
 struct jtimezone { int tz_minuteswest, tz_dsttime; };
 
 int jgettimeofday(struct jtimeval *tp, struct jtimezone * tzp)
@@ -31,4 +32,13 @@ int jgettimeofday(struct jtimeval *tp, struct jtimezone * tzp)
     tp->tv_usec = (long) (system_time.wMilliseconds * 1000);
     return 0;
 }
+
+//monotonic clock
+//alternative is QueryPerformanceCounter; it probably uses rdtsc, which is stable on recent processors, but it gives inconsistent results when the processor goes to sleep
+struct jtimespec jmtclk(){
+ UI t=GetTickCount64();
+ R(struct jtimeval){.tv_sec=t/1000,.tv_nsec=1000000*(t%1000)};}
+#else
+#include"j.h"
+struct jtimespec jmtclk(){struct timespec r; clock_gettime(CLOCK_MONOTONIC,&r);R r;}
 #endif

jsrc/ct.c

@@ -176,10 +176,14 @@ typedef struct pyxcondmutex{
  S pyxorigthread;  // thread number that is working on this pyx, or _1 if the value is available
  C errcode;  // 0 if no error, or error code
 #if PYXES
- WAITBLOK pyxwb;  // sync info
+ //WAITBLOK pyxwb;  // sync info
+ UI4 state;//one of the below pyx states.  Monotonically increases
 #endif
 } PYXBLOK;
-
+enum{
+ PYXEMPTY, //the pyx is not filled in, and no one is waiting
+ PYXWAIT,  //at least 1 thread is waiting, and the pyx is not filled in
+ PYXFULL}; //the pyx is filled in
 #if PYXES
 
 // Install a value/errcode into a (recursive) pyx, and broadcast to anyone waiting on it.  fa() the pyx to indicate that the thread has released the pyx
@@ -191,7 +195,7 @@ static I jtsetpyxval(J jt, A pyx, A z, C errcode){I res=1;
  if(likely(z!=0))ra(z);  // since the pyx is recursive, we must ra the result we store into it.  Could zap if inplaceable
  __atomic_store_n(&((PYXBLOK*)AAV0(pyx))->pyxvalue,z,__ATOMIC_RELEASE);  // set result value
  // broadcast to wake up any tasks waiting for the result
- WAITBLOKFLAG(&((PYXBLOK*)AAV0(pyx))->pyxwb);
+ if(PYXWAIT==xchga(&((PYXBLOK*)AAV0(pyx))->state,PYXFULL))jfutex_wakea(&((PYXBLOK*)AAV0(pyx))->state);
  // unprotect pyx.  It was raised when it was assigned to this owner; now it belongs to the system
  fa(pyx);
  R 1;
@@ -201,41 +205,33 @@ static I jtsetpyxval(J jt, A pyx, A z, C errcode){I res=1;
 static A jtcreatepyx(J jt, I thread,D timeout){A pyx;
  // Allocate.  Init value, cond, and mutex to idle
  GAT0(pyx,INT,((sizeof(PYXBLOK)+(SZI-1))>>LGSZI)+1,0); AAV0(pyx)[0]=0; // allocate the result pointer (1), and the cond/mutex for the pyx.
- WAITBLOKINIT(&((PYXBLOK*)AAV0(pyx))->pyxwb);
+ ((PYXBLOK*)AAV0(pyx))->state=PYXEMPTY;
  // Init the pyx to a recursive box, with raised usecount.  AN=1 always.  But set the value/errcode to NULL/no error and the thread# to the executing thread
  AT(pyx)=BOX+PYX; AFLAG(pyx)=BOX; ACINIT(pyx,ACUC2); AN(pyx)=1; ((PYXBLOK*)AAV0(pyx))->pyxvalue=0; ((PYXBLOK*)AAV0(pyx))->pyxorigthread=thread; ((PYXBLOK*)AAV0(pyx))->errcode=0;  ((PYXBLOK*)AAV0(pyx))->pyxmaxwt=timeout;
  // The pyx's usecount of 2 is one for the owning thread and one for the current thread, which has a tpop for the pyx that protects it until it is put into its box.  When the pyx is filled in the owner will fa().
  R pyx;
 }
 
 // w is an A holding a pyx value.  Return its value when it has been resolved.  If it times out
-A jtpyxval(J jt,A pyx){A res; C errcode;
- D maxtime=tod()+((PYXBLOK*)AAV0(pyx))->pyxmaxwt+0.000001;  // get the time when we have to give up on this pyx, min 1usec
- // read the pyx value.  Since the creating thread has a release barrier after creation and another after final resolution, we can be sure
- // that if we read nonzero the pyx has been resolved, even without an acquire barrier
- while((res=__atomic_load_n(&((PYXBLOK*)AAV0(pyx))->pyxvalue,__ATOMIC_ACQUIRE))==0&&(errcode=__atomic_load_n(&((PYXBLOK*)AAV0(pyx))->errcode,__ATOMIC_ACQUIRE))==0){  // repeat till defined
-  I adbreak=__atomic_load_n((US*)&JT(jt,adbreak)[0],__ATOMIC_ACQUIRE);  // break requests
+A jtpyxval(J jt,A pyx){ UI4 state;
+ if(PYXFULL==(state=lda(&((PYXBLOK*)AAV0(pyx))->state)))goto done;
+ if(state!=PYXWAIT)if(!casa(&((PYXBLOK*)AAV0(pyx))->state,&state,PYXWAIT))goto done;
+ UI ns=({D mwt=((PYXBLOK*)AAV0(pyx))->pyxmaxwt;mwt==inf?IMAX:(I)(mwt*1e9);});
+ struct jtimespec end=jtmtil(ns); // get the time when we have to give up on this pyx
+ while(1){ // repeat till defined
+  _jfutex_waitn(&((PYXBLOK*)AAV0(pyx))->state,PYXWAIT,ns);
+  if(lda(&((PYXBLOK*)AAV0(pyx))->state)==PYXFULL)break; // if pyx was filled, exit and return its value
+  I adbreak=lda((US*)&JT(jt,adbreak)[0]);  // break requests
   // wait till the value is defined.  We have to make one last check inside the lock to make sure the value is still unresolved
   // The wait may time out because another thread is requesting a system lock.  If so, we accept it now
   if(unlikely(adbreak>>8)!=0){jtsystemlockaccept(jt,LOCKPRISYM+LOCKPRIPATH+LOCKPRIDEBUG); continue;}  // process lock and keep waiting
   // or, the user may be requesting a BREAK interrupt for deadlock or other slow execution.  In that case fail the pyx.  It will not be deleted until the value has been stored
-  if(unlikely((adbreak&0xff)>1)){errcode=EVBREAK; break;}  // JBREAK: fail the pyx and exit
-  // if the pyx has a max time, see if that is exceeded
-  if(unlikely(maxtime<tod())){errcode=EVTIME; break;}  // timeout: fail the pyx and exit
-  pthread_mutex_lock(&((PYXBLOK*)AAV0(pyx))->pyxwb.mutex);
-  if((res=__atomic_load_n(&((PYXBLOK*)AAV0(pyx))->pyxvalue,__ATOMIC_ACQUIRE))==0&&(errcode=__atomic_load_n(&((PYXBLOK*)AAV0(pyx))->errcode,__ATOMIC_ACQUIRE))==0){
-   struct jtimeval nowtime;
-   jgettimeofday(&nowtime,0);  // system time now
-   I tousec=nowtime.tv_usec+200000;
-   struct timespec endtime={nowtime.tv_usec+(tousec>=1000000),tousec-1000000*(tousec>=1000000)};  // system time when we give up.  The struct says it uses nsec but it seems to use usec
-   pthread_cond_timedwait(&((PYXBLOK*)AAV0(pyx))->pyxwb.cond,&((PYXBLOK*)AAV0(pyx))->pyxwb.mutex,&endtime);
-  }
-  pthread_mutex_unlock(&((PYXBLOK*)AAV0(pyx))->pyxwb.mutex);
+  if(unlikely(adbreak&0xff))ASSERT(0,adbreak&0xff);  // JBREAK: fail the pyx and exit
+  ASSERT(-1!=(ns=jtmdif(end)),EVTIME);  // update timeout; potentially fail the pyx and exit
  }
- // res now contains the certified value of the pyx.
- if(likely(res!=0))R res;   // valid value, use it
- ASSERT(0,errcode)   // if error, return the error code
-}
+done:
+ if(likely(!!((PYXBLOK*)AAV0(pyx))->pyxvalue))R ((PYXBLOK*)AAV0(pyx))->pyxvalue; // valid value, use it
+ ASSERT(0,((PYXBLOK*)AAV0(pyx))->errcode);} // if error, return the error code
 
 // ************************************* Locks **************************************
 // take a readlock on *alock.  We come here only if a writelock was requested or running.  We have incremented the readlock
@@ -642,8 +638,9 @@ F2(jttcapdot2){A z;
   break;}
  case 5: { // create a user pyx.  y is the timeout in seconds
 #if PYXES
-  ASSERT(AN(w)==1,EVLENGTH) w=cvt(FL,w); D *atimeout=DAV(w); atimeout=*atimeout==0?&inf:atimeout;  // get the timeout value.  If 0, use infinity
-  z=box(jtcreatepyx(jt,THREADID(jt),*atimeout));  // create the recursive pyx, owned by this thread
+  ASSERT(AN(w)==1,EVLENGTH) w=cvt(FL,w); D atimeout=*DAV(w); // get the timeout value
+  ASSERT(atimeout==inf||atimeout<=9e9,EVLIMIT); // 9e9 is approx 63 bits of ns.  This leaves ~300y; should be ok
+  z=box(jtcreatepyx(jt,THREADID(jt),atimeout));  // create the recursive pyx, owned by this thread
 #else
 ASSERT(0,EVNONCE)
 #endif
@@ -737,6 +734,7 @@ ASSERT(0,EVNONCE)
    ASSERT(lockrc<=0,lockrc);  // positive is a hard failure
    lockfail=lockrc==-1;  // -1 is a soft failure
   }else{
+   ASSERT(timeout<=9e9,EVLIMIT); // 9e9 is approx 63 bits of ns.  This leaves ~300y; should be ok
    I lockrc=jtpthread_mutex_timedlock(jt,(jtpthread_mutex_t*)IAV0(mutex),1e9*timeout,1+THREADID(jt));
    ASSERT(lockrc<=0,lockrc);  // positive is a hard failure
    lockfail=lockrc==-1;  // -1 is a soft failure

jsrc/j.h

@@ -406,15 +406,18 @@ extern unsigned int __cdecl _clearfp (void);
 #endif
 
 #if SY_WIN32
-struct jtimeval { long tv_sec, tv_usec; };
+struct jtimespec { long long tv_sec, tv_nsec; };
+struct jtimeval { long long tv_sec, tv_usec; };
 struct jtimezone { int tz_minuteswest, tz_dsttime; };
 int jgettimeofday(struct jtimeval*, struct jtimezone*);
 #else
 #include <sys/time.h>
+#define jtimespec timespec
 #define jtimeval timeval
 #define jtimezone timezone
 #define jgettimeofday gettimeofday
 #endif
+struct jtimespec jmtclk(void); //monotonic clock.  Intended rel->abs conversions when sleeping; has poor granularity and slow on windows
 
 #if SY_64
 #if defined(MMSC_VER)  // SY_WIN32

jsrc/mt.c

@@ -1,11 +1,33 @@
-// jtpthread_mutex*: mutex implementation for macos/ios/..
-// loosly modeled after apple pthreads version 486.100.11 and ulrich drepper 'futexes are tricky', plus recursive mutexes
-// needed for T. because the mainline version does not support pthread_mutex_timedlock; glibc/pthreads4w are ok
-// _does not_ support condition variables (yet); so, use jtpthread_mutex_t if you need timedwait, but pthread_mutex_t if you need condition variables
-// _not_ signal-safe
+// concurrency primitives, including mutexes
+// a raft of reasons for this:
+// - return EINTR when interrupted
+// mutexes:
+//  - fast timedlock and recursive mutexes on macos
+//  - robust by default
+//    - various edge cases like lock on one thread/release on another, or acquire a lock you already hold, are UB in posix!
+//  - mutex requisition can be associated with a task, rather than a thread
+// condvars:
+//  - don't reacquire mutex on wake (_very_ slow)
+//  - wake n on linux
+// novel primitives, faster than would be possible with pthreads
+//  - mutex tokens
+//  - queue
+//  - pyx
+
+// mutexs are loosely modeled after ulrich drepper 'futexes are tricky'
 
 #include"j.h"
 
+struct jtimespec jtmtil(UI ns){
+ struct jtimespec r=jmtclk();
+ r.tv_sec+=ns/1000000000;r.tv_nsec+=ns%1000000000;
+ if(r.tv_nsec>=1000000000){r.tv_sec++;r.tv_nsec-=1000000000;}
+ R r;}
+I jtmdif(struct jtimespec w){
+ struct jtimespec t=jmtclk();
+ if(t.tv_sec>w.tv_sec||t.tv_sec==w.tv_sec&&t.tv_nsec>=w.tv_nsec)R -1;
+ R (w.tv_sec-t.tv_sec)*1000000000+w.tv_nsec-t.tv_nsec;}
+
 #if defined(__APPLE__) || defined(__linux__)
 enum{FREE=0,LOCK=1,WAIT=2};//values for mutex->v
 //todo consider storing owner in the high bits of v.  apple pthreads does this.  But it means we can't use xadd to unlock.  On the other hand, apple is mostly arm now, which doesn't have xadd anyway.
@@ -15,16 +37,17 @@ enum{FREE=0,LOCK=1,WAIT=2};//values for mutex->v
 
 // there is a flaw.  If t0 holds lock, t1 attempts to acquire it; when it eventually does, it will leave WAIT in v instead of LOCK, 
 
-// todo figure out ULF_WAIT_CANCEL_POINT (I think it allows implementing the desired behaviour for EVATTN)
+// todo what is ULF_WAIT_CANCEL_POINT?
 
 void jtpthread_mutex_init(jtpthread_mutex_t *m,B recursive){*m=(jtpthread_mutex_t){.recursive=recursive};}
 C jtpthread_mutex_lock(J jt,jtpthread_mutex_t *m,I self){
  if(uncommon(m->owner==self)){if(unlikely(!m->recursive))R EVCONCURRENCY; m->ct++;R 0;}
- UI4 e;if(likely((!(e=lda(&m->v)))&&((e=FREE),casa(&m->v,&e,LOCK)))){m->ct+=m->recursive;m->owner=self;R 0;} //success.  test-and-test-and-set is from glibc, mildly optimises the case when many threads swarm a locked mutex
+ UI4 e;if(likely((!(e=lda(&m->v)))&&((e=FREE),casa(&m->v,&e,LOCK))))goto success; //fast path.  test-and-test-and-set is from glibc, mildly optimises the case when many threads swarm a locked mutex.  Not sure if this is for the best, but after waffling for a bit I think it is
  if(e!=WAIT)e=xchga(&m->v,WAIT); //penalise the multi-waiters case, since it's slower anyway
  while(e!=FREE){
 #if __linux__
-  I i=_jfutex_waitn(&m->v,WAIT,(UI)-1); //bug? jfutex_wait doesn't get interrupted by signals on linux
+  I i=_jfutex_waitn(&m->v,WAIT,(UI)-1);
+  //bug? futex wait doesn't get interrupted by signals on linux if timeout is null
 #else
   I i=jfutex_wait(&m->v,WAIT);
 #endif
@@ -34,12 +57,11 @@ C jtpthread_mutex_lock(J jt,jtpthread_mutex_t *m,I self){
    else if(i==-ENOMEM)R EVWSFULL;//lol
    else R EVFACE;}
   e=xchga(&m->v,WAIT);} //exit when e==FREE; i.e., _we_ successfully installed WAIT in place of FREE
- m->ct+=m->recursive;m->owner=self;  R 0;}
+success:m->ct+=m->recursive;m->owner=self;  R 0;}
 I jtpthread_mutex_timedlock(J jt,jtpthread_mutex_t *m,UI ns,I self){
  if(uncommon(m->owner==self)){if(unlikely(!m->recursive))R EVCONCURRENCY; m->ct++;R 0;}
- UI4 e=0;if((e=lda(&m->v))!=FREE&&((e=FREE),casa(&m->v,&e,LOCK))){m->ct+=m->recursive;m->owner=self;R 0;} //success
- struct timespec tgt,now;if(clock_gettime(CLOCK_MONOTONIC,&now))R EVFACE;
- tgt.tv_sec=now.tv_sec+ns/1000000000;tgt.tv_nsec=now.tv_nsec+ns%1000000000;if(tgt.tv_nsec>=1000000000){tgt.tv_nsec-=1000000000;tgt.tv_sec++;};
+ UI4 e=0;if((e=lda(&m->v))!=FREE&&((e=FREE),casa(&m->v,&e,LOCK)))goto success;
+ struct timespec tgt=jtmtil(ns);
  if(common(e!=WAIT)){e=xchga(&m->v,WAIT);if(e==FREE)goto success;} //penalise the multi-waiters case, since it's slower anyway
  while(1){
   I i=_jfutex_waitn(&m->v,WAIT,ns);
@@ -52,11 +74,7 @@ I jtpthread_mutex_timedlock(J jt,jtpthread_mutex_t *m,UI ns,I self){
   e=xchga(&m->v,WAIT);
   if(e==FREE)goto success; //exit when e==FREE; i.e., _we_ successfully installed WAIT in place of FREE
   if(i==-ETIMEDOUT)R -1; //if the kernel says we timed out, trust it rather than doing another syscall to check the time
-  clock_gettime(CLOCK_MONOTONIC,&now);
-  if(now.tv_sec>=tgt.tv_sec || now.tv_sec==tgt.tv_sec&&now.tv_nsec>=tgt.tv_nsec)R -1;//timed out
-  ns=1000000000*(tgt.tv_sec-now.tv_sec);
-  if(now.tv_nsec<=tgt.tv_nsec)ns+=tgt.tv_nsec-now.tv_nsec;
-  else ns+=1000000000-(now.tv_nsec-tgt.tv_nsec);}
+  if(-1==(ns=jtmdif(tgt)))R -1;} //update delta, abort if timed out
 success:m->ct+=m->recursive;m->owner=self; R 0;}
 I jtpthread_mutex_trylock(jtpthread_mutex_t *m,I self){
  if(uncommon(m->recursive)&&m->owner){if(m->owner!=self)R -1; m->ct++;R 0;}

jsrc/mt.h

@@ -1,7 +1,11 @@
-// jtpthread_mutex*: mutex implementation for macos/ios/..
+// mt.h: mutex, sync, timing related interfaces
 // see mt.c
 
 #if PYXES
+struct jtimespec jtmtil(UI ns); //returns a time ns ns in the future
+I jtmdif(struct jtimespec when); //returns the time in ns between now and when.  If when is not in the future, the result will be -1
+//both of these are implemented in terms of mtclk and use its clock
+
 #if !defined(__APPLE__) && !defined(__linux__)
 #include <pthread.h>
 typedef pthread_mutex_t jtpthread_mutex_t;
@@ -56,11 +60,12 @@ C jtpthread_mutex_lock(J jt,jtpthread_mutex_t *m,I self);
 I jtpthread_mutex_timedlock(J jt,jtpthread_mutex_t*,UI ns,I self); //absolute timers suck; correct the interface.  -1=failure; 0=success; positive=error
 I jtpthread_mutex_trylock(jtpthread_mutex_t*,I self); //0=success -1=failure positive=error
 C jtpthread_mutex_unlock(jtpthread_mutex_t*,I self); //0 or error code
-
 //note: self must be non-zero
+
 #if defined(__linux__)
 #include <linux/futex.h>
 #include <sys/syscall.h>
+//glibc 'syscall': stupid errno
 static inline void jfutex_wake1(UI4 *p){
  __asm__ volatile("syscall" :: "a" (SYS_futex), //eax: syscall#
                                "D" (p), //rdi: ptr
@@ -77,17 +82,17 @@ static inline int jfutex_wait(UI4 *p,UI4 v){
                                   : "a" (SYS_futex), //eax: syscall#
                                     "D" (p), //rdi: ptr
                                     "S" (FUTEX_WAIT), //rsi: op
-                                    "d" (v), //rdx: val, espected
+                                    "d" (v), //rdx: espected
                                     "r" (pts)); //r10: timeout (null=no timeout)
- return r;}
+ R r;}
 static inline int _jfutex_waitn(UI4 *p,UI4 v,UI ns){
  struct timespec ts={.tv_sec=ns/1000000000, .tv_nsec=ns%1000000000};
  register struct timespec *pts asm("r10") = &ts;
  int r;__asm__ volatile("syscall" : "=a"(r) //result in rax
                                   : "a" (SYS_futex), //eax: syscall#
                                     "D" (p), //rdi: ptr
                                     "S" (FUTEX_WAIT), //rsi: op
-                                    "d" (v), //rdx: val, espected
+                                    "d" (v), //rdx: espected
                                     "r" (pts)); //r10: timeout (relative!)
  R r;}
 #elif defined(__APPLE__)