gloo/tcp: add a timeout for lazy listener connects

[arkadip-maitra]

Fixes #520

The root cause was that the TCP listener path dropped the caller-provided
timeout, so lazy subgroup connection setup could wait indefinitely on the
listener side even though the initiator side had timeout handling.

This change keeps timeout ownership in one place, avoids the earlier double-exception
race, and preserves clean failure behavior when the peer disappears during lazy
subgroup setup.

[d4l3k]

thanks for the fix!

[d4l3k]

rather than run a new loop can we add support for this to the standard loop.h? Maybe use timerfd w/ epoll?

[arkadip-maitra]

thanks. updated.

Instead of a separate timeout loop added support to existing tcp loop using timerfd

[d4l3k]

can we move Timer to a new file?

[ptdbot]

Thanks for iterating on this — moving to timerfd on the existing loop and splitting Timer into its own file both look good, and the root cause diagnosis in #520 is correct: connectAsListener() did drop the timeout.

That said, I think the current shape has a few issues that need addressing before this lands, the most significant being that the patch removes the initiator-side connect timeout entirely, which trades the reported hang for a different (and broader) class of hang.

1. Correctness

1a. pair.h: removing the cv_ connect timeout is a behavioral regression, not a refactor

waitUntil() is now unconditionally cv_.wait(lock, pred) — an infinite wait. Previously any failure to reach CONNECTED within timeout_ * 5 threw Connect timeout. That backstop covered at least three paths that this PR does not replace with a listener-side timer:

1. Device::connectAsInitiator() with disableConnectionRetries() (device.cc:369-378) — does a blocking socket->connect(sockaddr) followed by writeSeq, with no timeout anywhere in that branch.
2. The post-connect write<sequence_number_t>() phase on the initiator. connectLoop() honors the timeout for establishing the TCP connection, but once it succeeds the sequence-number write has no deadline. If the peer accepts the connection and then dies or stops reading, the write callback never fires, fn is never invoked, the pair stays CONNECTING, and Pair::connect() now blocks forever. This is the same failure class as #520, on the side the PR description says "already has retry + timeout behavior."
3. The listener path when timeout == kNoTimeout — listener.cc:148 only creates a timer if (timeout != kNoTimeout), so there is now no bound from either side.

The PR description frames this as "keeps timeout ownership in one place," but the listener timer only covers the passive path, so the net effect is a loss of coverage. Please keep the timeout_ * 5 guard in waitUntil() as a last-resort backstop and let the new listener timer be the fast, precise failure. The two are complementary — the cv guard is what makes a missed/lost callback recoverable instead of fatal.

Also note this changes the signature of a virtual (waitUntilConnected), which breaks any out-of-tree subclass.

1b. Listener::shutdown() drops pending callbacks without invoking them

shutdown() now does seqToCallback_.clear() (new in this PR) without calling the callbacks. Any Pair waiting on one is never notified, so combined with 1a a Pair::connect() in flight during device teardown blocks permanently. Please invoke each pending callback with an error (e.g. TimeoutError or a "listener shut down" error) before clearing, so waiters unwind.

1c. TLS is now inconsistent with TCP, and setSync() changes behavior

waitUntilSSLConnected() keeps the timeout_ * 5 + wait_for + throw logic, while the base waitUntil() loses it. If the guard is wrong for TCP, why is it right for SSL? As written the asymmetry looks unintentional.

More concretely, it is now applied unconditionally (if (timeout_ != kNoTimeout)) where it was previously gated on useTimeout. Pair::setSync() deliberately passed useTimeout=false with the comment "No need to wait for timeout here. If necessary, the connect path will timeout and signal this thread." After this change, setSync() on a TLS pair can throw SSL handshake timeout. That's a user-visible behavior change that isn't mentioned in the description.

1d. waitForConnection() silently drops the callback on a duplicate unresolved seq

seqToCallback_.emplace(seq, std::move(pending));

emplace is a no-op when the key already exists. The existing entry is only erased when pendingIt->second.resolved is true; if an unresolved wait for the same seq is already registered, the new pending — including the timer that was just armed — is discarded when it goes out of scope, and the caller's fn is never invoked or errored. Silent hang. Use insert_or_assign, or GLOO_ENFORCE that the seq is not already pending.

This is reachable rather than theoretical — see 2a.

2. Multiple communicators / process groups

2a. Sequence numbers are reused across process groups, and stale resolved entries break later subgroups

When useRankAsSeqNumber_ is set, nextAddress(int seq) uses the rank as the sequence number, so the same seq recurs for every process group sharing a Device — exactly the lazy-init subgroup scenario in #520. That interacts badly with the decision to keep timed-out entries in the map:

• Subgroup A times out on seq = N → entry stays with resolved = true.
• The peer's socket for subgroup A eventually arrives → haveConnection() erases the entry and drops the socket.
• Subgroup B then calls waitForConnection(N, ...) legitimately. The peer for B may have already connected and had its socket consumed/dropped by the step above, so B waits for a socket that will never be re-sent → timeout, or a permanent hang if timeout == kNoTimeout.

The resolved flag conflates "this seq timed out once" with "this seq is finished forever," which isn't true when seq is a rank. Please key the staleness on something that distinguishes generations, or explicitly document and bound the lifetime of these entries.

2b. Leak: timed-out entries are never reclaimed

timeoutConnection() intentionally leaves the entry in seqToCallback_. In the precise scenario this PR targets — peer died, socket never arrives — nothing ever erases it before shutdown(). That's unbounded growth across repeated subgroup creation. I understand the tradeoff (erasing means a late socket lands in seqToSocket_ and leaks an fd instead, which is worse), but it needs either a bound or, better, the generation-aware fix from 2a.

2c. Data race on Timer::fd_

armed_ and canceled_ are std::atomic, but fd_ is a plain int accessed from both threads:

• loop thread: handleEvents() → read(fd_, ...)
• other thread: cancel() → timerfd_settime(fd_, ...) → cleanup() → close(fd) / fd_ = -1

Besides being a data race, the concrete failure is: cleanup() closes the fd while the loop thread is between shared_from_this() and read(fd_), so read hits a closed — or worse, recycled — descriptor. GLOO_ENFORCE_NE(rv, -1, "read: ", ...) then throws out of handleEvents(), which propagates out of Loop::run() on the loop thread and terminates the process. Make fd_ atomic, or funnel all fd mutation onto the loop thread (you already have defer() for this).

2d. cleanup() can release the last reference to this and then keep using it

schedule() registers via the shared_ptr<Handler> overload, so Loop::handlers_ holds a strong reference. cleanup() calls loop->unregisterDescriptor(fd, this), which does handlers_.erase(fd) — potentially dropping the final reference — and then continues to touch this (close(fd), fd_ = -1). It only survives today because every current caller happens to hold its own strong ref. Please take auto self = shared_from_this() at the top of cancel(), or close the fd before unregistering. Worth a comment either way, since the Loop::handlers_ ↔ Timer ownership cycle is load-bearing and non-obvious.

2e. Blocking cross-thread cancel() is a lock-inversion hazard

unregisterDescriptor() from a non-loop thread blocks on cv_.wait(lock) until the loop ticks. So Timer::cancel() off the loop thread blocks while the loop thread runs callbacks. If the loop thread is inside timeoutConnection() → fn(...) → a pair callback that takes a lock held by the thread calling cancel(), that deadlocks. I could not construct this from current in-tree callers (shutdown() releases mutex_ before cancelling, which is good), so I'd call this "needs confirming" rather than a live bug — but it's worth a comment on cancel() stating it must not be called with a pair/context lock held.

2f. handleEvents() cancel-vs-fire is not decisive

canceled_ is loaded before fn_(), so a cancel arriving in that window still fires the timeout callback — i.e. a pair that just connected successfully can still be failed by a late timer. It's currently harmless only because timeoutConnection() re-checks resolved/*closed_ under mutex_. That re-check is load-bearing; please note that in a comment so it isn't optimized away later.

3. Main logic: minimal and clean

• waitUntil() is now a one-line template wrapper around cv_.wait(lock, pred) with no remaining variation. If 1a is resolved by restoring the timeout it becomes meaningful again; otherwise inline it and delete the indirection.
• haveConnection() sets it->second.resolved = true; on the line before seqToCallback_.erase(it) — a dead write. Remove it.
• Timer is one-shot but doesn't enforce it. cleanup() closes fd_ and sets it to -1, and schedule() never recreates it. Because cleanup() also resets armed_, the GLOO_ENFORCE(!armed_.exchange(true), "Timer is already armed") guard passes on a second schedule(), which then fails with a confusing timerfd_settime: Bad file descriptor. Either recreate the fd in schedule() or GLOO_ENFORCE(fd_ != -1, "Timer is one-shot and has already fired").
• schedule() requires shared ownership. It calls shared_from_this(), so a stack-allocated or unique_ptr Timer throws std::bad_weak_ptr. Since Loop::createTimer() is the intended entry point, make the constructor private with Loop as a friend, or add a static create().
• timeout == 0 semantics. GLOO_ENFORCE_GE(timeout.count(), 0) accepts 0, which is then silently rewritten to 1ms. The reason (an all-zero itimerspec disarms a timerfd) is not obvious and deserves a comment. Elsewhere gloo spells "no timeout" as kNoTimeout, so a caller passing 0 expecting "wait forever" gets an immediate failure instead — consider rejecting 0 outright.
• The timer callback captures raw this: loop_->createTimer([this, seq] { timeoutConnection(seq); }). Two functions above, the established pattern in this same file is [this, closed = closed_] (listener.cc:94), which is why closed_ is a shared_ptr<atomic<bool>> in the first place. This is safe today only because ~Listener() → shutdown() cancels every timer; please capture closed_ for consistency and defence in depth.
• handleEvents() retries EINTR exactly once. Use a loop, or note why one retry suffices.
• Nice cleanups worth calling out: the stray # in listener.cc is fixed, and moving success callbacks onto the loop thread in haveConnection() makes the callback threading consistent with the fast path.

4. Tests

There are no tests in this PR. For a timeout/lifetime fix in the transport layer with this much concurrency surface, that's the biggest gap — and there is already infrastructure to make it cheap:

gloo/test/tcp_test.cc has TEST(TcpTest, ConnectTimeout), which is almost exactly the harness needed: construct a Loop, drive one connect with a 100ms timeout, and assert the error via EXPECT_TRUE(dynamic_cast<const TimeoutError*>(&e)). Mirroring it should get you solid coverage in three compact tests, no new infrastructure:

1. ListenerTimeout — waitForConnection(seq, 100ms, fn) with no peer ever connecting; assert fn runs once with a TimeoutError. This is the actual regression and it's ~15 lines.
2. ListenerNoSpuriousTimeout — a peer connects well within the timeout; assert success and that no timeout callback fires afterwards (covers 2f and the cancel path).
3. ListenerLateSocketAfterTimeout — timeout first, then deliver the socket for the same seq; assert the late socket is dropped and, per 2a, that a subsequent waitForConnection() on that same seq still works. This is the multi-subgroup case and the one most likely to regress.

One gotcha: Loop::createTimer() calls shared_from_this(), so unlike the existing Loop loop; in tcp_test.cc these tests need std::make_shared<Loop>().

gloo/test/multiproc_test.h also already exists if you want to cover the actual #520 repro (a rank exiting before the subgroup's first collective) end-to-end. One test there would be worth more than the Python repro in the issue, since it would run in CI.

Please also confirm the TLS path still behaves — tls_tcp_test.cc exists, and 1c changes setSync() semantics for TLS pairs.

5. Summary of required changes

1. Restore the timeout_ * 5 backstop in waitUntil() (1a) — this is the blocking one.
2. Invoke pending callbacks with an error in shutdown() instead of dropping them (1b).
3. Fix the TLS/TCP inconsistency and the unintended setSync() timeout (1c).
4. Fix the duplicate-seq emplace no-op (1d).
5. Handle seq reuse across process groups; don't let a stale resolved entry break a later subgroup (2a), and bound the leak (2b).
6. Make fd_ race-free (2c) and keep this alive across cleanup() (2d).
7. Enforce one-shot Timer usage and shared-ownership construction; document the 0-timeout rewrite (§3).
8. Add the listener timeout tests to tcp_test.cc (§4).

Happy to re-review once these are addressed — the underlying fix is the right one, it just needs the initiator-side safety net kept and the seq-reuse case handled.

[arkadip-maitra]

I did not change the lazy seq-reuse behavior in this PR otherwise its all updated

[d4l3k]

LGTM