Merged in coinplugin/go-metadium (pull request #21)

Master

cmd/geth/main.go

@@ -183,6 +183,7 @@ var (
 		utils.PrefetchCount,
 		utils.LogFlag,
 		utils.MaxTxsPerBlock,
+		utils.Hub,
 	}
 )
 

cmd/geth/usage.go

@@ -250,6 +250,7 @@ var AppHelpFlagGroups = []flagGroup{
 			utils.PrefetchCount,
 			utils.LogFlag,
 			utils.MaxTxsPerBlock,
+			utils.Hub,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -690,6 +690,11 @@ var (
 		Usage: "Max # of transactions in a block",
 		Value: params.MaxTxsPerBlock,
 	}
+	Hub = cli.StringFlag{
+		Name:  "hub",
+		Usage: "Id of message hub",
+		Value: params.Hub,
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1385,6 +1390,9 @@ func SetMetadiumConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
 	if ctx.GlobalIsSet(MaxTxsPerBlock.Name) {
 		params.MaxTxsPerBlock = ctx.GlobalInt(MaxTxsPerBlock.Name)
 	}
+	if ctx.GlobalIsSet(Hub.Name) {
+		params.Hub = ctx.GlobalString(Hub.Name)
+	}
 
 	if params.ConsensusMethod == params.ConsensusInvalid {
 		params.ConsensusMethod = params.ConsensusPoW

common/batch/batch.go

@@ -0,0 +1,149 @@
+// batch.go
+
+package batch
+
+import (
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type Batch struct {
+	toInterval time.Duration
+	timeout    time.Duration
+	batchCount int
+	ch         chan interface{}
+	count      int32
+	f          func(interface{}, int) error
+}
+
+func NewBatch(toInterval, timeout time.Duration, batchCount int, f func(interface{}, int) error) *Batch {
+	return &Batch{
+		toInterval: toInterval,
+		timeout:    timeout,
+		batchCount: batchCount,
+		ch:         make(chan interface{}, batchCount*10),
+		count:      0,
+		f:          f,
+	}
+}
+
+func (b *Batch) Run() {
+	var (
+		data []interface{}
+		lt   time.Time = time.Now() // last time
+		ln   int       = 0          // last count
+	)
+
+	timer := time.NewTimer(0)
+	<-timer.C // drain the initial timeout
+
+	eod := false
+	for {
+		itstimer := false
+		fire := false
+
+		select {
+		case d := <-b.ch:
+			atomic.AddInt32(&b.count, -1)
+			if d == nil {
+				eod = true
+			} else {
+				data = append(data, d)
+			}
+		case <-timer.C:
+			itstimer = true
+		}
+		last := b.count == 0
+
+		if eod {
+			break
+		}
+
+		// when to fire
+		// 1. timer fired
+		//   1.1 no count change
+		//   1.2 more than 50 ms passed from the initial
+		// 2. count >= 100
+
+		if !itstimer {
+			if ln == 0 {
+				lt = time.Now()
+				ln = len(data)
+				timer.Stop()
+				timer.Reset(b.toInterval)
+			} else if len(data) >= b.batchCount {
+				fire = true
+			}
+		} else if last {
+			et := time.Since(lt)
+			if (len(data) == ln && et > b.toInterval) || et > b.timeout {
+				fire = true
+			}
+		}
+
+		if fire {
+			if len(data) < b.batchCount {
+				// do it
+				e := b.f(data, len(data))
+				if e != nil {
+					log.Error("Metadium Server", "Failed", e)
+				} else {
+					log.Debug("Metadium Server", "Count", len(data))
+				}
+				data = nil
+			} else {
+				for {
+					if len(data) < b.batchCount {
+						break
+					}
+
+					// do it
+					e := b.f(data, b.batchCount)
+					if e != nil {
+						log.Error("Metadium Server", "Failed", e)
+					} else {
+						log.Debug("Metadium Server", "Count", b.batchCount)
+					}
+					data = data[b.batchCount:]
+				}
+			}
+		}
+
+		lt = time.Now()
+		ln = len(data)
+
+		if itstimer && ln != 0 {
+			timer.Reset(b.toInterval)
+		} else if !itstimer && ln == 0 {
+			timer.Stop()
+		}
+	}
+
+	// got eod, flush the remaining data
+	for len(data) > 0 {
+		l := len(data)
+		if l > b.batchCount {
+			l = b.batchCount
+		}
+		e := b.f(data, l)
+		if e != nil {
+			log.Error("Metadium Server", "Failed", e)
+		} else {
+			log.Debug("Metadium Server", "Count", l)
+		}
+		data = data[l:]
+	}
+}
+
+func (b *Batch) Stop() {
+	b.ch <- nil
+}
+
+func (b *Batch) Put(data interface{}) {
+	b.ch <- data
+	atomic.AddInt32(&b.count, 1)
+}
+
+// EOF

metadium/miner/lrucache.go → common/lru/lrucache.go

@@ -1,6 +1,6 @@
 // lrucache.go
 
-package miner
+package lru
 
 import (
 	"container/list"
@@ -9,22 +9,31 @@ import (
 
 type LruCache struct {
 	lock  *sync.RWMutex
-	max   int
-	count int
+	max   int  // max count
+	count int  // current count
+	fifo  bool // if true, element order is not updated on access
 	lru   *list.List
 	data  map[interface{}]interface{}
 }
 
-func NewLruCache(max int) *LruCache {
+// NewLruCache creates LruCache
+func NewLruCache(max int, fifo bool) *LruCache {
 	return &LruCache{
 		lock:  &sync.RWMutex{},
 		max:   max,
 		count: 0,
+		fifo:  fifo,
 		lru:   list.New(),
 		data:  map[interface{}]interface{}{},
 	}
 }
 
+// Count returns the current count of elements
+func (c *LruCache) Count() int {
+	return c.count
+}
+
+// Put adds a key-value pair
 func (c *LruCache) Put(key, value interface{}) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
@@ -49,6 +58,7 @@ func (c *LruCache) Put(key, value interface{}) {
 	}
 }
 
+// Get returns a value with the given key if present, nil otherwise.
 func (c *LruCache) Get(key interface{}) interface{} {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
@@ -58,18 +68,22 @@ func (c *LruCache) Get(key interface{}) interface{} {
 		return nil
 	} else {
 		e := _e.(*list.Element)
-		c.lru.MoveToFront(e)
+		if !c.fifo {
+			c.lru.MoveToFront(e)
+		}
 		return e.Value.([]interface{})[1]
 	}
 }
 
+// Exists checks if a key exists.
 func (c *LruCache) Exists(key interface{}) bool {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 	_, ok := c.data[key]
 	return ok
 }
 
+// Del deletes a key-value pair if present. It returns true iff it's present.
 func (c *LruCache) Del(key interface{}) bool {
 	c.lock.Lock()
 	defer c.lock.Unlock()
@@ -87,4 +101,14 @@ func (c *LruCache) Del(key interface{}) bool {
 	}
 }
 
+// Clear resets the lru
+func (c *LruCache) Clear() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.count = 0
+	c.lru = list.New()
+	c.data = map[interface{}]interface{}{}
+}
+
 // EOF