Index out of range about kv HOT 13 CLOSED

// OpenBTree opens a store's BTree using handle. It returns the tree or an
// error, if any. The same tree may be opened more than once, but operations on
// the separate instances should not ever overlap or void the other instances.
// However, the intended API usage is to open the same tree handle only once
// (handled by some upper layer "dispatcher").
func OpenBTree(store *Allocator, collate func(a, b []byte) int, handle int64) (bt *BTree, err error) {
  r := &BTree{store: store, root: btree(handle), collate: collate}
  pb := buffer.Get(7)
  defer buffer.Put(pb)
  b := *pb
  if b, err = store.Get(b, handle); err != nil {
    return
  }
 // The problem arises here
  if len(b) != 7 {
    return nil, &ErrILSEQ{Off: h2off(handle), More: "btree.go:671"}
  }

  return r, nil
}

package main

import (
    "crypto/sha512"
    "encoding/hex"
    "log"
    "strconv"
    "sync"

    "github.com/cznic/kv"
)

const (
    dbName = "test.db"
)

type DB struct {
    sync.Mutex
    db       *kv.DB
    wg       *sync.WaitGroup
    indexNum int      // indexes per db
    numDB    int      // index number of db
    dbList   []string //list of all dbs
}

type DBWithoutMutex struct {
    db       *kv.DB
    wg       *sync.WaitGroup
    indexNum int      // indexes per db
    numDB    int      // index number of db
    dbList   []string //list of all dbs
}

func main() {
    done := make(chan struct{})
    closingDB := make(chan DBWithoutMutex)

    go queue(closingDB, done)

    var err error
    db := DB{}
    db.db, err = kv.Create(dbName+strconv.Itoa(db.numDB), &kv.Options{VerifyDbBeforeOpen: true})
    if err != nil {
        log.Fatal(err)
    }
    db.wg = &sync.WaitGroup{}
    db.dbList = append(db.dbList, dbName+strconv.Itoa(db.numDB))

    // All keys
    keys := struct {
        sync.Mutex
        keys [][]byte
    }{
        keys: [][]byte{},
    }

    wg := sync.WaitGroup{}
    startC := make(chan struct{})
    for count := 0; count < 100; count++ {
        wg.Add(1)
        go func(count int) {
            defer wg.Done()
            <-startC
            for i := 0; i < 100; i++ {

                h := sha512.New()
                h.Write([]byte(strconv.Itoa(i) + "_" + strconv.Itoa(count)))

                keys.Lock()
                keys.keys = append(keys.keys, h.Sum(nil))
                keys.Unlock()

                db.Lock()
                db.indexNum += 1
                if db.indexNum == 1000 {
                    db.indexNum = 0
                    closingDB <- DBWithoutMutex{
                        db:       db.db,
                        wg:       db.wg,
                        indexNum: db.indexNum,
                        numDB:    db.numDB,
                        dbList:   db.dbList,
                    }
                    db.numDB += 1
                    db.db, err = kv.Create(dbName+strconv.Itoa(db.numDB), &kv.Options{VerifyDbBeforeOpen: true})
                    if err != nil {
                        log.Fatalln("Error create", dbName+strconv.Itoa(db.numDB))
                    }
                    db.wg = &sync.WaitGroup{}
                    db.dbList = append(db.dbList, dbName+strconv.Itoa(db.numDB))
                }
                currentDB := DBWithoutMutex{
                    db:       db.db,
                    wg:       db.wg,
                    indexNum: db.indexNum,
                    numDB:    db.numDB,
                    dbList:   db.dbList,
                }
                currentDBname := dbName + strconv.Itoa(db.numDB)
                currentDB.wg.Add(1)
                db.Unlock()

                // time.Sleep(time.Duration(rand.Intn(100)) * time.Microsecond)
                val := []byte{1, 2, 3}

                _, _, err = currentDB.db.Put(nil, h.Sum(nil), func(k, v []byte) ([]byte, bool, error) { return val, true, nil })
                if err != nil {
                    log.Fatalf("Error put key %d to %s: %s\n", currentDB.indexNum, currentDBname, err)
                }

                _, err = currentDB.db.Get(nil, h.Sum(nil))
                if err != nil {
                    log.Fatalf("Error1 get key %d from %s: %s\n", currentDB.indexNum, currentDBname, err)
                }
                _, err = currentDB.db.Get(nil, h.Sum(nil))
                if err != nil {
                    log.Fatalf("Error2 get key %d from %s: %s\n", currentDB.indexNum, currentDBname, err)
                }
                currentDB.wg.Done()
            }
        }(count)
    }
    close(startC)
    wg.Wait()
    log.Println(len(keys.keys))
    db.db.Close()
    close(done)

    for _, path := range db.dbList {
        curDB, err := kv.Open(path, &kv.Options{VerifyDbBeforeOpen: true})
        if err != nil {
            log.Fatalf("Error open %s: %s\n", path, err)
        }
        for _, k := range keys.keys {
            _, err = curDB.Get(nil, k)
            if err != nil {
                log.Fatalf("Error get key %s from %s: %s\n", hex.EncodeToString(k), path, err)
            }
        }
        curDB.Close()
    }
}

func queue(closingDB <-chan DBWithoutMutex, done <-chan struct{}) {
    for {
        select {
        case db := <-closingDB:
            log.Println("Closing")
            go func() {
                db.wg.Wait()
                db.db.Close()
            }()
        case <-done:
            return
        }
    }
}