// Copyright 2018 Google Inc.
//
// 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 pipe

import (
	"math/rand"
	"reflect"
	"runtime"
	"sync"
	"testing"
)

func TestSimpleReadWrite(t *testing.T) {
	// Check that a simple write can be properly read from the rx side.
	tr := rand.New(rand.NewSource(99))
	rr := rand.New(rand.NewSource(99))

	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	wb := tx.Push(10)
	if wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	for i := range wb {
		wb[i] = byte(tr.Intn(256))
	}
	tx.Flush()

	var rx Rx
	rx.Init(b)
	rb := rx.Pull()
	if len(rb) != 10 {
		t.Fatalf("Bad buffer size returned: got %v, want %v", len(rb), 10)
	}

	for i := range rb {
		if v := byte(rr.Intn(256)); v != rb[i] {
			t.Fatalf("Bad read buffer at index %v: got %v, want %v", i, rb[i], v)
		}
	}
	rx.Flush()
}

func TestEmptyRead(t *testing.T) {
	// Check that pulling from an empty pipe fails.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on empty pipe")
	}
}

func TestTooLargeWrite(t *testing.T) {
	// Check that writes that are too large are properly rejected.
	b := make([]byte, 96)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(96); wb != nil {
		t.Fatalf("Write of 96 bytes succeeded on 96-byte pipe")
	}

	if wb := tx.Push(88); wb != nil {
		t.Fatalf("Write of 88 bytes succeeded on 96-byte pipe")
	}

	if wb := tx.Push(80); wb == nil {
		t.Fatalf("Write of 80 bytes failed on 96-byte pipe")
	}
}

func TestFullWrite(t *testing.T) {
	// Check that writes fail when the pipe is full.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(80); wb == nil {
		t.Fatalf("Write of 80 bytes failed on 96-byte pipe")
	}

	if wb := tx.Push(1); wb != nil {
		t.Fatalf("Write succeeded on full pipe")
	}
}

func TestFullAndFlushedWrite(t *testing.T) {
	// Check that writes fail when the pipe is full and has already been
	// flushed.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(80); wb == nil {
		t.Fatalf("Write of 80 bytes failed on 96-byte pipe")
	}

	tx.Flush()

	if wb := tx.Push(1); wb != nil {
		t.Fatalf("Write succeeded on full pipe")
	}
}

func TestTxFlushTwice(t *testing.T) {
	// Checks that a second consecutive tx flush is a no-op.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	// Make copy of original tx queue, flush it, then check that it didn't
	// change.
	orig := tx
	tx.Flush()

	if !reflect.DeepEqual(orig, tx) {
		t.Fatalf("Flush mutated tx pipe: got %v, want %v", tx, orig)
	}
}

func TestRxFlushTwice(t *testing.T) {
	// Checks that a second consecutive rx flush is a no-op.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
	rx.Flush()

	// Make copy of original rx queue, flush it, then check that it didn't
	// change.
	orig := rx
	rx.Flush()

	if !reflect.DeepEqual(orig, rx) {
		t.Fatalf("Flush mutated rx pipe: got %v, want %v", rx, orig)
	}
}

func TestWrapInMiddleOfTransaction(t *testing.T) {
	// Check that writes are not flushed when we need to wrap the buffer
	// around.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
	rx.Flush()

	// At this point the ring buffer is empty, but the write is at offset
	// 64 (50 + sizeOfSlotHeader + padding-for-8-byte-alignment).
	if wb := tx.Push(10); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on non-full pipe")
	}

	// We haven't flushed yet, so pull must return nil.
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on non-flushed pipe")
	}

	tx.Flush()

	// The two buffers must be available now.
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}

	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
}

func TestWriteAbort(t *testing.T) {
	// Check that a read fails on a pipe that has had data pushed to it but
	// has aborted the push.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(10); wb == nil {
		t.Fatalf("Write failed on empty pipe")
	}

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on empty pipe")
	}

	tx.Abort()
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on empty pipe")
	}
}

func TestWrappedWriteAbort(t *testing.T) {
	// Check that writes are properly aborted even if the writes wrap
	// around.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
	rx.Flush()

	// At this point the ring buffer is empty, but the write is at offset
	// 64 (50 + sizeOfSlotHeader + padding-for-8-byte-alignment).
	if wb := tx.Push(10); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on non-full pipe")
	}

	// We haven't flushed yet, so pull must return nil.
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on non-flushed pipe")
	}

	tx.Abort()

	// The pushes were aborted, so no data should be readable.
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on non-flushed pipe")
	}

	// Try the same transactions again, but flush this time.
	if wb := tx.Push(10); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on non-full pipe")
	}

	tx.Flush()

	// The two buffers must be available now.
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}

	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
}

func TestEmptyReadOnNonFlushedWrite(t *testing.T) {
	// Check that a read fails on a pipe that has had data pushed to it
	// but not yet flushed.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(10); wb == nil {
		t.Fatalf("Write failed on empty pipe")
	}

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on empty pipe")
	}

	tx.Flush()
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull on failed on non-empty pipe")
	}
}

func TestPullAfterPullingEntirePipe(t *testing.T) {
	// Check that Pull fails when the pipe is full, but all of it has
	// already been pulled but not yet flushed.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
	rx.Flush()

	// At this point the ring buffer is empty, but the write is at offset
	// 64 (50 + sizeOfSlotHeader + padding-for-8-byte-alignment). Write 3
	// buffers that will fill the pipe.
	if wb := tx.Push(10); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}

	if wb := tx.Push(20); wb == nil {
		t.Fatalf("Push failed on non-full pipe")
	}

	if wb := tx.Push(24); wb == nil {
		t.Fatalf("Push failed on non-full pipe")
	}

	tx.Flush()

	// The three buffers must be available now.
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}

	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}

	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}

	// Fourth pull must fail.
	if rb := rx.Pull(); rb != nil {
		t.Fatalf("Pull succeeded on empty pipe")
	}
}

func TestNoRoomToWrapOnPush(t *testing.T) {
	// Check that Push fails when it tries to allocate room to add a wrap
	// message.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	var rx Rx
	rx.Init(b)
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
	rx.Flush()

	// At this point the ring buffer is empty, but the write is at offset
	// 64 (50 + sizeOfSlotHeader + padding-for-8-byte-alignment). Write 20,
	// which won't fit (64+20+8+padding = 96, which wouldn't leave room for
	// the padding), so it wraps around.
	if wb := tx.Push(20); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}

	tx.Flush()

	// Buffer offset is at 28. Try to write 70, which would require a wrap
	// slot which cannot be created now.
	if wb := tx.Push(70); wb != nil {
		t.Fatalf("Push succeeded on pipe with no room for wrap message")
	}
}

func TestRxImplicitFlushOfWrapMessage(t *testing.T) {
	// Check if the first read is that of a wrapping message, that it gets
	// immediately flushed.
	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	if wb := tx.Push(50); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
	tx.Flush()

	// This will cause a wrapping message to written.
	if wb := tx.Push(60); wb != nil {
		t.Fatalf("Push succeeded when there is no room in pipe")
	}

	var rx Rx
	rx.Init(b)

	// Read the first message.
	if rb := rx.Pull(); rb == nil {
		t.Fatalf("Pull failed on non-empty pipe")
	}
	rx.Flush()

	// This should fail because of the wrapping message is taking up space.
	if wb := tx.Push(60); wb != nil {
		t.Fatalf("Push succeeded when there is no room in pipe")
	}

	// Try to read the next one. This should consume the wrapping message.
	rx.Pull()

	// This must now succeed.
	if wb := tx.Push(60); wb == nil {
		t.Fatalf("Push failed on empty pipe")
	}
}

func TestConcurrentReaderWriter(t *testing.T) {
	// Push a million buffers of random sizes and random contents. Check
	// that buffers read match what was written.
	tr := rand.New(rand.NewSource(99))
	rr := rand.New(rand.NewSource(99))

	b := make([]byte, 100)
	var tx Tx
	tx.Init(b)

	var rx Rx
	rx.Init(b)

	const count = 1000000
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		runtime.Gosched()
		for i := 0; i < count; i++ {
			n := 1 + tr.Intn(80)
			wb := tx.Push(uint64(n))
			for wb == nil {
				wb = tx.Push(uint64(n))
			}

			for j := range wb {
				wb[j] = byte(tr.Intn(256))
			}

			tx.Flush()
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		runtime.Gosched()
		for i := 0; i < count; i++ {
			n := 1 + rr.Intn(80)
			rb := rx.Pull()
			for rb == nil {
				rb = rx.Pull()
			}

			if n != len(rb) {
				t.Fatalf("Bad %v-th buffer length: got %v, want %v", i, len(rb), n)
			}

			for j := range rb {
				if v := byte(rr.Intn(256)); v != rb[j] {
					t.Fatalf("Bad %v-th read buffer at index %v: got %v, want %v", i, j, rb[j], v)
				}
			}

			rx.Flush()
		}
	}()

	wg.Wait()
}