diff --git a/cmd/rpc/main.go b/cmd/rpc/main.go index 7bced3f..e8d40fa 100644 --- a/cmd/rpc/main.go +++ b/cmd/rpc/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "log" "net" @@ -11,7 +12,8 @@ import ( ) func main() { - srv := crpc.NewServer() + srvDoneChannel := make(chan struct{}) + srv := crpc.NewServer(srvDoneChannel) rpc.Register(srv) rpc.HandleHTTP() listener, err := net.Listen("tcp", ":0") @@ -21,5 +23,20 @@ func main() { // tell the parent process what port we are listening on. port := listener.Addr().(*net.TCPAddr).Port fmt.Println(port) - fmt.Println(http.Serve(listener, nil)) + + // Start an HTTP server on the listener. It will send requests to the default HttpMux, which we've + // already told rpc to handle. + httpServer := &http.Server{} + + go httpServer.Serve(listener) + + // Wait for `Server.Close` to be called, then shut down the HTTP server. + <- srvDoneChannel + fmt.Println("Server starting clean shutdown...") + err = httpServer.Shutdown(context.Background()) + if err != nil { + log.Fatal("HTTP server Shutdown error: ", err) + } + + fmt.Println("Server shutdown complete") } diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index e2ed338..5830aa5 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -3,7 +3,7 @@ package rpc import ( "bufio" "fmt" - "log" + "io" "net/rpc" "os" "os/exec" @@ -53,73 +53,28 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio if _, err := os.Stat(r.binaryPath); err != nil { ct.Fatalf(t, "%s: RPC binary at %s does not exist or cannot be executed/read: %s", contextID, r.binaryPath, err) } + t.Logf("RPC (%s): starting RPC binary at %s", contextID, r.binaryPath) rpcCmd := exec.Command(r.binaryPath) stdout, err := rpcCmd.StdoutPipe() if err != nil { - ct.Fatalf(t, "%s: cannot pipe stdout of rpc binary: %s", contextID, err) + ct.Fatalf(t, "RPC (%s): cannot pipe stdout of rpc binary: %s", contextID, err) } rpcCmd.Stderr = rpcCmd.Stdout if err := rpcCmd.Start(); err != nil { // this calls NewRPCServer() effectively - ct.Fatalf(t, "%s: cannot start RPC binary %s: %s", contextID, r.binaryPath, err) + ct.Fatalf(t, "RPC (%s): cannot start RPC binary %s: %s", contextID, r.binaryPath, err) } // wait until we get a high-numbered port - portCh := make(chan struct { - port int - err error - }) - go func() { - rd := bufio.NewReader(stdout) - defer close(portCh) - defer func() { - // log stdout from the RPC server - go func() { - for { - str, err := rd.ReadString('\n') - if err != nil { - log.Print("RPC ERROR: " + err.Error()) - break - } - log.Printf(" RPC (%s): %s", contextID, str) - } - }() - // we need to .Wait to ensure we clean up resources when the RPC server dies. - rpcCmd.Wait() - }() + portCh := make(chan portChannelPayload) + logsFlushed := make(chan struct{}) + go readPortNumberAndLogOutput(t, contextID, portCh, rpcCmd, stdout, logsFlushed) - var port int - for { - str, err := rd.ReadString('\n') - if port == 0 { // we need a port - if err != nil { - portCh <- struct { - port int - err error - }{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} - return - } - port, err = strconv.Atoi(strings.TrimSpace(str)) - if err != nil { - log.Printf(" RPC (%s) stdout line isn't a port: %s", contextID, str) - continue - } - portCh <- struct { - port int - err error - }{ - port: port, - err: nil, - } - break - } - } - }() select { case p := <-portCh: rpcAddr := fmt.Sprintf("127.0.0.1:%d", p.port) var void int client, err := rpc.DialHTTP("tcp", rpcAddr) if err != nil { - t.Fatalf("RPC MustCreateClient DialHTTP: %s", err) + ct.Fatalf(t, "RPC (%s): MustCreateClient DialHTTP failed: %s", contextID, err) } err = client.Call("Server.MustCreateClient", ClientCreationOpts{ @@ -128,24 +83,82 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio Lang: r.clientType, }, &void) if err != nil { - ct.Fatalf(t, "%s: failed to create RPC client: %s", contextID, err) + ct.Fatalf(t, "RPC (%s): failed to create RPC client: %s", contextID, err) } return &RPCClient{ client: client, lang: r.clientType, rpcCmd: rpcCmd, + logsFlushed: logsFlushed, } case <-time.After(time.Second): - ct.Fatalf(t, "%s: timed out waiting for port number to be echoed to stdout. Did the RPC binary run, and is it actually the RPC binary? Path: %s", contextID, r.binaryPath) + ct.Fatalf(t, "RPC (%s): timed out waiting for port number to be echoed to stdout. Did the RPC binary run, and is it actually the RPC binary? Path: %s", contextID, r.binaryPath) } panic("unreachable") } +// portChannelPayload is the payload of the channel that we use to communicate the port number from the RPC server to the RPC client. +type portChannelPayload struct { + port int + err error +} + +// readPortNumberAndLogOutput waits for the RPC server to write the port number to stdout, and writes the port number to the `portCh` channel. +// Any other output is logged. +// +// Once the RPC server closes and all the logs are flushed, `logsFlushed` is closed. +func readPortNumberAndLogOutput(t ct.TestLike, contextID string, portCh chan portChannelPayload, rpcCmd *exec.Cmd, stdout io.ReadCloser, logsFlushed chan struct{}) { + rd := bufio.NewReader(stdout) + defer close(portCh) + + // XXX: why is this written as a `defer`, rather than simply happening after the `for` loop, given nobody is + // waiting for `readPortNumberAndLogOutput` to complete? + defer func() { + // log stdout from the RPC server + go func() { + defer close(logsFlushed) + for { + str, err := rd.ReadString('\n') + if err != nil { + t.Logf("RPC (%s): server closed: %s", contextID, err) + break + } + t.Logf("RPC (%s): server output: %s", contextID, str) + } + }() + // we need to .Wait to ensure we clean up resources when the RPC server dies. + rpcCmd.Wait() + }() + + var port int + for { + str, err := rd.ReadString('\n') + if port == 0 { // we need a port. XXX: why would it not be 0, given we `break` having successfully read the port? + if err != nil { + portCh <- portChannelPayload{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} + return + } + port, err = strconv.Atoi(strings.TrimSpace(str)) + if err != nil { + // Not a port number. Log it like a normal line. + t.Logf("RPC (%s): server output: %s", contextID, str) + continue + } + t.Logf("RPC (%s): server started on port %d", contextID, port) + portCh <- portChannelPayload{port: port, err: nil} + break + } + } +} + // RPCClient implements api.Client by making RPC calls to an RPC server, which actually has a concrete api.Client type RPCClient struct { client *rpc.Client lang api.ClientTypeLang rpcCmd *exec.Cmd + + // logsFlushed is a channel that is closed when the log copier goroutine has completed. + logsFlushed chan struct{} } func (c *RPCClient) ForceClose(t ct.TestLike) { @@ -163,12 +176,22 @@ func (c *RPCClient) ForceClose(t ct.TestLike) { func (c *RPCClient) Close(t ct.TestLike) { t.Helper() var void int - fmt.Println("RPCClient.Close") - err := c.client.Call("Server.Close", t.Name(), &void) - if err != nil { - t.Fatalf("RPCClient.Close: %s", err) + t.Logf("RPCClient.Close") + if err := c.client.Call("Server.Close", t.Name(), &void); err != nil { + // XXX this fails with "unexpected EOF" sometimes, and I don't understand why. + // The server won't actually stop until all clients have disconnected, so the RPC call should + // complete cleanly. + t.Logf("RPCClient.Close: Server.Close RPC call returned error: %s", err) + } + t.Logf("RPCClient.Close: disconnecting RPC client") + if err := c.client.Close(); err != nil { + t.Logf("RPCClient.Close: error disconnecting RPC client: %s", err) } - c.client.Close() + + // Wait for the goroutine that copies stdout to the logs to complete + t.Logf("RPCClient.Close: waiting for server to shut down") + <- c.logsFlushed + t.Logf("RPCClient.Close: done") } func (c *RPCClient) GetNotification(t ct.TestLike, roomID, eventID string) (*api.Notification, error) { diff --git a/internal/deploy/rpc/server.go b/internal/deploy/rpc/server.go index afe6bc3..59a7eaa 100644 --- a/internal/deploy/rpc/server.go +++ b/internal/deploy/rpc/server.go @@ -28,14 +28,21 @@ type Server struct { waitersMu *sync.Mutex lastCmdRecv time.Time lastCmdRecvMu *sync.Mutex + + // done is a chanel that is closed once Close() has been called. + done chan struct{} } -func NewServer() *Server { +// NewServer creates a new Server instance. +// +// doneChannel is a channel that will be closed once Close() has been called. +func NewServer(doneChannel chan struct{}) *Server { srv := &Server{ waiters: make(map[int]*RPCServerWaiter), waitersMu: &sync.Mutex{}, lastCmdRecv: time.Now(), lastCmdRecvMu: &sync.Mutex{}, + done: doneChannel, } go srv.checkKeepAlive() return srv @@ -91,6 +98,8 @@ func (s *Server) Close(testName string, void *int) error { s.activeClient.Close(&api.MockT{TestName: testName}) // write logs s.bindings.PostTestRun(s.contextID) + // Flag to main that we should shut down the HTTP server, once the client disconnects. + close(s.done) return nil } diff --git a/internal/tests/client_test.go b/internal/tests/client_test.go index bf4ed13..5ea847b 100644 --- a/internal/tests/client_test.go +++ b/internal/tests/client_test.go @@ -219,15 +219,20 @@ func TestSendingEvents(t *testing.T) { // run a subtest for each client factory func ForEachClient(t *testing.T, name string, deployment *deploy.ComplementCryptoDeployment, fn func(t *testing.T, client api.TestClient, csapi *client.CSAPI)) { - for _, createClient := range clientFactories { + testWrapper := func (createClient func (t *testing.T, cfg api.ClientCreationOpts) api.TestClient) { csapiAlice := deployment.Register(t, "hs1", helpers.RegistrationOpts{ LocalpartSuffix: "client", Password: "complement-crypto-password", }) opts := api.NewClientCreationOpts(csapiAlice) client := createClient(t, opts) + defer client.Close(t) t.Run(name+" "+string(client.Type()), func(t *testing.T) { fn(t, client, csapiAlice) }) } + + for _, createClient := range clientFactories { + testWrapper(createClient) + } }