buffer 管理问题

[dbgp]

设计说明

在数据库中用到 buffer 的地方很多（比如最常见的网络输入输出 buffer、redo log buffer)，并且是高频使用，实现得不好也严重影响性能。lealone 直到今天才完美解决 buffer 管理的问题，前后用了4种方案:

1. 最开始使用全局的 buffer pool，里面用一个支持多线程并发访问的队列实现，池里的 buffer 大小不是固定的，会先按需求的容量大小从队列里找到一个最适合的 buffer，用完后再放回池中。这个方案的缺点是多线程访问并不高效；另外 buffer 的大小不一样，有的 buffer 用不到，有的被拿去用了但是依然有空闲空间浪费了。

2. 为了解决方案1的问题，每个调度线程使用自己的 buffer pool，这样就不需要做并发控制了，只有单个线程访问，队列可以用普通的链表来实现。这个方案虽然访问队列快了，但是依然有出队和入队的开销，并且 buffer 内存空间浪费的问题也没有解决。

3. 不使用 buffer pool 了，给每条 tcp 连接分配两个专属 buffer，每个 buffer 默认是 4k，这样就没有访问队列的开销了，处理完一次请求后，两个 buffer 都能快速复位，所以也不浪费内存空间。这个方案的唯一缺点就是不能支持太多条 tcp 长连接，一旦数量多了依然很占用内存。

4. 这是目前的终极方案，给每个调度线程各分配两个专属 buffer，不给每条 tcp 连接分配 buffer 了。调度线程负责的所有 tcp 连接都使用调度线程的 buffer，每条 tcp 连接的输入输出包按顺序存放到调度线程的 输入包 buffer 和 输出包 buffer 中，调度线程解析完这些输入包或把输出包写到 socket 后，只需要把 buffer 的 pos 和 limit 设置一下就能复用了。不但没有内存碎片，也没有方案1和方案2访问队列的开销，也不像方案3那样 tcp 连接多了浪费内存，真的就是完美的解决方案。

这里的理解还是针对旧方案。

新方案应该是真的11月4日的所有的tcp连接都共用调度器的全局buffer开始的一系列实现。这次的学习就从这里先开始吧。之前的思路依然可以借鉴——

• lealone的启动过程，包含了server服务的启动
• client端如何与server端建立连接
• 连接创建之后如何传输数据

这三步完成，才可以看到buffer中网络传输中的使用场景。结果使用场景看buffer是如何管理的。

[dbgp]

客户端的创建过程，当时梳理时客户端还没采用调度器模式。这里可以看到“调度线程负责的所有 tcp 连接都使用调度线程的 buffer” 创建连接使用的buffer来自调度器

稍微修改一下lealoneclienttest，跟踪client端的创建过程，有个基本概念。可以看到上述断点是怎么进来的。

[dbgp]

客户端创建连接信息时，一步一步初始化调度器

这里会利用Class.forName的反射初始化调度器类的实例（个数根据设备的核数确定，为了测试可以指定1个调度线程）

构造工程类型为 RoundRobinFactory，在父类SchedulerFactoryBase的构造函数中会进行init的函数调用。并指定调度器的工厂为自己。

再下面时给连接对象绑定指定的调度器（随机指定一个，这里测试只有一个），在工厂的start方法里将所有的调度线程开启。（scheduler.start()调用之后，可以看到多了一个调度线程 ）;

这时候有主线程+一个调度线程。切换到调度线程可以看到已经在run方法里开始“做事”了（EventLoop开始Selector机制）——这个调度器的套路跟Server端类似

最终回到第一张图里，给调度器添加异步任务：“让调度线程负责创建session”

 ac = AsyncCallback.create(ci.isSingleThreadCallback());
// 让调度线程负责创建session
scheduler.handle(() -> {
    createSession(ci, allowRedirect, ac, config, netClient);
});

调度线程开始准备“真正地”建立连接请求——

这里通过selector端 key传递到 attachment，在client端 netClient.connectionEstablished(scheduler, this, key);中获得就回到了昨天上个答复中的过程。至此，客户端与server端的连接已经建立。

[dbgp]

上面的部分算是client端的连接建立过程。根据连接信息建立session时，创建出来ClientScheduler，用来管理实际的连接请求即可。

但server端肯定需要先建立自己的服务，等待client端端请求。相比较会复杂一些，下面看看server端服务的创建过程。这里梳理的部分已经过时了——例如NioServerAccepter类被重命名为NioNetServer类，这样目的也更明确——但基本过程还是相同的。

Server端初始化

在启动的主线程中执行初始化操作——

然后启动全局调度器线程，执行“杂七杂八的任务，数量不多，执行完就删除”——其中就包括启动过程中的动作：

scheduler.handle(() -> {
                // 提前触发对LealoneDatabase的初始化
                initLealoneDatabase();
                // 启动ProtocolServer
                if (!embedded) {
                    startProtocolServers();
                }
                // 等所有的Server启动完成后再启动Scheduler
                // 确保所有的初始PeriodicTask都在单线程中注册
                schedulerFactory.start();
                if (latch != null)
                    latch.countDown();
            })

中 startProtocolServers();过程中真正open一个ServerSocketChannel，然后设置RegisterAccepterTask对象的标志。
调度线程的runRegisterAccepterTasks方法会进行实际的注册。

然后通过Selector机制完成server端的建立。

如果通过Selector机制和自定义的EventLoop完成这一个关键步骤，下一篇继续。

[dbgp]

简化Accept事件的处理流程 里的逻辑比修改前“简化”了不少。

逻辑也更清晰：

• 直接声明 来自 asyncServer的ServerSocketChannel 的regiser方法
• 传递的attachment明确为asyncServer的 ProtocolServer，即 NioNetServer

在GlobalScheduler构造函数中初始化了netEventLoop。调度器的run方法中会循环执行不同的任务（调度器的核心）。
上面的register注册时，获取的Selector来自调度器（实际是netEventLoop中的Selector）

这样调度器执行runEventLoop方法时，会获取到Selector上是否有ready的key。当有client端连接到这个指定端口时，会触发 SelectionKey.OP_ACCEPT。完成 server.accept(scheduler)的动作。

[dbgp]

建立连接的过程参考旧的梳理:client端的连接创建、sessionInit和sessionInitAck的交互逻辑、数据是如何进行传输的、Server端的Session创建场景分析 重新看了一下之前的梳理。虽然也有一年多了，但可以说4月份之后的才有一点意义。更早之前的没啥价值了。

后续不同的专题都用不同的issue来跟踪学习。这里还是继续看buffer的问题，现在cs两端 还没建立完毕。创建完成之后才会发送数据。

[dbgp]

上面楼层提到Server端初始化，注册的过程没错。

然后通过Selector机制完成server端的建立。

这里有个前提是当有client端接入时，才会触发这个OP_ACCEPT事件，即——Selector将对应的SelectionKey添加到selectedKeys集合中，表示有一个新的连接可以被接受。

---

• client端建立连接之后会发送一个SessionInit请求
• 这个请求会被server端的调度器以一个SessionInitTask的任务进行处理。

这个任务依然由独立的调度器线程处理，首先会会创建出来ServerSession，然后发送给客户端一个 SessionInitAck响应

[dbgp]

在这个发送和响应的过程中，才真正涉及到本帖主题：

调度线程中的两个专属 buffer的使用。

这包含两个含义——Client 端的调度线程和 Server 端的调度线程。

[dbgp]

重新梳理一下client端创建连接的过程，有几个关键点：

1. 最外层connect方法返回的类型是 Future<JdbcConnection>，这里的 get方法会等待有结果的返回
2. 多个多层的CallBack的onComplete方法中的 handler最重要的作用就是定义并传递这个completeHandler，执行这个方法时，几乎都没有获取到异步的返回值。关键是当获取到返回值时，会再次调用这个 completeHandler.handle(asyncResult);方法
3. 客户端的回调方法设置会通过网络传递，当与服务端建立连接之后，才会设置异步的结果，然后值设置结果的方法里执行回调（也就是上一步的动作）

[dbgp]

正式进入buffer使用场景。首先需要掌握ByteBuffer的概念和用法——javaNIO中的核心概念ByteBuffer

然后看 “调度线程各分配两个专属 buffer”，使用这两个buffer来进行数据传输的维护动作——

DataBuffer

• 最底层的ByteBuffer对象封装中 DataBuffer对象中，属于 common模块中的 db包的内容
• 对象中包含一个 DataHandler 对象，在文件存储管理中会用到，这里网络传输场景下，这个对象时null
• 分配ByteBuffer的方式：direct——直接存储中OS中，不在JVM的堆内存中，不受gc管理，适用于网络IO场景

NetBuffer

针对网络传输场景，又封装了一个NetBuffer对象，调度器非配的专属buffer属于这个对象类型。属于comon模块中的net包

NetBuffer inBuffer = scheduler.getInputBuffer();
NetBuffer outBuffer = scheduler.getOutputBuffer();

TransferConnection

调度器非配的两个buffer，最终给到了这个抽象类，组成了 传输到 inputstream和outputstream
输入流相当于收到数据的流；输出流相当于发送数据的流，发送数据还需要一个channel概念，封装为WritableChannel概念

无论是c/s的那一段，都存在这样的情况。使用不同的子类进行扩展——

• client端使用的是 TcpClientConnection对象，也就是二楼里建立连接时构造的对象

 NetBuffer inBuffer = scheduler.getInputBuffer();
NetBuffer outBuffer = scheduler.getOutputBuffer();
conn = new TcpClientConnection(writableChannel, this, attachment.maxSharedSize, inBuffer,
                        outBuffer);

• server端使用的是TcpServerConnection对象，也就是五楼 完成 server.accept(scheduler)的动作时创建。由TcpServer对象调用 createConnection方法。server端除了有一个channel之后，自然还需要多一个 TcpServer对象

[dbgp]

跟踪一遍client端完整的“send”方法可以看出来buffer在这个场景下是如何管理的。核心代码只有以下三行：

//这里的out就是 TransferOutputStream
out.writeRequestHeader(this, packetId, packet.getType());
packet.encode(out, getProtocolVersion());
out.flush();

为实现这个效果，大概可以从“分配、写入、发送、回收”四个阶段理解。

分配

参加上面的 TransferConnection 这个对象中包含了使用buffer（NetBuffer）构造的 TransferInputStream 和 TransferOutputStream

分别实现自定义的 NetInputStream和 NetOutputStream接口
具体的实现分别由 DataInputStream 和 DataOutputStream构成，这样就可以使用DataStream标准库的各种读写方法。
DataInputStream 和 DataOutputStream 分别由自定义的 NetBufferInputStream 和 GlobalNetBufferOutputStream 作为底层的支持。

基于ByteBuffer的读写能力，可以使用相同类型的对象实现自定义的读写stream 。

• NetBufferInputStream 扩展了标准库的 InputStream， 重写的 available、read、close、reset方法都可以利用NetBuffer（ByteBuffer）进行实现
• GlobalNetBufferOutputStream 扩展了标准库的 OutputStream，重写了 wirte, flush方法

写入

写入请求头。

public TransferOutputStream writeRequestHeader(Session session, int packetId, PacketType packetType)
            throws IOException {
        this.session = session;
        // 1. 先写标记位，同时记录底层byteBuffer的position，表示开始写的起始位置。
        // 2. 同时写入协议包长度占位符 “ 协议包头占4个字节，最后flush时再回填” 
        startWrite(Session.STATUS_OK);
        //3. 写入具体协议内容：类型，包id，包类型，sessionId （包id和sessionId都由ClientSession的TcpConnection对象统一维护）
        writeByte(REQUEST).writeInt(packetId).writeInt(packetType.value).writeInt(session.getId());
        return this;
    }

写入具体的Packet。所有类型的Packet提供了统一的encode和decode方法。
encode方法向NetOutputStream中写入数据；
decode方法从NetInputStream中读取数据
这两个方法都在对应的Packet中实现。其中decode方法使用decoder静态类实现，并且在程序启动时完成注册（根据packet类型添加到数组中，需要调用decode方法时，可以根据packet类型快速找到对应的decoder）

发送

• 第一，先回填协议包长度，根据buffe.position方法 减去写入时的起始位置再减4，得到不同packet包的具体长度大小

public void flush() throws IOException {
            int pos = buffer.position();
            int length = buffer.position() - channel.startPos - 4;
            // 第一，先回填协议包长度，根据buffe.position方法 减去写入时的起始位置再减4，得到不同packet包的具体长度大小
            writePacketLength(channel.startPos, length);
            channel.flush(channel.startPos, pos);
        }

        // 按java.io.DataInputStream.readInt()的格式写
        private void writePacketLength(int pos, int v) {
            buffer.setByte(pos, (byte) ((v >>> 24) & 0xFF));
            buffer.setByte(pos + 1, (byte) ((v >>> 16) & 0xFF));
            buffer.setByte(pos + 2, (byte) ((v >>> 8) & 0xFF));
            buffer.setByte(pos + 3, (byte) (v & 0xFF));
        }

• 第二， 构造WritableBuffer。

从底层的ByteBuffer中记录原始的 position和limit（表示已经写入了数据的部分），然后slice出来一个新的ByteBuffer（只包含了此次packet的信息），构造的WritableBuffer对象包含了原始的底层buffer，称为parent；以及slice后的代表了当前packet的buffer

public void flush(int start, int end) {
            writableChannel.write(buffer.createWritableBuffer(start, end));
            written = true;
        }

// 从底层的ByteBuffer中记录原始的 position和limit（表示已经写入了数据的部分），然后slice出来一个新的ByteBuffer（只包含了此次packet的信息）
public ByteBuffer sliceByteBuffer(int start, int end) {
        int pos = buff.position();
        int limit = buff.limit();
        buff.position(start);
        buff.limit(end);
        ByteBuffer newBuffer = buff.slice();
        buff.position(pos);
        buff.limit(limit);
        return newBuffer;
    }

• 第三，调用NioEventLoop中的write方法，从上一步的WritableBuffer中取出此次packet对应的byteBuffer，从对应的SockeChannel中发送出去

回收

回收动作也是中flush方法中完成。recycleBuffer(buffer); 从这个 WritableBuffer中可以获取到parent，也就是最开始调度器分配的 NetBuffer对象，执行其底层对应的 ByteBuffer的clear方法（将数据清除，position恢复到0，limit恢复到capacity）

---

下面两张图就是client端发送 SessionInit协议包时的时间场景。

[dbgp]

从server端看如何利用buffer处理数据包的（依然以SessionInit包为例）

响应数据包的拆分逻辑——

判断

判断此次是否读取到了完整的数据。

连接建立之后，收到client端的数据包之后，server端自然需要读取数据。server端端NioEventLoop中read方法，读取用的buffer来自调度器的分配。可以看到从channel中读取了189字节的数据到bytebuffer中（写到了buffer中），然后调用flip方法，从写模式变为读模式。

处理数据

从attachment中获取到连接对象，进入handle方法

区分处理方式

因为c/s两端都有TransferConnection，需要知道这一次是处理请求还是处理响应

本次以SessionInit包为例，这时需要了解上面发送packet时的协议细节——依次读取协议头的内容：

• int 从buffer中读int值，得到此次包的长度。
• byte 请求类型
• int 包id
• int 包类型的值
• sessionId

读完协议头之后，处理协议包内容的讨论都是一致的：

• 根据packetTye获取decoder，
• 调用decode方法
• 构造具体的异步处理方法，添加到调度器中

SessinInit包：到来时比较特殊，还没有Session信息，单独处理SessionInit包：向调度器添加 SessionInitTask；执行时会创建ServerSession，对应这一个连接查询信息；然后发送给client端SessionInitAck包——SessionInitTask的run方法；
其他packet类型：会优先尝试是否可以直接执行，不能执行时，向调度器添加PacketHandleTask任务；调度执行时，执行对应packet注册的handler方法，执行结果时一个对应的packet包，然后将这个响应包再通过通道发送出去——PacketHandleTask的run方法。