diff --git a/cmd/slurp/slurp.go b/cmd/slurp/slurp.go index a000a779b..913580172 100644 --- a/cmd/slurp/slurp.go +++ b/cmd/slurp/slurp.go @@ -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. diff --git a/http/handler.go b/http/handler.go index cb42f5e52..f9339a751 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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) { diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go index 8bf8a2162..f13e14f4f 100644 --- a/rbf/cfg/cfg.go +++ b/rbf/cfg/cfg.go @@ -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") } diff --git a/rbf/db.go b/rbf/db.go index d37d0ff46..be12bbf53 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -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 diff --git a/rbf/tx.go b/rbf/tx.go index 7629bdaa0..e0d4e39f0 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -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 diff --git a/rbf/wal.go b/rbf/wal.go index 9304eed0d..4dd9e1966 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -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) { diff --git a/txfactory.go b/txfactory.go index 6d1431dd1..777fa03ae 100644 --- a/txfactory.go +++ b/txfactory.go @@ -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