Merge remote-tracking branch 'upstream/lighthorizon' into lite_search…

…_account
sreuland · Jul 21, 2022 · 223626e · 223626e
2 parents ff5b145 + 8c9eec3
commit 223626e
Show file tree

Hide file tree

Showing 5 changed files with 21 additions and 8 deletions.
diff --git a/exp/lighthorizon/index/backend/file.go b/exp/lighthorizon/index/backend/file.go
@@ -20,6 +20,10 @@ type FileBackend struct {
 }
 
 func NewFileBackend(dir string, parallel uint32) (*FileBackend, error) {
+	if parallel <= 0 {
+		parallel = 1
+	}
+
 	return &FileBackend{
 		dir:      dir,
 		parallel: parallel,

diff --git a/exp/lighthorizon/index/backend/parallel_flush.go b/exp/lighthorizon/index/backend/parallel_flush.go
@@ -23,7 +23,10 @@ func parallelFlush(parallel uint32, allIndexes map[string]types.NamedIndices, f
 
 	wg.Add(1)
 	go func() {
+		// forces this async func to be waited on also, otherwise the outer
+		// method returns before this finishes.
 		defer wg.Done()
+
 		for account, indexes := range allIndexes {
 			batches <- &batch{
 				account: account,
@@ -43,15 +46,17 @@ func parallelFlush(parallel uint32, allIndexes map[string]types.NamedIndices, f
 			defer wg.Done()
 			for batch := range batches {
 				if err := f(batch); err != nil {
-					log.Error(err)
+					log.Errorf("Error occurred writing batch: %v, retrying...", err)
 					time.Sleep(50 * time.Millisecond)
 					batches <- batch
 					continue
 				}
 
 				nwritten := atomic.AddUint64(&written, 1)
 				if nwritten%1000 == 0 {
-					log.Infof("Writing indexes... %d/%d %.2f%%", nwritten, len(allIndexes), (float64(nwritten)/float64(len(allIndexes)))*100)
+					log.Infof("Writing indexes... %d/%d %.2f%%", nwritten,
+						len(allIndexes),
+						(float64(nwritten)/float64(len(allIndexes)))*100)
 				}
 
 				if nwritten == uint64(len(allIndexes)) {

diff --git a/exp/lighthorizon/index/builder.go b/exp/lighthorizon/index/builder.go
@@ -127,7 +127,7 @@ func BuildIndices(
 				}
 
 				nprocessed := atomic.AddUint64(&processed, uint64(count))
-				if nprocessed%19 == 0 {
+				if nprocessed%97 == 0 {
 					printProgress("Reading ledgers", nprocessed, uint64(ledgerCount), startTime)
 				}
 

diff --git a/exp/lighthorizon/index/cmd/reduce.sh b/exp/lighthorizon/index/cmd/reduce.sh
@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/usr/bin/env bash
 # 
 # Combines indices that were built separately in different folders into a single
 # set of indices.
@@ -22,7 +22,7 @@ fi
 
 if [[ ! -d "$2" ]]; then 
     echo "Warning: index dest ('$2') does not exist, creating..."
-    mkdir -p $2
+    mkdir -p "$2"
 fi
 
 MAP_JOB_COUNT=$(ls $1 | grep -E 'job_[0-9]+' | wc -l)
@@ -50,8 +50,8 @@ do
     AWS_BATCH_JOB_ARRAY_INDEX=$i MAP_JOB_COUNT=$MAP_JOB_COUNT \
     REDUCE_JOB_COUNT=$REDUCE_JOB_COUNT WORKER_COUNT=4 \
     INDEX_SOURCE_ROOT=file://$1 INDEX_TARGET=file://$2 \
-        ./reduce &
-    
+        timeout -k 30s 10s ./reduce &
+
     echo "pid=$!"
     pids+=($!)
 done
@@ -72,4 +72,4 @@ done
 
 rm ./reduce # cleanup
 echo "All jobs succeeded!"
-exit 0
+exit 0
diff --git a/exp/lighthorizon/index/connect.go b/exp/lighthorizon/index/connect.go
@@ -15,6 +15,10 @@ func Connect(backendUrl string) (Store, error) {
 }
 
 func ConnectWithConfig(config StoreConfig) (Store, error) {
+	if config.Workers <= 0 {
+		config.Workers = 1
+	}
+
 	parsed, err := url.Parse(config.Url)
 	if err != nil {
 		return nil, err