implement generic worker pool

Add Job abstraction and define interface for workers. Implement concrete worker for blocks and transactions.

Author: philenius

export.go

@@ -4,13 +4,13 @@ import (
 	"bufio"
 	"fmt"
 	"os"
-	"philenius/ethereum-transaction-export/models"
+	"philenius/ethereum-transaction-export/work"
 	"time"
 
 	"github.com/mgutz/logxi/v1"
 )
 
-func exportAsCSV(txChan chan *models.Tx) {
+func exportAsCSV(jobs chan *work.Job) {
 
 	now := time.Now().Format("2006-01-02-15-04-05")
 	f, err := os.Create(fmt.Sprintf("geth_tx_export_%s.csv", now))
@@ -24,13 +24,14 @@ func exportAsCSV(txChan chan *models.Tx) {
 	w.Flush()
 
 	lineBuf := 0
-	for transaction := range txChan {
-		tx := transaction.Tx
+	for job := range jobs {
+		tx := job.Tx
 
 		lineBuf++
 		line := fmt.Sprintf(
 			"%s,%d,%s,%d,%d,%s,%s,%d,%d,%d,%s,%d\n",
-			tx.Hash, tx.Nonce, tx.BlockHash, *tx.BlockNumber, *tx.TransactionIndex, tx.From, tx.To, tx.Value.Int64(), tx.Gas, tx.GasPrice.Int64(), tx.Input, transaction.Timestamp,
+			tx.Hash, tx.Nonce, tx.BlockHash, *tx.BlockNumber, *tx.TransactionIndex, tx.From, tx.To,
+			tx.Value.Int64(), tx.Gas, tx.GasPrice.Int64(), tx.Input, job.Timestamp,
 		)
 		w.WriteString(line)
 
@@ -43,7 +44,7 @@ func exportAsCSV(txChan chan *models.Tx) {
 	w.Flush()
 }
 
-func exportFailedBlocks(blockChan chan int) {
+func exportFailedBlockJobs(jobs chan *work.Job) {
 
 	f, err := os.Create("failedBlocks.txt")
 	if err != nil {
@@ -52,16 +53,16 @@ func exportFailedBlocks(blockChan chan int) {
 
 	w := bufio.NewWriter(f)
 
-	for block := range blockChan {
-		w.WriteString(fmt.Sprintf("%d\n", block))
+	for job := range jobs {
+		w.WriteString(fmt.Sprintf("%d\n", job.BlockHeight))
 		w.Flush()
 	}
 
 	log.Info("finished writing to block error file")
 	w.Flush()
 }
 
-func exportFailedTx(txChan chan *models.TxHash) {
+func exportFailedTxJobs(jobs chan *work.Job) {
 
 	f, err := os.Create("failedTransactions.txt")
 	if err != nil {
@@ -70,8 +71,8 @@ func exportFailedTx(txChan chan *models.TxHash) {
 
 	w := bufio.NewWriter(f)
 
-	for tx := range txChan {
-		w.WriteString(fmt.Sprintf("%s\n", tx.Hash))
+	for job := range jobs {
+		w.WriteString(fmt.Sprintf("%s\n", job.TxHash))
 		w.Flush()
 	}
 

main.go

@@ -9,7 +9,6 @@ import (
 	"github.com/mgutz/logxi/v1"
 	"github.com/onrik/ethrpc"
 
-	"philenius/ethereum-transaction-export/models"
 	"philenius/ethereum-transaction-export/work"
 )
 
@@ -21,6 +20,7 @@ func main() {
 	port := flag.Int("port", 8545, "The port number of the Ethereum node.")
 	startHeight := flag.Int("start", 0, "The height / number of the block where to start.")
 	blockCount := flag.Int("count", 1000, "The total amount of blocks to fetch.")
+	blockConcurr := flag.Int("blockConcurr", 10, "The count of concurrent workers for fetching block.")
 	txConcurr := flag.Int("txConcurr", 10, "The count of concurrent workers for fetching transactions.")
 	flag.Parse()
 
@@ -32,11 +32,11 @@ func main() {
 	}
 	log.Info("successfully connected to Ethereum node", "host", *hostname, "port", *port, "version", version)
 
-	blockHeightChan := make(chan int, 10000)
-	txHashChan := make(chan *models.TxHash, 10000)
-	txChan := make(chan *models.Tx, 10000)
-	failedBlockChan := make(chan int, 10000)
-	failedTxChan := make(chan *models.TxHash, 10000)
+	blockHeightChan := make(chan *work.Job, 10000)
+	txHashChan := make(chan *work.Job, 10000)
+	txChan := make(chan *work.Job, 10000)
+	failedBlockChan := make(chan *work.Job, 10000)
+	failedTxChan := make(chan *work.Job, 10000)
 	latestBlock := 0
 	latestTransactionCount := int64(0)
 	wt := sync.WaitGroup{}
@@ -67,7 +67,7 @@ func main() {
 	go func() {
 		endHeight := *startHeight + *blockCount
 		for i := *startHeight; i < endHeight; i++ {
-			blockHeightChan <- i
+			blockHeightChan <- &work.Job{BlockHeight: i}
 		}
 		log.Info("finished listing block numbers")
 		close(blockHeightChan)
@@ -77,44 +77,28 @@ func main() {
 
 	// fetch all blocks
 	go func() {
-		for blockHeight := range blockHeightChan {
-			latestBlock = blockHeight
-			block, err := client.EthGetBlockByNumber(blockHeight, true)
-			if err != nil {
-				failedBlockChan <- blockHeight
-				log.Error("failed to get block", "blockNumber", *blockCount, "err", err.Error())
-				continue
-			}
-			if log.IsDebug() {
-				log.Debug("successfully got block", "blockNumber", block.Number)
-			}
-			for _, tx := range block.Transactions {
-				txHashChan <- &models.TxHash{block.Timestamp, tx.Hash}
-			}
-		}
-		close(txHashChan)
-		close(failedBlockChan)
-		log.Info("finished fetching all blocks")
+		p := work.NewBlockWorkerPool(*blockConcurr, clientAddr, blockHeightChan, txHashChan, failedBlockChan)
+		p.Run()
 		wt.Done()
 	}()
 	wt.Add(1)
 
 	// fetch all transactions
 	go func() {
-		p := work.NewPool(*txConcurr, clientAddr, txHashChan, txChan, failedTxChan)
+		p := work.NewTxWorkerPool(*txConcurr, clientAddr, txHashChan, txChan, failedTxChan)
 		p.Run()
 		wt.Done()
 	}()
 	wt.Add(1)
 
 	go func() {
-		exportFailedBlocks(failedBlockChan)
+		exportFailedBlockJobs(failedBlockChan)
 		wt.Done()
 	}()
 	wt.Add(1)
 
 	go func() {
-		exportFailedTx(failedTxChan)
+		exportFailedTxJobs(failedTxChan)
 		wt.Done()
 	}()
 	wt.Add(1)

models/wrapperTypes.go

@@ -0,0 +1,36 @@
+package work
+
+import (
+	"sync"
+
+	log "github.com/mgutz/logxi/v1"
+	"github.com/onrik/ethrpc"
+)
+
+type BlockWorker struct {
+	client     *ethrpc.EthRPC
+	jobs       chan *Job
+	result     chan *Job
+	failedJobs chan *Job
+	wt         *sync.WaitGroup
+}
+
+func (w *BlockWorker) doWork() {
+	for job := range w.jobs {
+		//latestBlock = blockHeight
+		block, err := w.client.EthGetBlockByNumber(job.BlockHeight, true)
+		if err != nil {
+			w.failedJobs <- job
+			log.Error("failed to get block", "blockNumber", job.BlockHeight, "err", err.Error())
+			continue
+		}
+		if log.IsDebug() {
+			log.Debug("successfully got block", "blockNumber", block.Number)
+		}
+		for _, tx := range block.Transactions {
+			w.result <- &Job{TxHash: tx.Hash, Timestamp: block.Timestamp}
+		}
+	}
+	log.Info("block worker finished")
+	w.wt.Done()
+}

work/blockWorker.go

@@ -0,0 +1,12 @@
+package work
+
+import (
+	"github.com/onrik/ethrpc"
+)
+
+type Job struct {
+	BlockHeight int
+	TxHash      string
+	Tx          *ethrpc.Transaction
+	Timestamp   int
+}

work/job.go

@@ -0,0 +1,38 @@
+package work
+
+import (
+	"sync"
+
+	log "github.com/mgutz/logxi/v1"
+	"github.com/onrik/ethrpc"
+)
+
+type TxWorker struct {
+	client     *ethrpc.EthRPC
+	jobs       chan *Job
+	result     chan *Job
+	failedJobs chan *Job
+	wt         *sync.WaitGroup
+}
+
+func (w *TxWorker) doWork() {
+	for job := range w.jobs {
+		//latestTransactionCount++
+		transaction, err := w.client.EthGetTransactionByHash(job.TxHash)
+		if err != nil || transaction.BlockNumber == nil || transaction.TransactionIndex == nil {
+			w.failedJobs <- job
+			log.Error("failed to get transaction", "txHash", job.TxHash)
+			continue
+		}
+
+		w.result <- &Job{
+			Timestamp: job.Timestamp,
+			Tx:        transaction,
+		}
+		if log.IsDebug() {
+			log.Debug("successfully got transaction", "txHash", transaction.Hash)
+		}
+	}
+	log.Info("transaction worker finished")
+	w.wt.Done()
+}

work/txWorker.go

@@ -1,39 +1,5 @@
 package work
 
-import (
-	"philenius/ethereum-transaction-export/models"
-	"sync"
-
-	log "github.com/mgutz/logxi/v1"
-	"github.com/onrik/ethrpc"
-)
-
-type Worker struct {
-	client     *ethrpc.EthRPC
-	jobs       chan *models.TxHash
-	result     chan *models.Tx
-	failedJobs chan *models.TxHash
-	wt         *sync.WaitGroup
-}
-
-func (w *Worker) doWork() {
-	for txHash := range w.jobs {
-		//latestTransactionCount++
-		transaction, err := w.client.EthGetTransactionByHash(txHash.Hash)
-		if err != nil || transaction.BlockNumber == nil || transaction.TransactionIndex == nil {
-			w.failedJobs <- txHash
-			log.Error("failed to get transaction", "txHash", txHash.Hash)
-			continue
-		}
-
-		w.result <- &models.Tx{
-			Timestamp: txHash.Timestamp,
-			Tx:        transaction,
-		}
-		if log.IsDebug() {
-			log.Debug("successfully got transaction", "txHash", transaction.Hash)
-		}
-	}
-	log.Info("finished fetching all transactions")
-	w.wt.Done()
+type Worker interface {
+	doWork()
 }

work/worker.go

@@ -1,7 +1,6 @@
 package work
 
 import (
-	"philenius/ethereum-transaction-export/models"
 	"sync"
 
 	log "github.com/mgutz/logxi/v1"
@@ -10,26 +9,43 @@ import (
 
 type Pool struct {
 	wt         *sync.WaitGroup
-	worker     []*Worker
-	jobs       chan *models.TxHash
-	results    chan *models.Tx
-	failedJobs chan *models.TxHash
+	workers    []Worker
+	results    chan *Job
+	failedJobs chan *Job
 }
 
-// NewPool instantiates a pool of workers with given concurrency
-func NewPool(concurrency int, clientAddr string, jobs chan *models.TxHash, results chan *models.Tx, failedJobs chan *models.TxHash) *Pool {
+// NewTxWorkerPool instantiates a pool of workers with given concurrency for fetching transactions.
+func NewTxWorkerPool(concurrency int, clientAddr string, jobs chan *Job, results chan *Job, failedJobs chan *Job) *Pool {
 
 	wt := &sync.WaitGroup{}
 
-	workerArr := make([]*Worker, concurrency)
+	workerArr := make([]Worker, concurrency)
 	for i := 0; i < concurrency; i++ {
-		workerArr[i] = &Worker{ethrpc.NewEthRPC(clientAddr), jobs, results, failedJobs, wt}
+		workerArr[i] = &TxWorker{ethrpc.NewEthRPC(clientAddr), jobs, results, failedJobs, wt}
 	}
 
 	return &Pool{
 		wt:         wt,
-		worker:     workerArr,
-		jobs:       jobs,
+		workers:    workerArr,
+		results:    results,
+		failedJobs: failedJobs,
+	}
+
+}
+
+// NewBlockWorkerPool instantiates a pool of workers with given concurrency for fetching blocks.
+func NewBlockWorkerPool(concurrency int, clientAddr string, jobs chan *Job, results chan *Job, failedJobs chan *Job) *Pool {
+
+	wt := &sync.WaitGroup{}
+
+	workerArr := make([]Worker, concurrency)
+	for i := 0; i < concurrency; i++ {
+		workerArr[i] = &BlockWorker{ethrpc.NewEthRPC(clientAddr), jobs, results, failedJobs, wt}
+	}
+
+	return &Pool{
+		wt:         wt,
+		workers:    workerArr,
 		results:    results,
 		failedJobs: failedJobs,
 	}
@@ -38,7 +54,7 @@ func NewPool(concurrency int, clientAddr string, jobs chan *models.TxHash, resul
 
 // Run starts the processing of incoming jobs
 func (p *Pool) Run() {
-	for _, w := range p.worker {
+	for _, w := range p.workers {
 		go w.doWork()
 		p.wt.Add(1)
 	}