Merge pull request #59 from fjl/feature/raceless-ethlog

Improve package ethlog

ethlog/README.md

-## Features
-
-- packages use tagged logger sending log messages to shared (process-wide) logging engine
-- log writers (interface ethlog.LogSystem) can be added to the logging engine by wrappers/guis/clients
-- shared logging engine dispatching to multiple log systems
-- log level can be set separately per log system
-- async logging thread: logging IO does not block main thread 
-- log messages are synchronously stringified to avoid incorrectly logging of changed states 
-- log level enum: ethlog.LogLevel:   Silence, ErrorLevel, WarnLevel, InfoLevel, DebugLevel, DebugDetailLevel
-
-## Usage
-
-In an ethereum component package:
-
-    import "github.com/ethereum/eth-go/ethlog"
-
-    // package-wide logger using tag
-    var logger = ethlog.NewLogger("TAG")
-
-Logger provides named Printf and Println style methods for all loglevels
-
-    logger.Infoln("this is info") # > [TAG] This is info
-    logger.Infof("this %v is info", object) # > [TAG] This object is info
-
-Ethereum wrappers should register log systems conforming to ethlog.LogSystem
-
-    import "github.com/ethereum/eth-go/ethlog"
-    
-    type CustomLogWriter struct {
-      logLevel ethlog.LogLevel
-    }
-
-    func (t *TestLogSystem) SetLogLevel(i LogLevel) {
-      t.level = i
-    }
-
-    func (t *TestLogSystem) GetLogLevel() LogLevel {
-      return t.level
-    }
-
-    func (c *CustomLogWriter) Printf(format string, v...interface{}) {
-      //....
-    }
-
-    func (c *CustomLogWriter) Println(v...interface{}) {
-      //....
-    }
-
-    ethlog.AddLogWriter(&CustomLogWriter{})
-
-ethlog also provides constructors for that wrap io.Writers into a standard logger with a settable level:
-
-    filename := "test.log"
-    file, _ := os.OpenFile(filename, os.O_RDWR|os.O_CREATE, os.ModePerm)
-    fileLogSystem := NewStdLogSystem(file, 0, WarnLevel)
-    AddLogSystem(fileLogSystem)
-    stdOutLogSystem := NewStdLogSystem(os.Stdout, 0, WarnLevel)
-    AddLogSystem(stdOutLogSystem)
-
-
-
-

ethlog/example_test.go

+package ethlog
+
+import "os"
+
+func ExampleLogger() {
+	logger := NewLogger("TAG")
+	logger.Infoln("so awesome")            // prints [TAG] so awesome
+	logger.Infof("this %q is raw", "coin") // prints [TAG] this "coin" is raw
+}
+
+func ExampleLogSystem() {
+	filename := "test.log"
+	file, _ := os.OpenFile(filename, os.O_RDWR|os.O_CREATE, os.ModePerm)
+	fileLog := NewStdLogSystem(file, 0, WarnLevel)
+	AddLogSystem(fileLog)
+
+	stdoutLog := NewStdLogSystem(os.Stdout, 0, WarnLevel)
+	AddLogSystem(stdoutLog)
+
+	NewLogger("TAG").Warnln("reactor meltdown") // writes to both logs
+}

ethlog/loggers.go

+/*
+Package ethlog implements a multi-output leveled logger.
+
+Other packages use tagged logger to send log messages to shared
+(process-wide) logging engine. The shared logging engine dispatches to
+multiple log systems. The log level can be set separately per log
+system.
+
+Logging is asynchronous and does not block the caller. Message
+formatting is performed by the caller goroutine to avoid incorrect
+logging of mutable state.
+*/
 package ethlog
 
 import (
@@ -6,46 +18,26 @@ import (
 	"log"
 	"os"
 	"sync"
+	"sync/atomic"
 )
 
+// LogSystem is implemented by log output devices.
+// All methods can be called concurrently from multiple goroutines.
 type LogSystem interface {
 	GetLogLevel() LogLevel
 	SetLogLevel(i LogLevel)
-	Println(v ...interface{})
-	Printf(format string, v ...interface{})
+	LogPrint(LogLevel, string)
 }
 
-type logMessage struct {
-	LogLevel LogLevel
-	format   bool
-	msg      string
+type message struct {
+	level LogLevel
+	msg   string
 }
 
-func newPrintlnLogMessage(level LogLevel, tag string, v ...interface{}) *logMessage {
-	return &logMessage{level, false, fmt.Sprintf("[%s] %s", tag, fmt.Sprint(v...))}
-}
-
-func newPrintfLogMessage(level LogLevel, tag string, format string, v ...interface{}) *logMessage {
-	return &logMessage{level, true, fmt.Sprintf("[%s] %s", tag, fmt.Sprintf(format, v...))}
-}
-
-func (msg *logMessage) send(logger LogSystem) {
-	if msg.format {
-		logger.Printf(msg.msg)
-	} else {
-		logger.Println(msg.msg)
-	}
-}
-
-var logMessages chan (*logMessage)
-var logSystems []LogSystem
-var quit chan chan error
-var drained chan bool
-var mutex = sync.Mutex{}
-
 type LogLevel uint8
 
 const (
+	// Standard log levels
 	Silence LogLevel = iota
 	ErrorLevel
 	WarnLevel
@@ -54,167 +46,203 @@ const (
 	DebugDetailLevel
 )
 
-func dispatch(msg *logMessage) {
-	for _, logSystem := range logSystems {
-		if logSystem.GetLogLevel() >= msg.LogLevel {
-			msg.send(logSystem)
-		}
+var (
+	logMessageC = make(chan message)
+	addSystemC  = make(chan LogSystem)
+	flushC      = make(chan chan struct{})
+	resetC      = make(chan chan struct{})
+)
+
+func init() {
+	go dispatchLoop()
+}
+
+// each system can buffer this many messages before
+// blocking incoming log messages.
+const sysBufferSize = 500
+
+func dispatchLoop() {
+	var (
+		systems  []LogSystem
+		systemIn []chan message
+		systemWG sync.WaitGroup
+	)
+	bootSystem := func(sys LogSystem) {
+		in := make(chan message, sysBufferSize)
+		systemIn = append(systemIn, in)
+		systemWG.Add(1)
+		go sysLoop(sys, in, &systemWG)
 	}
-}
 
-// log messages are dispatched to log writers
-func start() {
 	for {
 		select {
-		case status := <-quit:
-			status <- nil
-			return
-		case msg := <-logMessages:
-			dispatch(msg)
-		default:
-			drained <- true // this blocks until a message is sent to the queue
+		case msg := <-logMessageC:
+			for _, c := range systemIn {
+				c <- msg
+			}
+
+		case sys := <-addSystemC:
+			systems = append(systems, sys)
+			bootSystem(sys)
+
+		case waiter := <-resetC:
+			// reset means terminate all systems
+			for _, c := range systemIn {
+				close(c)
+			}
+			systems = nil
+			systemIn = nil
+			systemWG.Wait()
+			close(waiter)
+
+		case waiter := <-flushC:
+			// flush means reboot all systems
+			for _, c := range systemIn {
+				close(c)
+			}
+			systemIn = nil
+			systemWG.Wait()
+			for _, sys := range systems {
+				bootSystem(sys)
+			}
+			close(waiter)
 		}
 	}
 }
 
-func send(msg *logMessage) {
-	logMessages <- msg
-	select {
-	case <-drained:
-	default:
+func sysLoop(sys LogSystem, in <-chan message, wg *sync.WaitGroup) {
+	for msg := range in {
+		if sys.GetLogLevel() >= msg.level {
+			sys.LogPrint(msg.level, msg.msg)
+		}
 	}
+	wg.Done()
 }
 
+// Reset removes all active log systems.
+// It blocks until all current messages have been delivered.
 func Reset() {
-	mutex.Lock()
-	defer mutex.Unlock()
-	if logSystems != nil {
-		status := make(chan error)
-		quit <- status
-		select {
-		case <-drained:
-		default:
-		}
-		<-status
-	}
+	waiter := make(chan struct{})
+	resetC <- waiter
+	<-waiter
 }
 
-// waits until log messages are drained (dispatched to log writers)
+// Flush waits until all current log messages have been dispatched to
+// the active log systems.
 func Flush() {
-	if logSystems != nil {
-		<-drained
-	}
+	waiter := make(chan struct{})
+	flushC <- waiter
+	<-waiter
 }
 
+// AddLogSystem starts printing messages to the given LogSystem.
+func AddLogSystem(sys LogSystem) {
+	addSystemC <- sys
+}
+
+// A Logger prints messages prefixed by a given tag. It provides named
+// Printf and Println style methods for all loglevels. Each ethereum
+// component should have its own logger with a unique prefix.
 type Logger struct {
 	tag string
 }
 
 func NewLogger(tag string) *Logger {
-	return &Logger{tag}
-}
-
-func AddLogSystem(logSystem LogSystem) {
-	var mutex = &sync.Mutex{}
-	mutex.Lock()
-	defer mutex.Unlock()
-	if logSystems == nil {
-		logMessages = make(chan *logMessage, 10)
-		quit = make(chan chan error, 1)
-		drained = make(chan bool, 1)
-		go start()
-	}
-	logSystems = append(logSystems, logSystem)
+	return &Logger{"[" + tag + "] "}
 }
 
 func (logger *Logger) sendln(level LogLevel, v ...interface{}) {
-	if logMessages != nil {
-		msg := newPrintlnLogMessage(level, logger.tag, v...)
-		send(msg)
-	}
+	logMessageC <- message{level, logger.tag + fmt.Sprintln(v...)}
 }
 
 func (logger *Logger) sendf(level LogLevel, format string, v ...interface{}) {
-	if logMessages != nil {
-		msg := newPrintfLogMessage(level, logger.tag, format, v...)
-		send(msg)
-	}
+	logMessageC <- message{level, logger.tag + fmt.Sprintf(format, v...)}
 }
 
+// Errorln writes a message with ErrorLevel.
 func (logger *Logger) Errorln(v ...interface{}) {
 	logger.sendln(ErrorLevel, v...)
 }
 
+// Warnln writes a message with WarnLevel.
 func (logger *Logger) Warnln(v ...interface{}) {
 	logger.sendln(WarnLevel, v...)
 }
 
+// Infoln writes a message with InfoLevel.
 func (logger *Logger) Infoln(v ...interface{}) {
 	logger.sendln(InfoLevel, v...)
 }
 
+// Debugln writes a message with DebugLevel.
 func (logger *Logger) Debugln(v ...interface{}) {
 	logger.sendln(DebugLevel, v...)
 }
 
+// DebugDetailln writes a message with DebugDetailLevel.
 func (logger *Logger) DebugDetailln(v ...interface{}) {
 	logger.sendln(DebugDetailLevel, v...)
 }
 
+// Errorf writes a message with ErrorLevel.
 func (logger *Logger) Errorf(format string, v ...interface{}) {
 	logger.sendf(ErrorLevel, format, v...)
 }
 
+// Warnf writes a message with WarnLevel.
 func (logger *Logger) Warnf(format string, v ...interface{}) {
 	logger.sendf(WarnLevel, format, v...)
 }
 
+// Infof writes a message with InfoLevel.
 func (logger *Logger) Infof(format string, v ...interface{}) {
 	logger.sendf(InfoLevel, format, v...)
 }
 
+// Debugf writes a message with DebugLevel.
 func (logger *Logger) Debugf(format string, v ...interface{}) {
 	logger.sendf(DebugLevel, format, v...)
 }
 
+// DebugDetailf writes a message with DebugDetailLevel.
 func (logger *Logger) DebugDetailf(format string, v ...interface{}) {
 	logger.sendf(DebugDetailLevel, format, v...)
 }
 
+// Fatalln writes a message with ErrorLevel and exits the program.
 func (logger *Logger) Fatalln(v ...interface{}) {
 	logger.sendln(ErrorLevel, v...)
 	Flush()
 	os.Exit(0)
 }
 
+// Fatalf writes a message with ErrorLevel and exits the program.
 func (logger *Logger) Fatalf(format string, v ...interface{}) {
 	logger.sendf(ErrorLevel, format, v...)
 	Flush()
 	os.Exit(0)
 }
 
-type StdLogSystem struct {
-	logger *log.Logger
-	level  LogLevel
-}
-
-func (t *StdLogSystem) Println(v ...interface{}) {
-	t.logger.Println(v...)
+// NewStdLogSystem creates a LogSystem that prints to the given writer.
+// The flag values are defined package log.
+func NewStdLogSystem(writer io.Writer, flags int, level LogLevel) LogSystem {
+	logger := log.New(writer, "", flags)
+	return &stdLogSystem{logger, uint32(level)}
 }
 
-func (t *StdLogSystem) Printf(format string, v ...interface{}) {
-	t.logger.Printf(format, v...)
+type stdLogSystem struct {
+	logger *log.Logger
+	level  uint32
 }
 
-func (t *StdLogSystem) SetLogLevel(i LogLevel) {
-	t.level = i
+func (t *stdLogSystem) LogPrint(level LogLevel, msg string) {
+	t.logger.Print(msg)
 }
 
-func (t *StdLogSystem) GetLogLevel() LogLevel {
-	return t.level
+func (t *stdLogSystem) SetLogLevel(i LogLevel) {
+	atomic.StoreUint32(&t.level, uint32(i))
 }
 
-func NewStdLogSystem(writer io.Writer, flags int, level LogLevel) *StdLogSystem {
-	logger := log.New(writer, "", flags)
-	return &StdLogSystem{logger, level}
+func (t *stdLogSystem) GetLogLevel() LogLevel {
+	return LogLevel(atomic.LoadUint32(&t.level))
 }

ethlog/loggers_test.go

 package ethlog
 
 import (
-	"fmt"
 	"io/ioutil"
+	"math/rand"
 	"os"
+	"sync"
 	"testing"
+	"time"
 )
 
 type TestLogSystem struct {
-	Output string
+	mutex  sync.Mutex
+	output string
 	level  LogLevel
 }
 
-func (t *TestLogSystem) Println(v ...interface{}) {
-	t.Output += fmt.Sprintln(v...)
+func (ls *TestLogSystem) LogPrint(level LogLevel, msg string) {
+	ls.mutex.Lock()
+	ls.output += msg
+	ls.mutex.Unlock()
 }
 
-func (t *TestLogSystem) Printf(format string, v ...interface{}) {
-	t.Output += fmt.Sprintf(format, v...)
+func (ls *TestLogSystem) SetLogLevel(i LogLevel) {
+	ls.mutex.Lock()
+	ls.level = i
+	ls.mutex.Unlock()
 }
 
-func (t *TestLogSystem) SetLogLevel(i LogLevel) {
-	t.level = i
+func (ls *TestLogSystem) GetLogLevel() LogLevel {
+	ls.mutex.Lock()
+	defer ls.mutex.Unlock()
+	return ls.level
 }
 
-func (t *TestLogSystem) GetLogLevel() LogLevel {
-	return t.level
+func (ls *TestLogSystem) CheckOutput(t *testing.T, expected string) {
+	ls.mutex.Lock()
+	output := ls.output
+	ls.mutex.Unlock()
+	if output != expected {
+		t.Errorf("log output mismatch:\n   got: %q\n  want: %q\n", output, expected)
+	}
+}
+
+type blockedLogSystem struct {
+	LogSystem
+	unblock chan struct{}
+}
+
+func (ls blockedLogSystem) LogPrint(level LogLevel, msg string) {
+	<-ls.unblock
+	ls.LogSystem.LogPrint(level, msg)
 }
 
 func TestLoggerFlush(t *testing.T) {
+	Reset()
+
 	logger := NewLogger("TEST")
-	testLogSystem := &TestLogSystem{level: WarnLevel}
-	AddLogSystem(testLogSystem)
+	ls := blockedLogSystem{&TestLogSystem{level: WarnLevel}, make(chan struct{})}
+	AddLogSystem(ls)
 	for i := 0; i < 5; i++ {
+		// these writes shouldn't hang even though ls is blocked
 		logger.Errorf(".")
 	}
-	Flush()
-	Reset()
-	output := testLogSystem.Output
-	if output != "[TEST] .[TEST] .[TEST] .[TEST] .[TEST] ." {
-		t.Error("Expected complete logger output '[TEST] .[TEST] .[TEST] .[TEST] .[TEST] .', got ", output)
+
+	beforeFlush := time.Now()
+	time.AfterFunc(80*time.Millisecond, func() { close(ls.unblock) })
+	Flush() // this should hang for approx. 80ms
+	if blockd := time.Now().Sub(beforeFlush); blockd < 80*time.Millisecond {
+		t.Errorf("Flush didn't block long enough, blocked for %v, should've been >= 80ms", blockd)
 	}
+
+	ls.LogSystem.(*TestLogSystem).CheckOutput(t, "[TEST] .[TEST] .[TEST] .[TEST] .[TEST] .")
 }
 
 func TestLoggerPrintln(t *testing.T) {
+	Reset()
+
 	logger := NewLogger("TEST")
 	testLogSystem := &TestLogSystem{level: WarnLevel}
 	AddLogSystem(testLogSystem)
@@ -52,31 +84,27 @@ func TestLoggerPrintln(t *testing.T) {
 	logger.Infoln("info")
 	logger.Debugln("debug")
 	Flush()
-	Reset()
-	output := testLogSystem.Output
-	fmt.Println(quote(output))
-	if output != "[TEST] error\n[TEST] warn\n" {
-		t.Error("Expected logger output '[TEST] error\\n[TEST] warn\\n', got ", output)
-	}
+
+	testLogSystem.CheckOutput(t, "[TEST] error\n[TEST] warn\n")
 }
 
 func TestLoggerPrintf(t *testing.T) {
+	Reset()
+
 	logger := NewLogger("TEST")
 	testLogSystem := &TestLogSystem{level: WarnLevel}
 	AddLogSystem(testLogSystem)
-	logger.Errorf("error to %v\n", *testLogSystem)
-	logger.Warnf("warn")
+	logger.Errorf("error to %v\n", []int{1, 2, 3})
+	logger.Warnf("warn %%d %d", 5)
 	logger.Infof("info")
 	logger.Debugf("debug")
 	Flush()
-	Reset()
-	output := testLogSystem.Output
-	if output != "[TEST] error to { 2}\n[TEST] warn" {
-		t.Error("Expected logger output '[TEST] error to { 2}\\n[TEST] warn', got ", output)
-	}
+	testLogSystem.CheckOutput(t, "[TEST] error to [1 2 3]\n[TEST] warn %d 5")
 }
 
 func TestMultipleLogSystems(t *testing.T) {
+	Reset()
+
 	logger := NewLogger("TEST")
 	testLogSystem0 := &TestLogSystem{level: ErrorLevel}
 	testLogSystem1 := &TestLogSystem{level: WarnLevel}
@@ -85,18 +113,14 @@ func TestMultipleLogSystems(t *testing.T) {
 	logger.Errorln("error")
 	logger.Warnln("warn")
 	Flush()
-	Reset()
-	output0 := testLogSystem0.Output
-	output1 := testLogSystem1.Output
-	if output0 != "[TEST] error\n" {
-		t.Error("Expected logger 0 output '[TEST] error\\n', got ", output0)
-	}
-	if output1 != "[TEST] error\n[TEST] warn\n" {
-		t.Error("Expected logger 1 output '[TEST] error\\n[TEST] warn\\n', got ", output1)
-	}
+
+	testLogSystem0.CheckOutput(t, "[TEST] error\n")
+	testLogSystem1.CheckOutput(t, "[TEST] error\n[TEST] warn\n")
 }
 
 func TestFileLogSystem(t *testing.T) {
+	Reset()
+
 	logger := NewLogger("TEST")
 	filename := "test.log"
 	file, _ := os.OpenFile(filename, os.O_RDWR|os.O_CREATE, os.ModePerm)
@@ -115,7 +139,36 @@ func TestFileLogSystem(t *testing.T) {
 }
 
 func TestNoLogSystem(t *testing.T) {
+	Reset()
+
 	logger := NewLogger("TEST")
 	logger.Warnln("warn")
 	Flush()
 }
+
+func TestConcurrentAddSystem(t *testing.T) {
+	rand.Seed(time.Now().Unix())
+	Reset()
+
+	logger := NewLogger("TEST")
+	stop := make(chan struct{})
+	writer := func() {
+		select {
+		case <-stop:
+			return
+		default:
+			logger.Infoln("foo")
+			Flush()
+		}
+	}
+
+	go writer()
+	go writer()
+
+	stopTime := time.Now().Add(100 * time.Millisecond)
+	for time.Now().Before(stopTime) {
+		time.Sleep(time.Duration(rand.Intn(20)) * time.Millisecond)
+		AddLogSystem(NewStdLogSystem(ioutil.Discard, 0, InfoLevel))
+	}
+	close(stop)
+}

ethvm/vm_test.go

@@ -39,14 +39,14 @@ for i := 0; i < 10; i++ {
 
 return x`
 
-func setup(level int, typ Type) (*Closure, VirtualMachine) {
+func setup(level ethlog.LogLevel, typ Type) (*Closure, VirtualMachine) {
 	code, err := ethutil.Compile(mutcode, true)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	// Pipe output to /dev/null
-	ethlog.AddLogSystem(ethlog.NewStdLogSystem(ioutil.Discard, log.LstdFlags, ethlog.LogLevel(level)))
+	ethlog.AddLogSystem(ethlog.NewStdLogSystem(ioutil.Discard, log.LstdFlags, level))
 
 	ethutil.ReadConfig(".ethtest", "/tmp/ethtest", "")
 
@@ -57,7 +57,7 @@ func setup(level int, typ Type) (*Closure, VirtualMachine) {
 }
 
 func TestDebugVm(t *testing.T) {
-	closure, vm := setup(4, DebugVmTy)
+	closure, vm := setup(ethlog.DebugLevel, DebugVmTy)
 	ret, _, e := closure.Call(vm, nil)
 	if e != nil {
 		fmt.Println("error", e)
@@ -69,7 +69,7 @@ func TestDebugVm(t *testing.T) {
 }
 
 func TestVm(t *testing.T) {
-	closure, vm := setup(4, StandardVmTy)
+	closure, vm := setup(ethlog.DebugLevel, StandardVmTy)
 	ret, _, e := closure.Call(vm, nil)
 	if e != nil {
 		fmt.Println("error", e)
@@ -81,7 +81,7 @@ func TestVm(t *testing.T) {
 }
 
 func BenchmarkDebugVm(b *testing.B) {
-	closure, vm := setup(3, DebugVmTy)
+	closure, vm := setup(ethlog.InfoLevel, DebugVmTy)
 
 	b.ResetTimer()
 
@@ -91,7 +91,7 @@ func BenchmarkDebugVm(b *testing.B) {
 }
 
 func BenchmarkVm(b *testing.B) {
-	closure, vm := setup(3, StandardVmTy)
+	closure, vm := setup(ethlog.InfoLevel, StandardVmTy)
 
 	b.ResetTimer()
 
@@ -106,7 +106,7 @@ func RunCode(mutCode string, typ Type) []byte {
 		log.Fatal(err)
 	}
 
-	ethlog.AddLogSystem(ethlog.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlog.LogLevel(3)))
+	ethlog.AddLogSystem(ethlog.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlog.InfoLevel))
 
 	ethutil.ReadConfig(".ethtest", "/tmp/ethtest", "")
 

tests/helper/init.go

@@ -12,7 +12,7 @@ var Logger ethlog.LogSystem
 var Log = ethlog.NewLogger("TEST")
 
 func init() {
-	Logger = ethlog.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlog.LogLevel(3))
+	Logger = ethlog.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlog.InfoLevel)
 	ethlog.AddLogSystem(Logger)
 
 	ethutil.ReadConfig(".ethtest", "/tmp/ethtest", "")

tests/vm/gh_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"testing"
 
+	"github.com/ethereum/eth-go/ethlog"
 	"github.com/ethereum/eth-go/ethstate"
 	"github.com/ethereum/eth-go/ethutil"
 	"github.com/ethereum/eth-go/tests/helper"
@@ -87,6 +88,9 @@ func RunVmTest(url string, t *testing.T) {
 }
 
 // I've created a new function for each tests so it's easier to identify where the problem lies if any of them fail.
+func TestVMSha3(t *testing.T) {
+	helper.Logger.SetLogLevel(ethlog.Silence)
+	defer helper.Logger.SetLogLevel(ethlog.DebugLevel)
 
 func TestVMArithmetic(t *testing.T) {
 	const url = "https://raw.githubusercontent.com/ethereum/tests/develop/vmtests/vmArithmeticTest.json"