scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 08:10:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1070 from jaten-molecula/slurp_profile_rbf_comments

Browse source 
rbf default. Add TODO comments, slurp -profile returns a cpu profile
This commit is contained in:
tgruben 2020-11-03 19:52:54 -06:00 • committed by GitHub
commit bc26de63b0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 170 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

78
cmd/slurp/slurp.go
View file

@ -20,23 +20,18 @@ import (
"compress/gzip"
"context"
"flag"
"time"
//"fmt"
"fmt"
"io"
"io/ioutil"
gohttp "net/http"
"net/url"
"os"
"strconv"
"strings"
"time"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/http"
//"log"
"os"
//"path/filepath"
//"sort"
"strconv"
"strings"
)
// slurp: slurp is a load-tester for importing bulk data.
@ -51,6 +46,9 @@ type stateMachine struct {
client *http.InternalClient
start time.Time
direct bool
profile string
host string
}
func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
@ -86,6 +84,10 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
return err
}
vv("Finished import %v", time.Since(r.start))
if r.profile != "" {
stopProfile(r.host, r.profile)
}
}
//
uri := GetImportRoaringURI(r.lastIndex, r.lastShard)
@ -143,7 +145,7 @@ func (r *stateMachine) Upload() error {
return nil
}
func UploadTar(srcFile string, direct bool, client *http.InternalClient) error {
func UploadTar(srcFile string, direct bool, client *http.InternalClient, profile, host string) error {
f, err := os.Open(srcFile)
if err != nil {
@ -164,6 +166,8 @@ func UploadTar(srcFile string, direct bool, client *http.InternalClient) error {
viewData: make(map[string][]byte),
start: time.Now(),
direct: direct,
profile: profile,
host: host,
}
runner.client = client
for {
@ -184,9 +188,11 @@ func UploadTar(srcFile string, direct bool, client *http.InternalClient) error {
func main() {
var host string
var direct bool
var profile string
var tarSrcPath string
flag.StringVar(&host, "host", "127.0.0.1:10101", "host to import into")
flag.BoolVar(&direct, "direct", false, "direct write to database (unsafe)")
flag.StringVar(&profile, "profile", "", "profile and save a cpu profile of the import to this file")
flag.StringVar(&tarSrcPath, "src", "q2.tar.gz", "data to import")
flag.Parse()
@ -195,15 +201,63 @@ func main() {
globURI = uri
h := &gohttp.Client{}
if profile != "" {
startProfile(host)
}
c, err := http.NewInternalClient(host, h)
panicOn(err)
t0 := time.Now()
println("uploading", tarSrcPath)
panicOn(UploadTar(tarSrcPath, direct, c))
panicOn(UploadTar(tarSrcPath, direct, c, profile, host))
vv("total elapsed '%v'", time.Since(t0))
}
func startProfile(host string) {
cli := &gohttp.Client{}
req := &gohttp.Request{
Method: "GET",
URL: &url.URL{
Scheme: "http",
Host: host,
Path: "/cpu-profile/start",
},
}
resp, err := cli.Do(req)
if err != nil {
if resp != nil {
resp.Body.Close()
}
panic(err)
}
}
func stopProfile(host, outfile string) {
cli := &gohttp.Client{}
req := &gohttp.Request{
Method: "GET",
URL: &url.URL{
Scheme: "http",
Host: host,
Path: "/cpu-profile/stop",
},
}
resp, err := cli.Do(req)
if err != nil {
if resp != nil {
resp.Body.Close()
}
panic(err)
}
fd, err := os.Create(outfile)
panicOn(err)
defer fd.Close()
_, err = io.Copy(fd, resp.Body)
panicOn(err)
}
var globURI *pilosa.URI
// get correct node to go to.

56
http/handler.go
View file

@ -72,6 +72,8 @@ type Handler struct {
server *http.Server
middleware []func(http.Handler) http.Handler
pprofCPUProfileBuffer *bytes.Buffer
}
// externalPrefixFlag denotes endpoints that are intended to be exposed to clients.
@ -426,6 +428,12 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/internal/translate/field/{index}/{field}/keys/find", handler.handleFindFieldKeys).Methods("POST").Name("FindFieldKeys")
router.HandleFunc("/internal/translate/field/{index}/{field}/keys/create", handler.handleCreateFieldKeys).Methods("POST").Name("CreateFieldKeys")
// endpoints for collecting cpu profiles from a chosen begin point to
// when the client wants to stop. Used for profiling imports that
// could be long or short.
router.HandleFunc("/cpu-profile/start", handler.handleCPUProfileStart).Methods("GET").Name("CPUProfileStart")
router.HandleFunc("/cpu-profile/stop", handler.handleCPUProfileStop).Methods("GET").Name("CPUProfileStop")
// Endpoints to support lattice UI embedded via statik.
// The messiness here reflects the fact that assets live in a nontrivial
// directory structure that is controlled externally.
@ -876,6 +884,54 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
}
}
func (h *Handler) handleCPUProfileStart(w http.ResponseWriter, r *http.Request) {
if h.pprofCPUProfileBuffer == nil {
h.pprofCPUProfileBuffer = bytes.NewBuffer(nil)
} else {
http.Error(w, "cpu profile already in progress", http.StatusBadRequest)
return
}
err := pprof.StartCPUProfile(h.pprofCPUProfileBuffer)
if err != nil {
http.Error(w, fmt.Sprintf("%v", err), http.StatusBadRequest)
h.pprofCPUProfileBuffer = nil
return
}
w.WriteHeader(http.StatusOK)
}
func (h *Handler) handleCPUProfileStop(w http.ResponseWriter, r *http.Request) {
if h.pprofCPUProfileBuffer == nil {
http.Error(w, "no cpu profile in progress", http.StatusBadRequest)
return
}
pprof.StopCPUProfile()
// match what pprof usually returns:
// HTTP/1.1 200 OK
// Content-Disposition: attachment; filename="profile"
// Content-Type: application/octet-stream
// X-Content-Type-Options: nosniff
// Date: Tue, 03 Nov 2020 18:31:36 GMT
// Content-Length: 939
//Send the headers
by := h.pprofCPUProfileBuffer.Bytes()
w.Header().Set("Content-Disposition", "attachment; filename=\"profile\"")
w.Header().Set("Content-Type", "application/octet-stream")
w.Header().Set("Content-Length", fmt.Sprintf("%v", len(by)))
w.Header().Set("X-Content-Type-Options", "nosniff")
_, err := io.Copy(w, h.pprofCPUProfileBuffer)
h.pprofCPUProfileBuffer = nil
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
// handleGetShardsMax handles GET /internal/shards/max requests.
func (h *Handler) handleGetShardsMax(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {

14
rbf/cfg/cfg.go
View file

@ -47,7 +47,7 @@ func NewDefaultConfig() *Config {
return &Config{
MaxSize: DefaultMaxSize,
FsyncEnabled: true,
CheckpointEveryDur: 10 * time.Second,
CheckpointEveryDur: time.Millisecond,
MaxWALSegmentFileSize: 1 << 20,
}
}
@ -55,16 +55,8 @@ func NewDefaultConfig() *Config {
func (cfg *Config) DefineFlags(flags *pflag.FlagSet) {
default0 := NewDefaultConfig()
flags.IntVar(&cfg.MaxWALSegmentFileSize, "rbf-max-wal", default0.MaxWALSegmentFileSize, "RBF write-Ahead-Log file size in bytes")
// TODO: make delayed checkpointing work. Currently
// wal.go readWALPage() can fail to locate some pages
// when checkpointing does not happen after every Commit.
// Once that is done we can return to trying to checkpoint
// after some duration.
//flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur, "RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.")
flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", 0, "RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.")
flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur,
"RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.")
flags.Int64Var(&cfg.MaxSize, "rbf-max-db-size", default0.MaxSize, "RBF maximum size in bytes of a database file (distinct from a WAL file)")
flags.BoolVar(&cfg.FsyncEnabled, "rbf-fsync", default0.FsyncEnabled, "RBF: enable fsync fully safe flush-to-disk at each checkpoint")
}

34
rbf/db.go
View file

@ -219,13 +219,26 @@ func (db *DB) checkpoint(exclusive bool) error {
return err
}
walID := readMetaWALID(page)
// INVAR: walID represents everything already in the DB, and
// any wal page k > walID is in the WAL not the DB.
// Loop over each transaction
// We could be looking at a recovery. When there is no new
// meta page further down in the WAL, then there was power
// failure or the process was killed. So we have to search
// and find the next meta page, if present.
//
// Read ahead to the next meta page, if present, in the WAL. If we
// we find it, then we must ensure the pages between
// [walID, next_meta_page.walID] are committed.
// If there is NOT another meta page after, then those writes get
// rolled back.
walID++
for {
// Determine last page of transaction.
metaWALID, err := findNextWALMetaPage(db.segments, walID)
if err == io.EOF {
if err == ErrNoMetaFound {
break
} else if err != nil {
return err
@ -257,6 +270,25 @@ func (db *DB) checkpoint(exclusive bool) error {
}
}
// TODO: address this problem: if we write a WAL meta page to database page 0 before fsyncing
// the transactions updates from the WAL into the DB, then (upon
// power failure in the middle of a fsync), the meta page might
// get updated before all of the databases pages that included the changes
// that the meta page represents. The only way to have a strict
// ordering that the meta page is updated only after the other
// pages is to fsync it in a 2nd fsync that follows the
// the first. SSDs and HDs both exhibit these "unsynchronized writes".
// reference https://www.usenix.org/system/files/conference/fast13/fast13-final80.pdf
//
// needed pattern:
// 1) write tx-content pages;
// 2) fsync the tx-content pages;
// 3) write meta page;
// 4) fsync the meta page.
// The OS can also be inserting fsyncs at any point (e.g. due to memory pressure)
// and so we have to be certain that the meta page is written after a separate fsync.
// Write page data into main db file.
if err := db.writeDBPage(pgno, page); err != nil {
return err

12
rbf/tx.go
View file

@ -40,10 +40,14 @@ type Tx struct {
meta [PageSize]byte // copy of current meta page
walID int64 // max WAL ID at start of tx
rootRecords []*RootRecord // read-only cache of root records
pageMap *immutable.Map // mapping of database pages to WAL IDs
writable bool // if true, tx can write
exclusive bool // if true, tx writes directly to db file (no wal)
dirty bool // if true, changes have been made
// pageMap holds WAL pages that have not yet been transferred
// into the database pages. So it can be empty, if the whole previous
// WAL has been checkpointed back into the database.
pageMap *immutable.Map // mapping of database pages to WAL IDs
writable bool // if true, tx can write
exclusive bool // if true, tx writes directly to db file (no wal)
dirty bool // if true, changes have been made
wcache []byte // write cache

5
rbf/wal.go
View file

@ -16,7 +16,6 @@ package rbf
import (
"fmt"
"io"
"os"
"path/filepath"
"sort"
@ -199,6 +198,8 @@ func readWALPage(segments []WALSegment, walID int64) ([]byte, error) {
return nil, fmt.Errorf("cannot find segment containing WAL page: %d; over all supplied segments, minWALID=%v, maxWALID=%v; detail='%v'", walID, minWALID, maxWALID, detail)
}
var ErrNoMetaFound = fmt.Errorf("no meta page found")
func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, err error) {
maxWALID := maxWALID(segments)
@ -217,7 +218,7 @@ func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, e
}
}
return -1, io.EOF
return -1, ErrNoMetaFound
}
func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) {

2
txfactory.go
View file

@ -46,7 +46,7 @@ const (
// DefaultTxsrc is set here. pilosa/server/config.go references it
// to set the default for pilosa server exeutable.
// Can be overridden with env variable PILOSA_TXSRC for testing.
const DefaultTxsrc = BoltTxn
const DefaultTxsrc = RBFTxn
// DetectMemAccessPastTx true helps us catch places in api and executor
// where mmapped memory is being accessed after the point in time