imterah/hermes
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases 1 Packages Wiki Activity Actions

feature: Refactors backend runtime's communication mechanism to be more stable.

Browse source
This commit is contained in:
Tera << 8 2025-01-06 01:24:11 -05:00
parent 93f2f9cbee
commit 1e1a330a4b
Signed by: imterah
GPG key ID: 8FA7DD57BA6CEA37
10 changed files with 254 additions and 157 deletions
Download patch file Download diff file Expand all files Collapse all files

15
backend/api/backendruntime/core.go Normal file
View file

@ -0,0 +1,15 @@
package backendruntime
import "os"
var (
AvailableBackends []*Backend
RunningBackends map[uint]*Runtime
TempDir string
isDevelopmentMode bool
)
func init() {
RunningBackends = make(map[uint]*Runtime)
isDevelopmentMode = os.Getenv("HERMES_DEVELOPMENT_MODE") != ""
}

261
backend/api/backendruntime/runtime.go
View file

@ -7,6 +7,7 @@ import (
"os"
"os/exec"
"strings"
"sync"
"time"
"git.terah.dev/imterah/hermes/backend/backendlauncher"
@ -14,16 +15,6 @@ import (
"github.com/charmbracelet/log"
)
var (
AvailableBackends []*Backend
RunningBackends map[uint]*Runtime
TempDir string
)
func init() {
RunningBackends = make(map[uint]*Runtime)
}
func handleCommand(commandType string, command interface{}, sock net.Conn, rtcChan chan interface{}) error {
bytes, err := commonbackend.Marshal(commandType, command)
@ -101,7 +92,9 @@ func (runtime *Runtime) goRoutineHandler() error {
} else {
log.Debug("We have restarted. Running the restart callback...")
runtime.OnCrashCallback(sock)
log.Debug("Finished running the restart callback.")
log.Debug("Finished running the restart callback. Clearing caches...")
runtime.cleanUpPendingCommandProcessingJobs()
runtime.messageBufferLock = sync.Mutex{}
}
} else {
log.Debug("We have not restarted.")
@ -112,8 +105,6 @@ func (runtime *Runtime) goRoutineHandler() error {
log.Debugf("Setting up Hermes keepalive Goroutine")
hasFailedBackendRunningCheckAlready := false
time.Sleep(time.Second * 1)
for {
if !runtime.isRuntimeRunning {
return
@ -125,21 +116,23 @@ func (runtime *Runtime) goRoutineHandler() error {
// To be safe here, we have to use the proper (yet annoying) facilities to prevent cross-talk, since we're in
// a goroutine, and can't talk directly. This actually has benefits, as the OuterLoop should exit on its own, if we
// encounter a critical error.
runtime.RuntimeCommands <- &commonbackend.BackendStatusRequest{
statusResponse, err := runtime.ProcessCommand(&commonbackend.BackendStatusRequest{
Type: "backendStatusRequest",
}
})
statusResponse := <-runtime.RuntimeCommands
switch responseMessage := statusResponse.(type) {
case error:
log.Warnf("Failed to get response for backend (in backend runtime keep alive): %s", responseMessage.Error())
if err != nil {
log.Warnf("Failed to get response for backend (in backend runtime keep alive): %s", err.Error())
log.Debugf("Attempting to close socket...")
err := sock.Close()
if err != nil {
log.Debugf("Failed to close socket: %s", err.Error())
}
continue
}
switch responseMessage := statusResponse.(type) {
case *commonbackend.BackendStatusResponse:
if !responseMessage.IsRunning {
if hasFailedBackendRunningCheckAlready {
@ -168,112 +161,119 @@ func (runtime *Runtime) goRoutineHandler() error {
OuterLoop:
for {
commandRaw := <-runtime.RuntimeCommands
switch command := commandRaw.(type) {
case *commonbackend.AddProxy:
err := handleCommand("addProxy", command, sock, runtime.RuntimeCommands)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
for chanIndex, messageData := range runtime.messageBuffer {
if messageData == nil {
continue
}
case *commonbackend.BackendStatusRequest:
err := handleCommand("backendStatusRequest", command, sock, runtime.RuntimeCommands)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
// We don't use the Mutex here as the Mutex is only to prevent cross-talk between multiple ProcessCommand() calls.
switch command := messageData.Message.(type) {
case *commonbackend.AddProxy:
err := handleCommand("addProxy", command, sock, messageData.Channel)
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.CheckClientParameters:
err := handleCommand("checkClientParameters", command, sock, runtime.RuntimeCommands)
case *commonbackend.BackendStatusRequest:
err := handleCommand("backendStatusRequest", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.CheckServerParameters:
err := handleCommand("checkServerParameters", command, sock, runtime.RuntimeCommands)
case *commonbackend.CheckClientParameters:
err := handleCommand("checkClientParameters", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.ProxyConnectionsRequest:
err := handleCommand("proxyConnectionsRequest", command, sock, runtime.RuntimeCommands)
case *commonbackend.CheckServerParameters:
err := handleCommand("checkServerParameters", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.ProxyInstanceRequest:
err := handleCommand("proxyInstanceRequest", command, sock, runtime.RuntimeCommands)
case *commonbackend.ProxyConnectionsRequest:
err := handleCommand("proxyConnectionsRequest", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.ProxyStatusRequest:
err := handleCommand("proxyStatusRequest", command, sock, runtime.RuntimeCommands)
case *commonbackend.ProxyInstanceRequest:
err := handleCommand("proxyInstanceRequest", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.RemoveProxy:
err := handleCommand("removeProxy", command, sock, runtime.RuntimeCommands)
case *commonbackend.ProxyStatusRequest:
err := handleCommand("proxyStatusRequest", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.Start:
err := handleCommand("start", command, sock, runtime.RuntimeCommands)
case *commonbackend.RemoveProxy:
err := handleCommand("removeProxy", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
}
case *commonbackend.Stop:
err := handleCommand("stop", command, sock, runtime.RuntimeCommands)
case *commonbackend.Start:
err := handleCommand("start", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
case *commonbackend.Stop:
err := handleCommand("stop", command, sock, messageData.Channel)
if err != nil {
log.Warnf("failed to handle command in backend runtime instance: %s", err.Error())
if strings.HasPrefix(err.Error(), "failed to write message") {
break OuterLoop
}
}
default:
log.Warnf("Recieved unknown command type from channel: %q", command)
messageData.Channel <- fmt.Errorf("unknown command recieved")
}
default:
log.Warnf("Recieved unknown command type from channel: %q", command)
runtime.RuntimeCommands <- fmt.Errorf("unknown command recieved")
runtime.messageBuffer[chanIndex] = nil
}
}
@ -330,7 +330,9 @@ func (runtime *Runtime) Start() error {
return fmt.Errorf("runtime already running")
}
runtime.RuntimeCommands = make(chan interface{})
runtime.messageBuffer = make([]*messageForBuf, 10)
runtime.messageBufferLock = sync.Mutex{}
runtime.processRestartNotification = make(chan bool, 1)
runtime.logger = &writeLogger{
@ -379,6 +381,69 @@ func (runtime *Runtime) Stop() error {
return nil
}
func (runtime *Runtime) ProcessCommand(command interface{}) (interface{}, error) {
schedulingAttempts := 0
var commandChannel chan interface{}
SchedulingLoop:
for {
if !runtime.isRuntimeRunning {
time.Sleep(10 * time.Millisecond)
}
if schedulingAttempts > 50 {
return nil, fmt.Errorf("failed to schedule synchronous message transmission after 50 tries with 100ms intervals (REPORT THIS ISSUE)")
}
runtime.messageBufferLock.Lock()
// Attempt to find spot in buffer to schedule message transmission
for i, message := range runtime.messageBuffer {
if message != nil {
continue
}
commandChannel = make(chan interface{})
runtime.messageBuffer[i] = &messageForBuf{
Channel: commandChannel,
Message: command,
}
runtime.messageBufferLock.Unlock()
break SchedulingLoop
}
runtime.messageBufferLock.Unlock()
time.Sleep(100 * time.Millisecond)
schedulingAttempts++
}
// Fetch response and close Channel
response := <-commandChannel
close(commandChannel)
err, ok := response.(error)
if ok {
return nil, err
}
return response, nil
}
func (runtime *Runtime) cleanUpPendingCommandProcessingJobs() {
for messageIndex, message := range runtime.messageBuffer {
if message == nil {
continue
}
close(message.Channel)
runtime.messageBuffer[messageIndex] = nil
}
}
func NewBackend(path string) *Runtime {
return &Runtime{
ProcessPath: path,

27
backend/api/backendruntime/struct.go
View file

@ -4,6 +4,9 @@ import (
"net"
"os/exec"
"strings"
"sync"
"github.com/charmbracelet/log"
)
type Backend struct {
@ -11,6 +14,11 @@ type Backend struct {
Path string `validate:"required"`
}
type messageForBuf struct {
Channel chan interface{}
Message interface{}
}
type Runtime struct {
isRuntimeRunning bool
logger *writeLogger
@ -18,9 +26,11 @@ type Runtime struct {
currentListener net.Listener
processRestartNotification chan bool
ProcessPath string
Logs []string
RuntimeCommands chan interface{}
messageBufferLock sync.Mutex
messageBuffer []*messageForBuf
ProcessPath string
Logs []string
OnCrashCallback func(sock net.Conn)
}
@ -31,6 +41,17 @@ type writeLogger struct {
func (writer writeLogger) Write(p []byte) (n int, err error) {
logSplit := strings.Split(string(p), "\n")
if isDevelopmentMode {
for _, logLine := range logSplit {
if logLine == "" {
continue
}
log.Debug("spawned backend logs: " + logLine)
}
}
writer.Runtime.Logs = append(writer.Runtime.Logs, logSplit...)
return len(p), err