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

Merge pull request #889 from molecula/rbf-immutable-wal

Browse source
This commit is contained in:
Ben Johnson 2020-10-06 10:45:47 -06:00 • committed by GitHub
commit ab6581446d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 486 additions and 515 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -7,3 +7,5 @@ build
*~
lattice
release-pilosa-fsck.*.*.tar.gz
/log.*
/tourna.log.*

3
fragment_internal_test.go
View file

@ -33,13 +33,12 @@ import (
"testing"
"testing/quick"
"golang.org/x/sync/errgroup"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
// Test flags

8
rbf/cursor_test.go
View file

@ -76,8 +76,6 @@ func TestCursor_FirstNext_Quick(t *testing.T) {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
const n = 10000
@ -198,8 +196,6 @@ func TestCursor_LastPrev_Quick(t *testing.T) {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
const n = 10000
@ -312,8 +308,6 @@ func TestCursor_Union(t *testing.T) {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
@ -393,8 +387,6 @@ func TestCursor_Intersect(t *testing.T) {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {

382
rbf/db.go
View file

@ -41,12 +41,14 @@ const (
type DB struct {
data []byte // mmap data
file *os.File // file descriptor
segments []*WALSegment // write-ahead log
rootRecords []*RootRecord // cached root records
pageMap *immutable.Map // pgno-to-WALID mapping
txs map[*Tx]struct{} // active transactions
opened bool // true if open
wcache []byte // wal write cache
segments []WALSegment // write-ahead log
mu sync.RWMutex // general mutex
rwmu sync.Mutex // mutex for restricting single writer
exclmu sync.RWMutex // mutex for locking out everyone but a single writer
@ -71,6 +73,7 @@ func NewDB(path string) *DB {
db := &DB{
txs: make(map[*Tx]struct{}),
pageMap: immutable.NewMap(&uint32Hasher{}),
wcache: make([]byte, MaxWALSegmentFileSize+PageSize),
Path: path,
MaxSize: DefaultMaxSize,
}
@ -84,12 +87,10 @@ func (db *DB) DataPath() string {
// WALPath returns the path to the WAL directory.
func (db *DB) WALPath() string {
return filepath.Join(db.Path, "wal")
}
func CreateDirIfNotExist(path string) {
dir := filepath.Dir(path)
if _, err := os.Stat(dir); os.IsNotExist(err) {
err = os.MkdirAll(dir, 0755)
@ -99,10 +100,16 @@ func CreateDirIfNotExist(path string) {
}
}
// TxN returns the number of active transactions.
func (db *DB) TxN() int {
db.mu.RLock()
defer db.mu.RUnlock()
return len(db.txs)
}
// Open opens a database with the file specified in Path.
// Creates a new file if one does not already exist.
func (db *DB) Open() (err error) {
db.mu.Lock()
defer db.mu.Unlock()
@ -143,7 +150,7 @@ func (db *DB) Open() (err error) {
// Open write-ahead log & checkpoint to the end since no transactions are open.
if err := db.openWALSegments(); err != nil {
return fmt.Errorf("wal open: %w", err)
} else if err := db.checkpoint(true); err != nil {
} else if err := db.checkpoint(true, &nopLocker{}); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
@ -171,27 +178,66 @@ func (db *DB) openWALSegments() error {
}
// Truncate everything after the last successful meta page.
if walID, err := db.findLastWALMetaPage(); err != nil {
if walID, err := findLastWALMetaPage(db.segments); err != nil {
return err
} else if err := db.truncateWALAfter(walID); err != nil {
} else if db.segments, err = truncateWALAfter(db.segments, walID); err != nil {
return err
}
return nil
}
// checkpoint copies pages from WAL segments into the main DB file. This can
// updateWALSegment updates or adds a segment.
func (db *DB) updateWALSegment(s WALSegment) {
segments := make([]WALSegment, len(db.segments), len(db.segments)+1)
copy(segments, db.segments)
// Find the matching segment using the path.
segment := walSegmentByPath(segments, s.Path)
// Update existing segment if it already exists.
// Otherwise append segment to the end.
if segment != nil {
*segment = s
} else {
assert(len(segments) == 0 || segments[len(segments)-1].MinWALID < s.MinWALID)
segments = append(segments, s)
}
// Replace DB segment list.
db.segments = segments
}
// Checkpoint copies pages from WAL segments into the main DB file. This can
// only copy pages that aren't in use by an active transaction. The page map
// is rebuilt as well for all WAL pages still in use.
//
// If exclusive is true, all WAL writes are flushed to disk.
func (db *DB) checkpoint(exclusive bool) error {
if !db.opened {
func (db *DB) Checkpoint() error {
return db.checkpoint(false, &db.mu)
}
// checkpoint moves WAL segments to the main DB file.
//
// Note that mu should db.mu when called through DB.Checkpoint() but it
// can be &nopLocker if called under lock. The external API will be used
// to periodically checkpoint outside of a transaction and the locking
// must be used only in the beginning (to obtain the segment list) and at
// the end (when removing old segments from the list). If the entire function
// were to obtain a lock then it would block all new read & write transactions.
func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
// Obtain a snapshot of WAL segments at the start.
mu.Lock()
opened := db.opened
segments := db.segments
mu.Unlock()
if !opened {
return nil
}
// Determine last checkpointed WAL ID.
page, err := db.readPage(nil, 0)
page, err := db.readDBPage(0)
if err != nil {
return err
}
@ -206,7 +252,7 @@ func (db *DB) checkpoint(exclusive bool) error {
pageMap := immutable.NewMap(&uint32Hasher{})
for {
// Determine last page of transaction.
metaWALID, err := db.findNextWALMetaPage(walID)
metaWALID, err := findNextWALMetaPage(segments, walID)
if err == io.EOF {
break
} else if err != nil {
@ -215,9 +261,9 @@ func (db *DB) checkpoint(exclusive bool) error {
// Loop over pages in the transaction.
for ; walID <= metaWALID; walID++ {
canCheckpoint := exclusive || minActiveWALID == 0 || walID <= minActiveWALID
canCheckpoint := exclusive || minActiveWALID == 0 || walID < minActiveWALID
page, err := db.readWALPage(walID)
page, err := readWALPage(segments, walID)
if err != nil {
return err
}
@ -242,13 +288,13 @@ func (db *DB) checkpoint(exclusive bool) error {
// Ensure we actually read the bitmap data in when we checkpoint.
// NOTE: The walID variable is incremented above in the pgno check.
if isBitmapHeader {
if page, err = db.readWALPage(walID); err != nil {
if page, err = readWALPage(segments, walID); err != nil {
return err
}
}
// Write page data into main db file.
if err := db.writePage(pgno, page); err != nil {
if err := db.writeDBPage(pgno, page); err != nil {
return err
}
@ -259,100 +305,66 @@ func (db *DB) checkpoint(exclusive bool) error {
}
}
// Ensure WAL pages are fully copied & synced to DB file.
if err := fsync(db.file); err != nil {
return fmt.Errorf("db file sync: %w", err)
}
// Remove WAL segments that have been checkpointed.
if maxCheckpointedWALID != 0 {
for len(db.segments) > 0 {
segment := db.segments[0]
for _, segment := range segments {
if segment.MaxWALID() > maxCheckpointedWALID {
break
}
segpath := segment.Path()
if err := segment.Close(); err != nil {
return err
} else if err := os.Remove(segpath); err != nil {
if err := func() error {
mu.Lock()
defer mu.Unlock()
return db.removeWALSegment(segment.Path)
}(); err != nil {
return err
}
db.segments, db.segments[0] = db.segments[1:], nil
}
}
// Ensure all segments are flushed and there is no remapped pages.
if exclusive {
mu.Lock()
assert(len(db.segments) == 0)
assert(pageMap.Len() == 0)
mu.Unlock()
}
db.pageMap = pageMap
return nil
}
// truncateWALAfter removes all pages in the WAL after walID.
func (db *DB) truncateWALAfter(walID int64) error {
for i := len(db.segments) - 1; i >= 0; i-- {
segment := db.segments[i]
if segment.MaxWALID() <= walID {
break
}
// Drop entire segment if all pages are after WAL ID.
if walID < segment.MinWALID() {
// removeWALSegment closes and deletes the segment with the given path.
//
// The DB's segment list is entirely replaced so that transactions with
// a reference to the old list can continue to use it without a lock.
func (db *DB) removeWALSegment(path string) error {
newSegments := make([]WALSegment, 0, len(db.segments))
for _, segment := range db.segments {
// Close and remove if path matches.
if segment.Path == path {
if err := segment.Close(); err != nil {
return err
} else if err := os.Remove(segment.Path()); err != nil {
} else if err := os.Remove(segment.Path); err != nil {
return err
}
db.segments, db.segments[i] = db.segments[:len(db.segments)-1], nil
continue
}
// If we only remove some of the WAL pages then truncate and exit
// since segments before this will retain all their pages.
return segment.TruncateAfter(walID)
// Otherwise append to new slice of segments.
newSegments = append(newSegments, segment)
}
// Replace entire slice of segments.
db.segments = newSegments
return nil
}
func (db *DB) findNextWALMetaPage(walID int64) (metaWALID int64, err error) {
maxWALID := db.maxWALID()
for ; walID <= maxWALID; walID++ {
// Read page data from WAL and return if it is a meta page.
page, err := db.readWALPage(walID)
if err != nil {
return walID, err
} else if IsMetaPage(page) {
return walID, nil
}
// Skip over next page if this is a bitmap header.
if IsBitmapHeader(page) {
walID++
}
}
return -1, io.EOF
}
func (db *DB) findLastWALMetaPage() (walID int64, err error) {
if len(db.segments) == 0 {
return 0, nil
}
var maxMetaWALID int64
maxWALID := db.maxWALID()
for walID := db.minWALID(); walID <= maxWALID; walID++ {
if page, err := db.readWALPage(walID); err != nil {
return walID, err
} else if IsBitmapHeader(page) {
walID++ // skip next page for bitmap headers
} else if IsMetaPage(page) {
maxMetaWALID = walID // save max meta WAL ID
}
}
return maxMetaWALID, nil
}
// minActiveWALID returns the lowest WAL ID in use by any active transaction.
// Returns 0 if no transactions are active.
func (db *DB) minActiveWALID() int64 {
@ -365,147 +377,8 @@ func (db *DB) minActiveWALID() int64 {
return walID
}
// ActiveWALSegment returns the most recent WAL segment.
func (db *DB) ActiveWALSegment() *WALSegment {
db.mu.RLock()
defer db.mu.RUnlock()
return db.activeWALSegment()
}
func (db *DB) activeWALSegment() *WALSegment {
if len(db.segments) == 0 {
return nil
}
return db.segments[len(db.segments)-1]
}
// MinWALID returns the lowest WAL ID available in the WAL.
func (db *DB) MinWALID() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.minWALID()
}
func (db *DB) minWALID() int64 {
if len(db.segments) == 0 {
return 0
}
return db.segments[0].MinWALID()
}
// MaxWALID returns the highest WAL ID available in the WAL.
func (db *DB) MaxWALID() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.maxWALID()
}
func (db *DB) maxWALID() int64 {
if len(db.segments) == 0 {
return 0
}
s := db.segments[len(db.segments)-1]
return s.MaxWALID()
}
// WALPageN returns the number of pages across all segments.
func (db *DB) WALPageN() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
var n int64
for _, s := range db.segments {
n += int64(s.PageN())
}
return n
}
// SyncWAL flushes the active segment to disk.
func (db *DB) SyncWAL() error {
if s := db.ActiveWALSegment(); s != nil {
return s.Sync()
}
return nil
}
// readWALPage reads a single page at the given WAL ID.
func (db *DB) readWALPage(walID int64) ([]byte, error) {
// TODO(BBJ): Binary search for segment.
for _, s := range db.segments {
if walID >= s.MinWALID() && walID <= s.MaxWALID() {
return s.ReadWALPage(walID)
}
}
return nil, fmt.Errorf("cannot find segment containing WAL page: %d", walID)
}
func (db *DB) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
if err := db.ensureWritableWALSegment(); err != nil {
return 0, err
}
return db.activeWALSegment().WriteWALPage(page, isMeta)
}
func (db *DB) writeBitmapPage(pgno uint32, page []byte) (walID int64, err error) {
if err := db.ensureWritableWALSegment(); err != nil {
return 0, err
}
// Write header page for next bitmap page.
buf := make([]byte, PageSize)
writePageNo(buf[:], pgno)
writeFlags(buf[:], PageTypeBitmapHeader)
// TODO(BBJ): Write checksum.
if _, err := db.activeWALSegment().WriteWALPage(buf, false); err != nil {
return 0, fmt.Errorf("write bitmap header: %w", err)
}
// Write the bitmap page and return its WALID.
return db.activeWALSegment().WriteWALPage(page, false)
}
func (db *DB) ensureWritableWALSegment() error {
if s := db.activeWALSegment(); s != nil && s.Size() < MaxWALSegmentFileSize {
return nil
}
return db.addWALSegment()
}
// addWALSegment appends a new, writable segment and closing an existing segments for write.
func (db *DB) addWALSegment() error {
// If we have a current active WAL segment then close it and start the
// next segment from the next WAL ID. If there is no existing WAL segments,
// read the last checkpointed WAL ID from the DB and start after that.
var base int64
if s := db.activeWALSegment(); s != nil {
base = s.MaxWALID() + 1
if err := s.CloseForWrite(); err != nil {
return err
}
} else {
page, err := db.readPage(db.pageMap, 0)
if err != nil {
return err
}
base = readMetaWALID(page) + 1
}
// Create new segment file.
s := NewWALSegment(filepath.Join(db.WALPath(), FormatWALSegmentPath(base)))
if err := s.Open(); err != nil {
return fmt.Errorf("add wal segment: %w", err)
}
db.segments = append(db.segments, s)
return nil
}
// Close closes the database.
func (db *DB) Close() (err error) {
// TODO(bbj): Add wait group to hang until last Tx is complete.
// Wait for writer lock.
@ -542,12 +415,12 @@ func (db *DB) Close() (err error) {
// closeWALSegments closes the WAL and all its segments.
func (db *DB) closeWALSegments() (err error) {
for _, s := range db.segments {
if e := s.Close(); e != nil && err == nil {
err = e
}
}
db.segments = nil
return err
}
@ -613,7 +486,6 @@ func (db *DB) HasData(requireOneHotBit bool) (hasAnyRecords bool, err error) {
// Size returns the size of the database & WAL, in bytes.
func (db *DB) Size() (int64, error) {
db.mu.RLock()
defer db.mu.RUnlock()
@ -621,38 +493,27 @@ func (db *DB) Size() (int64, error) {
if err != nil {
return 0, err
}
return db.walSize() + fi.Size(), nil
return walSize(db.segments) + fi.Size(), nil
}
// WALSize returns the size of all WAL segments, in bytes.
func (db *DB) WALSize() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.walSize()
}
func (db *DB) walSize() int64 {
var sz int64
for _, s := range db.segments {
sz += s.Size()
}
return sz
return walSize(db.segments)
}
// WALSegments returns the WAL segments currently on the DB.
// This should only be used for debugging & testing purposes.
func (db *DB) WALSegments() []*WALSegment {
func (db *DB) WALSegments() []WALSegment {
db.mu.RLock()
defer db.mu.RUnlock()
return db.segments
other := make([]WALSegment, len(db.segments))
copy(other, db.segments)
return other
}
// init initializes a new database file.
func (db *DB) init() error {
if err := db.initMetaPage(); err != nil {
return fmt.Errorf("meta: %w", err)
} else if err := db.initRootRecordPage(); err != nil {
@ -713,6 +574,10 @@ func (db *DB) BeginWithExclusiveLock() (_ *Tx, err error) {
}
func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
if exclusive {
assert(writable) // exclusive transactions must be writable
}
if exclusive {
db.exclmu.Lock()
} else {
@ -749,7 +614,7 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
// Flush all WAL writes to disk before an exclusive writer so that we can
// work directly with the on-disk database.
if exclusive {
if err := db.checkpoint(true); err != nil {
if err := db.checkpoint(true, &nopLocker{}); err != nil {
cleanup()
return nil, err
}
@ -762,10 +627,21 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
writable: writable,
exclusive: exclusive,
}
if writable {
tx.wcache = db.wcache[:0]
}
// Copy list of WAL segments so they can be altered by the tx.
// Add last segment to the list of segments that will be updated/added.
if len(db.segments) != 0 {
tx.segments = make([]WALSegment, len(db.segments))
copy(tx.segments, db.segments)
tx.updatedSegmentPaths = []string{tx.segments[len(tx.segments)-1].Path}
}
// Copy meta page into transaction's buffer.
// This page is only written at the end of a dirty transaction.
page, err := db.readPage(db.pageMap, 0)
page, err := db.readMetaPage()
if err != nil {
tx.Rollback()
return nil, err
@ -802,8 +678,8 @@ func (db *DB) removeTx(tx *Tx) error {
// Write pages from WAL to DB.
// TODO(bbj): Move this to an async goroutine.
if tx.writable {
if err := db.checkpoint(false); err != nil {
return err
if err := db.checkpoint(false, &nopLocker{}); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
}
@ -824,21 +700,25 @@ func (db *DB) Check() error {
return tx.Check()
}
// writePage writes a page to the data file.
func (db *DB) writePage(pgno uint32, page []byte) error {
// writeDBPage writes a page to the data file.
func (db *DB) writeDBPage(pgno uint32, page []byte) error {
_, err := db.file.WriteAt(page, int64(pgno)*PageSize)
return err
}
func (db *DB) readPage(pageMap *immutable.Map, pgno uint32) ([]byte, error) {
// Check if page is currently in WAL.
if pageMap != nil {
if walID, ok := pageMap.Get(pgno); ok {
return db.readWALPage(walID.(int64))
}
}
// Otherwise read from the data file.
func (db *DB) readDBPage(pgno uint32) ([]byte, error) {
offset := int64(pgno) * PageSize
return db.data[offset : offset+PageSize], nil
}
func (db *DB) readMetaPage() ([]byte, error) {
if walID, ok := db.pageMap.Get(uint32(0)); ok {
return readWALPage(db.segments, walID.(int64))
}
return db.readDBPage(0)
}
type nopLocker struct{}
func (*nopLocker) Lock() {}
func (*nopLocker) Unlock() {}

5
rbf/db_test.go
View file

@ -112,10 +112,11 @@ func TestDB_Recovery(t *testing.T) {
tx1.Rollback()
// Close database & truncate WAL to remove commit page & bitmap data page.
segment := db.ActiveWALSegment()
segments := db.WALSegments()
segment := segments[len(segments)-1]
if err := db.Close(); err != nil {
t.Fatal(err)
} else if err := os.Truncate(segment.Path(), segment.Size()-(2*rbf.PageSize)); err != nil {
} else if err := os.Truncate(segment.Path, segment.Size()-(2*rbf.PageSize)); err != nil {
t.Fatal(err)
}

2
rbf/os.go
View file

@ -19,4 +19,4 @@ package rbf
// DefaultMaxSize is the default mmap size and therefore the maximum allowed
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 100 * (1 << 30) // 100GB
const DefaultMaxSize = 4 * (1 << 30)

28
rbf/rbf.go
View file

@ -91,6 +91,11 @@ var (
// Debug is just a temporary flag used for debugging.
var Debug bool
// Testing constants.
const (
SyncEnabled = true
)
// Magic32 returns the magic bytes as a big endian encoded uint32.
func Magic32() uint32 {
return binary.BigEndian.Uint32([]byte(Magic))
@ -648,3 +653,26 @@ func RowValues(b []uint64) []uint64 {
// _, file, line, _ := runtime.Caller(skip + 1)
// return fmt.Sprintf("%s:%d", file, line)
// }
// truncate truncates the file at path to sz bytes. File must exist.
func truncate(path string, sz int64) error {
f, err := os.OpenFile(path, os.O_WRONLY, 0666)
if err != nil {
return fmt.Errorf("open file: %w", err)
}
defer f.Close()
if err := f.Truncate(sz); err != nil {
return fmt.Errorf("truncate: %w", err)
} else if err := fsync(f); err != nil {
return fmt.Errorf("sync: %w", err)
}
return f.Close()
}
func fsync(f *os.File) error {
if !SyncEnabled {
return nil
}
return f.Sync()
}

20
rbf/rbf_norace.go
View file

@ -1,20 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// +build !race
package rbf
// RaceEnabled is true if the -race flag is enabled.
const RaceEnabled = false

20
rbf/rbf_race.go
View file

@ -1,20 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// +build race
package rbf
// RaceEnabled is true if the -race flag is enabled.
const RaceEnabled = true

2
rbf/rbf_test.go
View file

@ -86,6 +86,8 @@ func MustCloseDB(tb testing.TB, db *rbf.DB) {
tb.Helper()
if err := db.Check(); err != nil && err != rbf.ErrClosed {
tb.Fatal(err)
} else if n := db.TxN(); n != 0 {
tb.Fatalf("db still has %d active transactions; must closed before closing db", n)
} else if err := db.Close(); err != nil && err != rbf.ErrClosed {
tb.Fatal(err)
} else if err := os.RemoveAll(db.Path); err != nil {

180
rbf/tx.go
View file

@ -17,9 +17,9 @@ import (
"fmt"
"io"
"math"
"os"
"path/filepath"
"sort"
//"strconv"
"strings"
"sync"
@ -33,15 +33,19 @@ var _ = txkey.ToString
// Tx represents a transaction.
type Tx struct {
mu sync.RWMutex
db *DB // parent db
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
mu sync.RWMutex
db *DB // parent db
segments []WALSegment // copy of WAL segments
updatedSegmentPaths []string // updated or added segment paths
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
wcache []byte // write cache
// If Rollback() has already completed, don't do it again.
// Note db == nil means that commit has already been done.
@ -76,7 +80,7 @@ func (tx *Tx) Commit() error {
if tx.dirty {
if err := tx.writeMetaPage(MetaPageFlagCommit); err != nil {
return err
} else if err := tx.db.SyncWAL(); err != nil {
} else if err := tx.flushWALWriter(); err != nil {
return err
}
@ -89,6 +93,11 @@ func (tx *Tx) Commit() error {
tx.db.mu.Lock()
tx.db.rootRecords = tx.rootRecords
tx.db.pageMap = tx.pageMap
for _, path := range tx.updatedSegmentPaths {
segment := walSegmentByPath(tx.segments, path)
assert(segment != nil)
tx.db.updateWALSegment(*segment)
}
tx.db.mu.Unlock()
}
@ -112,11 +121,12 @@ func (tx *Tx) Rollback() {
// TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen?
// Remove all WAL pages that have been written by this transaction.
if tx.dirty {
if err := tx.db.truncateWALAfter(tx.walID); err != nil {
panic(err)
if _, err := truncateWALAfter(tx.segments, tx.walID); err != nil {
panicOn(err)
}
tx.segments = nil
tx.updatedSegmentPaths = nil
}
// Disconnect transaction from DB.
@ -915,11 +925,28 @@ func (tx *Tx) readPage(pgno uint32) ([]byte, error) {
return tx.meta[:], nil
}
// Verify page number requested is within current size of database.
pageN := readMetaPageN(tx.meta[:])
if pgno > pageN {
return nil, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
}
return tx.db.readPage(tx.pageMap, pgno)
// Check if page is remapped.
if walID, ok := tx.pageMap.Get(pgno); ok {
walID64 := walID.(int64)
// Read from write cache if not yet flushed to disk.
maxWALID := activeWALSegment(tx.segments).MaxWALID()
if walID64 > maxWALID {
offset := (walID64 - maxWALID - 1) * PageSize
return tx.wcache[offset : offset+PageSize], nil
}
// Otherwise return remapped page from WAL segment.
return readWALPage(tx.segments, walID64)
}
return tx.db.readDBPage(pgno)
}
func (tx *Tx) writePage(page []byte) error {
@ -928,11 +955,11 @@ func (tx *Tx) writePage(page []byte) error {
// If we are running in exclusive mode, directly write page to database.
if tx.exclusive {
return tx.db.writePage(readPageNo(page), page)
return tx.db.writeDBPage(readPageNo(page), page)
}
// Write page to WAL and obtain position in WAL.
walID, err := tx.db.writeWALPage(page, false)
walID, err := tx.writeWALPage(page, false)
if err != nil {
return err
}
@ -948,11 +975,11 @@ func (tx *Tx) writeBitmapPage(pgno uint32, page []byte) error {
// If we are running in exclusive mode, directly write page to database.
if tx.exclusive {
return tx.db.writePage(pgno, page)
return tx.db.writeDBPage(pgno, page)
}
// Write bitmap to WAL and obtain WAL position of the actual page data (not the prefix page).
walID, err := tx.db.writeBitmapPage(pgno, page)
walID, err := tx.writeBitmapWALPage(pgno, page)
if err != nil {
return err
}
@ -968,11 +995,11 @@ func (tx *Tx) writeMetaPage(flag uint32) error {
// If we are running in exclusive mode, directly write page to database.
if tx.exclusive {
return tx.db.writePage(0, tx.meta[:])
return tx.db.writeDBPage(0, tx.meta[:])
}
// Write page to WAL and obtain position in WAL.
walID, err := tx.db.writeWALPage(tx.meta[:], true)
walID, err := tx.writeWALPage(tx.meta[:], true)
if err != nil {
return err
}
@ -1562,3 +1589,112 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
}
return
}
func (tx *Tx) flushWALWriter() error {
// Ignore if we have no data in the write cache.
if len(tx.wcache) == 0 {
return nil
}
// Determine active WAL segment.
assert(len(tx.segments) != 0)
segment := &tx.segments[len(tx.segments)-1]
// Open write handle to active segment.
w, err := os.OpenFile(segment.Path, os.O_WRONLY, 0666)
if err != nil {
return fmt.Errorf("open wal segment write handle: %w", err)
}
defer w.Close()
// Flush cache to writer.
if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil {
return fmt.Errorf("write wal segment: %w", err)
} else if err := fsync(w); err != nil {
return fmt.Errorf("sync wal segment: %w", err)
} else if err := w.Close(); err != nil {
return fmt.Errorf("close wal segment: %w", err)
}
// Increase the size of the last WAL segment & clear cache.
assert(len(tx.wcache)%PageSize == 0)
segment.PageN += len(tx.wcache) / PageSize
tx.wcache = tx.wcache[:0]
return nil
}
func (tx *Tx) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
if err := tx.ensureWritableWALSegment(); err != nil {
return 0, err
}
// Determine next WAL ID from cached meta page.
walID = readMetaWALID(tx.meta[:]) + 1
// Update WAL ID on cached meta page.
writeMetaWALID(tx.meta[:], walID)
// Append write to write buffer.
tx.wcache = append(tx.wcache, page...)
return walID, nil
}
func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err error) {
if err := tx.ensureWritableWALSegment(); err != nil {
return 0, err
}
// Write header page for next bitmap page.
buf := make([]byte, PageSize)
writePageNo(buf[:], pgno)
writeFlags(buf[:], PageTypeBitmapHeader)
// TODO(BBJ): Write checksum.
if _, err := tx.writeWALPage(buf, false); err != nil {
return 0, fmt.Errorf("write bitmap header: %w", err)
}
// Write the bitmap page and return its WALID.
return tx.writeWALPage(page, false)
}
func (tx *Tx) ensureWritableWALSegment() error {
// Ignore if we still have space in the write cache.
writeCacheSize := int64(len(tx.wcache))
if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < MaxWALSegmentFileSize {
return nil
}
// Flush write cache out to file before adding new segment.
if err := tx.flushWALWriter(); err != nil {
return err
}
// If we have a current active WAL segment then close it and start the
// next segment from the next WAL ID. If there is no existing WAL segments,
// read the last checkpointed WAL ID from the DB and start after that.
var base int64
if len(tx.segments) != 0 {
base = activeWALSegment(tx.segments).MaxWALID() + 1
} else {
page, err := tx.readPage(0)
if err != nil {
return err
}
base = readMetaWALID(page) + 1
}
// Create new segment file.
s := NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base)))
if err := s.Open(); err != nil {
return fmt.Errorf("add wal segment: %w", err)
}
// Track all segments that need to be added back to DB.
// The DB can remove segments in the background so we don't want to replace.
tx.segments = append(tx.segments, s)
tx.updatedSegmentPaths = append(tx.updatedSegmentPaths, s.Path)
return nil
}

4
rbf/tx_test.go
View file

@ -218,8 +218,6 @@ func TestTx_Add_Quick(t *testing.T) {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
@ -257,8 +255,6 @@ func TestTx_AddRemove_Quick(t *testing.T) {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {

320
rbf/wal.go
View file

@ -16,9 +16,9 @@ package rbf
import (
"fmt"
"io"
"os"
"path/filepath"
"sync"
"syscall"
"github.com/pilosa/pilosa/v2/syswrap"
@ -26,66 +26,40 @@ import (
// WALSegment represents a single file in the WAL.
type WALSegment struct {
mu sync.RWMutex
minWALID int64 // base WALID; calculated from path
path string // path to file
w *os.File // write handle
data []byte // read-only mmap data
writeCache []byte // write buffer
pageN int // number of written pages
Path string // path to file
MinWALID int64 // base WALID; calculated from path
PageN int // number of written pages
data []byte // read-only mmap data
}
// NewWALSegment returns a new instance of WALSegment for a given path.
func NewWALSegment(path string) *WALSegment {
return &WALSegment{
path: path,
func NewWALSegment(path string) WALSegment {
return WALSegment{
Path: path,
}
}
// Path returns the path the segment was initialized with.
func (s *WALSegment) Path() string { return s.path }
// MinWALID returns the initial WAL ID of the segment. Only available after Open().
func (s *WALSegment) MinWALID() int64 {
s.mu.RLock()
defer s.mu.RUnlock()
return s.minWALID
}
// MaxWALID returns the maximum WAL ID of the segment. Only available after Open().
func (s *WALSegment) MaxWALID() int64 {
s.mu.RLock()
defer s.mu.RUnlock()
return s.minWALID + int64(s.pageN) - 1
}
// PageN returns the number of pages in the segment.
func (s *WALSegment) PageN() int {
s.mu.RLock()
defer s.mu.RUnlock()
return s.pageN
func (s WALSegment) MaxWALID() int64 {
return s.MinWALID + int64(s.PageN) - 1
}
// Size returns the current size of the segment, in bytes.
func (s *WALSegment) Size() int64 {
s.mu.RLock()
defer s.mu.RUnlock()
return int64(s.pageN) * PageSize
func (s WALSegment) Size() int64 {
return int64(s.PageN) * PageSize
}
func (s *WALSegment) Open() (err error) {
s.mu.Lock()
defer s.mu.Unlock()
// Extract base WAL ID and validate path.
if s.minWALID, err = ParseWALSegmentPath(s.path); err != nil {
if s.MinWALID, err = ParseWALSegmentPath(s.Path); err != nil {
return err
}
// Determine file size & create if necessary.
var sz int64
if fi, err := os.Stat(s.path); os.IsNotExist(err) {
if f, err := os.OpenFile(s.path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
if fi, err := os.Stat(s.Path); os.IsNotExist(err) {
if f, err := os.OpenFile(s.Path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("touch wal segment file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close touched wal segment file: %w", err)
@ -97,11 +71,11 @@ func (s *WALSegment) Open() (err error) {
}
// Determine page count & truncate if a partial page is written.
s.pageN = int(sz / PageSize)
s.PageN = int(sz / PageSize)
if sz%PageSize != 0 {
sz = int64(s.pageN * PageSize)
if err := os.Truncate(s.path, sz); err != nil {
return fmt.Errorf("truncate wal segment file: %w", err)
sz = int64(s.PageN * PageSize)
if err := truncate(s.Path, sz); err != nil {
return fmt.Errorf("truncate wal file: %w", err)
}
}
@ -113,7 +87,7 @@ func (s *WALSegment) Open() (err error) {
}
// Open file as a read-only memory map.
if f, err := os.OpenFile(s.path, os.O_RDONLY, 0666); err != nil {
if f, err := os.OpenFile(s.Path, os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open wal segment file: %w", err)
} else if s.data, err = syswrap.Mmap(int(f.Fd()), 0, int(mmapSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
@ -127,12 +101,6 @@ func (s *WALSegment) Open() (err error) {
// Close closes the write handle and the read-only mmap.
func (s *WALSegment) Close() error {
s.mu.Lock()
defer s.mu.Unlock()
if err := s.closeForWrite(); err != nil {
return err
}
if s.data != nil {
if err := syswrap.Munmap(s.data); err != nil {
return err
@ -142,144 +110,148 @@ func (s *WALSegment) Close() error {
return nil
}
// CloseForWrite closes the write handle, if initialized.
func (s *WALSegment) CloseForWrite() error {
s.mu.Lock()
defer s.mu.Unlock()
return s.closeForWrite()
}
func (s *WALSegment) closeForWrite() error {
// Ensure write buffer is flushed out.
if err := s.sync(); err != nil {
return err
}
s.writeCache = nil
// Close underlying file writer.
if s.w != nil {
if err := s.w.Close(); err != nil {
return err
}
s.w = nil
}
return nil
}
// ReadWALPage reads a single page at the given WAL ID.
func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
s.mu.RLock()
defer s.mu.RUnlock()
// Ensure requested ID is contained in this file.
if walID < s.minWALID || walID > s.minWALID+int64(s.pageN) {
return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.minWALID, s.pageN)
if walID < s.MinWALID || walID > s.MinWALID+int64(s.PageN) {
return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.MinWALID, s.PageN)
}
offset := (walID - s.minWALID) * PageSize
// If offset is within write buffer, return from write buffer.
writeBufferOffset := int64((s.pageN * PageSize) - len(s.writeCache))
if offset >= writeBufferOffset {
buf := s.writeCache[offset-writeBufferOffset:]
return buf[:PageSize:PageSize], nil
}
// Otherwise return from on-disk mmap.
offset := (walID - s.MinWALID) * PageSize
return s.data[offset : offset+PageSize], nil
}
// WriteWALPage writes a single page to the WAL segment and returns its WAL identifier.
func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) {
assert(len(page) == PageSize) // invalid page size
s.mu.Lock()
defer s.mu.Unlock()
// Initialize write file handle if not yet initialized.
if s.w == nil {
if s.w, err = os.OpenFile(s.path, os.O_WRONLY, 0666); err != nil {
return 0, fmt.Errorf("open wal segment write handle: %w", err)
func walSegmentByPath(segments []WALSegment, path string) *WALSegment {
for i := range segments {
if segments[i].Path == path {
return &segments[i]
}
}
// Determine current WAL position.
walID = s.minWALID + int64(s.pageN)
// Write WAL ID if this is a meta page.
if isMeta {
writeMetaWALID(page, walID)
// TODO: Write meta page checksum
}
// Append write to write buffer & increment page count.
if s.writeCache == nil {
s.writeCache = make([]byte, 0, MaxWALSegmentFileSize+PageSize)
}
s.writeCache = append(s.writeCache, page...)
s.pageN++
return walID, nil
}
// TruncateAfter removes all pages after a given WAL ID.
func (s *WALSegment) TruncateAfter(walID int64) error {
s.mu.Lock()
defer s.mu.Unlock()
// Ensure this is a partial truncation. Full truncation of a segment
// should be performed by the DB since it needs to remove the segment.
assert(walID > s.minWALID)
// Update to new page size.
newPageN := int((walID - s.minWALID) + 1) // new page count of segment
truncPageN := s.pageN - newPageN // number of pages removed
s.pageN = newPageN
// Check to see if we are only truncating from the write cache.
writeCachePageN := len(s.writeCache) / PageSize
if truncPageN <= int(writeCachePageN) {
s.writeCache = s.writeCache[:(writeCachePageN-truncPageN)*PageSize]
return nil
}
// Clear write cache.
s.writeCache = s.writeCache[:0]
// Remove on disk pages.
return os.Truncate(s.path, int64(s.pageN)*PageSize)
}
// Flush flushes the write buffer to the OS cache.
func (s *WALSegment) Flush() error {
s.mu.Lock()
defer s.mu.Unlock()
return s.flush()
}
func (s *WALSegment) flush() error {
if _, err := s.w.WriteAt(s.writeCache, int64((s.pageN*PageSize)-len(s.writeCache))); err != nil {
return fmt.Errorf("wal segment write: %w", err)
}
s.writeCache = s.writeCache[:0]
return nil
}
// Sync flushes the write buffer and invokes a file sync to flush data to disk.
func (s *WALSegment) Sync() error {
s.mu.Lock()
defer s.mu.Unlock()
return s.sync()
func activeWALSegment(segments []WALSegment) WALSegment {
if len(segments) == 0 {
return WALSegment{}
}
return segments[len(segments)-1]
}
func (s *WALSegment) sync() error {
if s.w == nil {
return nil
func minWALID(segments []WALSegment) int64 {
if len(segments) == 0 {
return 0
}
if err := s.flush(); err != nil {
return err
return segments[0].MinWALID
}
func maxWALID(segments []WALSegment) int64 {
if len(segments) == 0 {
return 0
}
s := segments[len(segments)-1]
return s.MaxWALID()
}
func walSize(segments []WALSegment) int64 {
var sz int64
for _, s := range segments {
sz += s.Size()
}
return sz
}
// readWALPage reads a single page at the given WAL ID.
func readWALPage(segments []WALSegment, walID int64) ([]byte, error) {
// TODO(BBJ): Binary search for segment.
for _, s := range segments {
if walID >= s.MinWALID && walID <= s.MaxWALID() {
return s.ReadWALPage(walID)
}
}
return nil, fmt.Errorf("cannot find segment containing WAL page: %d", walID)
}
func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, err error) {
maxWALID := maxWALID(segments)
for ; walID <= maxWALID; walID++ {
// Read page data from WAL and return if it is a meta page.
page, err := readWALPage(segments, walID)
if err != nil {
return walID, err
} else if IsMetaPage(page) {
return walID, nil
}
// Skip over next page if this is a bitmap header.
if IsBitmapHeader(page) {
walID++
}
}
return -1, io.EOF
}
func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) {
if len(segments) == 0 {
return 0, nil
}
var maxMetaWALID int64
maxWALID := maxWALID(segments)
for walID := minWALID(segments); walID <= maxWALID; walID++ {
if page, err := readWALPage(segments, walID); err != nil {
return walID, err
} else if IsBitmapHeader(page) {
walID++ // skip next page for bitmap headers
} else if IsMetaPage(page) {
maxMetaWALID = walID // save max meta WAL ID
}
}
return maxMetaWALID, nil
}
// truncateWALAfter removes all pages in the WAL after walID.
func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) {
var newSegments []WALSegment
for i := range segments {
segment := &segments[i]
// Append entire segment if WAL range entirely before target WAL ID.
if walID > segment.MaxWALID() {
newSegments = append(newSegments, *segment)
continue
}
// If we only remove some of the WAL pages then truncate and append.
if segment.MinWALID < walID {
newSegment := *segment
newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
if err := truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
return segments, err
}
newSegments = append(newSegments, newSegment)
continue
}
// Drop entire segment if all pages are after WAL ID.
if err := segment.Close(); err != nil {
return segments, err
} else if err := os.Remove(segment.Path); err != nil {
return segments, err
}
}
return newSegments, nil
}
func DumpWALSegments(segments []WALSegment) {
fmt.Printf("WAL (%d segments)\n", len(segments))
for i, s := range segments {
fmt.Printf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN)
}
return s.w.Sync()
}
// FormatWALSegmentPath returns a path for a WAL segment using a WAL ID.

20
rbf/wal_test.go
View file

@ -15,10 +15,10 @@
package rbf_test
import (
"bytes"
"encoding/hex"
// "bytes"
// "encoding/hex"
"io/ioutil"
"math/rand"
// "math/rand"
"os"
"path/filepath"
"testing"
@ -30,9 +30,9 @@ func TestWALSegment_Open(t *testing.T) {
t.Run("OK", func(t *testing.T) {
s := MustOpenWALSegment(t, 10)
defer MustCloseWALSegment(t, s)
if got, want := s.MinWALID(), int64(10); got != want {
if got, want := s.MinWALID, int64(10); got != want {
t.Fatalf("Base()=%d, want %d", got, want)
} else if got, want := s.PageN(), 0; got != want {
} else if got, want := s.PageN, 0; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
})
@ -40,6 +40,7 @@ func TestWALSegment_Open(t *testing.T) {
// TODO(BBJ): Test open w/ partially written pages.
}
/*
func TestWALSegment_WritePage(t *testing.T) {
rand := rand.New(rand.NewSource(0))
s := MustOpenWALSegment(t, 10)
@ -84,6 +85,7 @@ func TestWALSegment_WritePage(t *testing.T) {
t.Fatal("unexpected second page")
}
}
*/
func TestFormatWALSegmentPath(t *testing.T) {
if got, want := rbf.FormatWALSegmentPath(1234), "00000000000004d2.wal"; got != want {
@ -107,6 +109,7 @@ func TestParseWALSegmentPath(t *testing.T) {
})
}
/*
func BenchmarkWALSegment_WriteWALPage(b *testing.B) {
b.Run("8KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 8*(1<<10)) })
b.Run("16KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 16*(1<<10)) })
@ -147,9 +150,10 @@ func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) {
b.SetBytes(rbf.MaxWALSegmentFileSize)
}
*/
// MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
func MustOpenWALSegment(tb testing.TB, walID int64) *rbf.WALSegment {
func MustOpenWALSegment(tb testing.TB, walID int64) rbf.WALSegment {
tb.Helper()
dir, err := ioutil.TempDir("", "")
@ -169,11 +173,11 @@ func MustOpenWALSegment(tb testing.TB, walID int64) *rbf.WALSegment {
}
// MustCloseWALSegment closes s. Fails on error.
func MustCloseWALSegment(tb testing.TB, s *rbf.WALSegment) {
func MustCloseWALSegment(tb testing.TB, s rbf.WALSegment) {
tb.Helper()
if err := s.Close(); err != nil {
tb.Fatal(err)
} else if err := os.Remove(s.Path()); err != nil {
} else if err := os.Remove(s.Path); err != nil {
tb.Fatal(err)
}
}

3
server/cluster_test.go
View file

@ -25,9 +25,8 @@ import (
"testing"
"time"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
"golang.org/x/sync/errgroup"
)

2
test/pilosa.go
View file

@ -302,7 +302,7 @@ func Do(t *testing.T, method, urlStr string, body string) *httpResponse {
// set a timeout instead of allowing gohttp.Defaultclient to
// potentially hang forever.
hc := &gohttp.Client{
Timeout: time.Second * 10,
Timeout: time.Second * 30,
}
resp, err := hc.Do(req)