X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b0ba939812720869fca0a75b07d42518d4953345..aa29646b94556f244a2a8ef5f422d9e91b95ffe0:/services/ws/handler.go diff --git a/services/ws/handler.go b/services/ws/handler.go index d2c119acfe..8b6e9b9772 100644 --- a/services/ws/handler.go +++ b/services/ws/handler.go @@ -1,4 +1,8 @@ -package main +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + +package ws import ( "context" @@ -6,7 +10,9 @@ import ( "sync" "time" - "git.curoverse.com/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/stats" + "github.com/sirupsen/logrus" ) type handler struct { @@ -15,7 +21,7 @@ type handler struct { QueueSize int mtx sync.Mutex - queues map[chan interface{}]struct{} + lastDelay map[chan interface{}]stats.Duration setupOnce sync.Once } @@ -26,32 +32,32 @@ type handlerStats struct { EventCount uint64 } -func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsConn, chan<- interface{}) (session, error)) (stats handlerStats) { +func (h *handler) Handle(ws wsConn, logger logrus.FieldLogger, eventSource eventSource, newSession func(wsConn, chan<- interface{}) (session, error)) (hStats handlerStats) { h.setupOnce.Do(h.setup) ctx, cancel := context.WithCancel(ws.Request().Context()) - log := logger(ctx) - - incoming := eventSource.NewSink() - defer incoming.Stop() + defer cancel() queue := make(chan interface{}, h.QueueSize) h.mtx.Lock() - h.queues[queue] = struct{}{} + h.lastDelay[queue] = 0 h.mtx.Unlock() defer func() { h.mtx.Lock() - delete(h.queues, queue) + delete(h.lastDelay, queue) h.mtx.Unlock() }() sess, err := newSession(ws, queue) if err != nil { - log.WithError(err).Error("newSession failed") + logger.WithError(err).Error("newSession failed") return } + // Receive websocket frames from the client and pass them to + // sess.Receive(). go func() { + defer cancel() buf := make([]byte, 2<<20) for { select { @@ -62,27 +68,29 @@ func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsC ws.SetReadDeadline(time.Now().Add(24 * 365 * time.Hour)) n, err := ws.Read(buf) buf := buf[:n] - log.WithField("frame", string(buf[:n])).Debug("received frame") + logger.WithField("frame", string(buf[:n])).Debug("received frame") if err == nil && n == cap(buf) { err = errFrameTooBig } if err != nil { - if err != io.EOF { - log.WithError(err).Info("read error") + if err != io.EOF && ctx.Err() == nil { + logger.WithError(err).Info("read error") } - cancel() return } err = sess.Receive(buf) if err != nil { - log.WithError(err).Error("sess.Receive() failed") - cancel() + logger.WithError(err).Error("sess.Receive() failed") return } } }() + // Take items from the outgoing queue, serialize them using + // sess.EventMessage() as needed, and send them to the client + // as websocket frames. go func() { + defer cancel() for { var ok bool var data interface{} @@ -97,45 +105,48 @@ func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsC var e *event var buf []byte var err error - log := log + logger := logger switch data := data.(type) { case []byte: buf = data case *event: e = data - log = log.WithField("serial", e.Serial) + logger = logger.WithField("serial", e.Serial) buf, err = sess.EventMessage(e) if err != nil { - log.WithError(err).Error("EventMessage failed") - cancel() - break + logger.WithError(err).Error("EventMessage failed") + return } else if len(buf) == 0 { - log.Debug("skip") + logger.Debug("skip") continue } default: - log.WithField("data", data).Error("bad object in client queue") + logger.WithField("data", data).Error("bad object in client queue") continue } - log.WithField("frame", string(buf)).Debug("send event") + logger.WithField("frame", string(buf)).Debug("send event") ws.SetWriteDeadline(time.Now().Add(h.PingTimeout)) t0 := time.Now() _, err = ws.Write(buf) if err != nil { - log.WithError(err).Error("write failed") - cancel() - break + if ctx.Err() == nil { + logger.WithError(err).Error("write failed") + } + return } - log.Debug("sent") + logger.Debug("sent") if e != nil { - stats.QueueDelayNs += t0.Sub(e.Received) + hStats.QueueDelayNs += t0.Sub(e.Ready) + h.mtx.Lock() + h.lastDelay[queue] = stats.Duration(time.Since(e.Ready)) + h.mtx.Unlock() } - stats.WriteDelayNs += time.Since(t0) - stats.EventBytes += uint64(len(buf)) - stats.EventCount++ + hStats.WriteDelayNs += time.Since(t0) + hStats.EventBytes += uint64(len(buf)) + hStats.EventCount++ } }() @@ -145,9 +156,13 @@ func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsC // is done/cancelled or the incoming event stream ends. Shut // down the handler if the outgoing queue fills up. go func() { + defer cancel() ticker := time.NewTicker(h.PingTimeout) defer ticker.Stop() + incoming := eventSource.NewSink() + defer incoming.Stop() + for { select { case <-ctx.Done(): @@ -164,10 +179,8 @@ func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsC default: } } - continue case e, ok := <-incoming.Channel(): if !ok { - cancel() return } if !sess.Filter(e) { @@ -176,8 +189,7 @@ func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsC select { case queue <- e: default: - log.WithError(errQueueFull).Error("terminate") - cancel() + logger.WithError(errQueueFull).Error("terminate") return } } @@ -188,26 +200,38 @@ func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsC return } -func (h *handler) Status() interface{} { +func (h *handler) DebugStatus() interface{} { h.mtx.Lock() defer h.mtx.Unlock() var s struct { - QueueCount int - QueueMax int - QueueTotal uint64 + QueueCount int + QueueMin int + QueueMax int + QueueTotal uint64 + QueueDelayMin stats.Duration + QueueDelayMax stats.Duration } - for q := range h.queues { + for q, lastDelay := range h.lastDelay { + s.QueueCount++ n := len(q) s.QueueTotal += uint64(n) if s.QueueMax < n { s.QueueMax = n } + if s.QueueMin > n || s.QueueCount == 1 { + s.QueueMin = n + } + if (s.QueueDelayMin > lastDelay || s.QueueDelayMin == 0) && lastDelay > 0 { + s.QueueDelayMin = lastDelay + } + if s.QueueDelayMax < lastDelay { + s.QueueDelayMax = lastDelay + } } - s.QueueCount = len(h.queues) return &s } func (h *handler) setup() { - h.queues = make(map[chan interface{}]struct{}) + h.lastDelay = make(map[chan interface{}]stats.Duration) }