worker support; minor fixes

README.md

@@ -1,6 +1,7 @@
 # derohe-proxy
 
 Proxy to combine miners and to reduce network load.
+Long To-Do list, but this is a working release.
 
 **Features**
 - random nonces
@@ -8,16 +9,16 @@ Proxy to combine miners and to reduce network load.
 - notification of incoming and lost connections / submitted results / stats
 - user-defined logging interval
 - pool mining support (not stratum)
-- set user-defined wallet address
+- worker support (wallet_address.worker_name)
 
 **Usage**
 
-```derohe-proxy [--listen-address=<127.0.0.1:11111>] [--log-interval=<60>] [--minimal] [--nonce] [--pool] --daemon-address=<1.2.3.4:10100> --wallet-address=<deroi...>```
+```derohe-proxy [--listen-address=<127.0.0.1:11111>] [--log-interval=<60>] [--minimal] [--nonce] [--pool] --daemon-address=<1.2.3.4:10100>```
 
 ```--listen-address (optional): bind to address:port for incoming miner connections. By default, proxy listens on 0.0.0.0:10200
 --daemon-address: address:port of daemon
---wallet address (optional): use this wallet address for connection to node
 --log-interval (optional): logging every X seconds, where X >= 60. Default is 60 seconds
 --minimal (optional): forward only 2 jobs per block (1 for first 9 miniblocks, 1 for final miniblock)
 --nonce (optional): enable random nonces, disabled by default```
 --pool (optional): enable pool mining, disable keyhash replacement
+--wallet-address=<dero1....>   use this wallet address for all connections

config/config.go

@@ -4,12 +4,12 @@ var Command_line string = `derohe-proxy
 Proxy to combine all miners and to reduce network load
 
 Usage:
-  derohe-proxy [--listen-address=<127.0.0.1:10100>] [--log-interval=<60>] [--minimal] [--nonce] [--pool] --daemon-address=<1.2.3.4:10100> --wallet-address=<deroi...>
+  derohe-proxy --daemon-address=<1.2.3.4:10100> [--listen-address=<127.0.0.1:10100>] [--log-interval=<60>] [--minimal] [--nonce] [--pool] [--wallet-address=<dero1...>]
 
 Options:
  --listen-address=<127.0.0.1:10100>		bind to specific address:port, default is 0.0.0.0:10200
  --daemon-address=<1.2.3.4:10100>		connect to this daemon
- --wallet-address=<deroi....>   use this wallet address for all connections
+ --wallet-address=<dero1....>   use this wallet address for all connections
  --log-interval=<60>   set logging interval in seconds (range 60 - 3600), default is 60 seconds
  --minimal   forward only 2 jobs per block (1 for miniblocks and 1 for final miniblock), by default all jobs are forwarded
  --nonce   enable nonce editing, default is off

derohe-proxy.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/deroproject/derohe/globals"
 	"github.com/docopt/docopt-go"
 )
 
@@ -42,7 +43,11 @@ func main() {
 	}
 
 	if config.Arguments["--wallet-address"] != nil {
-		config.WalletAddr = config.Arguments["--wallet-address"].(string)
+		addr, err := globals.ParseValidateAddress(config.Arguments["--wallet-address"].(string))
+		if err != nil {
+			fmt.Printf("%v Wallet address is invalid!\n", time.Now().Format(time.Stamp))
+		}
+		config.WalletAddr = addr.String()
 		fmt.Printf("%v Using wallet %s for all connections\n", time.Now().Format(time.Stamp), config.WalletAddr)
 	}
 

proxy/client.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math/rand"
 	"net/url"
+	"sync"
 	"time"
 
 	"derohe-proxy/config"
@@ -38,7 +39,12 @@ type (
 	}
 )
 
-var connection *websocket.Conn
+type clientConn struct {
+	conn *websocket.Conn
+	sync.Mutex
+}
+
+var connection clientConn
 var Blocks uint64
 var Minis uint64
 var Rejected uint64
@@ -68,7 +74,7 @@ func Start_client(w string) {
 		} else {
 			fmt.Printf("%v Connected to node %v using wallet %v\n", time.Now().Format(time.Stamp), config.Daemon_address, w)
 		}
-		connection, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
+		connection.conn, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
 		if err != nil {
 			time.Sleep(5 * time.Second)
 			fmt.Println(err)
@@ -78,7 +84,7 @@ func Start_client(w string) {
 		var params GetBlockTemplate_Result
 
 		for {
-			msg_type, recv_data, err := connection.ReadMessage()
+			msg_type, recv_data, err := connection.conn.ReadMessage()
 			if err != nil {
 				break
 			}
@@ -128,7 +134,7 @@ func SendUpdateToDaemon() {
 				time.Sleep(60 * time.Second)
 			}
 
-			connection.WriteJSON(MinerInfo_Params{Wallet_Address: Address, Miner_Tag: "", Miner_Hashrate: Hashrate})
+			connection.conn.WriteJSON(MinerInfo_Params{Wallet_Address: Address, Miner_Tag: "", Miner_Hashrate: Hashrate})
 
 			count++
 		}
@@ -137,5 +143,9 @@ func SendUpdateToDaemon() {
 }
 
 func SendToDaemon(buffer []byte) {
-	connection.WriteMessage(websocket.TextMessage, buffer)
+	connection.Lock()
+	defer connection.Unlock()
+
+	connection.conn.WriteMessage(websocket.TextMessage, buffer)
+
 }

proxy/server.go

@@ -38,6 +38,7 @@ type user_session struct {
 	miniblocks    uint64
 	lasterr       string
 	address       rpc.Address
+	worker        string
 	orphans       uint64
 	hashrate      float64
 	valid_address bool
@@ -80,7 +81,7 @@ func Start_server() {
 		TLSConfig:               tlsConfig,
 		Handler:                 mux,
 		MaxLoad:                 10 * 1024,
-		MaxWriteBufferSize:      32 * 1024,
+		MaxWriteBufferSize:      5 * 1024 * 1024,
 		ReleaseWebsocketPayload: true,
 		KeepaliveTime:           240 * time.Hour, // we expects all miners to find a block every 10 days,
 		NPoller:                 runtime.NumCPU(),
@@ -107,7 +108,9 @@ func Start_server() {
 func CountMiners() int {
 	client_list_mutex.Lock()
 	defer client_list_mutex.Unlock()
+
 	miners_count = len(client_list)
+
 	return miners_count
 }
 
@@ -151,6 +154,15 @@ func onWebsocket(w http.ResponseWriter, r *http.Request) {
 	}
 	address := strings.TrimPrefix(r.URL.Path, "/ws/")
 
+	// check for worker suffix
+	var parseWorker []string
+	var worker string
+	if strings.Contains(address, ".") {
+		parseWorker = strings.Split(address, ".")
+		worker = parseWorker[1]
+		address = parseWorker[0]
+	}
+
 	addr, err := globals.ParseValidateAddress(address)
 	if err != nil {
 		fmt.Fprintf(w, "err: %s\n", err)
@@ -167,7 +179,7 @@ func onWebsocket(w http.ResponseWriter, r *http.Request) {
 	addr_raw := addr.PublicKey.EncodeCompressed()
 	wsConn := conn.(*websocket.Conn)
 
-	session := user_session{address: *addr, address_sum: graviton.Sum(addr_raw)}
+	session := user_session{address: *addr, address_sum: graviton.Sum(addr_raw), worker: worker}
 	wsConn.SetSession(&session)
 
 	client_list_mutex.Lock()
@@ -183,9 +195,9 @@ func onWebsocket(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if !config.Pool_mode {
-		fmt.Printf("%v Incoming connection: %v, Wallet: %v\n", time.Now().Format(time.Stamp), wsConn.RemoteAddr().String(), address)
+		fmt.Printf("%v Incoming connection: %v (%v), Wallet: %v\n", time.Now().Format(time.Stamp), wsConn.RemoteAddr().String(), worker, address)
 	} else {
-		fmt.Printf("%v Incoming connection: %v\n", time.Now().Format(time.Stamp), wsConn.RemoteAddr().String())
+		fmt.Printf("%v Incoming connection: %v (%v)\n", time.Now().Format(time.Stamp), wsConn.RemoteAddr().String(), worker)
 	}
 }
 
@@ -224,7 +236,7 @@ func newUpgrader() *websocket.Upgrader {
 		*/
 		SendToDaemon(data)
 		if !config.Pool_mode {
-			fmt.Printf("%v Submitting result from miner: %v, Wallet: %v\n", time.Now().Format(time.Stamp), c.RemoteAddr().String(), client_list[c].address.String())
+			fmt.Printf("%v Submitting result from miner: %v (%v), Wallet: %v\n", time.Now().Format(time.Stamp), c.RemoteAddr().String(), client_list[c].worker, client_list[c].address.String())
 		} else {
 			shares++
 			fmt.Printf("%v Shares submitted: %d\n", time.Now().Format(time.Stamp), shares)
@@ -236,7 +248,7 @@ func newUpgrader() *websocket.Upgrader {
 		client_list_mutex.Lock()
 		defer client_list_mutex.Unlock()
 		Wallet_count[client_list[c].address.String()]--
-		fmt.Printf("%v Lost connection: %v\n", time.Now().Format(time.Stamp), c.RemoteAddr().String())
+		fmt.Printf("%v Lost connection: %v (%v)\n", time.Now().Format(time.Stamp), c.RemoteAddr().String(), client_list[c].worker)
 		delete(client_list, c)
 	})