Merge branch '11221-always-restart-services'

[arvados.git] / services / crunch-run / logging.go

diff --git a/services/crunch-run/logging.go b/services/crunch-run/logging.go
index db9d101b581b560314374792421c134f918c47b0..5254ff671e6d3826356722259069515d450c1066 100644 (file)
--- a/services/crunch-run/logging.go
+++ b/services/crunch-run/logging.go
@@ -4,11 +4,12 @@ import (
        "bufio"
        "bytes"
        "fmt"
-       "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
        "io"
        "log"
        "sync"
        "time"
+
+       "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
 )
 
 // Timestamper is the signature for a function that takes a timestamp and
@@ -18,7 +19,7 @@ type Timestamper func(t time.Time) string
 // Logging plumbing:
 //
 // ThrottledLogger.Logger -> ThrottledLogger.Write ->
-// ThrottledLogger.buf -> ThrottledLogger.flusher -> goWriter ->
+// ThrottledLogger.buf -> ThrottledLogger.flusher ->
 // ArvLogWriter.Write -> CollectionFileWriter.Write | Api.Create
 //
 // For stdout/stderr ReadWriteLines additionally runs as a goroutine to pull
@@ -31,9 +32,9 @@ type ThrottledLogger struct {
        *log.Logger
        buf *bytes.Buffer
        sync.Mutex
-       writer      io.WriteCloser
-       stop        bool
-       flusherDone chan bool
+       writer   io.WriteCloser
+       stopping chan struct{}
+       stopped  chan struct{}
        Timestamper
        Immediate *log.Logger
 }
@@ -51,10 +52,11 @@ func RFC3339Timestamp(t time.Time) string {
 // tl.Immediate, if tl.Immediate is not nil.
 func (tl *ThrottledLogger) Write(p []byte) (n int, err error) {
        tl.Mutex.Lock()
+       defer tl.Mutex.Unlock()
+
        if tl.buf == nil {
                tl.buf = &bytes.Buffer{}
        }
-       defer tl.Mutex.Unlock()
 
        now := tl.Timestamper(time.Now().UTC())
        sc := bufio.NewScanner(bytes.NewBuffer(p))
@@ -77,48 +79,39 @@ func (tl *ThrottledLogger) Write(p []byte) (n int, err error) {
 // Periodically check the current buffer; if not empty, send it on the
 // channel to the goWriter goroutine.
 func (tl *ThrottledLogger) flusher() {
-       bufchan := make(chan *bytes.Buffer)
-       bufterm := make(chan bool)
-
-       // Use a separate goroutine for the actual write so that the writes are
-       // actually initiated closer every 1s instead of every
-       // 1s + (time to it takes to write).
-       go goWriter(tl.writer, bufchan, bufterm)
-       for {
-               if !tl.stop {
-                       time.Sleep(1 * time.Second)
+       ticker := time.NewTicker(time.Second)
+       defer ticker.Stop()
+       for stopping := false; !stopping; {
+               select {
+               case <-tl.stopping:
+                       // flush tl.buf, then exit the loop
+                       stopping = true
+               case <-ticker.C:
                }
+
+               var ready *bytes.Buffer
+
                tl.Mutex.Lock()
-               if tl.buf != nil && tl.buf.Len() > 0 {
-                       oldbuf := tl.buf
-                       tl.buf = nil
-                       tl.Mutex.Unlock()
-                       bufchan <- oldbuf
-               } else if tl.stop {
-                       tl.Mutex.Unlock()
-                       break
-               } else {
-                       tl.Mutex.Unlock()
-               }
-       }
-       close(bufchan)
-       <-bufterm
-       tl.flusherDone <- true
-}
+               ready, tl.buf = tl.buf, nil
+               tl.Mutex.Unlock()
 
-// Receive buffers from a channel and send to the underlying Writer
-func goWriter(writer io.Writer, c <-chan *bytes.Buffer, t chan<- bool) {
-       for b := range c {
-               writer.Write(b.Bytes())
+               if ready != nil && ready.Len() > 0 {
+                       tl.writer.Write(ready.Bytes())
+               }
        }
-       t <- true
+       close(tl.stopped)
 }
 
 // Close the flusher goroutine and wait for it to complete, then close the
 // underlying Writer.
 func (tl *ThrottledLogger) Close() error {
-       tl.stop = true
-       <-tl.flusherDone
+       select {
+       case <-tl.stopping:
+               // already stopped
+       default:
+               close(tl.stopping)
+       }
+       <-tl.stopped
        return tl.writer.Close()
 }
 
@@ -166,7 +159,8 @@ func ReadWriteLines(in io.Reader, writer io.Writer, done chan<- bool) {
 // per second.
 func NewThrottledLogger(writer io.WriteCloser) *ThrottledLogger {
        tl := &ThrottledLogger{}
-       tl.flusherDone = make(chan bool)
+       tl.stopping = make(chan struct{})
+       tl.stopped = make(chan struct{})
        tl.writer = writer
        tl.Logger = log.New(tl, "", 0)
        tl.Timestamper = RFC3339Timestamp