Krashanoff/websockets

backend/README.md

@@ -325,6 +325,24 @@ Status Code | Semantic
 401 | Unauthorized
 500 | Server error
 
+### `GET /live/<assignment_id>`
+
+#### Websocket Format
+
+This endpoint sends a continuous stream of the following object:
+
+```json
+{
+  "hidden": bool,
+  "testId": 0,
+  "testName": "name",
+  "score": 100.0,
+  "msg": "Error message or further information."
+}
+```
+
+If the `hidden` field is set to `true`, the `msg` and `testName` fields will be empty.
+
 ## Grading Scripts
 
 ### Output format

backend/go.mod

@@ -5,7 +5,10 @@ go 1.16
 require (
 	github.com/blockloop/scan v1.3.0
 	github.com/golang-jwt/jwt/v4 v4.2.0
+	github.com/google/uuid v1.3.0
 	github.com/labstack/echo/v4 v4.6.3
 	github.com/mattn/go-sqlite3 v1.14.12
+	github.com/stretchr/testify v1.7.0
 	golang.org/x/crypto v0.0.0-20220214200702-86341886e292
+	golang.org/x/net v0.0.0-20220225172249-27dd8689420f
 )

backend/go.sum

@@ -7,6 +7,8 @@ github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keL
 github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/golang-jwt/jwt/v4 v4.2.0 h1:besgBTC8w8HjP6NzQdxwKH9Z5oQMZ24ThTrHp3cZ8eU=
 github.com/golang-jwt/jwt/v4 v4.2.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/labstack/echo/v4 v4.6.3 h1:VhPuIZYxsbPmo4m9KAkMU/el2442eB7EBFFhNTTT9ac=
 github.com/labstack/echo/v4 v4.6.3/go.mod h1:Hk5OiHj0kDqmFq7aHe7eDqI7CUhuCrfpupQtLGGLm7A=
 github.com/labstack/gommon v0.3.1 h1:OomWaJXm7xR6L1HmEtGyQf26TEn7V6X88mktX9kee9o=
@@ -33,23 +35,27 @@ golang.org/x/crypto v0.0.0-20220214200702-86341886e292 h1:f+lwQ+GtmgoY+A2YaQxlSO
 golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210913180222-943fd674d43e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2 h1:CIJ76btIcR3eFI5EgSo6k1qKw9KJexJuRLI9G7Hp5wE=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20220225172249-27dd8689420f h1:oA4XRj0qtSt8Yo1Zms0CUlsT3KG69V2UGQWPBxujDmc=
+golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20211103235746-7861aae1554b h1:1VkfZQv42XQlA/jchYumAnv1UPo6RgF9rJFkTgZIxO4=
 golang.org/x/sys v0.0.0-20211103235746-7861aae1554b/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20201208040808-7e3f01d25324 h1:Hir2P/De0WpUhtrKGGjvSb2YxUgyZ7EFOSLIcSSpiwE=
 golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=

backend/grading/parser.go

@@ -23,7 +23,7 @@ func parseOutput(stdout io.Reader) ([]Result, error) {
 	for stdoutLines.Scan() {
 		hidden := false
 		testName := ""
-		resultScore := float64(-1)
+		resultScore := float64(0)
 		testIdStr := stdoutLines.Text()
 		testId, err := strconv.Atoi(testIdStr)
 		if err != nil {

backend/grading/parser_test.go

@@ -0,0 +1,22 @@
+package grading
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseOutput(t *testing.T) {
+	t.Run("BasicOutput", func(t *testing.T) {
+		results, err := parseOutput(strings.NewReader(`001
+HIDDEN
+NAME Test Name hi
+Some extra text
+SCORE 0.1 100`))
+		assert.NoError(t, err)
+		assert.NotEmpty(t, results)
+		assert.Len(t, results, 1)
+		assert.Equal(t, float64(10), results[0].Score)
+	})
+}

backend/grading/runner.go

@@ -3,18 +3,135 @@ package grading
 import (
 	"archive/tar"
 	"compress/gzip"
+	"context"
+	"database/sql"
 	"fmt"
 	"io"
 	"os"
-	"os/exec"
 	"path"
 	"strings"
+	"sync"
+	"time"
+
+	"github.com/blockloop/scan"
+	"github.com/google/uuid"
 )
 
+// Time to refresh results, in milliseconds.
+const REFRESH_MILLIS = 500
+
+// Multi-threaded job runner.
+type Runner struct {
+	store   *sql.DB
+	queue   chan jobWithID
+	running sync.Map
+}
+
+type jobWithID struct {
+	job Job
+	id  string
+}
+
+// Add a job to the runner, returning its associated job ID.
+func (r *Runner) Add(job Job) string {
+	withId := jobWithID{
+		job: job,
+		id:  uuid.NewString(),
+	}
+	r.queue <- withId
+	return withId.id
+}
+
+// Get a receive-only channel of results for a given job. If the job has
+// terminated, will return an already-closed channel of Results.
+func (r *Runner) Results(ctx context.Context, jobId string) <-chan Result {
+	output := make(chan Result, 5)
+	go func() {
+		conn, _ := r.store.Conn(ctx)
+		ticker := time.NewTicker(REFRESH_MILLIS * time.Millisecond)
+		for range ticker.C {
+			rows, err := conn.QueryContext(ctx, `
+			SELECT
+				hidden,
+				test_id,
+				test_name,
+				score,
+				message
+			FROM Submissions L
+			JOIN Results R
+			ON L.id = R.submission_id
+			WHERE id = $1
+			`, jobId)
+			if err != nil {
+				fmt.Println(err)
+				close(output)
+				return
+			}
+
+			var results []Result
+			if err := scan.Rows(&results, rows); err != nil {
+				fmt.Println(err)
+				close(output)
+				return
+			}
+			for _, result := range results {
+				output <- result
+			}
+		}
+	}()
+	return output
+}
+
+// Start a new work runner for controlling batch jobs.
+func Start(ctx context.Context, store *sql.DB) *Runner {
+	// Create a work queue for grading scripts, then spawn a task runner
+	// to execute grading script jobs in parallel.
+	occupied := make(chan bool, 5)
+	queue := make(chan jobWithID, 5)
+
+	runner := Runner{
+		store:   store,
+		queue:   queue,
+		running: sync.Map{},
+	}
+
+	go func() {
+		for jobAndID := range queue {
+			occupied <- true
+			conn, _ := store.Conn(ctx)
+			results := make(chan Result, 5)
+			go Grade(jobAndID.id, jobAndID.job, results)
+
+			// For each result of the grading script, write it back to the database.
+			go func(jobAndID jobWithID) {
+				for result := range results {
+					conn.ExecContext(ctx, `
+					INSERT INTO Results (
+						submission_id,
+						test_id,
+						hidden,
+						test_name,
+						score,
+						message
+					)
+					VALUES ($1, $2, $3, $4, $5, $6)
+					`, jobAndID.id,
+						result.TestID,
+						result.Hidden,
+						result.TestName,
+						result.Score,
+						result.Msg)
+				}
+			}(jobAndID)
+		}
+	}()
+
+	return &runner
+}
+
 // Channels are opened, fed objects from some other file.
-func Grade(job Job, jobQueue <-chan bool) {
-	defer func() { <-jobQueue }()
-	results := job.Results
+func Grade(id string, job Job, results chan<- Result) {
+	defer close(results)
 
 	// job.file will be hosted locally, somewhere.
 	// Not going to worry about it here.
@@ -27,7 +144,7 @@ func Grade(job Job, jobQueue <-chan bool) {
 
 	// Stick to bigmoney pattern for creating the temp assignment directory.
 	dirPfx := "./run"
-	dir := path.Join(dirPfx, dirPfx+"-"+strings.ReplaceAll(job.ID, "-", ""))
+	dir := path.Join(dirPfx, dirPfx+"-"+strings.ReplaceAll(id, "-", ""))
 	if err := os.MkdirAll(dir, 0640); err != nil {
 		panic(err)
 	}
@@ -47,7 +164,7 @@ func Grade(job Job, jobQueue <-chan bool) {
 	}
 
 	// Execute grading script driver.
-	cmd := exec.Command(job.Script, dir)
+	cmd := job.Script
 	output, err := cmd.StdoutPipe()
 	if err != nil {
 		fmt.Println(err)

backend/grading/types.go

@@ -2,24 +2,21 @@ package grading
 
 import (
 	"io"
+	"os/exec"
 )
 
 type Job struct {
-	ID   string
 	File io.Reader
 
 	// Path to the script to run on the Job.
-	Script string
-
-	// Channel to send results down to the Job requester.
-	Results chan<- Result
+	Script *exec.Cmd
 }
 
 // Result from a single test case in the grading script.
 type Result struct {
-	Hidden   bool
-	TestID   int
-	TestName string
-	Score    float64
-	Msg      string
+	Hidden   bool    `json:"hidden"`
+	TestID   int     `json:"testId"`
+	TestName string  `json:"testName"`
+	Score    float64 `json:"score"`
+	Msg      string  `json:"msg"`
 }

backend/handler/assignment.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"encoding/json"
 	"io"
 	"net/http"
 	"os"
@@ -9,7 +10,9 @@ import (
 	"time"
 
 	"github.com/blockloop/scan"
+	"github.com/cs130-w22/Group-A3/backend/grading"
 	"github.com/labstack/echo/v4"
+	"golang.org/x/net/websocket"
 )
 
 // Create a new assignment.
@@ -77,7 +80,10 @@ func UploadSubmission(cc echo.Context) error {
 	file, _ := submittedFile.Open()
 	defer file.Close()
 
-	return c.NoContent(http.StatusCreated)
+	return c.JSON(http.StatusCreated, echo.Map{
+		"id": c.Runner.Add(grading.Job{
+			File: file,
+		})})
 }
 
 // Get information about an assignment.
@@ -133,3 +139,27 @@ func GetAssignment(cc echo.Context) error {
 		"submissions": submissions,
 	})
 }
+
+// Get a live feed of results for the given submission ID.
+func LiveResults(cc echo.Context) error {
+	c := cc.(*Context)
+
+	websocket.Handler(func(ws *websocket.Conn) {
+		defer ws.Close()
+
+		// Receive submission ID
+		submissionId := c.Param("submissionId")
+
+		// Fetch the results channel and begin relaying results.
+		for result := range c.Runner.Results(c, submissionId) {
+			bytes, err := json.Marshal(result)
+			if err != nil {
+				c.Logger().Warn(err)
+			}
+			if err := websocket.Message.Send(ws, bytes); err != nil {
+				c.Logger().Error(err)
+			}
+		}
+	}).ServeHTTP(c.Response(), c.Request())
+	return nil
+}

backend/handler/class.go

@@ -23,7 +23,7 @@ func GetClass(cc echo.Context) error {
 	}
 
 	// Get the class' general information.
-	className := ""
+	className := "my class"
 	if err := c.Conn.QueryRowContext(c, `
 	SELECT name
 	FROM Courses
@@ -44,6 +44,7 @@ func GetClass(cc echo.Context) error {
 		c.Logger().Error(err)
 		return c.NoContent(http.StatusInternalServerError)
 	}
+	c.Logger().Error("end")
 	if err := scan.Rows(&assignments, rows); err != nil {
 		return c.NoContent(http.StatusInternalServerError)
 	}

backend/handler/context.go

@@ -17,9 +17,9 @@ import (
 // are logged in.
 type Context struct {
 	echo.Context
-	Conn     *sql.Conn
-	Claims   *jwt.Claims
-	JobQueue chan<- grading.Job
+	Conn   *sql.Conn
+	Claims *jwt.Claims
+	Runner *grading.Runner
 }
 
 func (c Context) Deadline() (time.Time, bool) {