Method Open

store.go:167–234

(ctx context.Context)

}

func (s *Store) Open(ctx context.Context) error {
	if err := s.levels.Validate(); err != nil {
		return err
	}

	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(50)
	for _, db := range s.dbs {
		db := db
		g.Go(func() error {
			select {
			case <-ctx.Done():
				return ctx.Err()
			default:
				return db.Open()
			}
		})
	}
	if err := g.Wait(); err != nil {
		return err
	}

	// Start monitors for compactions & snapshots.
	if s.CompactionMonitorEnabled {
		// Start compaction monitors for all levels except L0.
		for _, lvl := range s.levels {
			lvl := lvl
			if lvl.Level == 0 {
				continue
			}

			s.wg.Add(1)
			go func() {
				defer s.wg.Done()
				s.monitorCompactionLevel(s.ctx, lvl)
			}()
		}

		// Start snapshot monitor for snapshots.
		s.wg.Add(1)
		go func() {
			defer s.wg.Done()
			s.monitorCompactionLevel(s.ctx, s.SnapshotLevel())
		}()
	}

	if s.L0Retention > 0 && s.L0RetentionCheckInterval > 0 {
		s.wg.Add(1)
		go func() {
			defer s.wg.Done()
			s.monitorL0Retention(s.ctx)
		}()
	}

	// Start heartbeat monitor if any database has heartbeat configured.
	s.startHeartbeatMonitorIfNeeded()

	// Start validation monitor if configured.