diff options
| -rwxr-xr-x | internal/protocol.go | 9 | ||||
| -rwxr-xr-x | transport/pipe/pipe_factory.go | 21 | ||||
| -rwxr-xr-x | transport/socket/socket_factory.go | 5 |
3 files changed, 14 insertions, 21 deletions
diff --git a/internal/protocol.go b/internal/protocol.go index 78174118..73cb960e 100755 --- a/internal/protocol.go +++ b/internal/protocol.go @@ -68,8 +68,7 @@ func SendControl(rl relay.Relay, payload interface{}) error { fr.WritePayload(data) fr.WriteCRC(fr.Header()) - // hold a pointer to a frame - // Do we need a copy here???? + // we don't need a copy here, because frame copy the data before send err = rl.Send(fr) if err != nil { return errors.E(op, err) @@ -78,7 +77,7 @@ func SendControl(rl relay.Relay, payload interface{}) error { return nil } -func FetchPID(rl relay.Relay) (int64, error) { +func Pid(rl relay.Relay) (int64, error) { const op = errors.Op("fetch_pid") err := SendControl(rl, pidCommand{Pid: os.Getpid()}) if err != nil { @@ -111,5 +110,9 @@ func FetchPID(rl relay.Relay) (int64, error) { return 0, errors.E(op, err) } + if link.Pid <= 0 { + return 0, errors.E(op, errors.Str("pid should be greater than 0")) + } + return int64(link.Pid), nil } diff --git a/transport/pipe/pipe_factory.go b/transport/pipe/pipe_factory.go index 0d46f496..84a9d311 100755 --- a/transport/pipe/pipe_factory.go +++ b/transport/pipe/pipe_factory.go @@ -29,7 +29,7 @@ type sr struct { // SpawnWorkerWithTimeout creates new Process and connects it to goridge relay, // method Wait() must be handled on level above. -func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd, listeners ...events.Listener) (*worker.Process, error) { //nolint:gocognit +func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd, listeners ...events.Listener) (*worker.Process, error) { spCh := make(chan sr) const op = errors.Op("factory_spawn_worker_with_timeout") go func() { @@ -90,7 +90,8 @@ func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd, lis } } - pid, err := internal.FetchPID(relay) + // used as a ping + _, err = internal.Pid(relay) if err != nil { err = multierr.Combine( err, @@ -109,19 +110,6 @@ func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd, lis } } - if pid != w.Pid() { - select { - case spCh <- sr{ - w: nil, - err: errors.E(op, errors.Errorf("pid mismatches, get: %d, want: %d", pid, w.Pid())), - }: - return - default: - _ = w.Kill() - return - } - } - select { case // return worker @@ -177,7 +165,8 @@ func (f *Factory) SpawnWorker(cmd *exec.Cmd, listeners ...events.Listener) (*wor } // errors bundle - if pid, err := internal.FetchPID(relay); pid != w.Pid() { + _, err = internal.Pid(relay) + if err != nil { err = multierr.Combine( err, w.Kill(), diff --git a/transport/socket/socket_factory.go b/transport/socket/socket_factory.go index d98ce607..e5b36cfa 100755 --- a/transport/socket/socket_factory.go +++ b/transport/socket/socket_factory.go @@ -66,7 +66,7 @@ func (f *Factory) listen() error { } rl := socket.NewSocketRelay(conn) - pid, err := internal.FetchPID(rl) + pid, err := internal.Pid(rl) if err != nil { return err } @@ -189,7 +189,8 @@ func (f *Factory) SpawnWorker(cmd *exec.Cmd, listeners ...events.Listener) (*wor w.AttachRelay(rl) // errors bundle - if pid, err := internal.FetchPID(rl); pid != w.Pid() { + _, err = internal.Pid(rl) + if err != nil { err = multierr.Combine( err, w.Kill(), |