<commit_before>package transactionpool

import (
	"fmt"
	"os"
	"path/filepath"
	"time"

	"github.com/NebulousLabs/Sia/build"
	"github.com/NebulousLabs/Sia/modules"
	"github.com/NebulousLabs/Sia/persist"
	"github.com/NebulousLabs/Sia/types"

	"github.com/NebulousLabs/bolt"
)

const tpoolSyncRate = time.Minute * 2

// threadedRegularSync will make sure that sync gets called on the database
// every once in a while.
func (tp *TransactionPool) threadedRegularSync() {
	if err := tp.tg.Add(); err != nil {
		return
	}
	defer tp.tg.Done()
	for {
		select {
		case <-tp.tg.StopChan():
			// A queued AfterStop will close out the db properly.
			return
		case <-time.After(tpoolSyncRate):
			tp.mu.Lock()
			tp.syncDB()
			tp.mu.Unlock()
		}
	}
}

// syncDB commits the current global transaction and immediately begins a new
// one.
func (tp *TransactionPool) syncDB() {
	// Commit the existing tx.
	err := tp.dbTx.Commit()
	if err != nil {
		tp.log.Severe("ERROR: failed to apply database update:", err)
		tp.dbTx.Rollback()
	}
	// Begin a new tx
	tp.dbTx, err = tp.db.Begin(true)
	if err != nil {
		tp.log.Severe("ERROR: failed to initialize a db transaction:", err)
	}
	// Flush the cached DB pages from memory
	err = tp.dbTx.FlushDBPages()
	if err != nil {
		tp.log.Severe("ERROR: failed to flush db pages:", err)
	}
}

// resetDB deletes all consensus related persistence from the transaction pool.
func (tp *TransactionPool) resetDB(tx *bolt.Tx) error {
	err := tx.DeleteBucket(bucketConfirmedTransactions)
	if err != nil {
		return err
	}
	err = tp.putRecentBlockID(tx, types.BlockID{})
	if err != nil {
		return err
	}
	err = tp.putRecentConsensusChange(tx, modules.ConsensusChangeBeginning)
	if err != nil {
		return err
	}
	err = tp.putBlockHeight(tx, types.BlockHeight(0))
	if err != nil {
		return err
	}
	_, err = tx.CreateBucket(bucketConfirmedTransactions)
	return err
}

// initPersist creates buckets in the database
func (tp *TransactionPool) initPersist() error {
	// Create the persist directory if it does not yet exist.
	err := os.MkdirAll(tp.persistDir, 0700)
	if err != nil {
		return err
	}

	// Create the tpool logger.
	tp.log, err = persist.NewFileLogger(filepath.Join(tp.persistDir, logFile))
	if err != nil {
		return build.ExtendErr("unable to initialize the transaction pool logger", err)
	}
	tp.tg.AfterStop(func() {
		err := tp.log.Close()
		if err != nil {
			fmt.Println("Unable to close the transaction pool logger:", err)
		}
	})

	// Open the database file.
	tp.db, err = persist.OpenDatabase(dbMetadata, filepath.Join(tp.persistDir, dbFilename))
	if err != nil {
		return err
	}
	tp.tg.AfterStop(func() {
		err := tp.db.Close()
		if err != nil {
			tp.log.Println("Error while closing transaction pool database:", err)
		}
	})
	// Create the global tpool tx that will be used for most persist actions.
	tp.dbTx, err = tp.db.Begin(true)
	if err != nil {
		return build.ExtendErr("unable to begin tpool dbTx", err)
	}
	tp.tg.AfterStop(func() {
		err := tp.dbTx.Commit()
		if err != nil {
			tp.log.Println("Unable to close transaction properly during shutdown:", err)
		}
	})
	// Spin up the thread that occasionally syncrhonizes the database.
	go tp.threadedRegularSync()
<commit_msg>wrap Commit in mutex when closing tpool<commit_after>package transactionpool

import (
	"fmt"
	"os"
	"path/filepath"
	"time"

	"github.com/NebulousLabs/Sia/build"
	"github.com/NebulousLabs/Sia/modules"
	"github.com/NebulousLabs/Sia/persist"
	"github.com/NebulousLabs/Sia/types"

	"github.com/NebulousLabs/bolt"
)

const tpoolSyncRate = time.Minute * 2

// threadedRegularSync will make sure that sync gets called on the database
// every once in a while.
func (tp *TransactionPool) threadedRegularSync() {
	if err := tp.tg.Add(); err != nil {
		return
	}
	defer tp.tg.Done()
	for {
		select {
		case <-tp.tg.StopChan():
			// A queued AfterStop will close out the db properly.
			return
		case <-time.After(tpoolSyncRate):
			tp.mu.Lock()
			tp.syncDB()
			tp.mu.Unlock()
		}
	}
}

// syncDB commits the current global transaction and immediately begins a new
// one.
func (tp *TransactionPool) syncDB() {
	// Commit the existing tx.
	err := tp.dbTx.Commit()
	if err != nil {
		tp.log.Severe("ERROR: failed to apply database update:", err)
		tp.dbTx.Rollback()
	}
	// Begin a new tx
	tp.dbTx, err = tp.db.Begin(true)
	if err != nil {
		tp.log.Severe("ERROR: failed to initialize a db transaction:", err)
	}
	// Flush the cached DB pages from memory
	err = tp.dbTx.FlushDBPages()
	if err != nil {
		tp.log.Severe("ERROR: failed to flush db pages:", err)
	}
}

// resetDB deletes all consensus related persistence from the transaction pool.
func (tp *TransactionPool) resetDB(tx *bolt.Tx) error {
	err := tx.DeleteBucket(bucketConfirmedTransactions)
	if err != nil {
		return err
	}
	err = tp.putRecentBlockID(tx, types.BlockID{})
	if err != nil {
		return err
	}
	err = tp.putRecentConsensusChange(tx, modules.ConsensusChangeBeginning)
	if err != nil {
		return err
	}
	err = tp.putBlockHeight(tx, types.BlockHeight(0))
	if err != nil {
		return err
	}
	_, err = tx.CreateBucket(bucketConfirmedTransactions)
	return err
}

// initPersist creates buckets in the database
func (tp *TransactionPool) initPersist() error {
	// Create the persist directory if it does not yet exist.
	err := os.MkdirAll(tp.persistDir, 0700)
	if err != nil {
		return err
	}

	// Create the tpool logger.
	tp.log, err = persist.NewFileLogger(filepath.Join(tp.persistDir, logFile))
	if err != nil {
		return build.ExtendErr("unable to initialize the transaction pool logger", err)
	}
	tp.tg.AfterStop(func() {
		err := tp.log.Close()
		if err != nil {
			fmt.Println("Unable to close the transaction pool logger:", err)
		}
	})

	// Open the database file.
	tp.db, err = persist.OpenDatabase(dbMetadata, filepath.Join(tp.persistDir, dbFilename))
	if err != nil {
		return err
	}
	tp.tg.AfterStop(func() {
		err := tp.db.Close()
		if err != nil {
			tp.log.Println("Error while closing transaction pool database:", err)
		}
	})
	// Create the global tpool tx that will be used for most persist actions.
	tp.dbTx, err = tp.db.Begin(true)
	if err != nil {
		return build.ExtendErr("unable to begin tpool dbTx", err)
	}
	tp.tg.AfterStop(func() {
		tp.mu.Lock()
		err := tp.dbTx.Commit()
		tp.mu.Unlock()
		if err != nil {
			tp.log.Println("Unable to close transaction properly during shutdown:", err)
		}
	})
	// Spin up the thread that occasionally syncrhonizes the database.
	go tp.threadedRegularSync()

{'fraction_non_alphanumeric': 0.12803939673745768, 'fraction_numerical': 0.0018467220683287165, 'mean_word_length': 3.5543097407147863, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, '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': '5880786', 'n_tokens_mistral': 2415, 'n_tokens_neox': 2296, 'n_words': 658}