mats/go-cart-actor
1
0
Fork 0
Code Issues Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

pointer and ping misses
All checks were successful
Build and Publish / BuildAndDeploy (push) Successful in 1m49s
Details

Browse Source
This commit is contained in:
matst80
2024-11-09 20:13:47 +01:00
parent 198f24ede0
commit fa37db6f63
1 changed files with 16 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
synced-pool.go
View File

@@ -56,6 +56,7 @@ type Quorum interface {
type RemoteHost struct { type RemoteHost struct {
Host string Host string
MissedPings int
Pool *RemoteGrainPool Pool *RemoteGrainPool
connection net.Conn connection net.Conn
} }
@@ -65,7 +66,7 @@ type SyncedPool struct {
listener net.Listener listener net.Listener
Hostname string Hostname string
local *GrainLocalPool local *GrainLocalPool
remotes []RemoteHost remotes []*RemoteHost
remoteIndex map[CartId]*RemoteGrainPool remoteIndex map[CartId]*RemoteGrainPool
} }
@@ -81,7 +82,7 @@ func NewSyncedPool(local *GrainLocalPool, hostname string, d Discovery) (*Synced
Hostname: hostname, Hostname: hostname,
local: local, local: local,
listener: l, listener: l,
remotes: make([]RemoteHost, 0), remotes: make([]*RemoteHost, 0),
remoteIndex: make(map[CartId]*RemoteGrainPool), remoteIndex: make(map[CartId]*RemoteGrainPool),
} }
if d != nil { if d != nil {
@@ -126,6 +127,13 @@ func NewSyncedPool(local *GrainLocalPool, hostname string, d Discovery) (*Synced
log.Printf("Error adding undiscovered remote %s: %v\n", h, err) log.Printf("Error adding undiscovered remote %s: %v\n", h, err)
} }
} }
for _, r := range pool.remotes {
err := DoPing(r.connection)
if err != nil {
r.MissedPings++
log.Printf("Error pinging remote %s: %v\n, missed pings: %d", r.Host, err, r.MissedPings)
}
}
}() }()
} }
go func() { go func() {
@@ -324,7 +332,7 @@ func (p *SyncedPool) AddRemoteWithConnection(address string, connection net.Conn
Pool: pool, Pool: pool,
Host: address, Host: address,
} }
return p.addRemoteHost(address, remote) return p.addRemoteHost(address, &remote)
} }
func DoPing(connection net.Conn) error { func DoPing(connection net.Conn) error {
@@ -341,7 +349,7 @@ func DoPing(connection net.Conn) error {
return nil return nil
} }
func (p *SyncedPool) addRemoteHost(address string, remote RemoteHost) error { func (p *SyncedPool) addRemoteHost(address string, remote *RemoteHost) error {
for _, r := range p.remotes { for _, r := range p.remotes {
if r.Host == address { if r.Host == address {
log.Printf("Remote %s already exists\n", address) log.Printf("Remote %s already exists\n", address)
@@ -376,7 +384,7 @@ func (p *SyncedPool) AddRemote(address string) error {
Host: address, Host: address,
} }
return p.addRemoteHost(address, remote) return p.addRemoteHost(address, &remote)
} }
func (p *SyncedPool) Process(id CartId, messages ...Message) ([]byte, error) { func (p *SyncedPool) Process(id CartId, messages ...Message) ([]byte, error) {