Fix some race conditions.

expreduce/caslogger.go

@@ -5,18 +5,19 @@ import (
 	"github.com/op/go-logging"
 	"os"
 	"runtime/debug"
+	"sync"
 )
 
 var format = logging.MustStringFormatter(
-	//`%{color}%{time:15:04:05.000} %{callpath} ▶ %{level:.4s} %{id:03x}%{color:reset} %{message}`,
 	`%{color}%{time:15:04:05.000} %{callpath} ▶ %{id:03x}%{color:reset} %{message}`,
 )
+var goLoggingMutex = &sync.Mutex{}
 
 type CASLogger struct {
-	_log        *logging.Logger
-	leveled     logging.LeveledBackend
-	debugState  bool
-	isProfiling bool
+	_log		*logging.Logger
+	leveled		logging.LeveledBackend
+	debugState	bool
+	isProfiling	bool
 }
 
 func (this *CASLogger) Debugf(fmt string, args ...interface{}) {
@@ -69,9 +70,13 @@ func (this *CASLogger) Pre() string {
 }
 
 func (this *CASLogger) SetUpLogging() {
+	// go-logging appears to not be thread safe, so we have a mutex when
+	// configuring the logging.
+	goLoggingMutex.Lock()
 	this._log = logging.MustGetLogger("example")
 	backend := logging.NewLogBackend(os.Stderr, "", 0)
 	formatter := logging.NewBackendFormatter(backend, format)
 	this.leveled = logging.AddModuleLevel(formatter)
 	logging.SetBackend(this.leveled)
+	goLoggingMutex.Unlock()
 }

expreduce/definition_map.go

@@ -34,6 +34,16 @@ func (dm definitionMap) GetDef(key string) Def {
 	return value
 }
 
+func (dm definitionMap) LockKey(key string) {
+	shard := dm.internalMap.GetShard(key)
+	shard.Lock()
+}
+
+func (dm definitionMap) UnlockKey(key string) {
+	shard := dm.internalMap.GetShard(key)
+	shard.Unlock()
+}
+
 func (dm definitionMap) CopyDefs() definitionMap {
 	out := newDefinitionMap()
 	for mapTuple := range dm.internalMap.IterBuffered() {

expreduce/evalstate.go

@@ -488,12 +488,15 @@ func (this *EvalState) Define(lhs Ex, rhs Ex) {
 		existingRule := dv.rule
 		existingLhs := existingRule.Parts[1]
 		if IsSameQ(existingLhs, heldLhs, &this.CASLogger) {
+			this.defined.LockKey(name)
 			existingRhsCond := maskNonConditional(existingRule.Parts[2])
 			newRhsCond := maskNonConditional(rhs)
 			if IsSameQ(existingRhsCond, newRhsCond, &this.CASLogger) {
 				dv.rule.Parts[2] = rhs
+				this.defined.UnlockKey(name)
 				return
 			}
+			this.defined.UnlockKey(name)
 		}
 	}
 

expreduce/ex_expression.go

@@ -8,6 +8,7 @@ import "encoding/binary"
 import "time"
 import "flag"
 import "hash/fnv"
+import "sync/atomic"
 
 var printevals = flag.Bool("printevals", false, "")
 var checkhashes = flag.Bool("checkhashes", false, "")
@@ -553,7 +554,7 @@ func (this *Expression) NeedsEval() bool {
 }
 
 func (this *Expression) Hash() uint64 {
-	if this.cachedHash > 0 {
+	if atomic.LoadUint64(&this.cachedHash) > 0 {
 		return this.cachedHash
 	}
 	h := fnv.New64a()
@@ -563,7 +564,7 @@ func (this *Expression) Hash() uint64 {
 		binary.LittleEndian.PutUint64(b, part.Hash())
 		h.Write(b)
 	}
-	this.cachedHash = h.Sum64()
+	atomic.StoreUint64(&this.cachedHash, h.Sum64())
 	return h.Sum64()
 }