updateLoop is the main event loop of the indexer which pushes chain segments down into the processing backend.
()
| 250 | // updateLoop is the main event loop of the indexer which pushes chain segments |
| 251 | // down into the processing backend. |
| 252 | func (c *ChainIndexer) updateLoop() { |
| 253 | var ( |
| 254 | updating bool |
| 255 | updated time.Time |
| 256 | ) |
| 257 | |
| 258 | for { |
| 259 | select { |
| 260 | case errc := <-c.quit: |
| 261 | // Chain indexer terminating, report no failure and abort |
| 262 | errc <- nil |
| 263 | return |
| 264 | |
| 265 | case <-c.update: |
| 266 | // Section headers completed (or rolled back), update the index |
| 267 | c.lock.Lock() |
| 268 | if c.knownSections > c.storedSections { |
| 269 | // Periodically print an upgrade log message to the user |
| 270 | if time.Since(updated) > 8*time.Second { |
| 271 | if c.knownSections > c.storedSections+1 { |
| 272 | updating = true |
| 273 | c.log.Info("Upgrading chain index", "percentage", c.storedSections*100/c.knownSections) |
| 274 | } |
| 275 | updated = time.Now() |
| 276 | } |
| 277 | // Cache the current section count and head to allow unlocking the mutex |
| 278 | section := c.storedSections |
| 279 | var oldHead common.Hash |
| 280 | if section > 0 { |
| 281 | oldHead = c.SectionHead(section - 1) |
| 282 | } |
| 283 | // Process the newly defined section in the background |
| 284 | c.lock.Unlock() |
| 285 | newHead, err := c.processSection(section, oldHead) |
| 286 | if err != nil { |
| 287 | c.log.Error("Section processing failed", "error", err) |
| 288 | } |
| 289 | c.lock.Lock() |
| 290 | |
| 291 | // If processing succeeded and no reorgs occcurred, mark the section completed |
| 292 | if err == nil && oldHead == c.SectionHead(section-1) { |
| 293 | c.setSectionHead(section, newHead) |
| 294 | c.setValidSections(section + 1) |
| 295 | if c.storedSections == c.knownSections && updating { |
| 296 | updating = false |
| 297 | c.log.Info("Finished upgrading chain index") |
| 298 | } |
| 299 | |
| 300 | c.cascadedHead = c.storedSections*c.sectionSize - 1 |
| 301 | for _, child := range c.children { |
| 302 | c.log.Debug("Cascading chain index update", "head", c.cascadedHead) |
| 303 | child.newHead(c.cascadedHead, false) |
| 304 | } |
| 305 | } else { |
| 306 | // If processing failed, don't retry until further notification |
| 307 | c.log.Debug("Chain index processing failed", "section", section, "err", err) |
| 308 | c.knownSections = c.storedSections |
| 309 | } |
no test coverage detected