/*
Copyright 2020 Google LLC

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.
*/

package spanner

import (
	"context"
	"fmt"
	"math/rand"
	"reflect"
	"sync"
	"testing"
	"time"

	. "cloud.google.com/go/spanner/internal/testutil"
	"google.golang.org/api/iterator"
	sppb "google.golang.org/genproto/googleapis/spanner/v1"
)

const networkLatencyTime = 10 * time.Millisecond
const batchCreateSessionsMinTime = 10 * time.Millisecond
const batchCreateSessionsRndTime = 10 * time.Millisecond
const beginTransactionMinTime = 1 * time.Millisecond
const beginTransactionRndTime = 1 * time.Millisecond
const commitTransactionMinTime = 5 * time.Millisecond
const commitTransactionRndTime = 5 * time.Millisecond
const executeStreamingSqlMinTime = 10 * time.Millisecond
const executeStreamingSqlRndTime = 10 * time.Millisecond
const executeSqlMinTime = 10 * time.Millisecond
const executeSqlRndTime = 10 * time.Millisecond

const holdSessionTime = 100
const rndWaitTimeBetweenRequests = 10

var mu sync.Mutex
var rnd = rand.New(rand.NewSource(time.Now().UnixNano()))

func createBenchmarkServer(incStep uint64) (server *MockedSpannerInMemTestServer, client *Client, teardown func()) {
	t := &testing.T{}
	server, client, teardown = setupMockedTestServerWithConfig(t, ClientConfig{
		SessionPoolConfig: SessionPoolConfig{
			MinOpened:     100,
			MaxOpened:     400,
			WriteSessions: 0.2,
			incStep:       incStep,
		},
	})
	server.TestSpanner.PutExecutionTime(MethodBatchCreateSession, SimulatedExecutionTime{
		MinimumExecutionTime: networkLatencyTime + batchCreateSessionsMinTime,
		RandomExecutionTime:  batchCreateSessionsRndTime,
	})
	server.TestSpanner.PutExecutionTime(MethodCreateSession, SimulatedExecutionTime{
		MinimumExecutionTime: networkLatencyTime + batchCreateSessionsMinTime,
		RandomExecutionTime:  batchCreateSessionsRndTime,
	})
	server.TestSpanner.PutExecutionTime(MethodExecuteStreamingSql, SimulatedExecutionTime{
		MinimumExecutionTime: networkLatencyTime + executeStreamingSqlMinTime,
		RandomExecutionTime:  executeStreamingSqlRndTime,
	})
	server.TestSpanner.PutExecutionTime(MethodBeginTransaction, SimulatedExecutionTime{
		MinimumExecutionTime: networkLatencyTime + beginTransactionMinTime,
		RandomExecutionTime:  beginTransactionRndTime,
	})
	server.TestSpanner.PutExecutionTime(MethodCommitTransaction, SimulatedExecutionTime{
		MinimumExecutionTime: networkLatencyTime + commitTransactionMinTime,
		RandomExecutionTime:  commitTransactionRndTime,
	})
	server.TestSpanner.PutExecutionTime(MethodExecuteSql, SimulatedExecutionTime{
		MinimumExecutionTime: networkLatencyTime + executeSqlMinTime,
		RandomExecutionTime:  executeSqlRndTime,
	})
	// Wait until the session pool has been initialized.
	waitFor(t, func() error {
		if uint64(client.idleSessions.idleList.Len()+client.idleSessions.idleWriteList.Len()) == client.idleSessions.MinOpened {
			return nil
		}
		return fmt.Errorf("not yet initialized")
	})
	return
}

func readWorker(client *Client, b *testing.B, jobs <-chan int, results chan<- int) {
	for range jobs {
		mu.Lock()
		d := time.Millisecond * time.Duration(rnd.Int63n(rndWaitTimeBetweenRequests))
		mu.Unlock()
		time.Sleep(d)
		iter := client.Single().Query(context.Background(), NewStatement(SelectSingerIDAlbumIDAlbumTitleFromAlbums))
		row := 0
		for {
			_, err := iter.Next()
			if err == iterator.Done {
				break
			}
			if err != nil {
				b.Fatal(err)
			}
			row++
			if row == 1 {
				mu.Lock()
				d := time.Millisecond * time.Duration(rnd.Int63n(holdSessionTime))
				mu.Unlock()
				time.Sleep(d)
			}
		}
		iter.Stop()
		results <- row
	}
}

func writeWorker(client *Client, b *testing.B, jobs <-chan int, results chan<- int64) {
	for range jobs {
		mu.Lock()
		d := time.Millisecond * time.Duration(rnd.Int63n(rndWaitTimeBetweenRequests))
		mu.Unlock()
		time.Sleep(d)
		var updateCount int64
		var err error
		if _, err = client.ReadWriteTransaction(context.Background(), func(ctx context.Context, transaction *ReadWriteTransaction) error {
			if updateCount, err = transaction.Update(ctx, NewStatement(UpdateBarSetFoo)); err != nil {
				return err
			}
			return nil
		}); err != nil {
			b.Fatal(err)
		}
		results <- updateCount
	}
}

func Benchmark_Client_BurstRead_IncStep01(b *testing.B) {
	benchmarkClientBurstRead(b, 1)
}

func Benchmark_Client_BurstRead_IncStep10(b *testing.B) {
	benchmarkClientBurstRead(b, 10)
}

func Benchmark_Client_BurstRead_IncStep20(b *testing.B) {
	benchmarkClientBurstRead(b, 20)
}

func Benchmark_Client_BurstRead_IncStep25(b *testing.B) {
	benchmarkClientBurstRead(b, 25)
}

func Benchmark_Client_BurstRead_IncStep30(b *testing.B) {
	benchmarkClientBurstRead(b, 30)
}

func Benchmark_Client_BurstRead_IncStep40(b *testing.B) {
	benchmarkClientBurstRead(b, 40)
}

func Benchmark_Client_BurstRead_IncStep50(b *testing.B) {
	benchmarkClientBurstRead(b, 50)
}

func Benchmark_Client_BurstRead_IncStep100(b *testing.B) {
	benchmarkClientBurstRead(b, 100)
}

func benchmarkClientBurstRead(b *testing.B, incStep uint64) {
	for n := 0; n < b.N; n++ {
		server, client, teardown := createBenchmarkServer(incStep)
		sp := client.idleSessions
		if uint64(sp.idleList.Len()+sp.idleWriteList.Len()) != sp.MinOpened {
			b.Fatalf("session count mismatch\nGot: %d\nWant: %d", sp.idleList.Len()+sp.idleWriteList.Len(), sp.MinOpened)
		}

		totalQueries := int(sp.MaxOpened * 8)
		jobs := make(chan int, totalQueries)
		results := make(chan int, totalQueries)
		parallel := int(sp.MaxOpened * 2)

		for w := 0; w < parallel; w++ {
			go readWorker(client, b, jobs, results)
		}
		for j := 0; j < totalQueries; j++ {
			jobs <- j
		}
		close(jobs)
		totalRows := 0
		for a := 0; a < totalQueries; a++ {
			totalRows = totalRows + <-results
		}
		reportBenchmark(b, sp, server)
		teardown()
	}
}

func Benchmark_Client_BurstWrite01(b *testing.B) {
	benchmarkClientBurstWrite(b, 1)
}

func Benchmark_Client_BurstWrite10(b *testing.B) {
	benchmarkClientBurstWrite(b, 10)
}

func Benchmark_Client_BurstWrite20(b *testing.B) {
	benchmarkClientBurstWrite(b, 20)
}

func Benchmark_Client_BurstWrite25(b *testing.B) {
	benchmarkClientBurstWrite(b, 25)
}

func Benchmark_Client_BurstWrite30(b *testing.B) {
	benchmarkClientBurstWrite(b, 30)
}

func Benchmark_Client_BurstWrite40(b *testing.B) {
	benchmarkClientBurstWrite(b, 40)
}

func Benchmark_Client_BurstWrite50(b *testing.B) {
	benchmarkClientBurstWrite(b, 50)
}

func Benchmark_Client_BurstWrite100(b *testing.B) {
	benchmarkClientBurstWrite(b, 100)
}

func benchmarkClientBurstWrite(b *testing.B, incStep uint64) {
	for n := 0; n < b.N; n++ {
		server, client, teardown := createBenchmarkServer(incStep)
		sp := client.idleSessions
		if uint64(sp.idleList.Len()+sp.idleWriteList.Len()) != sp.MinOpened {
			b.Fatalf("session count mismatch\nGot: %d\nWant: %d", sp.idleList.Len()+sp.idleWriteList.Len(), sp.MinOpened)
		}

		totalUpdates := int(sp.MaxOpened * 8)
		jobs := make(chan int, totalUpdates)
		results := make(chan int64, totalUpdates)
		parallel := int(sp.MaxOpened * 2)

		for w := 0; w < parallel; w++ {
			go writeWorker(client, b, jobs, results)
		}
		for j := 0; j < totalUpdates; j++ {
			jobs <- j
		}
		close(jobs)
		totalRows := int64(0)
		for a := 0; a < totalUpdates; a++ {
			totalRows = totalRows + <-results
		}
		reportBenchmark(b, sp, server)
		teardown()
	}
}

func Benchmark_Client_BurstReadAndWrite01(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 1)
}

func Benchmark_Client_BurstReadAndWrite10(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 10)
}

func Benchmark_Client_BurstReadAndWrite20(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 20)
}

func Benchmark_Client_BurstReadAndWrite25(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 25)
}

func Benchmark_Client_BurstReadAndWrite30(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 30)
}

func Benchmark_Client_BurstReadAndWrite40(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 40)
}

func Benchmark_Client_BurstReadAndWrite50(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 50)
}

func Benchmark_Client_BurstReadAndWrite100(b *testing.B) {
	benchmarkClientBurstReadAndWrite(b, 100)
}

func benchmarkClientBurstReadAndWrite(b *testing.B, incStep uint64) {
	for n := 0; n < b.N; n++ {
		server, client, teardown := createBenchmarkServer(incStep)
		sp := client.idleSessions
		if uint64(sp.idleList.Len()+sp.idleWriteList.Len()) != sp.MinOpened {
			b.Fatalf("session count mismatch\nGot: %d\nWant: %d", sp.idleList.Len()+sp.idleWriteList.Len(), sp.MinOpened)
		}

		totalUpdates := int(sp.MaxOpened * 4)
		writeJobs := make(chan int, totalUpdates)
		writeResults := make(chan int64, totalUpdates)
		parallelWrites := int(sp.MaxOpened)

		totalQueries := int(sp.MaxOpened * 4)
		readJobs := make(chan int, totalQueries)
		readResults := make(chan int, totalQueries)
		parallelReads := int(sp.MaxOpened)

		for w := 0; w < parallelWrites; w++ {
			go writeWorker(client, b, writeJobs, writeResults)
		}
		for j := 0; j < totalUpdates; j++ {
			writeJobs <- j
		}
		for w := 0; w < parallelReads; w++ {
			go readWorker(client, b, readJobs, readResults)
		}
		for j := 0; j < totalQueries; j++ {
			readJobs <- j
		}

		close(writeJobs)
		close(readJobs)

		totalUpdatedRows := int64(0)
		for a := 0; a < totalUpdates; a++ {
			totalUpdatedRows = totalUpdatedRows + <-writeResults
		}
		totalReadRows := 0
		for a := 0; a < totalQueries; a++ {
			totalReadRows = totalReadRows + <-readResults
		}
		reportBenchmark(b, sp, server)
		teardown()
	}
}

func Benchmark_Client_SteadyIncrease01(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 1)
}

func Benchmark_Client_SteadyIncrease10(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 10)
}

func Benchmark_Client_SteadyIncrease20(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 20)
}

func Benchmark_Client_SteadyIncrease25(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 25)
}

func Benchmark_Client_SteadyIncrease30(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 30)
}

func Benchmark_Client_SteadyIncrease40(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 40)
}

func Benchmark_Client_SteadyIncrease50(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 50)
}

func Benchmark_Client_SteadyIncrease100(b *testing.B) {
	benchmarkClientSteadyIncrease(b, 100)
}

func benchmarkClientSteadyIncrease(b *testing.B, incStep uint64) {
	for n := 0; n < b.N; n++ {
		server, client, teardown := createBenchmarkServer(incStep)
		sp := client.idleSessions
		if uint64(sp.idleList.Len()+sp.idleWriteList.Len()) != sp.MinOpened {
			b.Fatalf("session count mismatch\nGot: %d\nWant: %d", sp.idleList.Len()+sp.idleWriteList.Len(), sp.MinOpened)
		}

		transactions := make([]*ReadOnlyTransaction, sp.MaxOpened)
		for i := uint64(0); i < sp.MaxOpened; i++ {
			transactions[i] = client.ReadOnlyTransaction()
			transactions[i].Query(context.Background(), NewStatement(SelectSingerIDAlbumIDAlbumTitleFromAlbums))
		}
		for i := uint64(0); i < sp.MaxOpened; i++ {
			transactions[i].Close()
		}
		reportBenchmark(b, sp, server)
		teardown()
	}
}

func reportBenchmark(b *testing.B, sp *sessionPool, server *MockedSpannerInMemTestServer) {
	sp.mu.Lock()
	defer sp.mu.Unlock()
	requests := drainRequestsFromServer(server.TestSpanner)
	// TODO(loite): Use b.ReportMetric when Go1.13 is the minimum required.
	b.Logf("BatchCreateSessions: %d\t", countRequests(requests, reflect.TypeOf(&sppb.BatchCreateSessionsRequest{})))
	b.Logf("CreateSession: %d\t", countRequests(requests, reflect.TypeOf(&sppb.CreateSessionRequest{})))
	b.Logf("BeginTransaction: %d\t", countRequests(requests, reflect.TypeOf(&sppb.BeginTransactionRequest{})))
	b.Logf("Commit: %d\t", countRequests(requests, reflect.TypeOf(&sppb.CommitRequest{})))
	b.Logf("ReadSessions: %d\t", sp.idleList.Len())
	b.Logf("WriteSessions: %d\n", sp.idleWriteList.Len())
}

func countRequests(requests []interface{}, tp reflect.Type) (count int) {
	for _, req := range requests {
		if tp == reflect.TypeOf(req) {
			count++
		}
	}
	return count
}