115 lines
3.3 KiB
Go
115 lines
3.3 KiB
Go
package tentacle
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"testing"
|
|
)
|
|
|
|
func TestPythonPeerUsesSameContract(t *testing.T) {
|
|
python, err := exec.LookPath("python3")
|
|
if err != nil {
|
|
t.Skip("Python 3 not installed")
|
|
}
|
|
fixture, err := filepath.Abs("testdata/peer.py")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
session, err := manager.Start(context.Background(), Spec{ID: "python", Program: []string{python, "-I", "-u", fixture}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wait_signal(t, session.Ready())
|
|
if err := session.Activate(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
arguments := json.RawMessage(`{"text":"é猫😀é","enabled":false,"count":0}`)
|
|
pending, err := session.Invoke(context.Background(), "echo", arguments)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
result := wait_result(t, pending)
|
|
var value struct {
|
|
Text string
|
|
Enabled bool
|
|
Count int
|
|
}
|
|
if result.Err != nil || json.Unmarshal(result.Value, &value) != nil || value.Text != "é猫😀é" || value.Enabled || value.Count != 0 {
|
|
t.Fatalf("Python round trip: %+v, error %v", value, result.Err)
|
|
}
|
|
if result.Session != session.Identity() {
|
|
t.Fatal("wrong session")
|
|
}
|
|
}
|
|
|
|
func TestManyTentaclesHaveAggregateBoundAndIndependentProgress(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
sessions := make([]*Session, 0, 16)
|
|
// Keep one healthy peer alongside fifteen peers that never reply.
|
|
healthy := start_active(t, manager, "healthy", "echo")
|
|
for index := range 15 {
|
|
sessions = append(sessions, start_active(t, manager, fmt.Sprintf("blocked-%d", index), "timeout"))
|
|
}
|
|
for index := range 63 {
|
|
if _, err := sessions[index%len(sessions)].Invoke(context.Background(), "echo", nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
result, err := healthy.Invoke(context.Background(), "echo", json.RawMessage(`{"healthy":true}`))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if reply := wait_result(t, result); reply.Err != nil || string(reply.Value) != `{"healthy":true}` {
|
|
t.Fatalf("healthy peer stalled: %+v", reply)
|
|
}
|
|
if _, err := sessions[0].Invoke(context.Background(), "echo", nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := healthy.Invoke(context.Background(), "echo", nil); !errors.Is(err, ErrBusy) {
|
|
t.Fatal("missing aggregate backpressure", err)
|
|
}
|
|
if stats := manager.Stats(); stats.Pending != 64 || stats.Processes != 16 {
|
|
t.Fatalf("%+v", stats)
|
|
}
|
|
for _, session := range sessions {
|
|
session.Disable()
|
|
}
|
|
result, err = healthy.Invoke(context.Background(), "echo", nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if reply := wait_result(t, result); reply.Err != nil {
|
|
t.Fatal(reply.Err)
|
|
}
|
|
}
|
|
|
|
func BenchmarkCommandRoundTrip(b *testing.B) {
|
|
for _, count := range []int{1, 8, 16} {
|
|
b.Run(fmt.Sprintf("processes-%d", count), func(b *testing.B) {
|
|
manager := test_manager(b, DefaultLimits(), nil)
|
|
sessions := make([]*Session, count)
|
|
for index := range count {
|
|
sessions[index] = start_active(b, manager, fmt.Sprintf("peer-%d", index), "echo")
|
|
}
|
|
arguments := json.RawMessage(`{"text":"sample"}`)
|
|
b.ReportAllocs()
|
|
b.ResetTimer()
|
|
for index := range b.N {
|
|
pending, err := sessions[index%count].Invoke(context.Background(), "echo", arguments)
|
|
if err != nil {
|
|
b.Fatal(err)
|
|
}
|
|
if result := <-pending; result.Err != nil {
|
|
b.Fatal(result.Err)
|
|
}
|
|
}
|
|
b.StopTimer()
|
|
})
|
|
}
|
|
}
|