torProvider now gracefuly handles tor process fails

connectivity/torProvider.go

@@ -29,11 +29,16 @@ const (
 
 type onionListenService struct {
 	os *tor.OnionService
+	tp *torProvider
 }
 
 type torProvider struct {
-	t    *tor.Tor
-	lock sync.Mutex
+	t               *tor.Tor
+	appDirectory    string
+	bundeledTorPath string
+	lock            sync.Mutex
+	breakChan       chan bool
+	childListeners  map[string]*onionListenService
 }
 
 func (ols *onionListenService) AddressFull() string {
@@ -49,6 +54,7 @@ func (ols *onionListenService) Accept() (net.Conn, error) {
 }
 
 func (ols *onionListenService) Close() {
+	ols.tp.unregisterListener(ols.AddressIdentity())
 	ols.os.Close()
 }
 
@@ -88,16 +94,27 @@ func (tp *torProvider) WaitTillBootstrapped() {
 }
 
 func (tp *torProvider) Listen(identity PrivateKey, port int) (ListenService, error) {
+	if tp.t == nil {
+		return nil, errors.New("Tor is offline")
+	}
 	tp.lock.Lock()
 	defer tp.lock.Unlock()
 	conf := &tor.ListenConf{NoWait: true, Version3: true, Key: identity, RemotePorts: []int{port}, Detach: true, DiscardKey: true}
 	os, err := tp.t.Listen(nil, conf)
-	return &onionListenService{os}, err
+	if err != nil {
+		return nil, err
+	}
+	ols := &onionListenService{os: os, tp: tp}
+	tp.childListeners[ols.AddressIdentity()] = ols
+	return ols, nil
 }
 
 func (tp *torProvider) Open(hostname string) (net.Conn, string, error) {
 	tp.lock.Lock()
 	defer tp.lock.Unlock()
+	if tp.t == nil {
+		return nil, hostname, errors.New("Tor is offline")
+	}
 	torDailer, err := tp.t.Dialer(nil, &tor.DialConf{})
 	if err != nil {
 		return nil, "", err
@@ -119,15 +136,23 @@ func (tp *torProvider) Open(hostname string) (net.Conn, string, error) {
 }
 
 func (tp *torProvider) Close() {
+	for _, child := range tp.childListeners {
+		child.Close()
+	}
+
 	tp.lock.Lock()
 	defer tp.lock.Unlock()
-	tp.t.Close()
+	tp.breakChan <- true
+	if tp.t != nil {
+		tp.t.Close()
+	}
 }
 
 // StartTor creates/starts a Tor ACN and returns a usable ACN object
 func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 	dataDir := path.Join(appDirectory, "tor")
 	os.MkdirAll(dataDir, 0700)
+	tp := &torProvider{appDirectory: appDirectory, bundeledTorPath: bundledTorPath, childListeners: make(map[string]*onionListenService), breakChan: make(chan bool)}
 
 	// attempt connect to system tor
 	log.Printf("dialing system tor control port\n")
@@ -141,7 +166,9 @@ func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 			pinfo, err := controlport.ProtocolInfo()
 			if err == nil && minTorVersionReqs(pinfo.TorVersion) {
 				log.Println("OK version " + pinfo.TorVersion)
-				return createFromExisting(controlport, dataDir), nil
+				tp.t = createFromExisting(controlport, dataDir)
+				go tp.monitorRestart()
+				return tp, nil
 			}
 			controlport.Close()
 		}
@@ -151,7 +178,9 @@ func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 	if checkCmdlineTorVersion("tor") {
 		t, err := tor.Start(nil, &tor.StartConf{DataDir: dataDir, DebugWriter: nil})
 		if err == nil {
-			return &torProvider{t: t}, err
+			tp.t = t
+			go tp.monitorRestart()
+			return tp, nil
 		}
 		log.Printf("Error connecting to self-run system tor: %v\n", err)
 	}
@@ -163,12 +192,61 @@ func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 		if err != nil {
 			log.Printf("Error running bundled tor: %v\n", err)
 		}
-		return &torProvider{t: t}, err
+		tp.t = t
+		if err == nil {
+			go tp.monitorRestart()
+		}
+		return tp, err
 	}
 	return nil, errors.New("Could not connect to or start Tor that met requirments")
 }
 
-func createFromExisting(controlport *control.Conn, datadir string) ACN {
+func (tp *torProvider) unregisterListener(id string) {
+	tp.lock.Lock()
+	defer tp.lock.Unlock()
+	delete(tp.childListeners, id)
+}
+
+func (tp *torProvider) monitorRestart() {
+	for {
+		select {
+		case <-time.After(time.Duration(30 * time.Second)):
+			tp.lock.Lock()
+			if tp.t != nil {
+				_, err := tp.t.Control.GetInfo("version")
+
+				if err != nil {
+					tp.lock.Unlock()
+					for _, child := range tp.childListeners {
+						child.Close()
+					}
+					tp.lock.Lock()
+					tp.t.Close()
+					tp.t = nil
+				}
+			}
+
+			if tp.t == nil {
+				newACN, err := StartTor(tp.appDirectory, tp.bundeledTorPath)
+				if err == nil {
+					switch newTp := newACN.(type) {
+					case *torProvider:
+						tp.t = newTp.t
+						// startTor will have started a new monitorRestart thread
+						tp.lock.Unlock()
+						return
+					}
+				}
+			}
+			tp.lock.Unlock()
+		case <-tp.breakChan:
+			return
+		}
+
+	}
+}
+
+func createFromExisting(controlport *control.Conn, datadir string) *tor.Tor {
 	t := &tor.Tor{
 		Process:              nil,
 		Control:              controlport,
@@ -184,13 +262,12 @@ func createFromExisting(controlport *control.Conn, datadir string) ACN {
 
 	t.EnableNetwork(nil, true)
 
-	return &torProvider{t: t}
+	return t
 }
 
 func checkCmdlineTorVersion(torCmd string) bool {
 	cmd := exec.Command(torCmd, "--version")
 	out, err := cmd.CombinedOutput()
-	log.Println("cmdline tor version: " + string(out))
 	re := regexp.MustCompile("[0-1]\\.[0-9]\\.[0-9]\\.[0-9]")
 	sysTorVersion := re.Find(out)
 	log.Println("cmdline tor version: " + string(sysTorVersion))

testing/integration_test.go

@@ -98,11 +98,13 @@ func (bot *ChatEchoBot) ChatMessageAck(messageID uint32, accepted bool) {
 }
 
 func TestApplicationIntegration(t *testing.T) {
+	startGoRoutines := runtime.NumGoroutine()
+
 	acn, err := connectivity.StartTor(".", "")
 	if err != nil {
 		t.Fatalf("Could not start tor: %v", err)
 	}
-	startGoRoutines := runtime.NumGoroutine()
+
 	messageStack := &Messages{}
 	messageStack.Init()