⚡ improve concurrency tolerance a bit by extending channel capacity

judge/worker.go

@@ -112,7 +112,7 @@ func (w *worker) Enqueue(sub types.Submission, subscribe bool, path string, f io
 		l: list.New(),
 	})
 	if subscribe {
-		c, element = w.Subscribe(sub.ID)
+		c, element = w.Subscribe(sub)
 	}
 	return c, element, w.p.Push(sub, false)
 }
@@ -216,14 +216,15 @@ func (w *worker) DestroySubscribers(id uint32) {
 	}
 }
 
-func (w *worker) Subscribe(id uint32) (chan interface{}, *list.Element) {
-	subscribers, ok := w.sm.Load(id)
+func (w *worker) Subscribe(sub types.Submission) (chan interface{}, *list.Element) {
+	subscribers, ok := w.sm.Load(sub.ID)
 	if !ok {
 		return nil, nil
 	}
 	subscribers.m.Lock()
 	defer subscribers.m.Unlock()
-	c := make(chan interface{}, 1)
+	// maximum number of messages = compile announcement + n test case announcements + n test case results + final result = (n + 1) * 2
+	c := make(chan interface{}, (sub.TestCount+1)*2)
 	return c, subscribers.l.PushBack(c)
 }
 

routes/problems/submit.go

@@ -113,12 +113,18 @@ func Submit(ctx *http.Context) http.Response {
 	}
 	if shouldStream && res != nil {
 		stream := ctx.StreamResponse()
-		go func() {
-			<-ctx.Request().Context().Done()
-			judge.Worker.Unsubscribe(sub.ID, element)
-		}()
-		for r := range res {
-			stream.Write(r)
+		done := ctx.Request().Context().Done()
+		for {
+			select {
+			case <-done:
+				judge.Worker.Unsubscribe(sub.ID, element)
+				return nil
+			case r, more := <-res:
+				if !more {
+					return nil
+				}
+				stream.Write(r)
+			}
 		}
 		return ctx.Success()
 	} else {

server/server.go

@@ -31,7 +31,7 @@ func createHandler(handler http.Handler) echo.HandlerFunc {
 func Register(e *echo.Echo) {
 	g := e.Group("/api", middlewares.Authentication())
 	if config.Config.Blizzard.RateLimit > 0 {
-		g.Use(middlewares.RateLimit())
+		//g.Use(middlewares.RateLimit())
 	}
 	if config.Config.Debug {
 		g.Use(middleware.BodyDump(func(c echo.Context, req, res []byte) {