.

go/neo/neotools/storage.go

@@ -26,6 +26,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"runtime"
 	"strings"
 
 	"lab.nexedi.com/kirr/neo/go/neo/server"
@@ -47,10 +48,6 @@ XXX currently storage is read-only.
 `)
 }
 
-// TODO set GOMAXPROCS *= N (a lot of file IO) + link
-// https://groups.google.com/forum/#!msg/golang-nuts/jPb_h3TvlKE/rQwbg-etCAAJ
-// https://github.com/golang/go/issues/6817
-
 func storageMain(argv []string) {
 	flags := flag.NewFlagSet("", flag.ExitOnError)
 	flags.Usage = func() { storageUsage(os.Stderr); flags.PrintDefaults() }	// XXX prettify
@@ -79,6 +76,15 @@ func storageMain(argv []string) {
 		zt.Exit(2)
 	}
 
+	// adjust GOMAXPROCS *= N (a lot of file IO) because file IO really consumes OS threads; details:
+	// https://groups.google.com/forum/#!msg/golang-nuts/jPb_h3TvlKE/rQwbg-etCAAJ
+	// https://github.com/golang/go/issues/6817
+	//
+	// XXX check how varying this affects performance
+	maxprocs := runtime.GOMAXPROCS(0)
+	runtime.GOMAXPROCS(maxprocs*8)		// XXX *8 is enough?
+
+
 	// XXX hack to use existing zodb storage for data
 	zstor, err := fs1.Open(context.Background(), argv[0])
 	if err != nil {

go/neo/t/t.sh

@@ -242,7 +242,7 @@ bench1() {
 		return
 	fi
 	go run zsha1.go --log_dir=$log $url
-	# TODO zsha1.go -prefetch=1
+	go run zsha1.go --log_dir=$log -useprefetch $url
 }
 
 echo -e "\n*** FileStorage"
@@ -275,50 +275,14 @@ done
 # xmysql -e "SHUTDOWN"
 # wait
 
-echo -e "\n*** NEO/go"
-NEOgo
-for i in $N; do
-	bench1 neo://$cluster@$Mbind
-done
-xneoctl set cluster stopping
-wait
+# echo -e "\n*** NEO/go"
+# NEOgo
+# for i in $N; do
+# 	bench1 neo://$cluster@$Mbind
+# done
+# xneoctl set cluster stopping
+# wait
 
 # all ok
 trap - EXIT
 exit
-
-# ----------------------------------------
-
-#Zpy $fs1/data.fs
-#sleep 1
-#time demo-zbigarray read zeo://$Zbind
-
-#Mpy --autostart=1
-## sleep 0.2
-#Sgo $fs1/data.fs
-
-
-
-NEOpylite
-#NEOpysql
-#time demo-zbigarray read neo://$cluster@$Mbind
-for i in `seq 2`; do
-	./zsha1.py neo://$cluster@$Mbind
-	go run zsha1.go neo://$cluster@$Mbind
-done
-xneoctl set cluster stopping
-#xmysql -e "SHUTDOWN"
-
-wait
-exit
-
-
-# --------
-
-# spawn Mpy + Sgo
-Mpy
-sleep 0.2 # XXX temp for debug: so master could start listening and first S connect try is not error
-Sgo ../../zodb/storage/fs1/testdata/1.fs
-Apy
-
-wait

go/neo/t/zsha1.go

@@ -17,23 +17,25 @@ import (
 
 	"lab.nexedi.com/kirr/neo/go/zodb"
 	_ "lab.nexedi.com/kirr/neo/go/zodb/wks"
+	"lab.nexedi.com/kirr/neo/go/zodb/storage"
 
 	"github.com/pkg/profile"
 )
 
 func main() {
 	defer log.Flush()
+	useprefetch := flag.Bool("useprefetch", false, "prefetch loaded objects")
 	flag.Parse()
 	url := flag.Args()[0]	// XXX dirty
 
 	ctx := context.Background()
-	err := zsha1(ctx, url)
+	err := zsha1(ctx, url, *useprefetch)
 	if err != nil {
 		log.Fatal(ctx, err)
 	}
 }
 
-func zsha1(ctx context.Context, url string) (err error) {
+func zsha1(ctx context.Context, url string, useprefetch bool) (err error) {
 	defer task.Running(&ctx, "zsha1")(&err)
 
 	stor, err := zodb.OpenStorageURL(ctx, url)
@@ -45,6 +47,52 @@ func zsha1(ctx context.Context, url string) (err error) {
 		err = xerr.First(err, err2)
 	}()
 
+	// XXX always open storage with cache by zodb.OpenStorageURL
+	var cache *storage.Cache
+	if useprefetch {
+		cache = storage.NewCache(stor, 16*1024*1024)
+	}
+
+	prefetch := func(ctx context.Context, xid zodb.Xid) {
+		if cache != nil {
+			//fmt.Printf("prefetch %v\n", xid)
+			cache.Prefetch(ctx, xid)
+		}
+	}
+
+	load := func(ctx context.Context, xid zodb.Xid) ([]byte, zodb.Tid, error) {
+		if cache != nil {
+			return cache.Load(ctx, xid)
+		} else {
+			return stor.Load(ctx, xid)
+		}
+	}
+
+	// prefetchBlk prefetches block of 512 objects starting from xid
+	var tprevLoadBlkStart time.Time
+	prefetchBlk := func(ctx context.Context, xid zodb.Xid) {
+		if cache == nil {
+			return
+		}
+
+		t1 := time.Now()
+		for i := 0; i < 512; i++ {
+			prefetch(ctx, xid)
+			xid.Oid++
+		}
+		t2 := time.Now()
+		δt := t2.Sub(t1)
+		fmt.Printf("tprefetch: %s", δt)
+		if !tprevLoadBlkStart.IsZero() {
+			fmt.Printf("\ttprevload: %s", t1.Sub(tprevLoadBlkStart))
+		}
+		fmt.Printf("\n")
+
+		tprevLoadBlkStart = t2
+	}
+
+
+
 	lastTid, err := stor.LastTid(ctx)
 	if err != nil {
 		return err
@@ -63,7 +111,10 @@ func zsha1(ctx context.Context, url string) (err error) {
 loop:
 	for {
 		xid := zodb.Xid{Oid: oid, XTid: zodb.XTid{Tid: before, TidBefore: true}}
-		data, _, err := stor.Load(ctx, xid)
+		if xid.Oid % 512 == 0 {
+			prefetchBlk(ctx, xid)
+		}
+		data, _, err := load(ctx, xid)
 		switch err.(type) {
 		case nil:
 			// ok
@@ -85,8 +136,12 @@ loop:
 	tend := time.Now()
 	δt := tend.Sub(tstart)
 
-	fmt.Printf("%x   ; oid=0..%d  nread=%d  t=%s (%s / object)  x=zsha1.go\n",
-		m.Sum(nil), oid-1, nread, δt, δt / time.Duration(oid))	// XXX /oid cast ?
+	x := "zsha1.go"
+	if useprefetch {
+		x += " +prefetch"
+	}
+	fmt.Printf("%x   ; oid=0..%d  nread=%d  t=%s (%s / object)  x=%s\n",
+		m.Sum(nil), oid-1, nread, δt, δt / time.Duration(oid), x) // XXX /oid cast ?
 
 	return nil
 }