解决MySql布尔型新旧版本兼容问题,采用枚举来表示布尔型的数据表。由正向工程赋值
大石头 authored at 2018-05-15 21:21:05
32.91 KiB
X
using System.Buffers;
using System.Net.Sockets;
using NewLife.Data;
using NewLife.Log;
using NewLife.Messaging;

namespace NewLife.Net;

/// <summary>增强TCP客户端(发送:写锁串行、直发核心与流式/文件/消息发送)</summary>
/// <remarks>
/// <para>发送出口共用一把写锁(<see cref="_writeLock"/>),任一时刻只有一位写者在写套接字;失败在锁外上报并按失败关闭会话。</para>
/// <para>直发:<see cref="SessionBase.Send(IPacket)"/> 锁内同步直写(0 分配 0 拷贝,链式包散列写一次系统调用提交整条链),
/// <see cref="TcpSession.SendAsync(IPacket, CancellationToken)"/> 锁内异步直写;流式/文件/消息发送见对应章节,出站队列(第二出口)与发送泵见 TcpSession.SendQueue。</para>
/// <para>数据流:入口 → 发送骨架 SendUnderLock/SendUnderLockAsync(等锁 → 写 → 释放 → 锁外上报)→ 写核心(SSL 流 / 散列写)→ 内核;
/// 平台差异(Span/数组、异步写降级、内核零拷贝)经 TcpSession.Platform。</para>
/// </remarks>
public partial class TcpSession
{
    #region 发送
    // 发送出口:默认直发,另有可选的发送队列作第二出口。两者共用一把写锁,任一时刻只有一位写者在写套接字。
    // 例外:SSL 握手写不经写锁(客户端握手发生在打开流程内,此时会话尚未对业务可用、不可能有并发发送);
    // 服务端握手则由 TcpSession.StartSslAsync 抢占写锁后才发起,握手期间到达的发送会等待,避免明文写进TLS流。
    //   Send(IPacket/byte[]/Span)  锁内同步直写:0 分配 0 拷贝,写完才返回,失败返回 -1
    //   SendAsync(IPacket)         锁内异步直写:0 分配 0 拷贝,等待可写期间不占线程
    //   SendAsync(Stream)          流式:读一块 → 写完(或挂起)再读下一块,内核缓冲即背压,内存有界
    //   SendFileAsync(文件)        锁内让内核零拷贝推文件(SendFile/TransmitFile)
    // 慢对端由内核发送缓冲形成天然背压;需要排队/限流的场合可在上层用 Actor 组合,不进发送核心。
    // 批量场景可选走 TcpSession.SendQueue(见 TcpSession.SendQueue.cs 出站队列与发送泵),由发送泵在锁内批量写出。
    // 直发入口共用发送骨架 SendUnderLock/SendUnderLockAsync:等锁 → 写 → 释放 → 锁外上报(失败上报/关闭必须在锁外,防回调内再 Send 死锁)。

    /// <summary>写锁。四个发送入口共用:任一时刻只有一位写者在写套接字,故多线程/多入口混用不会交错</summary>
    private readonly SemaphoreSlim _writeLock = new(1, 1);

    /// <summary>获取续发总预算的截止时间。返回 0 表示不限制(Timeout 未启用)</summary>
    private Int64 GetSendDeadline() => Timeout > 0 ? Runtime.TickCount64 + Timeout : 0;

    private Int32 _bsize;

    /// <summary>按本次发送量调优内核发送缓冲(_bsize 缓存,读取 SendBufferSize 耗时很大)</summary>
    /// <remarks>
    /// <para>调用方须已持有写锁:套接字参数调整不能与锁内的 Send/Write 并发。</para>
    /// <para>_bsize 只增不减,仅同步直写路径使用;异步直写与流式分块不调整内核缓冲。</para>
    /// </remarks>
    /// <param name="sock">目标套接字</param>
    /// <param name="count">本次发送字节数</param>
    private void TuneSendBufferSize(Socket sock, Int32 count)
    {
        if (_bsize == 0) _bsize = sock.SendBufferSize;
        if (_bsize < count) sock.SendBufferSize = _bsize = count;
    }

    /// <summary>发送一段数据,短计数自动续发。返回时全部字节已交给内核,失败抛异常</summary>
    /// <remarks>同步 Send 在接收方窗口受限时可能只发出一部分;循环续发直到发完,总耗时超过预算按超时失败。异常由调用方统一转为发送失败处理</remarks>
    /// <param name="sock">目标套接字</param>
    /// <param name="buffer">数据缓冲</param>
    /// <param name="offset">起始偏移</param>
    /// <param name="count">字节数</param>
    /// <param name="deadline">续发总预算截止时间(Runtime.TickCount64 毫秒),0 不限制</param>
    /// <returns>已发送字节数,等于 count</returns>
    private static Int32 SendAll(Socket sock, Byte[] buffer, Int32 offset, Int32 count, Int64 deadline)
    {
        var total = 0;
        while (total < count)
        {
            var sent = sock.Send(buffer, offset + total, count - total, SocketFlags.None);
            if (sent <= 0) throw new IOException(L.T("Send failed (result={0}), {1}/{2} bytes sent", sent, total, count));

            total += sent;

            // 续发前检查总预算:对端持续慢读时,单次 SendTimeout 约束不住整包发送的总耗时
            if (deadline > 0 && total < count && Runtime.TickCount64 > deadline)
                throw new TimeoutException(L.T("Send timeout, {0}/{1} bytes sent", total, count));
        }

        return total;
    }

    /// <summary>发送多段数据(scatter-gather),短计数后按段序跳过已发字节逐段续发</summary>
    /// <param name="sock">目标套接字</param>
    /// <param name="segments">数据段列表</param>
    /// <param name="total">总字节数</param>
    /// <param name="deadline">续发总预算截止时间(Runtime.TickCount64 毫秒),0 不限制</param>
    /// <returns>已发送字节数,等于 total</returns>
    private static Int32 SendAll(Socket sock, IList<ArraySegment<Byte>> segments, Int32 total, Int64 deadline)
    {
        // 先尝试一次多段发送(平台可合并系统调用);正常路径一次发完直接返回
        var sent = sock.Send(segments);
        if (sent <= 0) throw new IOException(L.T("Send failed (result={0}), 0/{1} bytes sent", sent, total));
        if (sent >= total) return sent;

        // 短计数:按段序跳过已发字节,逐段续发(复用原段缓冲,零拷贝)
        var skip = sent;
        foreach (var seg in segments)
        {
            if (skip >= seg.Count)
            {
                skip -= seg.Count;
                continue;
            }

            sent += SendAll(sock, seg.Array!, seg.Offset + skip, seg.Count - skip, deadline);
            skip = 0;
        }

        return sent;
    }

    /// <summary>发送骨架(同步):等写锁 → 写 → 释放 → 锁外上报/关闭</summary>
    /// <remarks>
    /// <para>失败上报与关闭必须放在锁外:OnError/Close 会触发用户事件回调,回调内再次 Send 会在同一线程二次进入写锁而死锁。</para>
    /// <para>写动作用静态 lambda 传入(state 携带数据包等输入),避免每次发送产生闭包分配;Span 入参无法经闭包传递,见下方重载。</para>
    /// <para>流式/文件/消息发送有额外的失败分类(调用方取消与数据错误不上报不关会话),不走本骨架,见各自方法注释。</para>
    /// </remarks>
    /// <param name="state">写动作的输入(数据包等)</param>
    /// <param name="write">锁内写动作(静态 lambda)</param>
    /// <param name="data">出错的发送数据,便于日志定位</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    private Int32 SendUnderLock<TState>(TState state, Func<TState, TcpSession, Int32> write, Object? data)
    {
        Exception? error = null;
        var rs = -1;

        _writeLock.Wait();
        try
        {
            rs = write(state, this);
        }
        catch (Exception ex)
        {
            error = ex;
        }
        finally
        {
            _writeLock.Release();
        }

        if (error != null)
        {
            ReportSendError(error, data);

            return -1;
        }

        return rs;
    }

    /// <summary>发送骨架(同步,内存段)。Span 不能被闭包捕获,单独提供重载;其余语义同 <see cref="SendUnderLock{TState}"/></summary>
    /// <param name="data">数据</param>
    /// <param name="write">锁内写动作(静态 lambda)</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    private Int32 SendUnderLock(ReadOnlySpan<Byte> data, MemoryWriter write)
    {
        Exception? error = null;
        var rs = -1;

        _writeLock.Wait();
        try
        {
            rs = write(data, this);
        }
        catch (Exception ex)
        {
            error = ex;
        }
        finally
        {
            _writeLock.Release();
        }

        if (error != null)
        {
            ReportSendError(error, null);

            return -1;
        }

        return rs;
    }

    /// <summary>内存段写动作(自定义委托:BCL 的 Func 不支持 ref struct 参数)</summary>
    /// <param name="data">数据</param>
    /// <param name="session">会话</param>
    /// <returns>已发送字节数</returns>
    private delegate Int32 MemoryWriter(ReadOnlySpan<Byte> data, TcpSession session);

    /// <summary>发送骨架(异步):等写锁 → 写 → 释放 → 锁外上报/关闭;等待可写期间不占线程</summary>
    /// <remarks>取消在等锁边界生效(WaitAsync);其余语义同 <see cref="SendUnderLock{TState}"/></remarks>
    /// <param name="state">写动作的输入(数据包等)</param>
    /// <param name="write">锁内写动作(静态 lambda)</param>
    /// <param name="data">出错的发送数据,便于日志定位</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    private async ValueTask<Int32> SendUnderLockAsync<TState>(TState state, Func<TState, TcpSession, ValueTask<Int32>> write, Object? data, CancellationToken cancellationToken)
    {
        Exception? error = null;
        var rs = -1;

        await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            rs = await write(state, this).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            error = ex;
        }
        finally
        {
            _writeLock.Release();
        }

        if (error != null)
        {
            ReportSendError(error, data);

            return -1;
        }

        return rs;
    }

    /// <summary>直发:锁内同步写出整个数据包(直接写数据包缓冲,0 拷贝 0 分配),写完才返回</summary>
    /// <remarks>写锁保证任一时刻只有一位写者;失败已上报并按失败关闭会话</remarks>
    /// <param name="pk">数据包。借用语义:调用方保留句柄,用完自行释放</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    private Int32 DirectSend(IPacket pk)
    {
        if (pk == null) return -1;

        if (Log != null && Log.Enable && LogSend) WriteLog("Send [{0}]: {1}", pk.Total, pk.ToHex(LogDataLength));

        return SendUnderLock(pk, static (p, s) => s.WritePacket(p), pk);
    }

    /// <summary>直发:锁内同步写出一段数据(直接写调用方缓冲,0 拷贝 0 分配),写完才返回</summary>
    /// <param name="data">数据。调用方需保证返回前不改写该缓冲</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    private Int32 DirectSend(ReadOnlySpan<Byte> data)
    {
        if (data.IsEmpty) return 0;

        if (Log != null && Log.Enable && LogSend) WriteLog("Send [{0}]: {1}", data.Length, data.ToHex(LogDataLength));

        return SendUnderLock(data, static (d, s) => s.WriteMemory(d));
    }

    /// <summary>直发:锁内异步写出整个数据包(0 拷贝 0 分配),等待可写期间不占线程</summary>
    /// <remarks>失败已上报并按失败关闭会话;同时使用多个发送入口时由写锁串行化</remarks>
    /// <param name="pk">数据包。借用语义:调用方保留句柄,用完自行释放</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    private async ValueTask<Int32> DirectSendAsync(IPacket pk, CancellationToken cancellationToken = default)
    {
        if (pk == null) return -1;

        cancellationToken.ThrowIfCancellationRequested();

        if (Log != null && Log.Enable && LogSend) WriteLog("Send [{0}]: {1}", pk.Total, pk.ToHex(LogDataLength));

        return await SendUnderLockAsync(pk, static (p, s) => s.WritePacketAsync(p), pk, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>写核心(同步):把整包(含链)写完才返回;短计数自动续发,失败抛异常</summary>
    /// <param name="pk">数据包</param>
    /// <returns>已发送字节数</returns>
    private Int32 WritePacket(IPacket pk)
    {
        var count = pk.Total;
        if (count == 0) return 0;

        var sock = Client ?? throw new InvalidOperationException(L.T("Session [{0}] is not open.", Name));

        TuneSendBufferSize(sock, count);

        // SSL 流内部处理部分写:整段写完才返回,无需续发循环
        if (_Stream is { } stream)
        {
            pk.CopyTo(stream);

            LastTime = DateTime.Now;

            return count;
        }

        // 一次系统调用写出整包:单节点直接写数据包自己的缓冲(0 拷贝 0 分配);
        // 链式包用散列写(scatter-gather)一次提交整条链——逐节点多次写会把一条逻辑帧拆成多个 TCP 段,
        // 对端“整帧同窗到达”的零拷贝快路径随之失效(同一次解读出内存视图而非流式体)
        SendPacketData(sock, pk, count, GetSendDeadline());

        LastTime = DateTime.Now;

        return count;
    }

    /// <summary>同步写一块内存(零拷贝:直接写调用方缓冲),短计数自动续发</summary>
    /// <param name="data">数据</param>
    /// <returns>已发送字节数</returns>
    private Int32 WriteMemory(ReadOnlySpan<Byte> data)
    {
        var sock = Client ?? throw new InvalidOperationException(L.T("Session [{0}] is not open.", Name));
        Int32 rs;

        TuneSendBufferSize(sock, data.Length);

        if (_Stream is { } stream)
        {
            // SSL 流内部处理部分写:整段写完才返回,失败直接抛异常,无需续发循环
            WriteSsl(stream, data);
            rs = data.Length;
        }
        else
        {
            // 同步 Send 在接收方窗口受限时可能只发出一部分,续发循环保证整段送出;总预算按会话 Timeout 计时
            rs = SendAllMemory(sock, data, GetSendDeadline());
        }

        LastTime = DateTime.Now;

        return rs;
    }

    /// <summary>发送失败上报:记错误日志并关闭会话</summary>
    /// <remarks>调用方必须已释放写锁:OnError/Close 会触发用户事件回调,回调内再次 Send 会二次进入写锁而死锁</remarks>
    /// <param name="error">发送异常</param>
    /// <param name="data">出错的发送数据,便于日志定位</param>
    private void ReportSendError(Exception error, Object? data)
    {
        if (error.IsDisposed()) return;

        if (Tracer is { } tracer)
        {
            using var span = tracer.NewSpan($"net:{Name}:Send");
            span?.SetError(error, data);
        }

        OnError("Send", error);

        // 发送异常可能是连接出了问题,需要关闭
        Close("SendError");
    }

    /// <summary>发送数据包(同步直发):写完才返回</summary>
    /// <param name="data">数据包。借用语义:调用方保留句柄,用完自行释放</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    protected override Int32 OnSend(IPacket data) => DirectSend(data);

    /// <summary>发送数据(同步直发):写完才返回</summary>
    /// <param name="data">数据</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    protected override Int32 OnSend(ArraySegment<Byte> data)
        => data.Array == null || data.Count <= 0 ? 0 : DirectSend(new ReadOnlySpan<Byte>(data.Array, data.Offset, data.Count));

    /// <summary>发送数据(同步直发):写完才返回</summary>
    /// <param name="data">数据</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    protected override Int32 OnSend(ReadOnlySpan<Byte> data) => DirectSend(data);
    #endregion

    #region 流式发送
    /// <summary>流式发送的读块大小</summary>
    /// <remarks>64KB:基准实测 16KB 分块在回环下的吞吐只有 64KB 分块的一半(小分块导致两端 syscall/唤醒次数成倍),64KB 与整段发送持平且仍远低于 LOH 阈值。</remarks>
    private const Int32 StreamChunkSize = 64 * 1024;

    /// <summary>流式发送:分块读入数据流并逐块写出,读一块写一块</summary>
    /// <remarks>
    /// <para>形态对齐主流网络框架的 SendAsync(Stream):默认 64KB 分块,每块由池化读块直接写出(0 拷贝),全程只在读块上驻留,不产生整段内存。</para>
    /// <para><b>背压</b>:每块写完(或在写锁/内核上挂起)才读下一块,慢速对端不会导致应用层积压——内存占用恒为一块读缓冲,水位由内核发送缓冲充当。</para>
    /// <para>与 <see cref="SessionBase.Send(IPacket)"/> 共用同一把写锁:并发调用不会交错。但两者是<b>各自独立加锁</b>,
    /// “头 + 流式体”先发头部再调用本方法只在单写者场景下保持一条逻辑消息;多写者并发时头体之间可能被插入其它数据,
    /// 需要整条消息原子时改用 <see cref="SendMessageLockedAsync"/>。</para>
    /// <para><b>打开时序</b>:本方法在取写锁之前先确保会话已打开。并发首访同一未打开会话时,打开流程(含 SSL 握手)可能被两个调用者同时进入;
    /// 写锁只串行化“写套接字”,不覆盖打开流程,调用方应在发送前先完成打开。</para>
    /// <para><b>超时</b>:异步写核心按“块”重设计时预算(每块 <see cref="SessionBase.Timeout"/>),故大流的总耗时上限约等于 块数 × Timeout;
    /// 与同步直写路径的“整段总预算”(<see cref="SessionBase.Send(IPacket)"/> 链式包续发)口径不同。</para>
    /// <para>流提前结束(不足 <paramref name="length"/>)或连接故障时抛出异常,已发出的部分不会回退。</para>
    /// </remarks>
    /// <param name="source">数据流</param>
    /// <param name="length">期望发送的字节数;负数表示读到流尾</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已发送的总字节数</returns>
    public async ValueTask<Int64> SendAsync(Stream source, Int64 length = -1, CancellationToken cancellationToken = default)
    {
        if (source == null) throw new ArgumentNullException(nameof(source));
        if (Disposed) throw new ObjectDisposedException(GetType().Name);
        if (!Open()) throw new InvalidOperationException(L.T("Session [{0}] is not open.", GetType().Name));

        Exception? error = null;
        var total = 0L;

        await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            total = await SendStreamCoreAsync(source, length, cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // 调用方主动取消:不是发送故障,不上报、不关会话(finally 会释放写锁)
            throw;
        }
        catch (InvalidDataException)
        {
            // 流提前结束属调用方数据问题,不是发送故障:不上报、不关会话,直接抛(finally 会释放写锁)
            throw;
        }
        catch (Exception ex)
        {
            error = ex;
        }
        finally
        {
            _writeLock.Release();
        }

        // 上报与关闭放锁外:OnError/Close 会触发用户事件回调,回调内再次 Send 会在同一把写锁上二次等待而死锁
        if (error != null)
        {
            ReportSendError(error, null);

            System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(error).Throw();
        }

        return total;
    }

    /// <summary>分块读流并写出。读一块写一块,内存恒为一块读缓冲(调用方须已持有写锁)</summary>
    /// <remarks>每块写完(或在写锁/内核上挂起)才读下一块,慢速对端不会导致应用层积压;流提前结束时报错,已发出的部分不回退。</remarks>
    /// <param name="source">数据流</param>
    /// <param name="length">期望发送的字节数;负数表示读到流尾</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已发送的总字节数</returns>
    private async ValueTask<Int64> SendStreamCoreAsync(Stream source, Int64 length, CancellationToken cancellationToken)
    {
        var chunk = StreamChunkSize;
        var buffer = ArrayPool<Byte>.Shared.Rent(chunk);
        try
        {
            var total = 0L;
            while (length < 0 || total < length)
            {
                var size = chunk;
                if (length > 0 && length - total < size) size = (Int32)(length - total);

                var count = await source.ReadAsync(buffer, 0, size, cancellationToken).ConfigureAwait(false);
                if (count <= 0) break;

                // 写完这一块才读下一块:内存恒为一块,慢对端由内核缓冲形成背压
                var rs = await WriteMemoryAsync(buffer.AsMemory(0, count)).ConfigureAwait(false);
                if (rs < 0) throw new IOException(L.T("Send failed on [{0}], {1} bytes sent before failure.", Name, total));

                total += count;
            }

            // 流提前结束:报错,避免调用方误以为已完整发送
            if (length > 0 && total < length) throw new InvalidDataException(L.T("Stream ended early. Expected {0} bytes but got {1}.", length, total));

            return total;
        }
        finally
        {
            ArrayPool<Byte>.Shared.Return(buffer);
        }
    }
    #endregion

    #region 文件发送
    /// <summary>发送文件:交给内核零拷贝推送(Linux sendfile / Windows TransmitFile),应用层不经过读块搬运</summary>
    /// <remarks>
    /// <para><b>内存</b>:文件内容全程不进入应用层缓冲,也没有 64KB 读块往返,大文件发送的 CPU 与内存开销都低于流式分块。</para>
    /// <para><b>背压</b>:等待可写期间不占线程;慢速对端由内核发送缓冲形成背压,应用层无积压。</para>
    /// <para>与 <see cref="SessionBase.Send(IPacket)"/> 共用同一把写锁:并发调用不会交错。但两者是<b>各自独立加锁</b>,
    /// “响应头 + 文件体”先发头部再调用本方法只在单写者场景下保持一条逻辑消息;多写者并发时头体之间可能被插入其它数据,
    /// 需要整条消息原子时改用 <see cref="SendMessageLockedAsync"/>。</para>
    /// <para><b>打开时序</b>:本方法在取写锁之前先确保会话已打开(与 <see cref="SendAsync(Stream, Int64, CancellationToken)"/> 同构)。并发首访同一未打开会话时,
    /// 打开流程(含 SSL 握手)可能被两个调用者同时进入;写锁只串行化“写套接字”,不覆盖打开流程,调用方应在发送前先完成打开。</para>
    /// <para>SSL 会话(<c>SslStream</c>)、Unix 域套接字(内核零拷贝在其中既不会推进也不会报错,实测永久挂起)与无零拷贝发送 API 的目标框架,均降级为 <see cref="SendAsync(Stream, Int64, CancellationToken)"/> 的分块读取发送,语义一致但多一次读块搬运。</para>
    /// <para>不支持偏移与长度:需要发送文件的某一段(如 HTTP Range)时,用 <see cref="SendAsync(Stream, Int64, CancellationToken)"/> 定位后分块发送。</para>
    /// </remarks>
    /// <param name="filePath">文件路径</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已发送字节数;发送失败返回 -1(已上报并按失败关闭会话)</returns>
    /// <exception cref="ArgumentNullException">路径为空</exception>
    /// <exception cref="FileNotFoundException">文件不存在</exception>
    /// <exception cref="InvalidOperationException">会话未打开</exception>
    public async ValueTask<Int64> SendFileAsync(String filePath, CancellationToken cancellationToken = default)
    {
        if (filePath.IsNullOrEmpty()) throw new ArgumentNullException(nameof(filePath));
        if (Disposed) throw new ObjectDisposedException(GetType().Name);

        var fi = filePath.AsFile();
        if (!fi.Exists) throw new FileNotFoundException(L.T("File not found: {0}", filePath), filePath);

        if (!Open()) throw new InvalidOperationException(L.T("Session [{0}] is not open.", GetType().Name));

        var length = fi.Length;

        Exception? error = null;
        var total = -1L;

        await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            // 非 SSL 会话优先尝试内核零拷贝(NET6+;SSL/Unix 域/低版本由桥内部排除并返回 -1 后降级)
            var kernel = await TrySendFileKernelAsync(filePath, length, cancellationToken).ConfigureAwait(false);
            if (kernel >= 0) return kernel;

            // SSL 会话或无零拷贝发送 API 的框架:分块读取发送(语义相同,多一次读块搬运)。
            // 显式启用异步 IO 与顺序扫描:默认 OpenRead 得到的 FileStream 未开异步,ReadAsync 会同步阻塞线程
            using var fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, FileOptions.Asynchronous | FileOptions.SequentialScan);

            total = await SendStreamCoreAsync(fs, length, cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // 调用方主动取消:不是连接故障,向上抛出(finally 会释放写锁)
            throw;
        }
        catch (InvalidDataException)
        {
            // 流提前结束属调用方数据问题,不是发送故障:不上报、不关会话
            throw;
        }
        catch (Exception ex)
        {
            error = ex;
        }
        finally
        {
            _writeLock.Release();
        }

        // 上报与关闭放锁外:OnError/Close 会触发用户事件回调,回调内再次 Send 会在同一把写锁上二次等待而死锁
        if (error != null)
        {
            ReportSendError(error, filePath);

            return -1;
        }

        return total;
    }
    #endregion

    #region 消息发送
    /// <summary>发送流式消息(协议模式):先发协议头部(声明体长),再把数据流内容读一块写一块地送出</summary>
    /// <param name="message">消息(头部字段就位)</param>
    /// <param name="body">消息体数据流</param>
    /// <param name="bodyLength">消息体字节数;负数时从可定位流推导(<see cref="Stream.CanSeek"/>)</param>
    /// <param name="cancellationToken">取消通知</param>
    /// <returns>已写出的流内容字节数</returns>
    /// <remarks>
    /// <para>头部与流内容在<strong>一次</strong>写锁持有期间写出(无交错),整条消息保持一条逻辑消息语义;大消息全程只在读块上驻留,不产生整段内存。</para>
    /// <para>流提前结束(不足声明长度)或连接故障时抛出异常,已发出的部分不回退。</para>
    /// </remarks>
    /// <exception cref="InvalidOperationException">未设置协议</exception>
    /// <exception cref="ArgumentException">体长未知且流不可定位</exception>
    public virtual async ValueTask<Int64> SendMessageAsync(IMessage message, Stream body, Int64 bodyLength = -1, CancellationToken cancellationToken = default)
    {
        if (message == null) throw new ArgumentNullException(nameof(message));
        if (body == null) throw new ArgumentNullException(nameof(body));

        var codec = Protocol ?? throw new InvalidOperationException(L.T("Protocol not set for session [{0}]", Name));

        // 长度未知:从可定位流推导(协议头须先声明长度)
        if (bodyLength < 0)
        {
            if (!body.CanSeek) throw new ArgumentException(L.T("Cannot determine the stream length. Provide bodyLength or use a seekable stream."), nameof(bodyLength));
            bodyLength = body.Length - body.Position;
        }

        // 头部先行:与流内容在一次写锁内(头体之间不会被其它写入者插入)
        var header = codec.BuildHeader(message, bodyLength);
        try
        {
            var rs = await SendMessageLockedAsync(header, body, bodyLength, cancellationToken).ConfigureAwait(false);

            // 失败返回 -1:底层已记错误日志并按失败关闭会话,这里转成异常,保持本方法“失败即抛”的契约
            if (rs < 0) throw new IOException(L.T("Send message failed on [{0}].", Name));

            return rs;
        }
        finally
        {
            // 写完才返回,此处释放头部构建产物
            header.TryDispose();
        }
    }

    /// <summary>锁内发送「头部 + 流内容」整条消息:一次加锁贯穿,头体之间不会被其它写入者插入</summary>
    /// <remarks>
    /// <para>与分别调用 <see cref="SessionBase.Send(IPacket)"/> 与 <see cref="SendAsync(Stream, Int64, CancellationToken)"/> 的区别:那两条各自加锁,
    /// 多写者并发时头体之间可能被插入其它数据;本方法一次持锁贯穿整条消息,保持“一条逻辑消息”语义。</para>
    /// <para>头部句柄借用(调用方释放);失败上报在释放写锁之后进行。</para>
    /// </remarks>
    /// <param name="header">协议头部数据包。借用语义:调用方保留句柄,用完自行释放</param>
    /// <param name="body">消息体数据流</param>
    /// <param name="length">消息体字节数</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已写出的流内容字节数;失败返回 -1(已上报)</returns>
    internal async ValueTask<Int64> SendMessageLockedAsync(IPacket header, Stream body, Int64 length, CancellationToken cancellationToken = default)
    {
        if (header == null) throw new ArgumentNullException(nameof(header));
        if (body == null) throw new ArgumentNullException(nameof(body));

        Exception? error = null;
        var total = -1L;

        await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            if (Log != null && Log.Enable && LogSend) WriteLog("Send [{0}]: {1}", header.Total, header.ToHex(LogDataLength));

            // 头部先行:与流内容在同一把写锁内,整条消息不会与其它写入者交错
            WritePacket(header);

            total = await SendStreamCoreAsync(body, length, cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // 调用方主动取消:不是发送故障,不上报、不关会话(finally 会释放写锁)
            throw;
        }
        catch (InvalidDataException)
        {
            // 流提前结束属调用方数据问题,不是发送故障:不上报、不关会话
            throw;
        }
        catch (Exception ex)
        {
            error = ex;
        }
        finally
        {
            _writeLock.Release();
        }

        // 上报与关闭放锁外:OnError/Close 会触发用户事件回调,回调内再次 Send 会在同一把写锁上二次等待而死锁
        if (error != null)
        {
            ReportSendError(error, null);

            return -1;
        }

        return total;
    }
    #endregion

    #region 异步发送
    /// <summary>异步发送数据包:写完才返回,等待可写期间不占线程(与同步 Send 共用写锁)</summary>
    /// <remarks>
    /// <para>与 <see cref="SessionBase.Send(IPacket)"/> 同义,只是异步执行:不排队、不拷贝,失败记录错误并关闭会话(返回 -1)。</para>
    /// <para>借用语义:调用方保留句柄,返回后即可释放自己的引用</para>
    /// </remarks>
    /// <param name="data">数据包</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>已发送字节数;失败返回 -1</returns>
    public ValueTask<Int32> SendAsync(IPacket data, CancellationToken cancellationToken = default)
    {
        if (data == null) throw new ArgumentNullException(nameof(data));
        if (Disposed) throw new ObjectDisposedException(GetType().Name);
        if (!Open()) throw new InvalidOperationException(L.T("Session [{0}] is not open.", GetType().Name));

        return DirectSendAsync(data, cancellationToken);
    }
    #endregion
}