joe/everytab
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

increase bundlegen producer host amount to ensure workers aren't starved

Browse source
This commit is contained in:
Joe Lothan 2026-05-25 16:21:49 -04:00
parent 4c7a0f54f7
commit 8ceb31bcbb
1 changed files with 6 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

9
pipeline/05_bundle_gen/main.go
View file

@ -121,15 +121,15 @@ func main() {
data []byte data []byte
} }
hostCh := make(chan HostRow, 6000) hostCh := make(chan HostRow, 50000)
entryCh := make(chan BundleEntry, 6000) entryCh := make(chan BundleEntry, 50000)
uploadCh := make(chan bundleJob, cfg.Uploaders*2) uploadCh := make(chan bundleJob, cfg.Uploaders*2)
// Stage 1: DB fetcher — continuously fetches pages into hostCh // Stage 1: DB fetcher — continuously fetches pages into hostCh
go func() { go func() {
defer close(hostCh) defer close(hostCh)
var lastRandom float64 = -1 var lastRandom float64 = -1
pageSize := cfg.EntriesPerBundle * 50 pageSize := 50000
fetched := 0 fetched := 0
for { for {
limit := pageSize limit := pageSize
@ -142,6 +142,7 @@ func main() {
limit = remaining limit = remaining
} }
} }
fetchStart := time.Now()
hosts, err := fetchHostsPage(ctx, pool, lastRandom, limit) hosts, err := fetchHostsPage(ctx, pool, lastRandom, limit)
if err != nil { if err != nil {
log.Fatalf("Failed to fetch hosts: %v", err) log.Fatalf("Failed to fetch hosts: %v", err)
@ -150,6 +151,8 @@ func main() {
break break
} }
lastRandom = hosts[len(hosts)-1].RandomOrder lastRandom = hosts[len(hosts)-1].RandomOrder
fmt.Printf("[fetcher] %d hosts in %dms (hostCh: %d/%d)\n",
len(hosts), time.Since(fetchStart).Milliseconds(), len(hostCh), cap(hostCh))
for _, h := range hosts { for _, h := range hosts {
hostCh <- h hostCh <- h
} }