mirror of
https://github.com/hanzoai/graph.git
synced 2026-10-30 18:35:37 +00:00
zeekay
f5aaeb712e
Derived from Dgraph (github.com/dgraph-io/dgraph) at commit 64804bd2, Apache-2.0. See NOTICE. WHAT THIS IS. The graph engine as a Go library: posting lists, the DQL parser, the type and tokenizer system, the index and schema layers, and the wire types. 39,170 lines of the upstream's 166,337. THE STORE is github.com/luxfi/zapdb, a Badger fork that kept the package name, the subpackages (options, pb, y) and the API — Open, OpenManaged, DB, Options — so the 74 import sites are aliased rather than rewritten. THE CODEC is ZAP. zapdb serializes with ZAP, not protobuf: its pb package builds the ZAP types by default and the protobuf ones only under `-tags grpc`, a path that does not compile (table/builder.go calls pb.Marshal, which that tag removes). One call site in posting needed converting; every other type here is Hanzo's own protobuf and is untouched. NO TRANSPORT. An embedded engine dials nothing, so nothing here imports google.golang.org/grpc. Removed: 20 transport functions in x, the generated service stubs for Raft, Zero and Worker (1,657 lines), and the JWT helper — identity belongs to IAM. The core called none of them. NOT KEPT: worker, conn, raftwal, the cluster commands, the GraphQL layer and the test harnesses. `query`, the DQL execution engine, went with them because it depends on worker; parsing DQL is available here, executing it is not. The history of the upstream project lives at github.com/dgraph-io/dgraph. This repository begins at the derivation because three quarters of that history describes code this fork does not carry.
217 lines
5.7 KiB
Go
217 lines
5.7 KiB
Go
/*
|
|
* SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc.
|
|
* SPDX-License-Identifier: Apache-2.0
|
|
*/
|
|
|
|
package posting
|
|
|
|
import (
|
|
"fmt"
|
|
"math"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
badger "github.com/luxfi/zapdb"
|
|
"github.com/luxfi/zapdb/options"
|
|
bpb "github.com/luxfi/zapdb/pb"
|
|
)
|
|
|
|
var val = make([]byte, 128)
|
|
|
|
func BenchmarkWriter(b *testing.B) {
|
|
createKVList := func() bpb.KVList {
|
|
var KVList bpb.KVList
|
|
for i := range 5000000 {
|
|
n := &bpb.KV{Key: []byte(fmt.Sprint(i)), Value: val, Version: 5}
|
|
KVList.Kv = append(KVList.Kv, n)
|
|
}
|
|
return KVList
|
|
}
|
|
|
|
// Creates separate writer for each thread
|
|
writeInBadgerMThreadsB := func(db *badger.DB, KVList *bpb.KVList, wg *sync.WaitGroup) {
|
|
defer wg.Done()
|
|
wb := db.NewManagedWriteBatch()
|
|
if err := wb.WriteList(KVList); err != nil {
|
|
panic(err)
|
|
}
|
|
require.NoError(b, wb.Flush())
|
|
|
|
}
|
|
|
|
// Resuses one writer for all threads
|
|
writeInBadgerMThreadsW := func(wb *badger.WriteBatch, KVList *bpb.KVList, wg *sync.WaitGroup) {
|
|
defer wg.Done()
|
|
|
|
if err := wb.WriteList(KVList); err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
}
|
|
// Creates separate writer for each thread
|
|
writeInBadgerSingleThreadB := func(db *badger.DB, KVList *bpb.KVList) {
|
|
wb := db.NewManagedWriteBatch()
|
|
if err := wb.WriteList(KVList); err != nil {
|
|
panic(err)
|
|
}
|
|
require.NoError(b, wb.Flush())
|
|
|
|
}
|
|
// Resuses one writer for all threads
|
|
writeInBadgerSingleThreadW := func(wb *badger.WriteBatch, KVList *bpb.KVList) {
|
|
if err := wb.WriteList(KVList); err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
}
|
|
|
|
dbOpts := badger.DefaultOptions("").
|
|
WithLogger(nil).
|
|
WithSyncWrites(false).
|
|
WithNumVersionsToKeep(math.MaxInt64).
|
|
WithCompression(options.None)
|
|
|
|
KVList := createKVList()
|
|
|
|
// Vanilla TxnWriter
|
|
b.Run("TxnWriter", func(b *testing.B) {
|
|
tmpIndexDir := b.TempDir()
|
|
|
|
dbOpts.Dir = tmpIndexDir
|
|
dbOpts.ValueDir = tmpIndexDir
|
|
var db, err2 = badger.OpenManaged(dbOpts)
|
|
require.NoError(b, err2)
|
|
defer db.Close()
|
|
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
w := NewTxnWriter(db)
|
|
for _, typ := range KVList.Kv {
|
|
k := typ.Key
|
|
v := typ.Value
|
|
require.NoError(b, w.SetAt(k, v, BitSchemaPosting, 1))
|
|
}
|
|
require.NoError(b, w.Flush())
|
|
|
|
}
|
|
})
|
|
// Single threaded BatchWriter
|
|
b.Run("WriteBatch1", func(b *testing.B) {
|
|
tmpIndexDir := b.TempDir()
|
|
|
|
dbOpts.Dir = tmpIndexDir
|
|
dbOpts.ValueDir = tmpIndexDir
|
|
|
|
var db, err2 = badger.OpenManaged(dbOpts)
|
|
require.NoError(b, err2)
|
|
defer db.Close()
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
wb := db.NewManagedWriteBatch()
|
|
if err := wb.WriteList(&KVList); err != nil {
|
|
panic(err)
|
|
}
|
|
require.NoError(b, wb.Flush())
|
|
}
|
|
})
|
|
// Multi threaded Batchwriter with thread contention in WriteBatch
|
|
b.Run("WriteBatchMultThreadDiffWB", func(b *testing.B) {
|
|
tmpIndexDir := b.TempDir()
|
|
|
|
dbOpts.Dir = tmpIndexDir
|
|
dbOpts.ValueDir = tmpIndexDir
|
|
|
|
var db, err2 = badger.OpenManaged(dbOpts)
|
|
require.NoError(b, err2)
|
|
defer db.Close()
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
var wg sync.WaitGroup
|
|
wg.Add(5)
|
|
|
|
go writeInBadgerMThreadsB(db, &bpb.KVList{Kv: KVList.Kv[:1000000]}, &wg)
|
|
go writeInBadgerMThreadsB(db, &bpb.KVList{Kv: KVList.Kv[1000001:2000000]}, &wg)
|
|
go writeInBadgerMThreadsB(db, &bpb.KVList{Kv: KVList.Kv[2000001:3000000]}, &wg)
|
|
go writeInBadgerMThreadsB(db, &bpb.KVList{Kv: KVList.Kv[3000001:4000000]}, &wg)
|
|
go writeInBadgerMThreadsB(db, &bpb.KVList{Kv: KVList.Kv[4000001:]}, &wg)
|
|
wg.Wait()
|
|
|
|
}
|
|
})
|
|
// Multi threaded Batchwriter with thread contention in SetEntry
|
|
b.Run("WriteBatchMultThreadSameWB", func(b *testing.B) {
|
|
tmpIndexDir := b.TempDir()
|
|
|
|
dbOpts.Dir = tmpIndexDir
|
|
dbOpts.ValueDir = tmpIndexDir
|
|
|
|
var db, err2 = badger.OpenManaged(dbOpts)
|
|
require.NoError(b, err2)
|
|
defer db.Close()
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
var wg sync.WaitGroup
|
|
wg.Add(5)
|
|
wb := db.NewManagedWriteBatch()
|
|
go writeInBadgerMThreadsW(wb, &bpb.KVList{Kv: KVList.Kv[:1000000]}, &wg)
|
|
go writeInBadgerMThreadsW(wb, &bpb.KVList{Kv: KVList.Kv[1000001:2000000]}, &wg)
|
|
go writeInBadgerMThreadsW(wb, &bpb.KVList{Kv: KVList.Kv[2000001:3000000]}, &wg)
|
|
go writeInBadgerMThreadsW(wb, &bpb.KVList{Kv: KVList.Kv[3000001:4000000]}, &wg)
|
|
go writeInBadgerMThreadsW(wb, &bpb.KVList{Kv: KVList.Kv[4000001:]}, &wg)
|
|
|
|
wg.Wait()
|
|
require.NoError(b, wb.Flush())
|
|
}
|
|
})
|
|
b.Run("WriteBatchSingleThreadDiffWB", func(b *testing.B) {
|
|
tmpIndexDir := b.TempDir()
|
|
|
|
dbOpts.Dir = tmpIndexDir
|
|
dbOpts.ValueDir = tmpIndexDir
|
|
|
|
var db, err2 = badger.OpenManaged(dbOpts)
|
|
require.NoError(b, err2)
|
|
defer db.Close()
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
writeInBadgerSingleThreadB(db, &bpb.KVList{Kv: KVList.Kv[:1000000]})
|
|
writeInBadgerSingleThreadB(db, &bpb.KVList{Kv: KVList.Kv[1000001:2000000]})
|
|
writeInBadgerSingleThreadB(db, &bpb.KVList{Kv: KVList.Kv[2000001:3000000]})
|
|
writeInBadgerSingleThreadB(db, &bpb.KVList{Kv: KVList.Kv[3000001:4000000]})
|
|
writeInBadgerSingleThreadB(db, &bpb.KVList{Kv: KVList.Kv[4000001:]})
|
|
}
|
|
})
|
|
b.Run("WriteBatchSingleThreadSameWB", func(b *testing.B) {
|
|
tmpIndexDir := b.TempDir()
|
|
|
|
dbOpts.Dir = tmpIndexDir
|
|
dbOpts.ValueDir = tmpIndexDir
|
|
|
|
var db, err2 = badger.OpenManaged(dbOpts)
|
|
require.NoError(b, err2)
|
|
defer db.Close()
|
|
|
|
b.ResetTimer()
|
|
|
|
for i := 0; i < b.N; i++ {
|
|
wb := db.NewManagedWriteBatch()
|
|
writeInBadgerSingleThreadW(wb, &bpb.KVList{Kv: KVList.Kv[:1000000]})
|
|
writeInBadgerSingleThreadW(wb, &bpb.KVList{Kv: KVList.Kv[1000001:2000000]})
|
|
writeInBadgerSingleThreadW(wb, &bpb.KVList{Kv: KVList.Kv[2000001:3000000]})
|
|
writeInBadgerSingleThreadW(wb, &bpb.KVList{Kv: KVList.Kv[3000001:4000000]})
|
|
writeInBadgerSingleThreadW(wb, &bpb.KVList{Kv: KVList.Kv[4000001:]})
|
|
require.NoError(b, wb.Flush())
|
|
}
|
|
})
|
|
}
|