status: even with 3 failing tests, this is the most useful current co…

…nfiguration. In this commit I shift to mangos to avoid the deadlock bug in C-libnanomsg. However this means there are three currently failing testts: TestSubmitDoesNotLeaveFileHandlesOpen001, TestSendToUnboundAddressShouldTimeout005, TestWorkerTimeout. I'm discussing these test suite failures with mangos' author Garrett at the moment. They result from slightly different implimentation semantics. Moreover they do not prevent goq from being used and useful under mangos. You can get full test suite to pass by switching nn to import C-nanomsg, but then the deadlock bug will prevent work from getting done when the server hangs.

goq.go

@@ -17,9 +17,9 @@ import (
 	"sync"
 	"time"
 
+	schema "github.com/glycerine/goq/schema"
 	//nn "github.com/glycerine/go-nanomsg"
 	nn "github.com/gdamore/mangos/compat"
-	schema "github.com/glycerine/goq/schema"
 )
 
 // In this model of work dispatch, there are three roles: submitter(s), a server, and worker(s).
@@ -836,7 +836,7 @@ func (js *JobServ) Start() {
 				js.CountDeaf++
 				resubJob, ok := js.RunQ[resubId]
 				if !ok {
-					// maybe it was cancelled in the meantime. panic(fmt.Sprintf("go resub for job id(%d) that isn't on our RunQ", resubId)
+					// maybe it was cancelled in the meantime. don't panic.
 					TSPrintf("**** [jobserver pid %d] got re-submit of job %d that is now not on our RunQ, so dropping it without re-queuing.\n", js.Pid, resubId)
 					continue
 				}

nextjob_test.go

@@ -5,8 +5,8 @@ import (
 	"os"
 	"testing"
 
-	cv "github.com/glycerine/goconvey/convey"
 	schema "github.com/glycerine/goq/schema"
+	cv "github.com/glycerine/goconvey/convey"
 )
 
 // next job test

sendtimeout_test.go

@@ -6,15 +6,16 @@ import (
 	"testing"
 	"time"
 
+	nn "github.com/gdamore/mangos/compat"
+	//nn "github.com/glycerine/go-nanomsg"
 	cv "github.com/glycerine/goconvey/convey"
-    nn "github.com/gdamore/mangos/compat"
 )
 
 //
 // simple test of mangos connect to unused address and send timeout
 //
 
-func TestSendShouldTimeout005(t *testing.T) {
+func TestSendToUnboundAddressShouldTimeout005(t *testing.T) {
 
 	cv.Convey("remotely, over nanomsg, a send to a non-existant address should timeout and fail", t, func() {
 
@@ -37,13 +38,15 @@ func TestSendShouldTimeout005(t *testing.T) {
 			cv.So(err, cv.ShouldNotEqual, nil)
 			return
 		}
-   	    cv.So(err, cv.ShouldNotEqual, nil)
+		// wrong? but err is nil.
+		//cv.So(err, cv.ShouldNotEqual, nil)
 		fmt.Printf("\n[pid %d] push socket made at '%s'.\n", os.Getpid(), unused_addr)
 
 		cy := []byte("hello")
-		_, err =push1.Send(cy, 0)
+		_, err = push1.Send(cy, 0)
 
+		// really should not be able to Send on a not-connected socket.
+		fmt.Printf("err was: '%s'\n", err) // nil under mangos. 'resource temporarily unavailable' when using C lib nanomsg.
 		cv.So(err, cv.ShouldNotEqual, nil)
-
 	})
 }

ser.go

@@ -8,8 +8,8 @@ import (
 	"io"
 	"time"
 
-	"github.com/glycerine/go-capnproto"
 	schema "github.com/glycerine/goq/schema"
+	"github.com/glycerine/go-capnproto"
 )
 
 func (js *JobServ) ServerToCapnp() (bytes.Buffer, *capn.Segment) {

signature_test.go

@@ -5,8 +5,8 @@ import (
 	"os"
 	"testing"
 
-	cv "github.com/glycerine/goconvey/convey"
 	schema "github.com/glycerine/goq/schema"
+	cv "github.com/glycerine/goconvey/convey"
 )
 
 // signature test

sub.go

@@ -5,9 +5,9 @@ import (
 	"os"
 	"time"
 
+	schema "github.com/glycerine/goq/schema"
 	//nn "github.com/glycerine/go-nanomsg"
 	nn "github.com/gdamore/mangos/compat"
-	schema "github.com/glycerine/goq/schema"
 )
 
 // Submitter represents all other queries beside those from workers.

wkto_test.go

@@ -63,8 +63,9 @@ func TestWorkerTimeout(t *testing.T) {
 			fmt.Printf("\n after worker.Destroy()\n")
 
 			// have to poll until everything gets done. Give ourselves 5 seconds.
-			const to = 5
-			timeout := time.After(to * time.Second)
+
+			to := time.Duration(cfg.SendTimeoutMsec) * 30 * time.Millisecond
+			timeout := time.After(to)
 			var deafcount int
 
 		OuterFor:
@@ -80,7 +81,7 @@ func TestWorkerTimeout(t *testing.T) {
 					}
 				case <-timeout:
 					cv.So(deafcount, cv.ShouldEqual, 1)
-					fmt.Printf("\nfailing test, no DeafChan 1 after %d seconds\n", to)
+					fmt.Printf("\nfailing test, no DeafChan 1 after... %v\n", to)
 					break OuterFor
 				}
 			}

worker.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"os"
 
+	schema "github.com/glycerine/goq/schema"
 	//nn "github.com/glycerine/go-nanomsg"
 	nn "github.com/gdamore/mangos/compat"
-	schema "github.com/glycerine/goq/schema"
 )
 
 // encapsulate the state that only NanomsgListener go routine should be touching