Merge branch '152-internal-changes-representation-is-not-working-as-expected' into 'develop'

Resolve "Internal Changes representation is not working as expected"

See merge request !218

nucleus/change.go

 package nucleus
 
 import (
+	"context"
 	"fmt"
 	"os"
+	"sync"
 	"time"
 
 	ppb "code.fbi.h-da.de/danet/api/go/gosdn/pnd"
@@ -36,16 +38,18 @@ func NewChange(device uuid.UUID, currentState ygot.GoStruct, change ygot.GoStruc
 	c := &Change{
 		cuid:          uuid.New(),
 		duid:          device,
+		state:         ppb.Change_PENDING,
 		timestamp:     time.Now(),
 		previousState: currentState,
 		intendedState: change,
 		callback:      callback,
 	}
-	stateIn, stateOut, requestState, errChan := stateManager(c, changeTimeout)
+	stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+	stateIn, stateOut, errChan := stateManager(stateManagerCtx, c, changeTimeout)
 	c.stateIn = stateIn
 	c.stateOut = stateOut
-	c.requestState = requestState
 	c.errChan = errChan
+	c.stateManagerCancel = stateManagerCancel
 	return c
 }
 
@@ -54,16 +58,18 @@ func NewChange(device uuid.UUID, currentState ygot.GoStruct, change ygot.GoStruc
 // state. It keeps track if the state is committed and confirmed. A callback
 // exists to acess the proper transport for the changed OND
 type Change struct {
-	cuid          uuid.UUID
-	duid          uuid.UUID
-	timestamp     time.Time
-	previousState ygot.GoStruct
-	intendedState ygot.GoStruct
-	callback      func(ygot.GoStruct, ygot.GoStruct) error
-	errChan       <-chan error
-	requestState  chan<- bool
-	stateIn       chan<- ppb.Change_State
-	stateOut      <-chan ppb.Change_State
+	cuid               uuid.UUID
+	duid               uuid.UUID
+	state              ppb.Change_State
+	timestamp          time.Time
+	previousState      ygot.GoStruct
+	intendedState      ygot.GoStruct
+	callback           func(ygot.GoStruct, ygot.GoStruct) error
+	stateMu            sync.RWMutex
+	errChan            <-chan error
+	stateIn            chan<- ppb.Change_State
+	stateOut           <-chan ppb.Change_State
+	stateManagerCancel context.CancelFunc
 }
 
 // ID returns the Change's UUID
@@ -75,9 +81,7 @@ func (c *Change) ID() uuid.UUID {
 // and starts the timeout-timer for the Change. If the timer expires
 // the change is rolled back.
 func (c *Change) Commit() error {
-	if c.State() == ppb.Change_COMMITTED {
-		return fmt.Errorf("change %v already committed", c.cuid)
-	}
+	//TODO: check if already commited
 	c.stateIn <- ppb.Change_COMMITTED
 	select {
 	case err := <-c.errChan:
@@ -89,9 +93,7 @@ func (c *Change) Commit() error {
 
 // Confirm confirms a committed Change and stops the rollback timer.
 func (c *Change) Confirm() error {
-	if c.State() != ppb.Change_COMMITTED {
-		return fmt.Errorf("cannot confirm uncommitted change %v", c.cuid)
-	}
+	//TODO: check if already confirmed
 	c.stateIn <- ppb.Change_CONFIRMED
 	select {
 	case err := <-c.errChan:
@@ -108,51 +110,91 @@ func (c *Change) Age() time.Duration {
 
 // State returns the changes's state.
 func (c *Change) State() ppb.Change_State {
-	c.requestState <- true
-	return <-c.stateOut
+	c.stateMu.RLock()
+	state := c.state
+	c.stateMu.RUnlock()
+	return state
 }
 
-func stateManager(ch *Change, timeout time.Duration) (chan<- ppb.Change_State, <-chan ppb.Change_State, chan<- bool, <-chan error) {
+func stateManager(ctx context.Context, ch *Change, timeout time.Duration) (chan<- ppb.Change_State, <-chan ppb.Change_State, <-chan error) {
 	stateIn := make(chan ppb.Change_State)
 	stateOut := make(chan ppb.Change_State)
-	stateRequest := make(chan bool)
+	// A Goroutine, which is created while a new Change is initialized acts as
+	// the reciever for errorChan
 	errChan := make(chan error)
 	// create ticker and make it wait for 1 week
 	// workaround for delayed ticker start and ugly housekeeping
 	ticker := time.NewTicker(time.Hour * 7 * 24)
+	ticker.Stop()
 
 	go func() {
-		state := ppb.Change_PENDING
+	running:
 		for {
+			//TODO: priority select? should ticker have priority?
 			select {
 			case <-ticker.C:
+				state := ch.State()
+				if state == ppb.Change_CONFIRMED {
+					continue
+				}
 				err := ch.callback(ch.intendedState, ch.previousState)
 				if err != nil {
-					state = ppb.Change_INCONSISTENT
-					errChan <- err
+					ch.stateMu.Lock()
+					ch.state = ppb.Change_INCONSISTENT
+					ch.stateMu.Unlock()
+					log.Errorf("change %v timed out", ch.cuid)
+					log.Error(err)
+					continue
 				}
-				errChan <- fmt.Errorf("change %v timed out", ch.cuid)
+				// A rollback has been executed and the timer is stopped
+				ticker.Stop()
+				// TODO: keep the Change as pending, or remove it?
+				ch.stateMu.Lock()
+				ch.state = ppb.Change_PENDING
+				ch.stateMu.Unlock()
+				log.Errorf("change %v timed out", ch.cuid)
 			case s := <-stateIn:
 				switch s {
 				case ppb.Change_COMMITTED:
+					state := ch.State()
+					if state == ppb.Change_COMMITTED || state == ppb.Change_CONFIRMED {
+						errChan <- fmt.Errorf("change %v already %s", ch.cuid, state.String())
+						continue
+					}
 					// reset ticker to enable activate the change timeout
 					ticker.Reset(timeout)
 					err := ch.callback(ch.previousState, ch.intendedState)
 					if err != nil {
-						state = ppb.Change_INCONSISTENT
+						ch.stateMu.Lock()
+						ch.state = ppb.Change_INCONSISTENT
+						ch.stateMu.Unlock()
 						errChan <- err
 						continue
 					}
-					state = ppb.Change_COMMITTED
+					ch.stateMu.Lock()
+					ch.state = ppb.Change_COMMITTED
+					ch.stateMu.Unlock()
 					stateOut <- state
 				case ppb.Change_CONFIRMED:
-					state = ppb.Change_CONFIRMED
+					state := ch.State()
+					if state != ppb.Change_COMMITTED {
+						errChan <- fmt.Errorf("cannot confirm uncommitted change %v", ch.cuid)
+						continue
+					}
+					// The change has been confirmed and the timer is stopped,
+					// since a rollback is not necessary anymore.
+					ch.stateMu.Lock()
+					ch.state = ppb.Change_CONFIRMED
+					ch.stateMu.Unlock()
 					stateOut <- state
+					ch.stateManagerCancel()
 				}
-			case <-stateRequest:
-				stateOut <- state
+			case <-ctx.Done():
+				ticker.Stop()
+				break running
 			}
 		}
+		log.Info("statemanager routine done for: ", ch.cuid)
 	}()
-	return stateIn, stateOut, stateRequest, errChan
+	return stateIn, stateOut, errChan
 }

nucleus/change_test.go

 package nucleus
 
 import (
+	"context"
 	"errors"
 	"reflect"
 	"sync"
@@ -50,11 +51,12 @@ func TestChange_CommitRollback(t *testing.T) {
 			return nil
 		},
 	}
-	stateIn, stateOut, requestState, errChan := stateManager(c, time.Millisecond*100)
+	stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+	stateIn, stateOut, errChan := stateManager(stateManagerCtx, c, time.Millisecond*100)
 	c.stateIn = stateIn
 	c.stateOut = stateOut
-	c.requestState = requestState
 	c.errChan = errChan
+	c.stateManagerCancel = stateManagerCancel
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -69,6 +71,7 @@ func TestChange_CommitRollback(t *testing.T) {
 		t.Errorf("Commit() = %v, want %v", got, want)
 	}
 	wg.Wait()
+	c.stateManagerCancel()
 }
 
 func TestChange_CommitRollbackError(t *testing.T) {
@@ -76,6 +79,7 @@ func TestChange_CommitRollbackError(t *testing.T) {
 	wg.Add(1)
 	wantErr := false
 	want := errors.New("this is an expected error")
+	rollbackErrChannel := make(chan error)
 	c := &Change{
 		cuid:          cuid,
 		duid:          did,
@@ -87,16 +91,17 @@ func TestChange_CommitRollbackError(t *testing.T) {
 			t.Logf("callback in test %v", t.Name())
 			switch hostname {
 			case rollbackHostname:
-				return errors.New("this is an expected error")
+				rollbackErrChannel <- errors.New("this is an expected error")
 			}
 			return nil
 		},
 	}
-	stateIn, stateOut, requestState, errChan := stateManager(c, time.Millisecond*100)
+	stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+	stateIn, stateOut, errChan := stateManager(stateManagerCtx, c, time.Millisecond*100)
 	c.stateIn = stateIn
 	c.stateOut = stateOut
-	c.requestState = requestState
 	c.errChan = errChan
+	c.stateManagerCancel = stateManagerCancel
 
 	go func() {
 		defer wg.Done()
@@ -106,11 +111,12 @@ func TestChange_CommitRollbackError(t *testing.T) {
 		}
 		time.Sleep(config.ChangeTimeout)
 	}()
-	got := <-c.errChan
+	got := <-rollbackErrChannel
 	if !reflect.DeepEqual(got, want) {
 		t.Errorf("Commit() = %v, want %v", got, want)
 	}
 	wg.Wait()
+	c.stateManagerCancel()
 }
 
 func TestChange_CommitError(t *testing.T) {
@@ -125,11 +131,12 @@ func TestChange_CommitError(t *testing.T) {
 			return errors.New("this is an expected error")
 		},
 	}
-	stateIn, stateOut, requestState, errChan := stateManager(c, time.Millisecond*100)
+	stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+	stateIn, stateOut, errChan := stateManager(stateManagerCtx, c, time.Millisecond*100)
 	c.stateIn = stateIn
 	c.stateOut = stateOut
-	c.requestState = requestState
 	c.errChan = errChan
+	c.stateManagerCancel = stateManagerCancel
 
 	time.Sleep(time.Millisecond * 10)
 	if err := c.Commit(); err == nil {
@@ -139,6 +146,7 @@ func TestChange_CommitError(t *testing.T) {
 	if !reflect.DeepEqual(got, want) {
 		t.Errorf("Commit() = %v, want %v", got, want)
 	}
+	c.stateManagerCancel()
 }
 
 func TestChange_Commit(t *testing.T) {
@@ -154,11 +162,12 @@ func TestChange_Commit(t *testing.T) {
 			return nil
 		},
 	}
-	stateIn, stateOut, requestState, errChan := stateManager(c, time.Millisecond*100)
+	stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+	stateIn, stateOut, errChan := stateManager(stateManagerCtx, c, time.Millisecond*100)
 	c.stateIn = stateIn
 	c.stateOut = stateOut
-	c.requestState = requestState
 	c.errChan = errChan
+	c.stateManagerCancel = stateManagerCancel
 
 	if err := c.Commit(); err != nil {
 		t.Errorf("Commit() error = %v", err)
@@ -204,11 +213,12 @@ func TestChange_Confirm(t *testing.T) {
 					return nil
 				},
 			}
-			stateIn, stateOut, requestState, errChan := stateManager(c, time.Millisecond*100)
+			stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+			stateIn, stateOut, errChan := stateManager(stateManagerCtx, c, time.Millisecond*100)
 			c.stateIn = stateIn
 			c.stateOut = stateOut
-			c.requestState = requestState
 			c.errChan = errChan
+			c.stateManagerCancel = stateManagerCancel
 
 			if tt.name == "committed" {
 				if err := c.Commit(); err != nil {
@@ -218,6 +228,7 @@ func TestChange_Confirm(t *testing.T) {
 			if err := c.Confirm(); (err != nil) != tt.wantErr {
 				t.Errorf("Confirm() error = %v, wantErr %v", err, tt.wantErr)
 			}
+			c.stateManagerCancel()
 		})
 	}
 }
@@ -269,8 +280,9 @@ func TestChange_State(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			testName := t.Name()
 			callback := func(first ygot.GoStruct, second ygot.GoStruct) error {
-				t.Logf("callback in test %v", t.Name())
+				t.Logf("callback in test %v", testName)
 				return nil
 			}
 			c := NewChange(did, rollbackDevice, commitDevice, callback)
@@ -287,6 +299,7 @@ func TestChange_State(t *testing.T) {
 			if got := c.State(); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("Change.State() = %v, want %v", got, tt.want)
 			}
+			c.stateManagerCancel()
 		})
 	}
 }