6 Commits
Author SHA1 Message Date
Abdul Rabbani bb8c861bfd Merge pull request #63 from vulcanize/develop
Improve entrypoint to be more helpful
2022-06-14 09:36:38 -04:00
Abdul Rabbani d5a9f999b3 Merge pull request #64 from vulcanize/feature/bug-remove-processed-slots
Send slots to processedCh and remove
2022-06-14 09:35:28 -04:00
Abdul Rabbani cdeb43177c Send slots to processedCh and remove 2022-06-14 09:27:59 -04:00
Abdul Rabbani 67c937418e Improve entrypoint to be more helpful 2022-06-14 09:06:51 -04:00
Abdul Rabbani 802f80057b Merge pull request #60 from vulcanize/develop
Fix datarace condition
2022-06-13 12:26:27 -04:00
Abdul Rabbani 64a89d8a46 Fix datarace condition
By creating local variables within each goroutine.
2022-06-13 12:15:18 -04:00
5 changed files with 23 additions and 10 deletions
+1
View File
@@ -8,3 +8,4 @@ temp/*
pkg/beaconclient/ssz-data/
*.test
ipld-eth-beacon-indexer.log
config/local.ipld-eth-beacon-indexer-config.json
+10 -6
View File
@@ -3,17 +3,21 @@
sleep 10
echo "Starting ipld-eth-beacon-indexer"
echo /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json
echo /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
/root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json
/root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
rv=$?
if [ $rv != 0 ]; then
echo "ipld-eth-beacon-indexer startup failed"
echo 1 > /root/HEALTH
echo "ipld-eth-beacon-indexer failed"
echo $rv > /root/HEALTH
echo $rv
cat /root/ipld-eth-beacon-indexer.output
else
echo "ipld-eth-beacon-indexer startup succeeded"
echo 0 > /root/HEALTH
echo "ipld-eth-beacon-indexer succeeded"
echo $rv > /root/HEALTH
echo $rv
cat /root/ipld-eth-beacon-indexer.output
fi
tail -f /dev/null
+3
View File
@@ -100,6 +100,8 @@ func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing,
// Start workers
for w := 1; w <= maxWorkers; w++ {
log.WithFields(log.Fields{"maxWorkers": maxWorkers}).Debug("Starting batch processing workers")
// Pass in function to increment metric! KnownGapProcessing or HistoricProcessing
go processSlotRangeWorker(ctx, workCh, errCh, db, serverEndpoint, metrics, checkDb)
}
@@ -125,6 +127,7 @@ func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing,
} else if slots.startSlot == slots.endSlot {
log.WithField("slot", slots.startSlot).Debug("Added new slot to workCh")
workCh <- slots.startSlot
processedCh <- slots
} else {
for i := slots.startSlot; i <= slots.endSlot; i++ {
workCh <- i
+2 -1
View File
@@ -545,7 +545,6 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
var (
isInBeaconState bool
isInSignedBeaconBlock bool
err error
)
errG, _ := errgroup.WithContext(context.Background())
errG.Go(func() error {
@@ -553,6 +552,7 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
case <-ctx.Done():
return nil
default:
var err error
isInBeaconState, err = checkSlotAndRoot(db, CheckBeaconStateStmt, slot, stateRoot)
if err != nil {
loghelper.LogError(err).Error("Unable to check if the slot and stateroot exist in eth_beacon.state")
@@ -565,6 +565,7 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
case <-ctx.Done():
return nil
default:
var err error
isInSignedBeaconBlock, err = checkSlotAndRoot(db, CheckSignedBeaconBlockStmt, slot, blockRoot)
if err != nil {
loghelper.LogError(err).Error("Unable to check if the slot and block_root exist in eth_beacon.signed_block")
+7 -3
View File
@@ -237,8 +237,11 @@ func removeRowPostProcess(ctx context.Context, db sql.Database, processCh <-chan
case slots := <-processCh:
// Make sure the start and end slot exist in the slots table.
go func() {
finishedProcess := false
for !finishedProcess {
log.WithFields(log.Fields{
"startSlot": slots.startSlot,
"endSlot": slots.endSlot,
}).Debug("Starting to check to see if the following slots have been processed")
for {
isStartProcess, err := isSlotProcessed(db, checkProcessedStmt, strconv.Itoa(slots.startSlot))
if err != nil {
errCh <- err
@@ -248,8 +251,9 @@ func removeRowPostProcess(ctx context.Context, db sql.Database, processCh <-chan
errCh <- err
}
if isStartProcess && isEndProcess {
finishedProcess = true
break
}
time.Sleep(1000 * time.Millisecond)
}
_, err := db.Exec(context.Background(), removeStmt, strconv.Itoa(slots.startSlot), strconv.Itoa(slots.endSlot))