added hub subsystem

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

eth/handler.go

@@ -998,7 +998,7 @@ func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) {
 
 	// Broadcast transactions to a batch of peers not knowing about it
 	for _, tx := range txs {
-		peers := pm.peers.PeersWithoutTx(tx.Hash())
+		peers := pm.peers.PeersWithoutTx2(tx.Hash())
 		for _, peer := range peers {
 			txset[peer] = append(txset[peer], tx)
 		}

eth/peer.go

@@ -28,7 +28,9 @@ import (
 	"github.com/ethereum/go-ethereum/common/lru"
 	"github.com/ethereum/go-ethereum/core/types"
 	metaapi "github.com/ethereum/go-ethereum/metadium/api"
+	metaminer "github.com/ethereum/go-ethereum/metadium/miner"
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
@@ -542,6 +544,42 @@ func (ps *peerSet) PeersWithoutTx(hash common.Hash) []*peer {
 	return list
 }
 
+// hub subsystem
+var (
+	isHub int = -1 //  -1: unset, 1: hub, 0: not hub
+)
+
+// PeersWithoutTx2 retrieves a list of peers that do not have a given transaction
+// in their set of known hashes.
+func (ps *peerSet) PeersWithoutTx2(hash common.Hash) []*peer {
+	if !metaminer.AmPartner() {
+		return ps.PeersWithoutTx(hash)
+	}
+
+	if isHub == -1 {
+		isHub = metaminer.AmHub(params.Hub)
+	}
+
+	if isHub == 0 {
+		// send it to the hub if it did not come from there
+		ps.lock.RLock()
+		for _, p := range ps.peers {
+			if p.id == params.Hub {
+				var list []*peer
+				if !p.knownTxs.Exists(hash) {
+					list = append(list, p)
+				}
+				ps.lock.RUnlock()
+				return list
+			}
+		}
+		ps.lock.RUnlock()
+	}
+
+	// fall back
+	return ps.PeersWithoutTx(hash)
+}
+
 // BestPeer retrieves the known peer with the currently highest total difficulty.
 func (ps *peerSet) BestPeer() *peer {
 	ps.lock.RLock()

metadium/admin.go

@@ -1043,6 +1043,21 @@ func IsPartner(id string) bool {
 	return true
 }
 
+// id is v4 id
+func AmHub(id string) int {
+	if admin == nil || admin.self == nil {
+		return -1
+	}
+
+	admin.lock.Lock()
+	defer admin.lock.Unlock()
+	if strings.HasPrefix(strings.ToUpper(admin.self.Id), strings.ToUpper(id)) {
+		return 1
+	} else {
+		return 0
+	}
+}
+
 func (ma *metaAdmin) pendingEmpty() bool {
 	type txpool_status struct {
 		Pending hexutil.Uint `json:"pending"`
@@ -1425,6 +1440,7 @@ func init() {
 	metaminer.IsMinerFunc = IsMiner
 	metaminer.AmPartnerFunc = AmPartner
 	metaminer.IsPartnerFunc = IsPartner
+	metaminer.AmHubFunc = AmHub
 	metaminer.LogBlockFunc = LogBlock
 	metaminer.SuggestGasPriceFunc = suggestGasPrice
 	metaminer.CalculateRewardsFunc = calculateRewards

metadium/miner/miner.go

@@ -14,6 +14,7 @@ var (
 	IsMinerFunc            func() bool
 	AmPartnerFunc          func() bool
 	IsPartnerFunc          func(string) bool
+	AmHubFunc              func(string) int
 	LogBlockFunc           func(int64, common.Hash)
 	CalculateRewardsFunc   func(*big.Int, *big.Int, *big.Int, func(common.Address, *big.Int)) (*common.Address, []byte, error)
 	VerifyRewardsFunc      func(*big.Int, string) error
@@ -48,6 +49,14 @@ func AmPartner() bool {
 	}
 }
 
+func AmHub(id string) int {
+	if AmHubFunc == nil {
+		return -1
+	} else {
+		return AmHubFunc(id)
+	}
+}
+
 func LogBlock(height int64, hash common.Hash) {
 	if LogBlockFunc != nil {
 		LogBlockFunc(height, hash)

metadium/scripts/gmet.sh

@@ -175,6 +175,7 @@ function start ()
 
     OPTS="$COINBASE $DISCOVER $RPCOPT $BOOT_NODES $NONCE_LIMIT $TESTNET ${GMET_OPTS}"
     [ "$PORT" = "" ] || OPTS="${OPTS} --port $(($PORT + 1))"
+    [ "$HUB" = "" ] || OPTS="${OPTS} --hub ${HUB}"
 
     [ -d "$d/logs" ] || mkdir -p $d/logs