You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mynewt.apache.org by cc...@apache.org on 2017/04/11 23:37:31 UTC
incubator-mynewt-newtmgr git commit: nmxact - Ensure all fsm
goroutines done
Repository: incubator-mynewt-newtmgr
Updated Branches:
refs/heads/master 5fce9f8d2 -> b6629b106
nmxact - Ensure all fsm goroutines done
Project: http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/commit/b6629b10
Tree: http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/tree/b6629b10
Diff: http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/diff/b6629b10
Branch: refs/heads/master
Commit: b6629b106f4a1b74b13a44c8d93dbcc670523730
Parents: 5fce9f8
Author: Christopher Collins <cc...@apache.org>
Authored: Tue Apr 11 16:35:58 2017 -0700
Committer: Christopher Collins <cc...@apache.org>
Committed: Tue Apr 11 16:36:24 2017 -0700
----------------------------------------------------------------------
nmxact/example/ble_dual/ble_dual.go | 72 ++++++++++++++++----------------
nmxact/nmble/ble_fsm.go | 17 +++++++-
nmxact/nmxutil/nmxutil.go | 61 ++++++++++++---------------
3 files changed, 79 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/blob/b6629b10/nmxact/example/ble_dual/ble_dual.go
----------------------------------------------------------------------
diff --git a/nmxact/example/ble_dual/ble_dual.go b/nmxact/example/ble_dual/ble_dual.go
index 958aaa1..e4615fe 100644
--- a/nmxact/example/ble_dual/ble_dual.go
+++ b/nmxact/example/ble_dual/ble_dual.go
@@ -26,15 +26,18 @@ import (
"sync"
"syscall"
+ log "github.com/Sirupsen/logrus"
+
"mynewt.apache.org/newt/util"
"mynewt.apache.org/newtmgr/nmxact/bledefs"
"mynewt.apache.org/newtmgr/nmxact/nmble"
+ "mynewt.apache.org/newtmgr/nmxact/nmxutil"
"mynewt.apache.org/newtmgr/nmxact/sesn"
"mynewt.apache.org/newtmgr/nmxact/xact"
"mynewt.apache.org/newtmgr/nmxact/xport"
)
-func configExitHandler(x xport.Xport, s sesn.Sesn) {
+func configExitHandler(x xport.Xport) {
onExit := func() {
x.Stop()
}
@@ -95,6 +98,8 @@ func sendOne(s sesn.Sesn) {
}
func main() {
+ nmxutil.SetLogLevel(log.InfoLevel)
+
// Initialize the BLE transport.
params := nmble.NewXportCfg()
params.SockPath = "/tmp/blehostd-uds"
@@ -116,48 +121,45 @@ func main() {
}
defer x.Stop()
- // Prepare a BLE session:
- // * Plain NMP (not tunnelled over OIC).
- // * We use a random address.
- // * Peer has name "nimble-bleprph".
- sc1 := sesn.NewSesnCfg()
- sc1.MgmtProto = sesn.MGMT_PROTO_NMP
- sc1.Ble.OwnAddrType = bledefs.BLE_ADDR_TYPE_RANDOM
- sc1.Ble.PeerSpec = sesn.BlePeerSpecName("ccollins")
+ configExitHandler(x)
- s1, err := x.BuildSesn(sc1)
- if err != nil {
- fmt.Fprintf(os.Stderr, "error creating BLE session: %s1\n", err.Error())
- os.Exit(1)
+ peerNames := []string{
+ "ccollins",
+ "ccollins2",
+ "ccollins3",
}
- sc2 := sesn.NewSesnCfg()
- sc2.MgmtProto = sesn.MGMT_PROTO_NMP
- sc2.Ble.OwnAddrType = bledefs.BLE_ADDR_TYPE_RANDOM
- sc2.Ble.PeerSpec = sesn.BlePeerSpecName("ccollins2")
+ sesns := []sesn.Sesn{}
+ for _, n := range peerNames {
+ // Prepare a BLE session:
+ // * Plain NMP (not tunnelled over OIC).
+ // * We use a random address.
+ // * Peer has name "nimble-bleprph".
+ sc := sesn.NewSesnCfg()
+ sc.MgmtProto = sesn.MGMT_PROTO_NMP
+ sc.Ble.OwnAddrType = bledefs.BLE_ADDR_TYPE_RANDOM
+ sc.Ble.PeerSpec = sesn.BlePeerSpecName(n)
+
+ s, err := x.BuildSesn(sc)
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "error creating BLE session: %s\n",
+ err.Error())
+ os.Exit(1)
+ }
- s2, err := x.BuildSesn(sc2)
- if err != nil {
- fmt.Fprintf(os.Stderr, "error creating BLE session: %s2\n", err.Error())
- os.Exit(1)
+ sesns = append(sesns, s)
}
- configExitHandler(x, s1)
-
var wg sync.WaitGroup
- wg.Add(1)
- go func() {
- for {
- sendOne(s1)
- }
- }()
- wg.Add(1)
- go func() {
- for {
- sendOne(s2)
- }
- }()
+ for _, s := range sesns {
+ wg.Add(1)
+ go func(x sesn.Sesn) {
+ for {
+ sendOne(x)
+ }
+ }(s)
+ }
wg.Wait()
}
http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/blob/b6629b10/nmxact/nmble/ble_fsm.go
----------------------------------------------------------------------
diff --git a/nmxact/nmble/ble_fsm.go b/nmxact/nmble/ble_fsm.go
index ef5e69d..9a01a70 100644
--- a/nmxact/nmble/ble_fsm.go
+++ b/nmxact/nmble/ble_fsm.go
@@ -49,6 +49,11 @@ const (
FSM_DISCONNECT_TYPE_REQUESTED
)
+type BleFsmListener struct {
+ bl *BleListener
+ abortChan chan struct{}
+}
+
type BleRxNmpFn func(data []byte)
type BleDisconnectFn func(dt BleFsmDisconnectType, peer BleDev, err error)
@@ -78,6 +83,7 @@ type BleFsm struct {
state BleSesnState
errFunnel nmxutil.ErrFunnel
id uint32
+ wg sync.WaitGroup
// Protects all accesses to the FSM state variable.
stateMtx sync.Mutex
@@ -145,6 +151,7 @@ func (bf *BleFsm) addBleListener(name string, base BleMsgBase) (
return nil, err
}
+ bf.wg.Add(1)
bf.bls[bl] = struct{}{}
return bl, nil
}
@@ -178,6 +185,8 @@ func (bf *BleFsm) removeBleListener(name string, base BleMsgBase) {
bl := bf.params.Bx.Bd.RemoveListener(base)
delete(bf.bls, bl)
+
+ bf.wg.Done()
}
func (bf *BleFsm) removeBleBaseListener(name string, base BleMsgBase) {
@@ -266,7 +275,10 @@ func (bf *BleFsm) processErr(err error) {
bf.peerDev = nil
- go bf.params.DisconnectCb(dt, peer, err)
+ bf.wg.Wait()
+
+ bf.errFunnel.Reset()
+ bf.params.DisconnectCb(dt, peer, err)
}
func (bf *BleFsm) connectListen(seq BleSeq) error {
@@ -766,9 +778,10 @@ func (bf *BleFsm) executeState() (bool, error) {
}
func (bf *BleFsm) startOnce() (bool, error) {
- bf.errFunnel.BlockUntilReset()
+ bf.errFunnel.Start()
if !bf.IsClosed() {
+ bf.errFunnel.Reset()
return false, nmxutil.NewSesnAlreadyOpenError(fmt.Sprintf(
"Attempt to open an already-open BLE session (state=%d)",
bf.getState()))
http://git-wip-us.apache.org/repos/asf/incubator-mynewt-newtmgr/blob/b6629b10/nmxact/nmxutil/nmxutil.go
----------------------------------------------------------------------
diff --git a/nmxact/nmxutil/nmxutil.go b/nmxact/nmxutil/nmxutil.go
index 4b232e9..2f967a0 100644
--- a/nmxact/nmxutil/nmxutil.go
+++ b/nmxact/nmxutil/nmxutil.go
@@ -51,17 +51,6 @@ func NewSingleResource() SingleResource {
}
}
-func (s *SingleResource) removeWaiter(waiter chan error) {
- s.mtx.Lock()
- defer s.mtx.Unlock()
-
- for i, w := range s.waitQueue {
- if w == waiter {
- s.waitQueue = append(s.waitQueue[:i], s.waitQueue[i+1:]...)
- }
- }
-}
-
func (s *SingleResource) Acquire() error {
s.mtx.Lock()
@@ -78,7 +67,6 @@ func (s *SingleResource) Acquire() error {
err := <-w
if err != nil {
- s.removeWaiter(w)
return err
}
@@ -89,6 +77,7 @@ func (s *SingleResource) Release() {
s.mtx.Lock()
if !s.acquired {
+ panic("SingleResource release without acquire")
s.mtx.Unlock()
return
}
@@ -114,6 +103,7 @@ func (s *SingleResource) Abort(err error) {
for _, w := range s.waitQueue {
w <- err
}
+ s.waitQueue = [](chan error){}
}
type ErrLessFn func(a error, b error) bool
@@ -130,6 +120,16 @@ type ErrFunnel struct {
resetMtx sync.Mutex
curErr error
errTimer *time.Timer
+ started bool
+}
+
+func (f *ErrFunnel) Start() {
+ f.resetMtx.Lock()
+
+ f.mtx.Lock()
+ defer f.mtx.Unlock()
+
+ f.started = true
}
func (f *ErrFunnel) Insert(err error) {
@@ -140,10 +140,11 @@ func (f *ErrFunnel) Insert(err error) {
f.mtx.Lock()
defer f.mtx.Unlock()
- if f.curErr == nil {
- // Subsequent use attempts will block until the funnel is inactive.
- f.resetMtx.Lock()
+ if !f.started {
+ panic("ErrFunnel insert without start")
+ }
+ if f.curErr == nil {
f.curErr = err
f.errTimer = time.AfterFunc(f.AccumDelay, func() {
f.timerExp()
@@ -159,35 +160,27 @@ func (f *ErrFunnel) Insert(err error) {
}
}
-func (f *ErrFunnel) resetNoLock() {
- if f.curErr != nil {
- f.curErr = nil
- f.errTimer.Stop()
- f.resetMtx.Unlock()
- }
-}
-
func (f *ErrFunnel) Reset() {
f.mtx.Lock()
defer f.mtx.Unlock()
- f.resetNoLock()
-}
-
-func (f *ErrFunnel) BlockUntilReset() {
- f.resetMtx.Lock()
- f.resetMtx.Unlock()
+ if f.started {
+ f.started = false
+ f.curErr = nil
+ f.errTimer.Stop()
+ f.resetMtx.Unlock()
+ }
}
func (f *ErrFunnel) timerExp() {
f.mtx.Lock()
- defer f.mtx.Unlock()
+ err := f.curErr
+ f.curErr = nil
+ f.mtx.Unlock()
- if f.curErr == nil {
+ if err == nil {
panic("ErrFunnel timer expired but no error")
}
- f.ProcCb(f.curErr)
-
- f.resetNoLock()
+ f.ProcCb(err)
}