Merge pull request #21 from wsc1/master

cb cleanup/test

libsio/cb.c

@@ -4,6 +4,8 @@
 #include <stdatomic.h>
 #include <string.h>
 #include <stdint.h>
+#include <time.h>
+#include <stdio.h>
 
 #include "cb.h"
 
@@ -19,6 +21,10 @@ Cb * newCb(int bufSz) {
 
 	cb->out = 0;
 	cb->outF = 0;
+	cb->time.tv_sec = 0;
+	cb->time.tv_nsec = 1000;
+	cb->inGo = 0;
+	fprintf(stderr, "cb c addr %p inGo c addr on new: %p\n", cb, &cb->inGo);
 	return cb;
 }
 
@@ -44,7 +50,6 @@ int getOutF(Cb *cb) {
 
 // forward decl.
 static void toGoAndBack(Cb *cb);
-static int onF(Cb *cb);
 
 /*
  From Ian Lance Taylor: in C11 stdatomic terms Go atomic.CompareAndSwap is like
@@ -86,7 +91,7 @@ void duplexCb(Cb *cb, void *out, int *onF, void *in, int inF) {
 
 
 static void toGoAndBack(Cb *cb) {
-	_Atomic uint32_t * gp = &cb->inGo;
+	_Atomic uint32_t * gp = &(cb->inGo);
 	uint32_t b; 
 	for (;;) {
 		b = atomic_load_explicit(gp, memory_order_acquire);
@@ -95,6 +100,7 @@ static void toGoAndBack(Cb *cb) {
 			// can't happen unless the underlying API executes callbacks
 			// on more than one thread, as anyhow the current thread
 			// is in this function (no setjmp/longjmp).
+			fprintf(stderr, "toGoAndBack: %d > 0, does the API guarantee one callback at a time?\n", b);
 			continue;
 		}
 		if (atomic_compare_exchange_weak_explicit(gp, &b, b+1, memory_order_acq_rel, memory_order_relaxed)) {
@@ -103,9 +109,28 @@ static void toGoAndBack(Cb *cb) {
 	}
 	b++;
 	uint32_t cmp;
-	for (;;) {
+	int i;
+	for (i=1; i<=1000000;i++) {
 		cmp = atomic_load_explicit(gp, memory_order_acquire);
 		if (cmp != b) {
+			return;
+		}
+		if (i >= 1000 && i%50 == 0) {
+			// sleep is 1us, but involves syscall so system latency is involved.
+			// avoid as much as possible without entirely eating the CPU.
+			// TBD(wsc) make this buffer size real time dependent rather than by cycle
+			// counts.  If the buffer time is large, then we can sleep as in 
+			// cb.go, otherwise either we're in a slack time or contention is causing the 
+			// atomic to fail and it might help to back off.
+			nanosleep(&cb->time, NULL);
+		}
+	}
+	// paranoid code to reset state to as if Go was running properly
+	// equivalent of libsio.ErrCApiLost
+	fprintf(stderr, "atomic failed after 1000000 tries (resetting), did Go die?\n");
+	while (b>0) {
+		b = atomic_load_explicit(gp, memory_order_acquire);
+		if (atomic_compare_exchange_weak_explicit(gp, &b, 0, memory_order_acq_rel, memory_order_relaxed)) {
 			break;
 		}
 	}

libsio/cb.go

@@ -6,9 +6,11 @@ package libsio
 // #include "cb.h"
 import "C"
 import (
+	"errors"
 	"io"
 	"runtime"
 	"sync/atomic"
+	"time"
 	"unsafe"
 
 	"zikichombo.org/sound"
@@ -24,7 +26,7 @@ import (
 // as well:
 //
 //  1. The synchronisation mechanism here assumes that at most one C callback
-//  thread is executing a callback at a time, so as to avoid syscalls.
+//  thread is executing a callback at a time.
 //
 //  2. The C API must accept configuration by buffer size and
 //  never present the user with a buffer which exceeds this size.
@@ -35,7 +37,8 @@ import (
 //    - for input, there will be latency and CPU overhead
 //    - for output, the C API must allow the callback to inform the
 //      underlying system of the actual number of frames provided, even
-//      for non EOF conditions.
+//      for non EOF conditions.  Normally, this means the C API has
+//      latency associated with alignment.
 //
 //  4. For best reliability, the Go code should be run on a thread with the same
 //     priority as the C API.
@@ -70,18 +73,40 @@ type Cb struct {
 	// latency and cpu overhead, there is nothing that can be done as any regular alignment
 	// of bursts of irregular length data will have this effect.
 	over []float64
+
+	// time tracking
+	lastTime time.Time
+	bufDur   time.Duration
 }
 
 func NewCb(v sound.Form, sco sample.Codec, b int) *Cb {
 	return &Cb{
-		Form: v,
-		sco:  sco,
-		bsz:  b,
-		il:   cil.New(v.Channels(), b),
-		over: make([]float64, b),
-		c:    C.newCb(C.int(b))}
+		Form:   v,
+		sco:    sco,
+		bsz:    b,
+		il:     cil.New(v.Channels(), b),
+		over:   make([]float64, 0, b),
+		c:      C.newCb(C.int(b)),
+		bufDur: time.Duration(b) * v.SampleRate().Period()}
 }
 
+const (
+	// amount of slack we give between ask for wake up and
+	// pseudo-spin.  guestimated for general OS scheduling
+	// latency of worst case 1 preempting task + general Go GC
+	// latency.
+	sleepSlack = 5 * time.Millisecond
+
+	// nb of times to try an atomic before defaulting to
+	// runtime.Gosched, as the later might on some systems
+	// and some circumstances invoke a syscall.
+	atomicTryLen = 10
+
+	// max number of tries before we assume something killed
+	// the C thread
+	atomicTryLim = 100000000
+)
+
 func (r *Cb) Close() error {
 	C.closeCb(r.c)
 	C.freeCb(r.c)
@@ -106,19 +131,28 @@ func (r *Cb) Receive(d []float64) (int, error) {
 		start = len(r.over) / nC
 		r.over = r.over[:0]
 	}
+	r.maybeSleep()
 
 	var sl []float64 // per cb subslice of d
 	addr := (*uint32)(unsafe.Pointer(&r.c.inGo))
 	bps := r.sco.Bytes()
 	var nf, onf int  // frame counter and overlap frame count
 	var cbBuf []byte // cast from C pointer callback data
+	orgTime := r.lastTime
 	for start < nF {
-		r.fromC(addr)
+		if err := r.fromC(addr); err != nil {
+			return 0, ErrCApiLost
+		}
+		if orgTime == r.lastTime {
+			r.lastTime = time.Now()
+		}
 
 		nf = int(r.c.inF)
 		if nf == 0 {
 			r.toC(addr)
-			break
+			if err := r.toC(addr); err != nil {
+				return 0, ErrCApiLost
+			}
 		}
 
 		// in case the C cb doesn't align to the buffer size
@@ -140,14 +174,17 @@ func (r *Cb) Receive(d []float64) (int, error) {
 			r.sco.Decode(r.over, cbBuf[nf*bps*nC:])
 		}
 
-		r.toC(addr)
+		if err := r.toC(addr); err != nil {
+			return 0, ErrCApiLost
+		}
 		start += nf
 	}
 
 	r.il.Deinter(d[:start*nC])
 	return start, nil
 }
 
+// Send is as in sound.Sink.Send
 func (r *Cb) Send(d []float64) error {
 	N := len(d)
 	nC := r.Channels()
@@ -166,24 +203,34 @@ func (r *Cb) Send(d []float64) error {
 	bps := r.sco.Bytes()
 	var nf int
 	var cbBuf []byte
+	orgTime := r.lastTime
 	for start < nF {
-		r.fromC(addr)
+		if err := r.fromC(addr); err != nil {
+			return ErrCApiLost
+		}
+		if orgTime == r.lastTime {
+			r.lastTime = time.Now()
+		}
 		// get the slice at buffer size
 		nf = int(r.c.outF)
 		if nf == 0 {
-			r.toC(addr)
+			if err := r.toC(addr); err != nil {
+				return ErrCApiLost
+			}
 			return io.EOF
 		}
 		if start+nf > nF {
 			nf = nF - start
 		}
 		sl = d[start*nC : (start+nf)*nC]
 		// "render"
-		cbBuf = (*[1 << 30]byte)(unsafe.Pointer(C.getIn(r.c)))[:nf*bps*nC]
+		cbBuf = (*[1 << 30]byte)(unsafe.Pointer(r.c.out))[:nf*bps*nC]
 		r.sco.Encode(cbBuf, sl)
 		// tell the API about any truncation that happened.
 		r.c.outF = C.int(nf)
-		r.toC(addr)
+		if err := r.toC(addr); err != nil {
+			return ErrCApiLost
+		}
 		start += nf
 	}
 	return nil
@@ -193,29 +240,60 @@ func (r *Cb) SendReceive(out, in []float64) (int, error) {
 	return 0, nil
 }
 
-func (r *Cb) setOutF(nf int) {
-	addr := (*uint32)(unsafe.Pointer(&r.c.outF))
-	atomic.StoreUint32(addr, uint32(nf))
+func (r *Cb) maybeSleep() {
+	var t time.Time
+	if r.lastTime == t { // don't sleep on first call.
+		return
+	}
+	// sleep a conservative amount of time if
+	// latency is high enough (see sleepSlack)
+	passed := time.Since(r.lastTime)
+	if passed+sleepSlack < r.bufDur {
+		time.Sleep(r.bufDur - (passed + sleepSlack))
+	}
 }
 
-func (r *Cb) fromC(addr *uint32) {
+// ErrCApiLost can be returned if the thread running the C API
+// is somehow killed or the hardware causes the callbacks to
+// block.
+var ErrCApiLost = errors.New("too many atomic tries, C callbacks aren't happening.")
+
+func (r *Cb) fromC(addr *uint32) error {
 	var sz uint32
+	i := 0
 	for {
 		sz = atomic.LoadUint32(addr)
 		if sz != 0 {
-			return
+			return nil
+		}
+		i++
+		if i%atomicTryLen == 0 {
+			if i >= atomicTryLim {
+				return ErrCApiLost
+			}
+			// runtime.Gosched may invoke a syscall if many g's on m
+			// use sparingly
+			runtime.Gosched()
 		}
-		runtime.Gosched()
 	}
 }
 
-func (r *Cb) toC(addr *uint32) {
+func (r *Cb) toC(addr *uint32) error {
 	var sz uint32
+	i := 0
 	for {
 		sz = atomic.LoadUint32(addr)
 		if atomic.CompareAndSwapUint32(addr, sz, sz-1) {
-			return
+			return nil
+		}
+		i++
+		if i%atomicTryLen == 0 {
+			if i >= atomicTryLim {
+				return ErrCApiLost
+			}
+			// runtime.Gosched may invoke a syscall if many g's on m
+			// use sparingly
+			runtime.Gosched()
 		}
-		runtime.Gosched()
 	}
 }

libsio/cb.h

@@ -3,6 +3,7 @@
 
 #include <stdatomic.h>
 #include <stdint.h>
+#include <time.h>
 
 typedef struct Cb {
 	int bufSz;
@@ -11,6 +12,7 @@ typedef struct Cb {
 	int inF;
 	void * out;
 	int outF;
+	struct timespec time;
 } Cb;
 
 

libsio/cb_test.go

@@ -0,0 +1,45 @@
+package libsio
+
+import (
+	"fmt"
+	"testing"
+
+	"zikichombo.org/sound"
+	"zikichombo.org/sound/sample"
+)
+
+func TestCbCapture(t *testing.T) {
+	N := 1024
+	v := sound.MonoCd()
+	c := sample.SFloat32L
+	b := 512
+	cb := NewCb(v, c, b)
+	go runcbsCapture(cb, N, b, c.Bytes())
+	d := make([]float64, b)
+	for i := 0; i < N; i++ {
+		fmt.Printf("cb receive %d\n", i)
+		n, err := cb.Receive(d)
+		if err != nil {
+			t.Error(err)
+		} else if n != b {
+			t.Errorf("expected %d got %d\n", b, n)
+		}
+	}
+}
+
+func TestCbPlay(t *testing.T) {
+	N := 1024
+	v := sound.MonoCd()
+	c := sample.SFloat32L
+	b := 512
+	cb := NewCb(v, c, b)
+	go runcbsPlay(cb, N, b, c.Bytes())
+	d := make([]float64, b)
+	for i := 0; i < N; i++ {
+		fmt.Printf("cb send %d\n", i)
+		err := cb.Send(d)
+		if err != nil {
+			t.Error(err)
+		}
+	}
+}

libsio/doc.go

@@ -1,7 +1,7 @@
 // Copyright 2018 The ZikiChombo Authors. All rights reserved.  Use of this source
 // code is governed by a license that can be found in the License file.
 
-// Package libsio provides support for the different ports.
+// Package libsio provides some support for implementing the different ports.
 //
 // Package libsio is part of http://zikichombo.org
 package libsio /* import "zikichombo.org/sio/libsio" */