feat(cli): make it possible to step through the scheduler

.gitignore

@@ -9,4 +9,5 @@ reports/
 vendor/
 result*
 __pycache__/
-bazel-*
+bazel-*
+.\#*

src/cli/cmd/root.go

@@ -24,6 +24,11 @@ func init() {
 	schedulerCmd.AddCommand(schedulerUpCmd)
 	schedulerCmd.AddCommand(schedulerDownCmd)
 	schedulerCmd.AddCommand(schedulerStatusCmd)
+	schedulerCmd.AddCommand(schedulerResetCmd)
+	schedulerCmd.AddCommand(schedulerLoadCmd)
+	schedulerCmd.AddCommand(schedulerRegisterCmd)
+	schedulerCmd.AddCommand(schedulerCreateRunCmd)
+	schedulerCmd.AddCommand(schedulerStepCmd)
 	rootCmd.AddCommand(generateCmd)
 	rootCmd.AddCommand(versionsCmd)
 }

src/cli/cmd/scheduler.go

@@ -112,3 +112,70 @@ var schedulerStatusCmd = &cobra.Command{
 		fmt.Println(string(json))
 	},
 }
+
+var schedulerResetCmd = &cobra.Command{
+	Use:   "reset",
+	Short: "Reset the state of the scheduler",
+	Long:  ``,
+	Args:  cobra.NoArgs,
+	Run: func(cmd *cobra.Command, args []string) {
+		lib.Reset()
+	},
+}
+
+var schedulerLoadCmd = &cobra.Command{
+	Use:   "load",
+	Short: "Load a test case into the scheduler",
+	Long:  ``,
+	Args:  cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		testId, err := lib.ParseTestId(args[0])
+		if err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+		queueSize := lib.LoadTest(testId)
+		fmt.Printf("Test case loaded, current queue size: %d\n", queueSize)
+	},
+}
+
+var schedulerRegisterCmd = &cobra.Command{
+	Use:   "register",
+	Short: "Register the executor in the scheduler",
+	Long:  ``,
+	Args:  cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		testId, err := lib.ParseTestId(args[0])
+		if err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+		lib.Register(testId)
+	},
+}
+
+var schedulerCreateRunCmd = &cobra.Command{
+	Use:   "create-run",
+	Short: "Create a new run id",
+	Long:  ``,
+	Args:  cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		testId, err := lib.ParseTestId(args[0])
+		if err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+		runId := lib.CreateRun(testId)
+		fmt.Printf("Created run id: %v\n", runId)
+	},
+}
+
+var schedulerStepCmd = &cobra.Command{
+	Use:   "step",
+	Short: "Execute a single step of the currently loaded test case",
+	Long:  ``,
+	Args:  cobra.NoArgs,
+	Run: func(cmd *cobra.Command, args []string) {
+		fmt.Println(string(lib.Step()))
+	},
+}

src/cli/default.nix

@@ -20,7 +20,7 @@ buildGoModule rec {
 
   # This hash should be the output of:
   #   go mod vendor && nix-hash --base32 --type sha256 vendor
-  vendorSha256 = "0d7nwxx0i834w0kvixl4map8wb6kw4cz30ckvk394cxm4sblzipd";
+  vendorSha256 = "12bc4m3pqn8x6r2kik5sawgdncjj6sz9yfsjqi2sp6w098qw51ky";
 
   buildFlagsArray =
     [ "-ldflags=-X main.version=${lib.commitIdFromGitRepo ./../../.git}" ];
@@ -30,8 +30,8 @@ buildGoModule rec {
     # that's where `go.mod` says to go look for it.
     cp -R ${detsysLib}/src ../lib
 
-    # Static linking.
-    export CGO_ENABLED=0
+    # Need cgo for sqlite3.
+    export CGO_ENABLED=1
   '';
 
   # Rename the resulting binary. (We can't use buildFlags with `-o`, because

src/cli/go.sum

@@ -105,6 +105,8 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
 github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
+github.com/mattn/go-sqlite3 v1.14.5 h1:1IdxlwTNazvbKJQSxoJ5/9ECbEeaTTyeU7sEAZ5KKTQ=
+github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=

src/db/migrations/1605014343_init.sql

@@ -64,6 +64,7 @@ CREATE TABLE IF NOT EXISTS heap_trace (
 CREATE TABLE IF NOT EXISTS deployment (
   test_id      INTEGER   NOT NULL,
   component    TEXT      NOT NULL,
+  type         TEXT      NOT NULL,
   args         JSON      NOT NULL,
   PRIMARY KEY(test_id, component),
   FOREIGN KEY(test_id) REFERENCES test(id));

src/executor/executor.go

@@ -113,15 +113,35 @@ func handleTimer(db *sql.DB, testId lib.TestId, topology Topology, m lib.Marshal
 	}
 }
 
-func Register(topology Topology) {
-	// TODO(stevan): Make executorUrl part of topology.
-	const executorUrl string = "http://localhost:3001/api/v1/"
+func handleInits(topology Topology, m lib.Marshaler) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
+		if r.Method != "GET" {
+			http.Error(w, jsonError("Method is not supported."),
+				http.StatusNotFound)
+			return
+		}
+
+		var inits []lib.Event
+		for component, reactor := range topology {
+			inits = append(inits,
+				lib.OutEventsToEvents(component, reactor.Init())...)
+		}
 
-	components := make([]string, 0, len(topology))
-	for component, _ := range topology {
-		components = append(components, component)
+		// Use `[]` for no events, rather than `null`, in the JSON encoding.
+		if inits == nil {
+			inits = []lib.Event{}
+		}
+
+		bs, err := json.Marshal(struct {
+			Events []lib.Event `json:"events"`
+		}{inits})
+		if err != nil {
+			panic(err)
+		}
+
+		fmt.Fprint(w, string(bs))
 	}
-	lib.RegisterExecutor(executorUrl, components)
 }
 
 func DeployWithComponentUpdate(srv *http.Server, testId lib.TestId, topology Topology, m lib.Marshaler, cu ComponentUpdate) {
@@ -133,35 +153,59 @@ func DeployWithComponentUpdate(srv *http.Server, testId lib.TestId, topology Top
 	mux.HandleFunc("/api/v1/event", handler(db, testId, topology, m, cu))
 	mux.HandleFunc("/api/v1/tick", handleTick(topology, m, cu))
 	mux.HandleFunc("/api/v1/timer", handleTimer(db, testId, topology, m, cu))
+	mux.HandleFunc("/api/v1/inits", handleInits(topology, m))
 	srv.Addr = ":3001"
 	srv.Handler = mux
 	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
 		panic(err)
 	}
-
-	var inits [][]lib.OutEvent
-	for _, reactor := range topology {
-		oevs := reactor.Init()
-		inits = append(inits, oevs)
-	}
-	// TODO(stevan): Randomise the order of the init events for better
-	// coverage? This should be done in the Scheduler using the seed to
-	// avoid making the Executor non-deterministic.
-	for _, oevs := range inits {
-		lib.EnqueueInitEvents(oevs)
-	}
 }
 
 func Deploy(srv *http.Server, testId lib.TestId, topology Topology, m lib.Marshaler) {
 	DeployWithComponentUpdate(srv, testId, topology, m, func(string, time.Time) {})
 }
 
-func DeployRaw(srv *http.Server, testId lib.TestId, topology map[string]string, m lib.Marshaler, constructor func(string) lib.Reactor) {
-	topologyCooked := make(Topology, len(topology))
-	for name, component := range topology {
-		topologyCooked[name] = constructor(component)
+func topologyFromDeployment(testId lib.TestId, constructor func(string) lib.Reactor) (Topology, error) {
+	query := fmt.Sprintf(`SELECT component,type
+                              FROM deployment
+                              WHERE test_id = %d`, testId.TestId)
+
+	db := lib.OpenDB()
+	defer db.Close()
+
+	rows, err := db.Query(query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	topologyRaw := make(map[string]string)
+	type Column struct {
+		Component string
+		Type      string
 	}
+	for rows.Next() {
+		column := Column{}
+		err := rows.Scan(&column.Component, &column.Type)
+		if err != nil {
+			return nil, err
+		}
+		topologyRaw[column.Component] = column.Type
+	}
+
+	topologyCooked := make(Topology)
+	for component, typ := range topologyRaw {
+		topologyCooked[component] = constructor(typ)
+	}
+	return topologyCooked, nil
+}
 
+func DeployRaw(srv *http.Server, testId lib.TestId, m lib.Marshaler, constructor func(string) lib.Reactor) {
+	topologyCooked, err := topologyFromDeployment(testId, constructor)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Printf("Deploying topology: %+v\n", topologyCooked)
 	Deploy(srv, testId, topologyCooked, m)
 }
 
@@ -257,7 +301,7 @@ func (e *Executor) Deploy(srv *http.Server) {
 }
 
 func (e *Executor) Register() {
-	Register(e.topology)
+	lib.Register(e.testId)
 }
 
 func (e *Executor) Reset(runId lib.RunId) {

src/generator/detsys-generator.sh

@@ -17,15 +17,15 @@ INSERT INTO agenda (test_id, id, kind, event, args, \`from\`, \`to\`, at)
 VALUES
   (${TEST_ID}, 0, "invoke", "write", '{"value": 1}', "client:0", "frontend", "1970-01-01T00:00:00Z"),
   (${TEST_ID}, 1, "invoke", "read",  "{}",           "client:0", "frontend", "1970-01-01T00:00:10Z");
-INSERT INTO deployment VALUES(${TEST_ID}, "frontend", '{"inFlight":{},"inFlightSessionToClient":{},"nextSessionId":0}');
-INSERT INTO deployment VALUES(${TEST_ID}, "register1", '{"value":[]}');
-INSERT INTO deployment VALUES(${TEST_ID}, "register2", '{"value":[]}');
+INSERT INTO deployment VALUES(${TEST_ID}, "frontend", "frontend", '{"inFlight":{},"inFlightSessionToClient":{},"nextSessionId":0}');
+INSERT INTO deployment VALUES(${TEST_ID}, "register1", "register", '{"value":[]}');
+INSERT INTO deployment VALUES(${TEST_ID}, "register2", "register", '{"value":[]}');
 EOF
 elif [ "${TEST}" == "broadcast" ]; then
     sqlite3 "${DETSYS_DB}" <<EOF
-INSERT INTO deployment VALUES(${TEST_ID}, "A", '{"log":"Hello world!","neighbours":{}}');
-INSERT INTO deployment VALUES(${TEST_ID}, "B", '{"log":"","neighbours":{}}');
-INSERT INTO deployment VALUES(${TEST_ID}, "C", '{"log":"","neighbours":{}}');
+INSERT INTO deployment VALUES(${TEST_ID}, "A", "node", '{"log":"Hello world!","neighbours":{}}');
+INSERT INTO deployment VALUES(${TEST_ID}, "B", "node", '{"log":"","neighbours":{}}');
+INSERT INTO deployment VALUES(${TEST_ID}, "C", "node", '{"log":"","neighbours":{}}');
 EOF
 fi
 

src/lib/default.nix

@@ -11,9 +11,9 @@ in
 buildGoModule rec {
   pname = "lib";
   version = "latest";
-  goPackagePath = "github.com/symbiont-io/detsys-testkit/${pname}";
+  goPackagePath = "github.com/symbiont-io/detsys-testkit/src/${pname}";
   src = gitignoreSource ./.;
-  vendorSha256 = null;
+  vendorSha256 = "1dq4w81g3s26chpfqad3mqpybg4900646h2b2k6xsz295ds9qk7q";
 
   postInstall = ''
     mkdir -p $out

src/lib/go.mod

@@ -1,3 +1,5 @@
 module github.com/symbiont-io/detsys-testkit/src/lib
 
 go 1.15
+
+require github.com/mattn/go-sqlite3 v1.14.5

src/lib/go.sum

@@ -0,0 +1,2 @@
+github.com/mattn/go-sqlite3 v1.14.5 h1:1IdxlwTNazvbKJQSxoJ5/9ECbEeaTTyeU7sEAZ5KKTQ=
+github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=

src/lib/marshaler.go

@@ -104,7 +104,7 @@ type event interface{ IsEvent() }
 func (_ unscheduledEvent) IsEvent() {}
 func (_ timerEvent) IsEvent()       {}
 
-func MarshalUnscheduledEvents(from string, oevs []OutEvent) json.RawMessage {
+func OutEventsToEvents(from string, oevs []OutEvent) []event {
 	usevs := make([]event, len(oevs))
 	for index, oev := range oevs {
 		var event event
@@ -137,6 +137,11 @@ func MarshalUnscheduledEvents(from string, oevs []OutEvent) json.RawMessage {
 		}
 		usevs[index] = event
 	}
+	return usevs
+}
+
+func MarshalUnscheduledEvents(from string, oevs []OutEvent) json.RawMessage {
+	usevs := OutEventsToEvents(from, oevs)
 	bs, err := json.Marshal(struct {
 		Events []event `json:"events"`
 	}{usevs})