Linux Networking: Understanding TCP_CLOSE State

2.5. Reasons for socket transitioning to <span>TCP_CLOSE</span> state

Continuing with bpfsnoop:

# bpfsnoop \
        -k sk_stream_wait_close \
        -k '(s)tcp_reset' \
        -k '(s)tcp_done' \
        -k '(s)tcp_done_with_error' \
        -k '(b)__tcp_close' \
                --output-arg 'sk->sk_receive_queue.qlen' \
                --output-arg '&sk->sk_receive_queue' \
                --output-arg 'sk->sk_receive_queue.next' \
                --output-arg 'sk->sk_receive_queue.prev' \
                --output-arg 'sk->sk_receive_queue.next->pp_recycle' \
                --output-arg 'ip42(&sk->__sk_common.skc_daddr)' \
                --output-arg 'port(&sk->__sk_common.skc_dport)' \
                --output-arg 'sk->__sk_common.skc_num' \
        -o tcp_close.bpfsnoop.4.log

← tcp_done_with_error args=((struct sock *)sk=0xff1100011a2f0900, (int)err=32) retval=(void) cpu=45 process=(0:swapper/45) timestamp=15:06:41.253928962
Arg attrs: (__u32)'sk->sk_receive_queue.qlen'=0x1/1, (struct sk_buff_head *)'&sk->sk_receive_queue'=0xff1100011a2f09d8, (struct sk_buff *)'sk->sk_receive_queue.next'=0xff1100012619a100, (struct sk_buff *)'sk->sk_receive_queue.prev'=0xff1100012619a100, (__u8)'sk->sk_receive_queue.next->pp_recycle'=0x1, (unsigned int *)'ip42(&sk->__sk_common.skc_daddr)'=[10.x.x.x,10.y.y.y], (short unsigned int *)'port(&sk->__sk_common.skc_dport)'=9092, (__u16)'sk->__sk_common.skc_num'=0x0/0
Func stack:
  tcp_done_with_error+0x5                               ; net/ipv4/tcp_input.c:4441
  bpf_trampoline_6442479037+0x39
  tcp_reset+0x5                                         ; net/ipv4/tcp_input.c:4456
  tcp_rcv_state_process+0x20e                           ; net/ipv4/tcp_input.c:6696
  tcp_v4_do_rcv+0xd3                                    ; net/ipv4/tcp_ipv4.c:1757
  tcp_v4_rcv+0xbe2                                      ; net/ipv4/tcp_ipv4.c:2166
  ip_protocol_deliver_rcu+0x3c                          ; net/ipv4/ip_input.c:205
  ip_local_deliver_finish+0x72                          ; net/ipv4/ip_input.c:237
  ip_local_deliver+0x6c                                 ; net/ipv4/ip_input.c:257
  ip_sublist_rcv_finish+0x77                            ; include/net/dst.h:477
  ip_sublist_rcv+0x17c                                  ; net/ipv4/ip_input.c:640
  ip_list_rcv+0x102                                     ; net/ipv4/ip_input.c:675
  __netif_receive_skb_list_core+0x22d                   ; net/core/dev.c:5639
  netif_receive_skb_list_internal+0x19e                 ; net/core/dev.c:5741
  napi_complete_done+0x74                               ; include/net/gro.h:448
  mlx5e_napi_poll+0x173
  __napi_poll+0x33                                      ; net/core/dev.c:6600
  net_rx_action+0x18a                                   ; net/core/dev.c:6669
  handle_softirqs+0xe8                                  ; arch/x86/include/asm/jump_label.h:27
  __irq_exit_rcu+0x77                                   ; kernel/softirq.c:612
  irq_exit_rcu+0xe                                      ; kernel/softirq.c:676
  common_interrupt+0xa4
  asm_common_interrupt+0x27                             ; arch/x86/include/asm/idtentry.h:678
  cpuidle_enter_state+0xda                              ; drivers/cpuidle/cpuidle.c:291
  cpuidle_enter+0x2e
  call_cpuidle+0x23                                     ; kernel/sched/idle.c:135
  cpuidle_idle_call+0x11d                               ; kernel/sched/idle.c:219
  do_idle+0x82                                          ; kernel/sched/idle.c:284
  cpu_startup_entry+0x2a                                ; kernel/sched/idle.c:379
  start_secondary+0x129                                 ; arch/x86/kernel/smpboot.c:211
  secondary_startup_64_no_verify+0x18f                  ; arch/x86/kernel/head_64.S:449

Here, <span>err=32</span> corresponds to <span>err = EPIPE</span>.

// net/ipv4/tcp_input.c

void tcp_done_with_error(struct sock *sk, int err)
{
    /* This barrier is coupled with smp_rmb() in tcp_poll() */
    WRITE_ONCE(sk->sk_err, err);
    smp_wmb();

    tcp_write_queue_purge(sk);
    tcp_done(sk);

    if (!sock_flag(sk, SOCK_DEAD))
        sk_error_report(sk);
}
EXPORT_SYMBOL(tcp_done_with_error);

/* When we get a reset we do this. */
void tcp_reset(struct sock *sk, struct sk_buff *skb)
{
    int err;

    trace_tcp_receive_reset(sk);

    /* mptcp can't tell us to ignore reset pkts,
     * so just ignore the return value of mptcp_incoming_options().
     */
    if (sk_is_mptcp(sk))
        mptcp_incoming_options(sk, skb);

    /* We want the right error as BSD sees it (and indeed as we do). */
    switch (sk->sk_state) {
    case TCP_SYN_SENT:
        err = ECONNREFUSED;
        break;
    case TCP_CLOSE_WAIT:
        err = EPIPE;
        break;
    case TCP_CLOSE:
        return;
    default:
        err = ECONNRESET;
    }
    tcp_done_with_error(sk, err);
}

Thus:

1. The socket first receives a FIN packet, transitioning to <span>TCP_CLOSE_WAIT</span>.
2. The socket then receives a FINACK packet, which is buffered in <span>sk->sk_receive_queue</span>.
3. Finally, the socket receives a RST packet, transitioning to <span>TCP_CLOSE</span>.

3. Sarama has a TCP connection leak issue

After discussing with colleagues, it was confirmed that 9093 is the port for the Kafka server, and the library used to connect to Kafka is Sarama.

Checking the Sarama issue list, a related issue has already been raised:

• Client SeekBroker Connection Leak^[1]

However, this issue requires further investigation.

3.1. Adding more logs to Sarama

Directly modifying Sarama code:

1. Add logs in the broker where TCP connections are involved.
2. Add logs in the client where brokers are involved.
3. Enable Sarama’s logger in user-space processes.

diff --git a/broker.go b/broker.go
index d0d5b87..9cc4ca2 100644
--- a/broker.go
+++ b/broker.go
@@ -155,6 +155,36 @@ func NewBroker(addr string) *Broker {
    return &Broker{id: -1, addr: addr}
 }

+func (b *Broker) connInfo(conn net.Conn) string {
+   if conn == nil {
+       return "not connected"
+   }
+
+   laddr, raddr := conn.LocalAddr(), conn.RemoteAddr()
+   return fmt.Sprintf("laddr=%s raddr=%s", laddr.String(), raddr.String())
+}
+
+func (b *Broker) debugInfo() string {
+   var sb strings.Builder
+
+   fmt.Fprintf(&sb, "id=0x%x addr=%s %s", b.id, b.addr, b.connInfo(b.conn))
+
+   return sb.String()
+}

 // Open tries to connect to the Broker if it is not already connected or connecting, but does not block
 // waiting for the connection to complete. This means that any subsequent operations on the broker will
 // block waiting for the connection to succeed or fail. To get the effect of a fully synchronous Open call,
@@ -202,16 +232,26 @@ func (b *Broker) Open(conf *Config) error {
        };
        dialer := conf.getDialer();
        b.conn, b.connErr = dialer.Dial("tcp", b.addr)
+       Logger.Printf("[DBG] [BROKER] connecting conn[%s] err[%v]\n", b.connInfo(b.conn), b.connErr)
        if b.connErr != nil {
            Logger.Printf("Failed to connect to broker %s: %s\n", b.addr, b.connErr)
            b.conn = nil;
            atomic.StoreInt32(&b.opened, 0);
            return;
        }
        if conf.Net.TLS.Enable {
            b.conn = tls.Client(b.conn, validServerNameTLS(b.addr, conf.Net.TLS.Config));
        }

+       Logger.Printf("[DBG] [BROKER] connected %s\n", b.debugInfo());
+
        b.conn = newBufConn(b.conn);
        b.conf = conf;

@@ -241,7 +281,7 @@ func (b *Broker) Open(conf *Config) error {
            b.connErr = b.authenticateViaSASLv0();

            if b.connErr != nil {
-               err = b.conn.Close()
+               err = b.closeConn();
                if err == nil {
                    DebugLogger.Printf("Closed connection to broker %s\n", b.addr);
                } else {
@@ -262,7 +302,7 @@ func (b *Broker) Open(conf *Config) error {
            if b.connErr != nil {
                close(b.responses);
                <-b.done;
-               err = b.conn.Close()
+               err = b.closeConn();
                if err == nil {
                    DebugLogger.Printf("Closed connection to broker %s\n", b.addr);
                } else {
@@ -278,6 +318,8 @@ func (b *Broker) Open(conf *Config) error {
        } else {
            DebugLogger.Printf("Connected to broker at %s (unregistered)\n", b.addr);
        }
+
+       Logger.Printf("[DBG] [BROKER] opened %s\n", b.debugInfo());
    })

    return nil;
@@ -290,6 +332,13 @@ func (b *Broker) ResponseSize() int {
    return len(b.responses);
 }

+func (b *Broker) closeConn() error {
+   dbgInfo := b.debugInfo();
+   err := b.conn.Close();
+   Logger.Printf("[DBG] [BROKER] closed %s (err=%v)\n", dbgInfo, err);
+   return err;
+}
+
 // Connected returns true if the broker is connected and false otherwise. If the broker is not
 // connected but it had tried to connect, the error from that connection attempt is also returned.
 func (b *Broker) Connected() (bool, error) {
    return b.conn != nil, b.connErr;
}

Through the above kernel module confirmation:

1. The seed broker TCP connection has transitioned to TCP_CLOSE state, but the user-space has not closed it.
2. The registered broker TCP connection has the same issue.

Additionally, the following log was observed:

client/metadata got error from broker 1193 while fetching metadata: write tcp 10.x.x.x:24180->10.y.y.y.y:9093: write: broken pipe

This log explains the reduction or even disappearance of the above dmesg logs: Sarama randomly selects brokers to send requests, and these brokers’ TCP connections have become unusable, leading to the kernel returning <span>-EPIPE</span> error when writing data, causing Sarama to close that broker’s TCP connection.

However, why does the Kafka server actively close the TCP connection?This requires further research.

3.2. Fixing Sarama

At appropriate places, check if the socket corresponding to the TCP connection has <span>sk->err</span>:

func (b *Broker) getSockOpt(conn *net.TCPConn, opt int) (int, error) {
    file, err := conn.File();
    if err != nil {
        return 0, fmt.Errorf("failed to get file from connection: %w", err);
    }
    defer file.Close(); // Close the duplicated file descriptor

    sockErr, err := syscall.GetsockoptInt(int(file.Fd()), syscall.SOL_SOCKET, opt);
    if err != nil {
        return 0, fmt.Errorf("failed to get sockopt %d: %w", opt, err);
    }
    return sockErr, nil;
}

func (b *Broker) getSockError() error {
    b.lock.Lock();
    defer b.lock.Unlock();
    if b.connTCP == nil {
        return nil;
    }

    sockErr, err := b.getSockOpt(b.connTCP, syscall.SO_ERROR);
    if err != nil {
        return fmt.Errorf("failed to get socket error: %w", err);
    }
    if sockErr != 0 {
        return syscall.Errno(sockErr);
    }
    return nil;
}

func (client *client) checkSeedBrokersHealth(brokers []*Broker) []*Broker {
    if len(brokers) == 0 {
        return nil;
    }

    healthyBrokers := make([]*Broker, 0, len(brokers));
    for _, broker := range brokers {
        if err := broker.getSockError(); err != nil {
            Logger.Printf("client/seedbrokers close seed broker #%d at %s due to socket error: %v", broker.ID(), broker.Addr(), err);
            safeAsyncClose(broker);
            continue;
        }

        healthyBrokers = append(healthyBrokers, broker);
    }

    return healthyBrokers;
}

func (client *client) checkBrokersHealth() {
    for id, broker := range client.brokers {
        if err := broker.getSockError(); err != nil {
            Logger.Printf("client/brokers close broker #%d at %s due to socket error: %v", broker.ID(), broker.Addr(), err);
            safeAsyncClose(broker);
            delete(client.brokers, id);
        }
    }

    client.seedBrokers = client.checkSeedBrokersHealth(client.seedBrokers);
    client.deadSeeds = client.checkSeedBrokersHealth(client.deadSeeds);
}

func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bool) (retry bool, err error) {
    if client.Closed() {
        return;
    }
    client.lock.Lock();
    defer client.lock.Unlock();

    // Check health of existing brokers, including seed brokers, dead
    // seed brokers, and registered brokers.
    // - if error occurred on broker's tcp socket, close the tcp
    //   connection.
    // - if it's seed broker or dead seed broker, remove it from
    //   the list.
    client.checkBrokersHealth();
}

After adding <span>checkBrokersHealth()</span> in <span>updateMetadata()</span>, Sarama will close TCP connections that are in <span>TCP_CLOSE</span> state every 10 minutes.

Through the above kernel module confirmation, only a small number of TCP sockets remain in <span>TCP_CLOSE</span> state, and they will be closed within 10 minutes.

For more fix details, please refer to the Sarama PR:

• fix: close broken tcp connections^[2]

4. Improving the Kernel Protocol Stack

1. If user-space behavior cannot be controlled, can the kernel be modified to allow system administrators to handle such issues?
2. Can the dmesg logs be reduced? Or even completely eliminated?

4.1. Plan to add <span>net.ipv4.tcp_purge_receive_queue</span> sysctl

The core changes are as follows:

diff --git a/net/ipv4/tcp_input.c b/net/ipv4/tcp_input.c
index 198f8a0d37be0..a01d39c37be2f 100644
--- a/net/ipv4/tcp_input.c
+++ b/net/ipv4/tcp_input.c
@@ -4648,6 +4648,7 @@ EXPORT_IPV6_MOD(tcp_done_with_error);
 /* When we get a reset we do this. */
 void tcp_reset(struct sock *sk, struct sk_buff *skb)
 {
+   const struct net *net = sock_net(sk);
     int err;

     trace_tcp_receive_reset(sk);
@@ -4664,6 +4665,21 @@ void tcp_reset(struct sock *sk, struct sk_buff *skb)
         err = ECONNREFUSED;
         break;
     case TCP_CLOSE_WAIT:
+       /* RFC9293 3.10.7.4. Other States
+        *   Second, check the RST bit:
+        *     CLOSE-WAIT STATE
+        *
+        * If the RST bit is set, then any outstanding RECEIVEs and
+        * SEND should receive "reset" responses.  All segment queues
+        * should be flushed.  Users should also receive an unsolicited
+        * general "connection reset" signal.  Enter the CLOSED state,
+        * delete the TCB, and return.
+        *
+        * If net.ipv4.tcp_purge_receive_queue is enabled,
+        * sk_receive_queue will be flushed too.
+        */
+       if (unlikely(net->ipv4.sysctl_tcp_purge_receive_queue))
+           skb_queue_purge(&sk->sk_receive_queue);
         err = EPIPE;
         break;
     case TCP_CLOSE:

Indeed, the reference for such changes is section <span>3.10.7.4</span> of RFC9293.

4.2. Plan to add <span>page_pool_release_stalled</span> tracepoint

The core changes are as follows:

diff --git a/net/core/page_pool.c b/net/core/page_pool.c
index 1a5edec485f1..59a85887921c 100644
--- a/net/core/page_pool.c
+++ b/net/core/page_pool.c
@@ -1218,8 +1218,11 @@ static void page_pool_release_retry(struct work_struct *wq)
            (!netdev || netdev == NET_PTR_POISON)) {
                int sec = (s32)((u32)jiffies - (u32)pool->defer_start) / HZ;

-               pr_warn("%s() stalled pool shutdown: id %u, %d inflight %d sec\n",
-                       __func__, pool->user.id, inflight, sec);
+               if (sec >= DEFER_WARN_INTERVAL / HZ && sec < DEFER_WARN_INTERVAL * 2 / HZ)
+                       pr_warn("%s() stalled pool shutdown: id %u, %d inflight %d sec\n",
+                               __func__, pool->user.id, inflight, sec);
+               else
+                       trace_page_pool_release_stalled(pool, inflight, sec);
                pool->defer_warn = jiffies + DEFER_WARN_INTERVAL;
        }

The motivation for such changes is:

1. At least one <span>pr_warn()</span>.
2. Subsequently, confirm if there are still inflight pages through <span>page_pool_release_stalled</span> tracepoint.

Perhaps there are better improvement methods, which require further research.

5. Conclusion

First, I would like to thank the following individuals for their enthusiastic participation in discussions during my troubleshooting process:

• Jason Xing
• Lance Yang
• Jiayuan Chen
• Gray Liang

Next, bpfsnoop^[3] played a crucial role in troubleshooting, eliminating the need to write complex bpftrace scripts.

After confirming the behavior of the protocol stack, I quickly wrote a Python script using AI to replicate the behavior of the protocol stack.

Finally, once the Sarama PR is merged, please update Sarama to the latest commit to avoid TCP connection leak issues.

References[1]

Client SeekBroker Connection Leak: https://github.com/IBM/sarama/issues/3143

[2]

fix: close broken tcp connections: https://github.com/IBM/sarama/pull/3384

[3]

bpfsnoop: https://github.com/bpfsnoop/bpfsnoop