diff --git a/ccrawl/crawl.go b/ccrawl/crawl.go index c49baa7..d744ce9 100644 --- a/ccrawl/crawl.go +++ b/ccrawl/crawl.go @@ -4,7 +4,6 @@ import ( "bufio" "bytes" "compress/gzip" - "container/heap" "context" "crypto/sha1" "encoding/hex" @@ -19,114 +18,8 @@ import ( "github.com/andybalholm/brotli" ) -// ── frontier ────────────────────────────────────────────────────────────────── - -// FrontierEntry is one URL in the crawl frontier. -type FrontierEntry struct { - URL string // normalized URL - Host string // hostname for politeness grouping - Priority float32 // harmonic centrality (higher = crawl sooner) - NextAt int64 // earliest Unix timestamp to fetch - Depth uint8 // BFS depth from seed - Retries uint8 -} - -// frontierHeap implements a max-heap on Priority. -type frontierHeap []FrontierEntry - -func (h frontierHeap) Len() int { return len(h) } -func (h frontierHeap) Less(i, j int) bool { return h[i].Priority > h[j].Priority } -func (h frontierHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } -func (h *frontierHeap) Push(x any) { *h = append(*h, x.(FrontierEntry)) } -func (h *frontierHeap) Pop() any { - old := *h - n := len(old) - x := old[n-1] - *h = old[:n-1] - return x -} - -// Frontier is an in-memory URL frontier with a priority heap and per-host -// politeness (minimum delay between requests to the same host). -type Frontier struct { - mu sync.Mutex - heap frontierHeap - seen map[string]struct{} // URL SHA-1 dedup - hostAt map[string]int64 // last-fetch Unix timestamp per host - delay time.Duration // minimum inter-request delay per host -} - -// NewFrontier creates a Frontier with the given per-host politeness delay. -func NewFrontier(delay time.Duration) *Frontier { - f := &Frontier{ - seen: make(map[string]struct{}), - hostAt: make(map[string]int64), - delay: delay, - } - heap.Init(&f.heap) - return f -} - -// Add enqueues a URL if it has not been seen before. Returns true if added. -func (f *Frontier) Add(e FrontierEntry) bool { - f.mu.Lock() - defer f.mu.Unlock() - key := urlSHA1(e.URL) - if _, ok := f.seen[key]; ok { - return false - } - f.seen[key] = struct{}{} - heap.Push(&f.heap, e) - return true -} - -// Len returns the number of queued URLs. -func (f *Frontier) Len() int { - f.mu.Lock() - defer f.mu.Unlock() - return f.heap.Len() -} - -// Pop removes and returns the highest-priority URL that is eligible to crawl -// now (host delay has elapsed). Returns false if no eligible URL exists. -// -// To avoid starvation when the top-priority host is in its politeness window, -// Pop scans up to scanLimit candidates before giving up. Skipped (ineligible) -// entries are temporarily held and re-pushed so the heap remains consistent. -func (f *Frontier) Pop(now int64) (FrontierEntry, bool) { - const scanLimit = 16 - f.mu.Lock() - defer f.mu.Unlock() - - delayNanos := int64(f.delay) - var skipped []FrontierEntry - - for f.heap.Len() > 0 && len(skipped) < scanLimit { - e := heap.Pop(&f.heap).(FrontierEntry) - lastAt := f.hostAt[e.Host] - minAt := lastAt + delayNanos/int64(time.Second) // delay in seconds - if now >= minAt && now >= e.NextAt { - for _, s := range skipped { - heap.Push(&f.heap, s) - } - f.hostAt[e.Host] = now - return e, true - } - skipped = append(skipped, e) - } - // restore skipped entries - for _, s := range skipped { - heap.Push(&f.heap, s) - } - return FrontierEntry{}, false -} - -// urlSHA1 returns the hex SHA-1 of a URL string (used for deduplication). -func urlSHA1(rawURL string) string { - h := sha1.New() - _, _ = io.WriteString(h, rawURL) - return hex.EncodeToString(h.Sum(nil)) -} +// The frontier itself lives in frontier.go, on disk. It used to be a heap and +// a map right here, which is fine until the seed set is bigger than memory. // ContentSHA1 returns the hex SHA-1 of raw content bytes (matches CC's digest // field in CDX records). diff --git a/ccrawl/crawl_test.go b/ccrawl/crawl_test.go index 86d1d4a..941375d 100644 --- a/ccrawl/crawl_test.go +++ b/ccrawl/crawl_test.go @@ -12,6 +12,8 @@ import ( func TestFrontierAddDedup(t *testing.T) { f := NewFrontier(time.Second) + defer func() { _ = f.Close() }() + added := f.Add(FrontierEntry{URL: "https://example.com/", Host: "example.com", Priority: 1.0}) if !added { t.Error("first Add should return true") @@ -20,19 +22,30 @@ func TestFrontierAddDedup(t *testing.T) { if added2 { t.Error("duplicate URL should not be added again") } + if err := f.Flush(); err != nil { + t.Fatalf("Flush: %v", err) + } if f.Len() != 1 { t.Errorf("Frontier.Len = %d, want 1", f.Len()) } + if s := f.Stats(); s.Admitted != 1 || s.Duplicates != 1 { + t.Errorf("Stats = %+v, want Admitted 1 and Duplicates 1", s) + } } func TestFrontierPriority(t *testing.T) { f := NewFrontier(0) // no delay for test + defer func() { _ = f.Close() }() + f.Add(FrontierEntry{URL: "https://low.com/", Host: "low.com", Priority: 1.0}) f.Add(FrontierEntry{URL: "https://high.com/", Host: "high.com", Priority: 100.0}) f.Add(FrontierEntry{URL: "https://mid.com/", Host: "mid.com", Priority: 50.0}) now := time.Now().Unix() - e1, ok := f.Pop(now) + e1, ok, err := f.Pop(now) + if err != nil { + t.Fatalf("Pop: %v", err) + } if !ok { t.Fatal("Pop returned nothing") } @@ -43,11 +56,16 @@ func TestFrontierPriority(t *testing.T) { func TestFrontierPoliteness(t *testing.T) { f := NewFrontier(10 * time.Second) + defer func() { _ = f.Close() }() + f.Add(FrontierEntry{URL: "https://example.com/a", Host: "example.com", Priority: 1.0}) f.Add(FrontierEntry{URL: "https://example.com/b", Host: "example.com", Priority: 0.5}) now := time.Now().Unix() - e1, ok := f.Pop(now) + e1, ok, err := f.Pop(now) + if err != nil { + t.Fatalf("Pop: %v", err) + } if !ok { t.Fatal("first Pop should succeed") } @@ -55,12 +73,16 @@ func TestFrontierPoliteness(t *testing.T) { t.Fatalf("unexpected host %s", e1.Host) } // second Pop for same host should fail (delay not elapsed) - _, ok2 := f.Pop(now) - if ok2 { + if _, ok2, err := f.Pop(now); err != nil { + t.Fatalf("Pop: %v", err) + } else if ok2 { t.Error("second Pop for same host within delay should fail") } // after delay has elapsed it should succeed - e3, ok3 := f.Pop(now + 15) + e3, ok3, err := f.Pop(now + 15) + if err != nil { + t.Fatalf("Pop: %v", err) + } if !ok3 { t.Error("Pop after delay elapsed should succeed") } diff --git a/ccrawl/frontier.go b/ccrawl/frontier.go new file mode 100644 index 0000000..5175db7 --- /dev/null +++ b/ccrawl/frontier.go @@ -0,0 +1,811 @@ +package ccrawl + +import ( + "bytes" + "container/list" + "context" + "crypto/sha1" + "database/sql" + "encoding/binary" + "fmt" + "sort" + "strings" + "sync" + "time" + + _ "modernc.org/sqlite" // pure Go driver, no cgo +) + +// Frontier states. A URL is pending until something claims it, claimed while a +// worker holds it, and then done or failed for good. +const ( + frontierPending = 0 + frontierClaimed = 1 + frontierDone = 2 + frontierFailed = 3 +) + +// FrontierEntry is one URL in the crawl frontier. +type FrontierEntry struct { + URL string // normalized URL + Host string // hostname for politeness grouping + Priority float32 // harmonic centrality (higher = crawl sooner) + NextAt int64 // earliest Unix timestamp to fetch + Depth uint8 // BFS depth from seed + Retries uint8 +} + +// FrontierConfig configures a frontier. The zero value is a usable in memory +// frontier, which is what the tests and small runs want. +type FrontierConfig struct { + // Path is the SQLite file holding the state. Empty means a private in + // memory database, which is gone when the process is. + Path string + // Delay is the minimum spacing between two requests to the same host. + Delay time.Duration + // MergeSize is how many staged URLs pile up before they are merged into the + // frontier. The merge is one ordered sweep of the frontier B-tree whatever it + // carries, so a bigger number is cheaper per URL and costs a longer wait + // before a newly discovered URL becomes poppable. + MergeSize int + // SeenURLs is how many URL hashes the in memory seen cache holds per + // generation, so the resident set is between one and two times it. Under + // sizing it does not lose URLs, it just sends more of the seen check to the + // disk. Roughly 50 bytes a key, so the default is around 100 MB. + SeenURLs int + // BatchSize is how many admissions accumulate before a transaction. One + // insert per URL is roughly two orders of magnitude slower than a batch. + BatchSize int + // ClaimSize is how many rows one refill claims. Larger amortises the query + // over more pops and leaves more work to reclaim after a crash. + ClaimSize int + // HotHosts is the size of the in memory host politeness cache. + HotHosts int + // SyncEvery is how many completions buffer before they are written, and so + // how many URLs a kill can cost. It defaults to ClaimSize. Set it to 1 for a + // frontier where a kill must never cause a refetch, at roughly a quarter of + // the pop rate. + SyncEvery int + // CacheMB is the SQLite page cache in megabytes. The frontier is a B-tree + // that a crawl walks in priority order while inserting in URL hash order, so + // the cache is what keeps the upper levels of both off the disk. + CacheMB int +} + +// FrontierStats counts what the frontier did, as opposed to what it holds. +type FrontierStats struct { + Admitted int64 // URLs that were new and went in + Duplicates int64 // URLs already known, dropped by the primary key + Claimed int64 // pops handed out + Deferred int64 // claims written back because their host was not eligible + Reclaimed int64 // rows a previous run left claimed and this one took back + Filtered int64 // URLs the seen cache rejected without touching the disk +} + +// Frontier is a disk backed URL frontier with per host politeness. +// +// The state lives in SQLite because the alternative does not survive contact +// with the job it is for: the gao seed set is around 280M URLs, and a Go map +// with 280M string keys is tens of gigabytes of live heap and a garbage +// collector that never finishes. On disk the same set is a file, and the parts +// that have to be fast, the seen check and the pop, are held in front of it by +// bounded memory caches. +// +// The three moving parts: +// +// - Admission batches. Add buffers rows and writes them in transactions of +// BatchSize, because SQLite's cost is per transaction far more than per row. +// A bounded exact cache answers the rediscoveries without touching the disk +// and the primary key resolves whatever falls through it. +// - Claim buffering. Pop refills from one query that claims ClaimSize rows at +// once, then serves from memory. The pop rate is therefore a memory +// operation with a query amortised over hundreds of it. +// - Politeness write back. A claimed URL whose host is inside its delay is +// written back with next_at set to when the host is next eligible, so it +// stops being a candidate until then rather than being rescanned. +// +// A Frontier is safe for concurrent use. +type Frontier struct { + mu sync.Mutex + db *sql.DB + cfg FrontierConfig + delay int64 // politeness delay in seconds + + seen *seenCache + pending []FrontierEntry // admissions not yet written + ready []FrontierEntry // claimed rows not yet handed out + readyAt int // how far into ready the scan has walked + defer_ []FrontierEntry // claimed rows going back because of politeness + + finished []finishedURL // completions not yet written + + hosts map[string]*list.Element // host -> LRU element holding hostAt + hostLRU *list.List + dirtyHosts map[string]int64 // politeness clocks not yet written + + staged int64 // rows in the staging table, not yet merged + queued int64 // pending rows, tracked rather than counted + stats FrontierStats +} + +// finishedURL is one buffered completion. +type finishedURL struct { + hash []byte + state int +} + +// hostAt is one host's next eligible time, held in the LRU. +type hostAt struct { + host string + nextAt int64 +} + +// NewFrontier opens an in memory frontier with the given politeness delay. +// It is the small case: a test, or a crawl that fits in one process lifetime. +func NewFrontier(delay time.Duration) *Frontier { + f, err := OpenFrontier(FrontierConfig{Delay: delay}) + if err != nil { + // An in memory SQLite database with a fixed schema does not fail to + // open for any reason a caller could handle. + panic(fmt.Sprintf("ccrawl: in memory frontier: %v", err)) + } + return f +} + +// OpenFrontier opens or creates a frontier at cfg.Path. +// +// Reopening an existing file resumes it. Rows a previous run left claimed are +// returned to pending, because a claim that outlived its process is a URL that +// was fetched or was not, and the only safe reading is that it was not. Rows +// already marked done stay done, which is what stops a restart refetching. +func OpenFrontier(cfg FrontierConfig) (*Frontier, error) { + if cfg.BatchSize <= 0 { + cfg.BatchSize = 10000 + } + if cfg.ClaimSize <= 0 { + cfg.ClaimSize = 512 + } + if cfg.HotHosts <= 0 { + cfg.HotHosts = 1 << 16 + } + if cfg.SeenURLs <= 0 { + cfg.SeenURLs = 1 << 21 + } + if cfg.MergeSize <= 0 { + cfg.MergeSize = 1 << 20 + } + if cfg.SyncEvery <= 0 { + cfg.SyncEvery = cfg.ClaimSize + } + if cfg.CacheMB <= 0 { + cfg.CacheMB = 128 + } + + dsn := cfg.Path + if dsn == "" { + dsn = ":memory:" + } + db, err := sql.Open("sqlite", dsn) + if err != nil { + return nil, fmt.Errorf("open frontier %s: %w", dsn, err) + } + // One connection, because an in memory database is per connection and + // because a frontier is written by one process that already serialises + // itself behind a mutex. WAL and NORMAL are the usual pair: a crash can + // lose the last transaction, which for a frontier means recrawling a few + // URLs, and that is a much better trade than an fsync per batch. + db.SetMaxOpenConns(1) + // wal_autocheckpoint is the one that matters. The default checkpoints every + // 1000 pages, which on a write heavy frontier means copying the WAL back + // into the database several times a second and costs about five times the + // throughput. Thirty two megabytes of WAL is a cheap thing to trade for it. + if _, err := db.Exec(fmt.Sprintf(` + PRAGMA journal_mode=WAL; + PRAGMA synchronous=NORMAL; + PRAGMA temp_store=MEMORY; + PRAGMA cache_size=-%d; + PRAGMA wal_autocheckpoint=8192;`, cfg.CacheMB*1024) + ` + CREATE TABLE IF NOT EXISTS frontier ( + url_hash BLOB PRIMARY KEY, + url TEXT NOT NULL, + host TEXT NOT NULL, + priority REAL NOT NULL DEFAULT 0, + depth INTEGER NOT NULL DEFAULT 0, + retries INTEGER NOT NULL DEFAULT 0, + next_at INTEGER NOT NULL DEFAULT 0, + state INTEGER NOT NULL DEFAULT 0 + ) WITHOUT ROWID; + -- Priority before next_at on purpose. The claim query wants the highest + -- priority eligible rows, and an index that sorts by next_at first makes + -- SQLite sort every pending row on every refill, which is quadratic over + -- a crawl. This order lets it walk in priority order and stop at the + -- limit. + CREATE INDEX IF NOT EXISTS frontier_claim ON frontier(state, priority DESC, next_at); + -- Admissions land here first and are merged in url_hash order. A rowid + -- table with no index on it, so writing to it is an append. + CREATE TABLE IF NOT EXISTS stage ( + seq INTEGER PRIMARY KEY, + url_hash BLOB NOT NULL, + url TEXT NOT NULL, + host TEXT NOT NULL, + priority REAL NOT NULL DEFAULT 0, + depth INTEGER NOT NULL DEFAULT 0, + retries INTEGER NOT NULL DEFAULT 0, + next_at INTEGER NOT NULL DEFAULT 0 + ); + CREATE TABLE IF NOT EXISTS hosts ( + host TEXT PRIMARY KEY, + next_at INTEGER NOT NULL DEFAULT 0 + ) WITHOUT ROWID; + `); err != nil { + _ = db.Close() + return nil, fmt.Errorf("frontier schema: %w", err) + } + + f := &Frontier{ + db: db, + cfg: cfg, + delay: int64(cfg.Delay / time.Second), + seen: newSeenCache(cfg.SeenURLs), + hosts: make(map[string]*list.Element), + hostLRU: list.New(), + dirtyHosts: make(map[string]int64), + } + + res, err := db.Exec(`UPDATE frontier SET state = ? WHERE state = ?`, frontierPending, frontierClaimed) + if err != nil { + _ = db.Close() + return nil, fmt.Errorf("reclaim in flight rows: %w", err) + } + if n, err := res.RowsAffected(); err == nil { + f.stats.Reclaimed = n + } + // One count at open, then tracked by delta. Counting 280M rows on every + // call to Len would make the progress line the slowest thing in the crawl. + if err := db.QueryRow(`SELECT count(*) FROM frontier WHERE state = ?`, frontierPending).Scan(&f.queued); err != nil { + _ = db.Close() + return nil, fmt.Errorf("count pending: %w", err) + } + // A run that died between a stage and a merge left rows here. They are + // admissions that were never deduped, so the merge on the next flush picks + // them up as though nothing had happened. + if err := db.QueryRow(`SELECT count(*) FROM stage`).Scan(&f.staged); err != nil { + _ = db.Close() + return nil, fmt.Errorf("count staged: %w", err) + } + return f, nil +} + +// Close flushes pending admissions and closes the database. +func (f *Frontier) Close() error { + f.mu.Lock() + defer f.mu.Unlock() + err := f.flushLocked() + if merr := f.mergeLocked(); err == nil { + err = merr + } + if derr := f.flushFinishedLocked(); err == nil { + err = derr + } + if herr := f.flushHostsLocked(); err == nil { + err = herr + } + // Only the part of the claim buffer nothing has looked at yet, plus the + // write backs. Anything before readyAt was handed to a caller and is that + // caller's business, not the frontier's. + if rerr := f.releaseLocked(append(f.ready[f.readyAt:], f.defer_...)); err == nil { + err = rerr + } + f.ready, f.readyAt, f.defer_ = nil, 0, nil + if cerr := f.db.Close(); err == nil { + err = cerr + } + return err +} + +// Add admits a URL. It reports whether the URL looks new. +// +// The answer is exact in one direction and provisional in the other. False +// means the seen cache has this URL and nothing is written, which is a fact +// rather than a guess. True means the cache has not got it, so the row is +// buffered and the primary key decides at flush time, and a URL the cache had +// forgotten is counted as a duplicate then. Callers that want the exact totals +// read FrontierStats after a Flush. +// +// No return value drops a URL. That is the property worth stating plainly, +// because the usual structure in this position is a Bloom filter, and a Bloom +// filter here would discard genuinely new URLs at its false positive rate. +func (f *Frontier) Add(e FrontierEntry) bool { + f.mu.Lock() + defer f.mu.Unlock() + if f.seen.add(seenKey(e.URL)) { + f.stats.Duplicates++ + f.stats.Filtered++ + return false + } + f.pending = append(f.pending, e) + if len(f.pending) >= f.cfg.BatchSize { + if err := f.flushLocked(); err != nil { + return true + } + if f.staged >= f.mergeThreshold() { + _ = f.mergeLocked() + } + } + return true +} + +// Flush stages buffered admissions and merges them into the frontier, so that +// afterwards every admitted URL is queryable and Stats is exact. Call it when +// the seeding is done or before reading Stats. Add stages on its own every +// BatchSize URLs and merges every MergeSize, so a long seed load does not need +// this at all until the end. +func (f *Frontier) Flush() error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.flushLocked(); err != nil { + return err + } + return f.mergeLocked() +} + +// flushLocked appends the buffer to the staging table. +// +// Appends, not inserts into the frontier. This is the whole reason the staging +// table exists: url_hash order has nothing to do with the order URLs are +// discovered in, so inserting straight into the frontier writes to a random +// page of a B-tree that is far larger than any page cache, and the admit rate +// falls off a cliff as the table grows. Measured on this design, four million +// rows went in at under 3k a second and still slowing, against 170k a second +// for the first two hundred thousand. Staging is a rowid table written in +// increasing rowid order, which is an append to the right hand edge and costs +// the same at four million rows as at four hundred. +func (f *Frontier) flushLocked() error { + if len(f.pending) == 0 { + return nil + } + tx, err := f.db.Begin() + if err != nil { + return fmt.Errorf("frontier flush: %w", err) + } + defer func() { _ = tx.Rollback() }() + + stmt, err := tx.Prepare(`INSERT INTO stage + (url_hash, url, host, priority, depth, retries, next_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`) + if err != nil { + return fmt.Errorf("frontier flush: %w", err) + } + defer func() { _ = stmt.Close() }() + + for _, e := range f.pending { + if _, err := stmt.Exec(urlHashBytes(e.URL), e.URL, e.Host, e.Priority, e.Depth, e.Retries, e.NextAt); err != nil { + return fmt.Errorf("frontier stage %s: %w", e.URL, err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("frontier flush: %w", err) + } + f.staged += int64(len(f.pending)) + f.pending = f.pending[:0] + return nil +} + +// mergeThreshold is how many staged rows are worth a merge. +// +// A sweep costs a pass over the whole frontier, so a fixed threshold makes the +// total merge cost grow with the square of the crawl: ten merges into a table +// that is ten times bigger by the end. Scaling the threshold with what is +// already queued keeps each sweep carrying a fixed fraction of the table, which +// turns that into a handful of sweeps over the life of a crawl. +func (f *Frontier) mergeThreshold() int64 { + n := int64(f.cfg.MergeSize) + if eighth := f.queued / 8; eighth > n { + return eighth + } + return n +} + +// mergeLocked moves the staging table into the frontier in url_hash order. +// +// One ORDER BY turns a pile of random writes into a single left to right sweep +// of the frontier's B-tree, so every leaf page is read and dirtied once instead +// of once per row that happens to land on it. The sweep costs the same whether +// it carries ten thousand rows or ten million, which is why the merge is worth +// deferring until MergeSize rows have piled up rather than running per batch. +// +// The dedup is here too, and it is exact: INSERT OR IGNORE against the primary +// key resolves both URLs already in the frontier and URLs staged twice. +func (f *Frontier) mergeLocked() error { + if f.staged == 0 { + return nil + } + tx, err := f.db.Begin() + if err != nil { + return fmt.Errorf("frontier merge: %w", err) + } + defer func() { _ = tx.Rollback() }() + + res, err := tx.Exec(`INSERT OR IGNORE INTO frontier + (url_hash, url, host, priority, depth, retries, next_at, state) + SELECT url_hash, url, host, priority, depth, retries, next_at, ? + FROM stage ORDER BY url_hash`, frontierPending) + if err != nil { + return fmt.Errorf("frontier merge: %w", err) + } + admitted, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("frontier merge: %w", err) + } + if _, err := tx.Exec(`DELETE FROM stage`); err != nil { + return fmt.Errorf("frontier merge: %w", err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("frontier merge: %w", err) + } + dups := f.staged - admitted + f.staged = 0 + f.stats.Admitted += admitted + f.stats.Duplicates += dups + f.queued += admitted + return nil +} + +// Len returns the number of URLs waiting to be crawled, counting buffered +// admissions and claimed rows nothing has been handed yet. +// +// The counter is maintained rather than queried. Counting hundreds of millions +// of rows would make the progress line the slowest thing in the crawl, and a +// crawler asks how much is left far more often than it needs the answer to be +// exact to the row. +func (f *Frontier) Len() int { + f.mu.Lock() + defer f.mu.Unlock() + // queued still counts rows sitting in the claim buffer, since a pop is what + // decrements it, so adding the buffer here would count it twice. Staged rows + // have not been deduped yet, so this is an upper bound until the next merge. + return int(f.queued) + int(f.staged) + len(f.pending) +} + +// Stats returns the counters. Admitted and Duplicates only count URLs that have +// been merged, so read them after a Flush. +func (f *Frontier) Stats() FrontierStats { + f.mu.Lock() + defer f.mu.Unlock() + return f.stats +} + +// Pop returns the highest priority URL that is eligible to crawl at now, and +// records the fetch against its host's politeness clock. +// +// A false second return means nothing is eligible yet, which is a wait rather +// than an end: check Len to tell "come back in a moment" from "finished". +func (f *Frontier) Pop(now int64) (FrontierEntry, bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + + for { + // readyAt walks the claim buffer once. An entry it steps over was either + // handed out or written back, so there is nothing to compact and the scan + // stays O(1) per pop rather than O(claim size). + for f.readyAt < len(f.ready) { + e := f.ready[f.readyAt] + f.readyAt++ + at := f.hostNextAt(e.Host) + if now < at || now < e.NextAt { + e.NextAt = maxInt64(at, e.NextAt) + f.defer_ = append(f.defer_, e) + continue + } + f.setHostNextAt(e.Host, now+f.delay) + f.stats.Claimed++ + f.queued-- + return e, true, nil + } + f.ready, f.readyAt = f.ready[:0], 0 + if err := f.releaseLocked(f.defer_); err != nil { + return FrontierEntry{}, false, err + } + f.defer_ = f.defer_[:0] + + // Buffered admissions have to land before the query looks, or a crawler + // that adds an outlink and immediately pops would be told the frontier + // is empty for as long as it takes to fill a batch. + if err := f.flushLocked(); err != nil { + return FrontierEntry{}, false, err + } + n, err := f.refillLocked(now) + if err != nil { + return FrontierEntry{}, false, err + } + if n == 0 { + return FrontierEntry{}, false, nil + } + } +} + +// refillLocked claims a batch of eligible rows and returns how many it got. +// +// Claiming marks the rows so a second worker, or a second process pointed at +// the same file, cannot hand out the same URL. It is also what makes a crash +// recoverable: whatever is still marked claimed when the process dies is +// exactly the set the next run has to put back. +func (f *Frontier) refillLocked(now int64) (int, error) { + ctx := context.Background() + // Completions first. A row that finished has to stop being claimed before + // the next claim query runs, or the frontier keeps a growing tail of rows + // that are done in memory and in flight on disk. + if err := f.flushFinishedLocked(); err != nil { + return 0, err + } + if err := f.flushHostsLocked(); err != nil { + return 0, err + } + rows, err := f.db.QueryContext(ctx, `SELECT url, host, priority, depth, retries, next_at + FROM frontier WHERE state = ? AND next_at <= ? + ORDER BY priority DESC LIMIT ?`, + frontierPending, now, f.cfg.ClaimSize) + if err != nil { + return 0, fmt.Errorf("frontier refill: %w", err) + } + var batch []FrontierEntry + for rows.Next() { + var e FrontierEntry + var depth, retries int64 + if err := rows.Scan(&e.URL, &e.Host, &e.Priority, &depth, &retries, &e.NextAt); err != nil { + _ = rows.Close() + return 0, fmt.Errorf("frontier refill: %w", err) + } + e.Depth, e.Retries = uint8(depth), uint8(retries) + batch = append(batch, e) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return 0, fmt.Errorf("frontier refill: %w", err) + } + if err := rows.Close(); err != nil { + return 0, fmt.Errorf("frontier refill: %w", err) + } + if len(batch) == 0 { + // Nothing eligible on disk. If admissions are still sitting in staging, + // merge them and look again, so a crawl that is adding as fast as it pops + // never reports itself finished with work in hand. + if f.staged > 0 { + if err := f.mergeLocked(); err != nil { + return 0, err + } + return f.refillLocked(now) + } + return 0, nil + } + + hashes := make([]any, len(batch)) + for i, e := range batch { + hashes[i] = urlHashBytes(e.URL) + } + args := append([]any{frontierClaimed}, hashes...) + q := fmt.Sprintf(`UPDATE frontier SET state = ? WHERE url_hash IN (%s)`, + strings.TrimSuffix(strings.Repeat("?,", len(hashes)), ",")) + if _, err := f.db.ExecContext(ctx, q, args...); err != nil { + return 0, fmt.Errorf("frontier claim: %w", err) + } + + // Highest priority first, so the in memory scan hands out in the same order + // the query chose rather than in whatever order politeness leaves behind. + sort.SliceStable(batch, func(i, j int) bool { return batch[i].Priority > batch[j].Priority }) + f.ready = append(f.ready, batch...) + return len(batch), nil +} + +// releaseLocked puts claimed rows back as pending, carrying whatever next_at +// politeness decided for them. +func (f *Frontier) releaseLocked(entries []FrontierEntry) error { + if len(entries) == 0 { + return nil + } + tx, err := f.db.Begin() + if err != nil { + return fmt.Errorf("frontier release: %w", err) + } + defer func() { _ = tx.Rollback() }() + stmt, err := tx.Prepare(`UPDATE frontier SET state = ?, next_at = ? WHERE url_hash = ?`) + if err != nil { + return fmt.Errorf("frontier release: %w", err) + } + defer func() { _ = stmt.Close() }() + for _, e := range entries { + if e.URL == "" { + continue + } + if _, err := stmt.Exec(frontierPending, e.NextAt, urlHashBytes(e.URL)); err != nil { + return fmt.Errorf("frontier release %s: %w", e.URL, err) + } + f.stats.Deferred++ + } + return tx.Commit() +} + +// Done marks a URL as crawled. It is what a restart reads to know not to fetch +// the URL again. +// +// The write is buffered up to SyncEvery completions, so it is exact across a +// clean Close and bounded across a kill: at most SyncEvery URLs get fetched a +// second time by the run that follows one. Buffering is worth roughly four +// times the pop rate, and a caller who would rather have the exactness sets +// SyncEvery to 1. +func (f *Frontier) Done(url string) error { return f.finish(url, frontierDone) } + +// Fail marks a URL as given up on. Same effect as Done for the crawl, kept +// apart so a run can report the difference. +func (f *Frontier) Fail(url string) error { return f.finish(url, frontierFailed) } + +func (f *Frontier) finish(url string, state int) error { + f.mu.Lock() + defer f.mu.Unlock() + f.finished = append(f.finished, finishedURL{hash: urlHashBytes(url), state: state}) + if len(f.finished) >= f.cfg.SyncEvery { + return f.flushFinishedLocked() + } + return nil +} + +// Sync writes every buffered completion. A caller that has just been asked to +// stop, or that is about to do something it cannot undo, calls this to make the +// frontier's idea of what is done match its own. +func (f *Frontier) Sync() error { + f.mu.Lock() + defer f.mu.Unlock() + return f.flushFinishedLocked() +} + +func (f *Frontier) flushFinishedLocked() error { + if len(f.finished) == 0 { + return nil + } + tx, err := f.db.Begin() + if err != nil { + return fmt.Errorf("frontier finish: %w", err) + } + defer func() { _ = tx.Rollback() }() + stmt, err := tx.Prepare(`UPDATE frontier SET state = ? WHERE url_hash = ?`) + if err != nil { + return fmt.Errorf("frontier finish: %w", err) + } + defer func() { _ = stmt.Close() }() + // Same reason the merge sorts: completions arrive in priority order, which + // is unrelated to where the rows live, and updating in key order walks the + // B-tree once instead of jumping around it. + sort.Slice(f.finished, func(i, j int) bool { + return bytes.Compare(f.finished[i].hash, f.finished[j].hash) < 0 + }) + for _, d := range f.finished { + if _, err := stmt.Exec(d.state, d.hash); err != nil { + return fmt.Errorf("frontier finish: %w", err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("frontier finish: %w", err) + } + f.finished = f.finished[:0] + return nil +} + +// Retry puts a URL back in the queue at a later time with its retry count +// bumped, which is what a transient fetch failure deserves. +func (f *Frontier) Retry(e FrontierEntry, at int64) error { + f.mu.Lock() + defer f.mu.Unlock() + // Straight through rather than buffered. A retry is rare next to a + // completion, and a lost one is a URL dropped from the crawl rather than a + // URL fetched twice. + if _, err := f.db.Exec(`UPDATE frontier SET state = ?, next_at = ?, retries = retries + 1 WHERE url_hash = ?`, + frontierPending, at, urlHashBytes(e.URL)); err != nil { + return fmt.Errorf("frontier retry %s: %w", e.URL, err) + } + f.queued++ + return nil +} + +// hostNextAt reads a host's next eligible time, from the LRU when it is hot, +// from the unflushed writes when it is not, and from the database otherwise. A +// crawl touches a small set of hosts hard and a long tail once, which is the +// access pattern an LRU is for. +func (f *Frontier) hostNextAt(host string) int64 { + if el, ok := f.hosts[host]; ok { + f.hostLRU.MoveToFront(el) + return el.Value.(*hostAt).nextAt + } + if at, ok := f.dirtyHosts[host]; ok { + f.cacheHost(host, at) + return at + } + var at int64 + _ = f.db.QueryRow(`SELECT next_at FROM hosts WHERE host = ?`, host).Scan(&at) + f.cacheHost(host, at) + return at +} + +// setHostNextAt records a fetch against a host. +// +// The write is buffered rather than issued, because a statement per pop is the +// one thing that would put the pop rate back where SQLite can be seen. The +// buffer is drained once per refill, so a kill loses at most one claim batch of +// host timings; the cost of that is a restart being a little eager with the +// hosts it was mid delay on, which is a much smaller price than a write per +// URL. +func (f *Frontier) setHostNextAt(host string, at int64) { + if el, ok := f.hosts[host]; ok { + el.Value.(*hostAt).nextAt = at + f.hostLRU.MoveToFront(el) + } else { + f.cacheHost(host, at) + } + f.dirtyHosts[host] = at +} + +// flushHostsLocked writes buffered politeness clocks in one transaction. +func (f *Frontier) flushHostsLocked() error { + if len(f.dirtyHosts) == 0 { + return nil + } + tx, err := f.db.Begin() + if err != nil { + return fmt.Errorf("frontier hosts: %w", err) + } + defer func() { _ = tx.Rollback() }() + stmt, err := tx.Prepare(`INSERT INTO hosts (host, next_at) VALUES (?, ?) + ON CONFLICT(host) DO UPDATE SET next_at = excluded.next_at`) + if err != nil { + return fmt.Errorf("frontier hosts: %w", err) + } + defer func() { _ = stmt.Close() }() + for host, at := range f.dirtyHosts { + if _, err := stmt.Exec(host, at); err != nil { + return fmt.Errorf("frontier host %s: %w", host, err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("frontier hosts: %w", err) + } + clear(f.dirtyHosts) + return nil +} + +func (f *Frontier) cacheHost(host string, at int64) { + el := f.hostLRU.PushFront(&hostAt{host: host, nextAt: at}) + f.hosts[host] = el + for f.hostLRU.Len() > f.cfg.HotHosts { + old := f.hostLRU.Back() + f.hostLRU.Remove(old) + delete(f.hosts, old.Value.(*hostAt).host) + } +} + +// urlHashBytes is the frontier's primary key: the first 16 bytes of the URL's +// SHA-1. Sixteen bytes is 128 bits, so a collision across the 280M URLs this +// is sized for is not something that happens, and it is four bytes shorter per +// row than the full digest across a table with hundreds of millions of them. +func urlHashBytes(rawURL string) []byte { + sum := sha1.Sum([]byte(rawURL)) + return sum[:16] +} + +// seenKey folds a URL into the 64 bits the seen cache holds. +// +// A collision here would drop a URL, the same way a Bloom filter false positive +// would, so the number matters. The cache holds a few million keys at a time, +// not the whole frontier, and the odds of any collision among four million 64 +// bit hashes are about one in two million. That is six orders of magnitude +// below the one percent a conventionally sized filter would cost, which is the +// difference between a bound worth stating and a bound worth avoiding. +func seenKey(rawURL string) uint64 { + sum := sha1.Sum([]byte(rawURL)) + return binary.BigEndian.Uint64(sum[:8]) +} + +func maxInt64(a, b int64) int64 { + if a > b { + return a + } + return b +} diff --git a/ccrawl/frontier_scale_test.go b/ccrawl/frontier_scale_test.go new file mode 100644 index 0000000..c891aba --- /dev/null +++ b/ccrawl/frontier_scale_test.go @@ -0,0 +1,97 @@ +package ccrawl + +import ( + "fmt" + "os" + "path/filepath" + "runtime" + "strconv" + "testing" + "time" +) + +// TestFrontierScale is the harness behind the numbers in the frontier's design +// notes. It is off by default because it writes gigabytes and takes minutes. +// +// CCRAWL_SCALE_URLS=10000000 go test ./ccrawl -run TestFrontierScale -v -timeout 2h +// +// Set CCRAWL_SCALE_DIR to put the database somewhere with room for it. Ten +// million URLs is about 1.2 GB, and it is close enough to linear to extrapolate +// from. +func TestFrontierScale(t *testing.T) { + raw := os.Getenv("CCRAWL_SCALE_URLS") + if raw == "" { + t.Skip("set CCRAWL_SCALE_URLS to run the scale harness") + } + n, err := strconv.Atoi(raw) + if err != nil || n <= 0 { + t.Fatalf("CCRAWL_SCALE_URLS=%q is not a positive count", raw) + } + dir := os.Getenv("CCRAWL_SCALE_DIR") + if dir == "" { + dir = t.TempDir() + } + path := filepath.Join(dir, "frontier-scale.db") + _ = os.Remove(path) + + f, err := OpenFrontier(FrontierConfig{Path: path, SeenURLs: 1 << 22}) + if err != nil { + t.Fatalf("OpenFrontier: %v", err) + } + report := func(stage string, done int, since time.Duration) { + var m runtime.MemStats + runtime.ReadMemStats(&m) + t.Logf("%-8s %9d %7.0f/s heap %4.0f MB sys %4.0f MB", + stage, done, float64(done)/since.Seconds(), + float64(m.HeapAlloc)/(1<<20), float64(m.Sys)/(1<<20)) + } + + // Two million hosts, so the politeness cache is exercised well past its + // capacity rather than answering everything out of the hot set. + start := time.Now() + for i := 0; i < n; i++ { + host := fmt.Sprintf("h%d.example", i%2000000) + f.Add(FrontierEntry{ + URL: fmt.Sprintf("https://%s/page/%d", host, i), + Host: host, + Priority: float32(i % 100000), + }) + if i > 0 && i%2000000 == 0 { + report("admit", i, time.Since(start)) + } + } + if err := f.Flush(); err != nil { + t.Fatalf("Flush: %v", err) + } + report("admit", n, time.Since(start)) + if fi, err := os.Stat(path); err == nil { + t.Logf("database %.2f GB for %d URLs", float64(fi.Size())/(1<<30), n) + } + + now := time.Now().Unix() + start = time.Now() + var popped int + for popped < n { + e, ok, err := f.Pop(now) + if err != nil { + t.Fatalf("Pop: %v", err) + } + if !ok { + break + } + popped++ + if err := f.Done(e.URL); err != nil { + t.Fatalf("Done: %v", err) + } + if popped%2000000 == 0 { + report("pop", popped, time.Since(start)) + } + } + report("pop", popped, time.Since(start)) + if popped != n { + t.Errorf("popped %d of %d admitted, the frontier lost URLs", popped, n) + } + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } +} diff --git a/ccrawl/frontier_test.go b/ccrawl/frontier_test.go new file mode 100644 index 0000000..4405268 --- /dev/null +++ b/ccrawl/frontier_test.go @@ -0,0 +1,415 @@ +package ccrawl + +import ( + "fmt" + "path/filepath" + "runtime" + "testing" + "time" +) + +// seedFrontier admits n URLs spread over hosts hosts and flushes them. +func seedFrontier(tb testing.TB, f *Frontier, n, hosts int) { + tb.Helper() + for i := 0; i < n; i++ { + host := fmt.Sprintf("h%d.example", i%hosts) + f.Add(FrontierEntry{ + URL: fmt.Sprintf("https://%s/page/%d", host, i), + Host: host, + Priority: float32(n - i), + }) + } + if err := f.Flush(); err != nil { + tb.Fatalf("Flush: %v", err) + } +} + +// drain pops until nothing is eligible and returns the URLs in order. +func drain(tb testing.TB, f *Frontier, now int64) []string { + tb.Helper() + var got []string + for { + e, ok, err := f.Pop(now) + if err != nil { + tb.Fatalf("Pop: %v", err) + } + if !ok { + return got + } + got = append(got, e.URL) + if err := f.Done(e.URL); err != nil { + tb.Fatalf("Done: %v", err) + } + } +} + +// TestFrontierResumesWithoutRefetching is the property the whole thing is for: +// a run that dies mid crawl and starts again picks up what it had not done and +// does not hand back what it had. +func TestFrontierResumesWithoutRefetching(t *testing.T) { + path := filepath.Join(t.TempDir(), "frontier.db") + now := time.Now().Unix() + + f, err := OpenFrontier(FrontierConfig{Path: path}) + if err != nil { + t.Fatalf("OpenFrontier: %v", err) + } + seedFrontier(t, f, 200, 40) + + done := map[string]bool{} + for i := 0; i < 60; i++ { + e, ok, err := f.Pop(now) + if err != nil || !ok { + t.Fatalf("Pop %d: ok=%v err=%v", i, ok, err) + } + if err := f.Done(e.URL); err != nil { + t.Fatalf("Done: %v", err) + } + done[e.URL] = true + } + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + f2, err := OpenFrontier(FrontierConfig{Path: path}) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer func() { _ = f2.Close() }() + + if got := f2.Len(); got != 140 { + t.Errorf("reopened Len = %d, want 140 (200 seeded minus 60 done)", got) + } + rest := drain(t, f2, now) + if len(rest) != 140 { + t.Fatalf("second run popped %d URLs, want 140", len(rest)) + } + seen := map[string]bool{} + for _, u := range rest { + if done[u] { + t.Fatalf("refetched %s, which the first run already finished", u) + } + if seen[u] { + t.Fatalf("popped %s twice in one run", u) + } + seen[u] = true + } +} + +// TestFrontierReclaimsClaimsLeftInFlight covers the other half of a crash: URLs +// a worker had taken but not finished. They were fetched or they were not, and +// only one of those readings is safe, so they go back in the queue. +func TestFrontierReclaimsClaimsLeftInFlight(t *testing.T) { + path := filepath.Join(t.TempDir(), "frontier.db") + now := time.Now().Unix() + + f, err := OpenFrontier(FrontierConfig{Path: path}) + if err != nil { + t.Fatalf("OpenFrontier: %v", err) + } + seedFrontier(t, f, 20, 20) + var claimed []string + for i := 0; i < 5; i++ { + e, ok, err := f.Pop(now) // popped and never marked done, as a kill would leave it + if err != nil || !ok { + t.Fatalf("Pop: ok=%v err=%v", ok, err) + } + claimed = append(claimed, e.URL) + } + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + f2, err := OpenFrontier(FrontierConfig{Path: path}) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer func() { _ = f2.Close() }() + if got := f2.Stats().Reclaimed; got != 5 { + t.Errorf("Reclaimed = %d, want 5", got) + } + back := map[string]bool{} + for _, u := range drain(t, f2, now) { + back[u] = true + } + for _, u := range claimed { + if !back[u] { + t.Errorf("%s was claimed when the run died and never came back", u) + } + } +} + +// TestFrontierDedupsAcrossRestart pins that the dedup is the primary key rather +// than the memory cache. The cache is empty on open, so a reseed after a +// restart is exactly the case where a cache only design quietly doubles the +// queue. +func TestFrontierDedupsAcrossRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "frontier.db") + + f, err := OpenFrontier(FrontierConfig{Path: path}) + if err != nil { + t.Fatalf("OpenFrontier: %v", err) + } + seedFrontier(t, f, 100, 10) + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + f2, err := OpenFrontier(FrontierConfig{Path: path}) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer func() { _ = f2.Close() }() + seedFrontier(t, f2, 100, 10) // the same 100 URLs again + if s := f2.Stats(); s.Admitted != 0 || s.Duplicates != 100 { + t.Errorf("reseed Stats = %+v, want Admitted 0 and Duplicates 100", s) + } + if got := f2.Len(); got != 100 { + t.Errorf("Len after reseed = %d, want 100", got) + } +} + +// TestFrontierPolitenessSurvivesRestart checks that a restart does not hit a +// host it was mid delay on. The clocks are buffered and drained per claim +// batch, so this only holds for a clean Close, which is what the test does. +func TestFrontierPolitenessSurvivesRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "frontier.db") + now := time.Now().Unix() + + f, err := OpenFrontier(FrontierConfig{Path: path, Delay: 30 * time.Second}) + if err != nil { + t.Fatalf("OpenFrontier: %v", err) + } + f.Add(FrontierEntry{URL: "https://slow.example/a", Host: "slow.example"}) + f.Add(FrontierEntry{URL: "https://slow.example/b", Host: "slow.example"}) + e, ok, err := f.Pop(now) + if err != nil || !ok { + t.Fatalf("Pop: ok=%v err=%v", ok, err) + } + if err := f.Done(e.URL); err != nil { + t.Fatalf("Done: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + f2, err := OpenFrontier(FrontierConfig{Path: path, Delay: 30 * time.Second}) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer func() { _ = f2.Close() }() + if _, ok, err := f2.Pop(now + 5); err != nil { + t.Fatalf("Pop: %v", err) + } else if ok { + t.Error("a restart popped a host it was 5 seconds into a 30 second delay on") + } + if _, ok, err := f2.Pop(now + 31); err != nil { + t.Fatalf("Pop: %v", err) + } else if !ok { + t.Error("Pop after the delay elapsed should succeed") + } +} + +// TestFrontierDeferredURLsAreNotLost is the politeness write back: an entry +// claimed while its host is inside the delay goes back to pending rather than +// evaporating with the claim buffer. +func TestFrontierDeferredURLsAreNotLost(t *testing.T) { + f := NewFrontier(60 * time.Second) + defer func() { _ = f.Close() }() + now := time.Now().Unix() + + // One busy host and one quiet one. The busy host's second URL has to be + // deferred to reach the quiet host's, and then still be there afterwards. + f.Add(FrontierEntry{URL: "https://busy.example/1", Host: "busy.example", Priority: 10}) + f.Add(FrontierEntry{URL: "https://busy.example/2", Host: "busy.example", Priority: 9}) + f.Add(FrontierEntry{URL: "https://quiet.example/1", Host: "quiet.example", Priority: 1}) + + got := drain(t, f, now) + if len(got) != 2 || got[0] != "https://busy.example/1" || got[1] != "https://quiet.example/1" { + t.Fatalf("first pass popped %v, want busy/1 then quiet/1", got) + } + if f.Len() != 1 { + t.Fatalf("Len = %d, want 1 deferred URL still queued", f.Len()) + } + if s := f.Stats(); s.Deferred == 0 { + t.Error("Stats.Deferred = 0, want the write back to have been counted") + } + later := drain(t, f, now+61) + if len(later) != 1 || later[0] != "https://busy.example/2" { + t.Fatalf("after the delay popped %v, want busy/2", later) + } +} + +func TestFrontierRetryRequeues(t *testing.T) { + f := NewFrontier(0) + defer func() { _ = f.Close() }() + now := time.Now().Unix() + + f.Add(FrontierEntry{URL: "https://flaky.example/", Host: "flaky.example"}) + e, ok, err := f.Pop(now) + if err != nil || !ok { + t.Fatalf("Pop: ok=%v err=%v", ok, err) + } + if err := f.Retry(e, now+10); err != nil { + t.Fatalf("Retry: %v", err) + } + if _, ok, _ := f.Pop(now); ok { + t.Error("a URL retried for later should not be eligible now") + } + e2, ok, err := f.Pop(now + 11) + if err != nil || !ok { + t.Fatalf("Pop after retry time: ok=%v err=%v", ok, err) + } + if e2.Retries != 1 { + t.Errorf("Retries = %d, want 1", e2.Retries) + } +} + +// TestFrontierThroughput is the done when box from the issue, at the scale this +// machine has disk for. The rates are asserted well under what the run reports +// so the test fails on a regression rather than on a busy CI runner. +func TestFrontierThroughput(t *testing.T) { + if testing.Short() { + t.Skip("throughput measurement, skipped under -short") + } + const n = 200000 + path := filepath.Join(t.TempDir(), "frontier.db") + f, err := OpenFrontier(FrontierConfig{Path: path, SeenURLs: n}) + if err != nil { + t.Fatalf("OpenFrontier: %v", err) + } + defer func() { _ = f.Close() }() + + var before, after runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&before) + + start := time.Now() + seedFrontier(t, f, n, 5000) + admit := time.Since(start) + + now := time.Now().Unix() + start = time.Now() + var popped int + for { + e, ok, err := f.Pop(now) + if err != nil { + t.Fatalf("Pop: %v", err) + } + if !ok { + break + } + popped++ + if err := f.Done(e.URL); err != nil { + t.Fatalf("Done: %v", err) + } + } + pop := time.Since(start) + + runtime.ReadMemStats(&after) + admitRate := float64(n) / admit.Seconds() + popRate := float64(popped) / pop.Seconds() + t.Logf("admitted %d in %s (%.0f/s), popped %d in %s (%.0f/s), heap %.1f MB for %d URLs", + n, admit.Round(time.Millisecond), admitRate, + popped, pop.Round(time.Millisecond), popRate, + float64(after.HeapAlloc-before.HeapAlloc)/(1<<20), n) + + if popped != n { + t.Fatalf("popped %d of %d admitted", popped, n) + } + if popRate < 5000 { + t.Errorf("pop rate %.0f/s, want at least 5000/s", popRate) + } + if admitRate < 20000 { + t.Errorf("admit rate %.0f/s, want at least 20000/s", admitRate) + } +} + +func BenchmarkFrontierAdd(b *testing.B) { + f, err := OpenFrontier(FrontierConfig{Path: filepath.Join(b.TempDir(), "f.db"), SeenURLs: b.N}) + if err != nil { + b.Fatal(err) + } + defer func() { _ = f.Close() }() + b.ResetTimer() + for i := 0; i < b.N; i++ { + host := fmt.Sprintf("h%d.example", i%5000) + f.Add(FrontierEntry{URL: fmt.Sprintf("https://%s/p/%d", host, i), Host: host, Priority: float32(i)}) + } + if err := f.Flush(); err != nil { + b.Fatal(err) + } +} + +func BenchmarkFrontierPop(b *testing.B) { + f, err := OpenFrontier(FrontierConfig{Path: filepath.Join(b.TempDir(), "f.db"), SeenURLs: b.N}) + if err != nil { + b.Fatal(err) + } + defer func() { _ = f.Close() }() + seedFrontier(b, f, b.N, 5000) + now := time.Now().Unix() + b.ResetTimer() + for i := 0; i < b.N; i++ { + e, ok, err := f.Pop(now) + if err != nil { + b.Fatal(err) + } + if !ok { + b.Fatalf("frontier ran dry after %d of %d pops", i, b.N) + } + if err := f.Done(e.URL); err != nil { + b.Fatal(err) + } + } +} + +// TestSeenCacheNeverInventsAHit is the property that made this an exact cache +// instead of a Bloom filter. A key that was never added has to come back false, +// every time, because the caller drops the URL on a true. +func TestSeenCacheNeverInventsAHit(t *testing.T) { + const n = 50000 + c := newSeenCache(n) + for i := 0; i < n; i++ { + if c.add(uint64(i) * 0x9e3779b97f4a7c15) { + t.Fatalf("key %d was reported as seen the first time it was added", i) + } + } + for i := n; i < 2*n; i++ { + if c.add(uint64(i) * 0x9e3779b97f4a7c15) { + t.Fatalf("key %d was never added and the cache claimed it had been", i) + } + } +} + +func TestSeenCacheRemembersAndBounds(t *testing.T) { + c := newSeenCache(1000) + if c.add(42) { + t.Error("first add reported the key as already seen") + } + if !c.add(42) { + t.Error("second add of the same key reported it as new") + } + // Well past the limit: the cache forgets rather than growing, and forgetting + // is safe because the primary key still catches what falls out. + for i := 0; i < 10000; i++ { + c.add(uint64(i) | 1<<40) + } + if got := c.len(); got > 2000 { + t.Errorf("resident keys = %d, want at most twice the 1000 limit", got) + } +} + +// TestSeenCacheKeepsHotKeysResident checks the promotion path: a URL that keeps +// being rediscovered should not age out just because the generation turned. +func TestSeenCacheKeepsHotKeysResident(t *testing.T) { + c := newSeenCache(100) + const hot = uint64(0xdeadbeef) + c.add(hot) + for i := 0; i < 500; i++ { + c.add(uint64(i)) + if !c.add(hot) { + t.Fatalf("hot key was forgotten after %d other insertions", i) + } + } +} diff --git a/ccrawl/seencache.go b/ccrawl/seencache.go new file mode 100644 index 0000000..8ae5ad3 --- /dev/null +++ b/ccrawl/seencache.go @@ -0,0 +1,60 @@ +package ccrawl + +// seenCache is a memory bounded exact set of URL hashes. +// +// It answers "this URL was definitely admitted already" without a disk read, +// which is the answer that matters at crawl scale: a crawl rediscovers the same +// URLs over and over from outlinks, and a lookup per rediscovery is what turns +// a frontier into the slowest part of the crawler. +// +// The obvious structure here is a Bloom filter, and it is the wrong one. A +// Bloom filter's error is a false positive, which in front of a seen check +// means reporting a brand new URL as already known, and a frontier that does +// that drops the URL. At the usual one percent that is one page in a hundred +// gone from the crawl, silently, with nothing downstream able to tell. Trading +// coverage for memory is a decision a crawl operator might make on purpose, but +// it is not one a data structure should make on their behalf. +// +// So this is exact for what it holds and simply forgets the coldest entries +// instead of lying about them. A forgotten entry costs one wasted row in the +// next insert batch, where the primary key drops it, and nothing else. +// +// Eviction is two generations rather than an LRU list, because an LRU needs a +// list node per entry and the node costs more than the key. When the hot +// generation fills it becomes the cold one and the previous cold one is +// dropped, so the set holds between limit and two times limit entries and the +// bookkeeping is a map assignment. +type seenCache struct { + hot, cold map[uint64]struct{} + limit int +} + +func newSeenCache(limit int) *seenCache { + if limit < 1 { + limit = 1 + } + return &seenCache{hot: make(map[uint64]struct{}, limit/8), limit: limit} +} + +// add records a key and reports whether it was already known. A true return is +// exact. A false return means the cache does not know, which is either a new +// URL or one evicted long enough ago to be cold. +func (c *seenCache) add(key uint64) bool { + if _, ok := c.hot[key]; ok { + return true + } + if _, ok := c.cold[key]; ok { + // Promote, so a URL that keeps being rediscovered stays resident rather + // than aging out on a fixed schedule. + c.hot[key] = struct{}{} + return true + } + if len(c.hot) >= c.limit { + c.cold, c.hot = c.hot, make(map[uint64]struct{}, c.limit/8) + } + c.hot[key] = struct{}{} + return false +} + +// len reports how many keys are resident, for tests and stats. +func (c *seenCache) len() int { return len(c.hot) + len(c.cold) } diff --git a/go.mod b/go.mod index f3fa2d1..8536635 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/tamnd/yomi v0.3.1 golang.org/x/net v0.56.0 golang.org/x/sync v0.21.0 + modernc.org/sqlite v1.52.0 ) require ( @@ -49,7 +50,6 @@ require ( modernc.org/libc v1.72.3 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect - modernc.org/sqlite v1.52.0 // indirect ) require (