From cdeb43177cc8424bf44bff1f32f7ac0f15bbb449 Mon Sep 17 00:00:00 2001 From: Abdul Rabbani Date: Tue, 14 Jun 2022 09:27:59 -0400 Subject: [PATCH] Send slots to processedCh and remove --- pkg/beaconclient/capturehistoric.go | 3 +++ pkg/beaconclient/processhistoric.go | 10 +++++++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/pkg/beaconclient/capturehistoric.go b/pkg/beaconclient/capturehistoric.go index a6764db..af3bd69 100644 --- a/pkg/beaconclient/capturehistoric.go +++ b/pkg/beaconclient/capturehistoric.go @@ -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 diff --git a/pkg/beaconclient/processhistoric.go b/pkg/beaconclient/processhistoric.go index dcff8f4..7f2096d 100644 --- a/pkg/beaconclient/processhistoric.go +++ b/pkg/beaconclient/processhistoric.go @@ -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)) -- 2.45.2