<commit_before>// Copyright (c) 2012, <NAME> <<EMAIL>>
// All rights reserved.
//
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

package leveldb

import (
	"errors"
	"sync"
	"time"

	"github.com/conformal/goleveldb/leveldb/memdb"
)

var (
	errTransactExiting = errors.New("leveldb: transact exiting")
)

type cStats struct {
	sync.Mutex
	duration time.Duration
	read     uint64
	write    uint64
}

func (p *cStats) add(n *cStatsStaging) {
	p.Lock()
	p.duration += n.duration
	p.read += n.read
	p.write += n.write
	p.Unlock()
}

func (p *cStats) get() (duration time.Duration, read, write uint64) {
	p.Lock()
	defer p.Unlock()
	return p.duration, p.read, p.write
}

type cStatsStaging struct {
	start    time.Time
	duration time.Duration
	on       bool
	read     uint64
	write    uint64
}

func (p *cStatsStaging) startTimer() {
	if !p.on {
		p.start = time.Now()
		p.on = true
	}
}

func (p *cStatsStaging) stopTimer() {
	if p.on {
		p.duration += time.Since(p.start)
		p.on = false
	}
}

type cReq struct {
	level    int
	min, max iKey
	cch      chan<- struct{}
}

type cMem struct {
	s     *session
	level int
	rec   *sessionRecord
}

func newCMem(s *session) *cMem {
	return &cMem{s: s, rec: new(sessionRecord)}
}

func (c *cMem) flush(mem *memdb.DB, level int) error {
	s := c.s

	// Write memdb to table
	t, n, err := s.tops.createFrom(mem.NewIterator(nil))
	if err != nil {
		return err
	}

	if level < 0 {
		level = s.version_NB().pickLevel(t.min.ukey(), t.max.ukey())
	}
	c.rec.addTableFile(level, t)

	s.logf("mem@flush created L%d@%d N·%d S·%s %q:%q", level, t.file.Num(), n, shortenb(int(t.size)), t.min, t.max)

	c.level = level
	return nil
}

func (c *cMem) reset() {
	c.rec = new(sessionRecord)
}

func (c *cMem) commit(journal, seq uint64) error {
	c.rec.setJournalNum(journal)
	c.rec.setSeq(seq)
	// Commit changes
	return c.s.commit(c.rec)
}

func (d *DB) compactionError() {
	var err error
noerr:
	for {
		select {
		case _, _ = <-d.closeCh:
			return
		case err = <-d.compErrSetCh:
			if err != nil {
				goto haserr
			}
		}
	}
haserr:
	for {
		select {
		case _, _ = <-d.closeCh:
			return
		case err = <-d.compErrSetCh:
			if err == nil {
				goto noerr
			}
		case d.compErrCh <- err:
		}
	}
}

func (d *DB) transact(name string, exec, rollback func() error) {
	s := d.s
	defer func() {
		if x := recover(); x != nil {
			if x == errTransactExiting && rollback != nil {
				if err := rollback(); err != nil {
					s.logf("%s rollback error %q", name, err)
				}
			}
			panic(x)
		}
	}()
	for n := 0; ; n++ {
		if d.isClosed() {
			s.logf("%s exiting", name)
			panic(errTransactExiting)
		} else if n > 0 {
			s.logf("%s retrying N·%d", name, n)
		}
		err := exec()
		select {
		case _, _ = <-d.closeCh:
			s.logf("%s exiting", name)
			panic(errTransactExiting)
		case d.compErrSetCh <- err:
		}
		if err == nil {
			return
		}
		s.logf("%s error %q", name, err)
		time.Sleep(time.Second)
	}
}

func (d *DB) memCompaction() {
	s := d.s
	c := newCMem(s)
	stats := new(cStatsStaging)
	mem := d.getFrozenMem()

	s.logf("mem@flush N·%d S·%s", mem.Len(), shortenb(mem.Size()))

	d.transact("mem@flush", func() (err error) {
		stats.startTimer()
		defer stats.stopTimer()
		return c.flush(mem, -1)
	}, func() error {
		for _, r := range c.rec.addedTables {
			s.logf("mem@flush rollback @%d", r.num)
			f := s.getTableFile(r.num)
			if err := f.Remove(); err != nil {<commit_msg>leveldb: compaction: Skip compaction of empty memdb<commit_after>// Copyright (c) 2012, <NAME> <<EMAIL>>
// All rights reserved.
//
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

package leveldb

import (
	"errors"
	"sync"
	"time"

	"github.com/conformal/goleveldb/leveldb/memdb"
)

var (
	errTransactExiting = errors.New("leveldb: transact exiting")
)

type cStats struct {
	sync.Mutex
	duration time.Duration
	read     uint64
	write    uint64
}

func (p *cStats) add(n *cStatsStaging) {
	p.Lock()
	p.duration += n.duration
	p.read += n.read
	p.write += n.write
	p.Unlock()
}

func (p *cStats) get() (duration time.Duration, read, write uint64) {
	p.Lock()
	defer p.Unlock()
	return p.duration, p.read, p.write
}

type cStatsStaging struct {
	start    time.Time
	duration time.Duration
	on       bool
	read     uint64
	write    uint64
}

func (p *cStatsStaging) startTimer() {
	if !p.on {
		p.start = time.Now()
		p.on = true
	}
}

func (p *cStatsStaging) stopTimer() {
	if p.on {
		p.duration += time.Since(p.start)
		p.on = false
	}
}

type cReq struct {
	level    int
	min, max iKey
	cch      chan<- struct{}
}

type cMem struct {
	s     *session
	level int
	rec   *sessionRecord
}

func newCMem(s *session) *cMem {
	return &cMem{s: s, rec: new(sessionRecord)}
}

func (c *cMem) flush(mem *memdb.DB, level int) error {
	s := c.s

	// Write memdb to table
	t, n, err := s.tops.createFrom(mem.NewIterator(nil))
	if err != nil {
		return err
	}

	if level < 0 {
		level = s.version_NB().pickLevel(t.min.ukey(), t.max.ukey())
	}
	c.rec.addTableFile(level, t)

	s.logf("mem@flush created L%d@%d N·%d S·%s %q:%q", level, t.file.Num(), n, shortenb(int(t.size)), t.min, t.max)

	c.level = level
	return nil
}

func (c *cMem) reset() {
	c.rec = new(sessionRecord)
}

func (c *cMem) commit(journal, seq uint64) error {
	c.rec.setJournalNum(journal)
	c.rec.setSeq(seq)
	// Commit changes
	return c.s.commit(c.rec)
}

func (d *DB) compactionError() {
	var err error
noerr:
	for {
		select {
		case _, _ = <-d.closeCh:
			return
		case err = <-d.compErrSetCh:
			if err != nil {
				goto haserr
			}
		}
	}
haserr:
	for {
		select {
		case _, _ = <-d.closeCh:
			return
		case err = <-d.compErrSetCh:
			if err == nil {
				goto noerr
			}
		case d.compErrCh <- err:
		}
	}
}

func (d *DB) transact(name string, exec, rollback func() error) {
	s := d.s
	defer func() {
		if x := recover(); x != nil {
			if x == errTransactExiting && rollback != nil {
				if err := rollback(); err != nil {
					s.logf("%s rollback error %q", name, err)
				}
			}
			panic(x)
		}
	}()
	for n := 0; ; n++ {
		if d.isClosed() {
			s.logf("%s exiting", name)
			panic(errTransactExiting)
		} else if n > 0 {
			s.logf("%s retrying N·%d", name, n)
		}
		err := exec()
		select {
		case _, _ = <-d.closeCh:
			s.logf("%s exiting", name)
			panic(errTransactExiting)
		case d.compErrSetCh <- err:
		}
		if err == nil {
			return
		}
		s.logf("%s error %q", name, err)
		time.Sleep(time.Second)
	}
}

func (d *DB) memCompaction() {
	s := d.s
	c := newCMem(s)
	stats := new(cStatsStaging)
	mem := d.getFrozenMem()

	s.logf("mem@flush N·%d S·%s", mem.Len(), shortenb(mem.Size()))

	// Don't compact empty memdb.
	if mem.Len() == 0 {
		s.logf("mem@flush skipping")
		// drop frozen mem
		d.dropFrozenMem()
		return
	}

	d.transact("mem@flush", func() (err error) {
		stats.startTimer()
		defer stats.stopTimer()
		return c.flush(mem, -1)
	}, func() error {
		for _, r := range c.rec.addedTables {
			s.logf("mem@flush rollback @%d", r.num)
			f := s.getTableFile(r.num)
			if err := f.Remove(); err != nil {
{'fraction_non_alphanumeric': 0.1659376322844645, 'fraction_numerical': 0.005785240581346127, 'mean_word_length': 2.8987898789878987, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3677242', 'n_tokens_mistral': 3157, 'n_tokens_neox': 2904, 'n_words': 750}