MYDB项目笔记

public interface TransactionManager {
    long begin(); //开启事务
    void commit(long xid);  //提交事务
    void abort(long xid);   //撤销或回滚事务
    boolean isActive(long xid); //查询一个事务的状态是否正在运行
    boolean isCommitted(long xid);  //查询一个事务的状态是否已经提交
    boolean isAborted(long xid);    //查询一个事务的状态是否撤销或回滚
    void close();   //关闭事务
}

public class TransactionManagerImpl implements TransactionManager {

    // XID文件头长度
    static final int LEN_XID_HEADER_LENGTH = 8;
    // 每个事务的占用长度
    private static final int XID_FIELD_SIZE = 1;

    // 事务的三种状态
    private static final byte FIELD_TRAN_ACTIVE = 0;//正在进行，尚未结束
    private static final byte FIELD_TRAN_COMMITTED = 1;//已提交
    private static final byte FIELD_TRAN_ABORTED = 2;//已撤销(回滚)

    // 超级事务，永远为commited状态
    public static final long SUPER_XID = 0;

    // XID文件后缀
    static final String XID_SUFFIX = ".xid";
    
}

/**
 * 检查XID文件是否合法
 * 读取XID_FILE_HEADER中的xidcounter，根据它计算文件的理论长度，对比实际长度
 */
private void checkXIDCounter() {
    // 初始化文件长度为0
    long fileLen = 0;
    try {
        // 获取文件的长度，RandomAccessFile在构造函数中赋值
        fileLen = file.length();
    } catch (IOException e1) {
        // 如果出现异常，抛出BadXIDFileException错误
        Panic.panic(Error.BadXIDFileException);
    }

    // 如果文件长度小于XID头部长度，抛出BadXIDFileException错误
    if (fileLen < LEN_XID_HEADER_LENGTH) {
        Panic.panic(Error.BadXIDFileException);
    }

    // 分配一个长度为XID头部长度的ByteBuffer
    ByteBuffer buf = ByteBuffer.allocate(LEN_XID_HEADER_LENGTH);
    try {
        // 将文件通道的位置设置为0
        fc.position(0);
        // 从文件通道读取数据到ByteBuffer
        fc.read(buf);
    } catch (IOException e) {
        // 如果出现异常，抛出错误
        Panic.panic(e);
    }
    // 将ByteBuffer的内容解析为长整型，作为xidCounter
    this.xidCounter = Parser.parseLong(buf.array()); 
    // 计算xidCounter+1对应的XID位置
    long end = getXidPosition(this.xidCounter + 1);
    // 如果计算出的XID位置与文件长度不符，抛出BadXIDFileException错误
    if (end != fileLen) {
        Panic.panic(Error.BadXIDFileException);
    }
}

/**
 * 将数组前八位转换成长整数
 * @param buf 需要转换的字节数组
 * @return 转换后的数据
 */
public static long parseLong(byte[] buf) {
    ByteBuffer buffer = ByteBuffer.wrap(buf, 0, 8);
    return buffer.getLong();
}

@Test
public void testBufferGetLong(){
    
    // 创建一个包含8个字节的字节数组
    //因为long 在Java中占用8个字节，每个字节占用8位，一下数组可以转换成一个long数字
    // 00000000 00000000 00000000 00000000 00000000 00000000 00001010 00000001
    // 1010 00000001 --> 2561
    byte[] byteArray = new byte[]{0, 0, 0, 0, 0, 0, 10, 1};
    // 使用ByteBuffer.wrap方法将字节数组包装为一个ByteBuffer对象
    ByteBuffer buffer = ByteBuffer.wrap(byteArray);

    // 使用getLong方法从ByteBuffer中读取一个长整型数
    long longValue = buffer.getLong();

    // 输出读取的长整型数
    System.out.println("The long value is: " + longValue);
}

// 根据事务xid取得其在xid文件中对应的位置
private long getXidPosition(long xid) {
    return LEN_XID_HEADER_LENGTH + (xid - 1) * XID_FIELD_SIZE;
}

// 开始一个事务，并返回XID
public long begin() {
    // 锁定计数器，防止并发问题
    counterLock.lock();
    try {
        // xidCounter是当前事务的计数器，每开始一个新的事务，就将其加1
        long xid = xidCounter + 1;
        // 调用updateXID方法，将新的事务ID和事务状态（这里是活动状态）写入到XID文件中
        updateXID(xid, FIELD_TRAN_ACTIVE);
        // 调用incrXIDCounter方法，将事务计数器加1，并更新XID文件的头部信息
        incrXIDCounter();
        // 返回新的事务ID
        return xid;
    } finally {
        // 释放锁
        counterLock.unlock();
    }
}

// 更新xid事务的状态为status
private void updateXID(long xid, byte status) {
    // 获取事务xid在xid文件中对应的位置
    long offset = getXidPosition(xid);
    // 创建一个长度为XID_FIELD_SIZE的字节数组
    byte[] tmp = new byte[XID_FIELD_SIZE];
    // 将事务状态设置为status
    tmp[0] = status;
    // 使用字节数组创建一个ByteBuffer
    ByteBuffer buf = ByteBuffer.wrap(tmp);
    try {
        // 将文件通道的位置设置为offset
        fc.position(offset);
        // 将ByteBuffer中的数据写入到文件通道
        fc.write(buf);
    } catch (IOException e) {
        // 如果出现异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
    try {
        // 强制将文件通道中的所有未写入的数据写入到磁盘
        fc.force(false);
    } catch (IOException e) {
        // 如果出现异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
}

// 将XID加一，并更新XID Header
private void incrXIDCounter() {
    // 事务总数加一
    xidCounter++;
    // 将新的事务总数转换为字节数组，并用ByteBuffer包装
    ByteBuffer buf = ByteBuffer.wrap(Parser.long2Byte(xidCounter));
    try {
        // 将文件通道的位置设置为0，即文件的开始位置
        fc.position(0);
        // 将ByteBuffer中的数据写入到文件通道，即更新了XID文件的头部信息
        fc.write(buf);
    } catch (IOException e) {
        // 如果出现异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
    try {
        // 强制将文件通道中的所有未写入的数据写入到磁盘
        fc.force(false);
    } catch (IOException e) {
        // 如果出现异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
}


/**
 * 将长整型值写入到字节缓冲区，将其转成为8字节的二进制形式，然后将这个8个字节写入到字节缓冲区
 * @param value
 * @return
 */
public static byte[] long2Byte(long value) {
    return ByteBuffer.allocate(Long.SIZE / Byte.SIZE).putLong(value).array();
}

// 定义一个方法，接收一个事务ID（xid）和一个状态（status）作为参数
private boolean checkXID(long xid, byte status) {
    // 计算事务ID在XID文件中的位置
    long offset = getXidPosition(xid);
    // 创建一个新的字节缓冲区（ByteBuffer），长度为XID_FIELD_SIZE
    ByteBuffer buf = ByteBuffer.wrap(new byte[XID_FIELD_SIZE]);
    try {
        // 将文件通道的位置设置为offset
        fc.position(offset);
        // 从文件通道读取数据到字节缓冲区
        fc.read(buf);
    } catch (IOException e) {
        // 如果出现异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
    // 检查字节缓冲区的第一个字节是否等于给定的状态
    // 如果等于，返回true，否则返回false
    return buf.array()[0] == status;
}

/**
 * 当资源不在缓存时的获取行为
 */
protected abstract T getForCache(long key) throws Exception;
/**
 * 当资源被驱逐时的写回行为
 */
protected abstract void releaseForCache(T obj);

//从缓存中获取资源
protected T get(long key) throws Exception {
    // 循环直到获取资源
    while (true) {
        // 获取锁
        lock.lock();
        if (getting.containsKey(key)) {
            // 如果其他线程正在获取这个资源，那么当前线程将等待一毫秒然后继续循环
            lock.unlock();
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
                continue;
            }
            continue;
        }

        if (cache.containsKey(key)) {
            // 如果资源已经在缓存中，直接返回资源，并增加引用计数
            T obj = cache.get(key);
            references.put(key, references.get(key) + 1);
            lock.unlock();
            return obj;
        }

        // 如果资源不在缓存中，尝试获取资源。如果缓存已满，抛出异常
        if (maxResource > 0 && count == maxResource) {
            lock.unlock();
            throw Error.CacheFullException;
        }
        count++;
        getting.put(key, true);
        lock.unlock();
        break;
    }

    // 尝试获取资源
    T obj = null;
    try {
        obj = getForCache(key);
    } catch (Exception e) {
        lock.lock();
        count--;
        getting.remove(key);
        lock.unlock();
        throw e;
    }

    // 将获取到的资源添加到缓存中，并设置引用计数为1
    lock.lock();
    getting.remove(key);
    cache.put(key, obj);
    references.put(key, 1);
    lock.unlock();

    return obj;
}

/**
 * 强行释放一个缓存
 */
protected void release(long key) {
    lock.lock(); // 获取锁
    try {
        int ref = references.get(key) - 1; // 获取资源的引用计数并减一
        if (ref == 0) { // 如果引用计数为0
            T obj = cache.get(key); // 从缓存中获取资源
            releaseForCache(obj); // 处理资源的释放
            references.remove(key); // 从引用计数的映射中移除资源
            cache.remove(key); // 从缓存中移除资源
            count--; // 将缓存中的资源计数减一
        } else { // 如果引用计数不为0
            references.put(key, ref); // 更新资源的引用计数
        }
    } finally {
        lock.unlock(); // 释放锁
    }
}

/**
 * 关闭缓存，写回所有资源
 */
protected void close() {
    lock.lock();
    try {
        //获取所有资源key
        Set<Long> keys = cache.keySet();
        for (long key : keys) {
            //获取缓存
            T obj = cache.get(key);
            //释放缓存
            releaseForCache(obj);
            //引用计数移除缓存
            references.remove(key);
            //实际缓存移除缓存
            cache.remove(key);
        }
    } finally {
        //释放锁
        lock.unlock();
    }
}

public class SubArray {
    public byte[] raw;
    public int start;
    public int end;

    public SubArray(byte[] raw, int start, int end) {
        this.raw = raw;
        this.start = start;
        this.end = end;
    }
}

@Test
public void testSubArray(){
    //创建一个1~10的数组
    byte[] subArray = new byte[10];
    for (int i = 0; i < subArray.length; i++) {
        subArray[i] = (byte) (i+1);
    }
    //创建两个SubArray
    SubArray sub1 = new SubArray(subArray,3,7);
    SubArray sub2 = new SubArray(subArray,6,9);

    //修改共享数组数据
    sub1.raw[4] = (byte)44;

    //打印原始数组
    System.out.println("Original Array: ");
    printArray(subArray);

    //打印共享数组
    System.out.println("SubArray1: ");
    printSubArray(sub1);
    System.out.println("SubArray2: ");
    printSubArray(sub2);
}

private void printArray(byte[] array){
    System.out.println(Arrays.toString(array));
}

private void printSubArray(SubArray subArray){
    for (int i = subArray.start; i <= subArray.end; i++) {
        System.out.print(subArray.raw[i] + "\t");
    }
    System.out.println();
}


-------------------------演示结果----------------------------
Original Array: 
[1, 2, 3, 4, 44, 6, 7, 8, 9, 10]
SubArray1: 
4	44	6	7	8	
SubArray2: 
7	8	9	10	

public class PageImpl implements Page {
    private int pageNumber;
    private byte[] data;
    private boolean dirty;
    private Lock lock;
    private PageCache pc;
}

/**
 * 根据pageNumber从数据库文件中读取页数据，并包裹成Page
 */
@Override
protected Page getForCache(long key) throws Exception {
    // 将key转换为页码
    int pgno = (int) key;
    // 计算页码对应的偏移量
    long offset = PageCacheImpl.pageOffset(pgno);

    // 分配一个大小为PAGE_SIZE的ByteBuffer
    ByteBuffer buf = ByteBuffer.allocate(PAGE_SIZE);
    // 锁定文件，确保线程安全
    fileLock.lock();
    try {
        // 设置文件通道的位置为计算出的偏移量
        fc.position(offset);
        // 从文件通道读取数据到ByteBuffer
        fc.read(buf);
    } catch (IOException e) {
        // 如果发生异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
    // 无论是否发生异常，都要解锁
    fileLock.unlock();
    // 使用读取到的数据、页码和当前对象创建一个新的PageImpl对象并返回
    return new PageImpl(pgno, buf.array(), this);
}

public PageImpl(int pageNumber, byte[] data, PageCache pc) {
    this.pageNumber = pageNumber; // 设置页面的页号
    this.data = data; // 设置页面实际包含的字节数据
    this.pc = pc; // 设置页面缓存
    lock = new ReentrantLock(); // 初始化一个新的可重入锁
}

@Override
protected void releaseForCache(Page pg) {
    if (pg.isDirty()) {
        flush(pg);
        pg.setDirty(false);
    }
}

 private void flush(Page pg) {
    int pgno = pg.getPageNumber(); // 获取Page的页码
    long offset = pageOffset(pgno); // 计算Page在文件中的偏移量

    fileLock.lock(); // 加锁，确保线程安全
    try {
        ByteBuffer buf = ByteBuffer.wrap(pg.getData()); // 将Page的数据包装成ByteBuffer
        fc.position(offset); // 设置文件通道的位置
        fc.write(buf); // 将数据写入到文件中
        fc.force(false); // 强制将数据从操作系统的缓存刷新到磁盘
    } catch (IOException e) {
        Panic.panic(e); // 如果发生异常，调用Panic.panic方法处理
    } finally {
        fileLock.unlock(); // 最后，无论是否发生异常，都要解锁
    }
}

// 将raw插入pg中的offset位置，并将pg的offset设置为较大的offset
public static void recoverInsert(Page pg, byte[] raw, short offset) {
    pg.setDirty(true); // 将pg的dirty标志设置为true，表示pg的数据已经被修改
    System.arraycopy(raw, 0, pg.getData(), offset, raw.length); // 将raw的数据复制到pg的数据中的offset位置

    short rawFSO = getFSO(pg.getData()); // 获取pg的当前空闲空间偏移量
    if (rawFSO < offset + raw.length) { // 如果当前的空闲空间偏移量小于offset + raw.length
        setFSO(pg.getData(), (short) (offset + raw.length)); // 将pg的空闲空间偏移量设置为offset + raw.length
    }
}

// 将raw插入pg中的offset位置，不更新update
public static void recoverUpdate(Page pg, byte[] raw, short offset) {
    pg.setDirty(true); // 将pg的dirty标志设置为true，表示pg的数据已经被修改
    System.arraycopy(raw, 0, pg.getData(), offset, raw.length); // 将raw的数据复制到pg的数据中的offset位置
}

// 设置"ValidCheck"为打开状态
public static void setVcOpen(Page pg) {
    pg.setDirty(true);
    setVcOpen(pg.getData());
}

private static void setVcOpen(byte[] raw) {
    // 随机生成8字节的数据，并拷贝到第一页的 100~107 字节
    System.arraycopy(RandomUtil.randomBytes(LEN_VC), 0, raw, OF_VC, LEN_VC);
}

// 设置"ValidCheck"为关闭状态
public static void setVcClose(Page pg) {
    pg.setDirty(true);
    setVcClose(pg.getData());
}

// 设置"ValidCheck"为关闭状态
private static void setVcClose(byte[] raw) {
    // 将"ValidCheck"设置为关闭状态
    // 通过复制raw数组中的一部分元素来实现
    // 具体来说，就是将raw数组中从OF_VC开始的LEN_VC个元素复制到raw数组中从OF_VC+LEN_VC开始的位置
    // 即 100~107 拷贝到 108~115
    System.arraycopy(raw, OF_VC, raw, OF_VC + LEN_VC, LEN_VC);
}

// 检查"ValidCheck"是否有效
public static boolean checkVc(Page pg) {
    return checkVc(pg.getData());
}

// 检查"ValidCheck"是否有效
private static boolean checkVc(byte[] raw) {
    // 比较 100~107 和 108~115 处字节是否相等
    return Arrays.equals(Arrays.copyOfRange(raw, OF_VC, OF_VC + LEN_VC), Arrays.copyOfRange(raw, OF_VC + LEN_VC, OF_VC + 2 * LEN_VC));
}

// 设置空闲空间偏移量
private static void setFSO(byte[] raw, short ofData) {
    // 将空闲空间偏移量的值复制到字节数组的指定位置
    System.arraycopy(Parser.short2Byte(ofData), 0, raw, OF_FREE, OF_DATA); 
}

// 获取pg的空闲空间偏移量
public static short getFSO(Page pg) {
    return getFSO(pg.getData()); // 返回pg的数据的空闲空间偏移量
}

// 获取空闲空间偏移量
private static short getFSO(byte[] raw) {
    return Parser.parseShort(Arrays.copyOfRange(raw, 0, 2)); // 返回字节数组的前两个字节表示的短整数值
}


// 获取页面的空闲空间大小
public static int getFreeSpace(Page pg) {
    return PageCache.PAGE_SIZE - (int) getFSO(pg.getData()); // 返回页面的空闲空间大小
}

// 将raw插入pg中，返回插入位置
public static short insert(Page pg, byte[] raw) {
    pg.setDirty(true); // 将pg的dirty标志设置为true，表示pg的数据已经被修改
    short offset = getFSO(pg.getData()); // 获取pg的空闲空间偏移量
    System.arraycopy(raw, 0, pg.getData(), offset, raw.length); // 将raw的数据复制到pg的数据中的offset位置
    setFSO(pg.getData(), (short) (offset + raw.length)); // 更新pg的空闲空间偏移量
    return offset; // 返回插入位置
}

private int calChecksum(int xCheck, byte[] log) {
    for (byte b : log) {
        xCheck = xCheck * SEED + b; //seed = 13331
    }
    return xCheck;
}

// class: src/main/java/top/guoziyang/mydb/backend/dm/logger/Logger.java
ByteBuffer buf = ByteBuffer.wrap(Parser.int2Byte(0)); // 将0转换成四字节的数字
try {
    fc.position(0);
    fc.write(buf); //将其写入到文件
    fc.force(false);
} catch (IOException e) {
    Panic.panic(e);
}

void init() {
    long size = 0;
    try {
        size = file.length(); //读取文件大小
    } catch (IOException e) {
        Panic.panic(e);
    }
    if (size < 4) { //若文件大小小于4，证明日志文件创建出现问题
        Panic.panic(Error.BadLogFileException);
    }

    ByteBuffer raw = ByteBuffer.allocate(4); //创建一个容量为4的ByteBuffer
    try {
        fc.position(0);
        fc.read(raw); //读取四字节大小的内容
    } catch (IOException e) {
        Panic.panic(e);
    }
    int xChecksum = Parser.parseInt(raw.array()); //将其转换成int整数
    this.fileSize = size;
    this.xChecksum = xChecksum; //赋值给当前对象

    checkAndRemoveTail(); //检查是否需要去除BadTail
}

/**
 * 获取下一条日志
 */
private byte[] internNext() {
    // 检查当前位置是否已经超过了文件的大小，如果超过了，说明没有更多的日志可以读取，返回 null
    if (position + OF_DATA >= fileSize) {
        return null;
    }

    // 创建一个大小为 4 的 ByteBuffer，用于读取日志的大小
    ByteBuffer tmp = ByteBuffer.allocate(4);
    try {
        // 将文件通道的位置设置为当前位置
        fc.position(position);
        // 从文件通道中读取 4 个字节的数据到 ByteBuffer 中，即Size日志文件的大小
        fc.read(tmp);
    } catch (IOException e) {
        // 如果发生 IO 异常，调用 Panic.panic 方法处理异常
        Panic.panic(e);
    }

    // 使用 Parser.parseInt 方法将读取到的 4 个字节的数据转换为 int 类型，得到日志的大小
    int size = Parser.parseInt(tmp.array());

    // 检查当前位置加上日志的大小是否超过了文件的大小，如果超过了，说明日志不完整，返回 null
    if (position + size + OF_DATA > fileSize) {
        return null;
    }

    // 创建一个大小为 OF_DATA + size 的 ByteBuffer，用于读取完整的日志
    ByteBuffer buf = ByteBuffer.allocate(OF_DATA + size);
    try {
        // 将文件通道的位置设置为当前位置
        fc.position(position);
        // 从文件通道中读取 OF_DATA + size 个字节的数据到 ByteBuffer 中
        // 读取整条日志 [Size][Checksum][Data]
        fc.read(buf);
    } catch (IOException e) {
        // 如果发生 IO 异常，调用 Panic.panic 方法处理异常
        Panic.panic(e);
    }

    // 将 ByteBuffer 中的数据转换为字节数组
    byte[] log = buf.array();

    // 计算日志数据的校验和
    int checkSum1 = calChecksum(0, Arrays.copyOfRange(log, OF_DATA, log.length));
    // 从日志中读取校验和
    int checkSum2 = Parser.parseInt(Arrays.copyOfRange(log, OF_CHECKSUM, OF_DATA));

    // 比较计算得到的校验和和日志中的校验和，如果不相等，说明日志已经被破坏，返回 null
    if (checkSum1 != checkSum2) {
        return null;
    }

    // 更新当前位置
    position += log.length;

    // 返回读取到的日志
    return log;
}

@Override
public void log(byte[] data) {
    // 解析成一条完整的log日志
    byte[] log = wrapLog(data);
    ByteBuffer buf = ByteBuffer.wrap(log);
    lock.lock();
    try {
        //写入到指定位置
        fc.position(fc.size());
        fc.write(buf);
    } catch (IOException e) {
        Panic.panic(e);
    } finally {
        lock.unlock();
    }
    // 更新总校验值
    updateXChecksum(log);
}

/**
 * 更新总校验值
 */
private void updateXChecksum(byte[] log) {
    //计算总校验值
    this.xChecksum = calChecksum(this.xChecksum, log);
    try {
        fc.position(0);
        fc.write(ByteBuffer.wrap(Parser.int2Byte(xChecksum)));
        fc.force(false);
    } catch (IOException e) {
        Panic.panic(e);
    }
}

/**
* 将数据解析成完整log
*/
private byte[] wrapLog(byte[] data) {
    // 使用 calChecksum 方法计算数据的校验和，然后将校验和转换为字节数组
    byte[] checksum = Parser.int2Byte(calChecksum(0, data));
    // 将数据的长度转换为字节数组
    byte[] size = Parser.int2Byte(data.length);
    // 使用 Bytes.concat 方法将 size、checksum 和 data 连接成一个新的字节数组，然后返回这个字节数组
    return Bytes.concat(size, checksum, data);
}

public class Recover {
    private static final byte LOG_TYPE_INSERT = 0;
    private static final byte LOG_TYPE_UPDATE = 1;

    // updateLog:
    // [LogType] [XID] [UID] [OldRaw] [NewRaw]
    
    // insertLog:
    // [LogType] [XID] [Pgno] [Offset] [Raw]
}

private static void redoTranscations(TransactionManager tm, Logger lg, PageCache pc) {
    // 重置日志文件的读取位置到开始
    lg.rewind();
    // 循环读取日志文件中的所有日志记录
    while (true) {
        // 读取下一条日志记录
        byte[] log = lg.next();
        // 如果读取到的日志记录为空，表示已经读取到日志文件的末尾，跳出循环
        if (log == null) break;
        // 判断日志记录的类型
        if (isInsertLog(log)) {
            // 如果是插入日志，解析日志记录，获取插入日志信息
            InsertLogInfo li = parseInsertLog(log);
            // 获取事务ID
            long xid = li.xid;
            // 如果当前事务已经提交，进行重做操作
            if (!tm.isActive(xid)) {
                doInsertLog(pc, log, REDO);
            }
        } else {
            // 如果是更新日志，解析日志记录，获取更新日志信息
            UpdateLogInfo xi = parseUpdateLog(log);
            // 获取事务ID
            long xid = xi.xid;
            // 如果当前事务已经提交，进行重做操作
            if (!tm.isActive(xid)) {
                doUpdateLog(pc, log, REDO);
            }
        }
    }
}

private static void undoTranscations(TransactionManager tm, Logger lg, PageCache pc) {
    // 创建一个用于存储日志的映射，键为事务ID，值为日志列表
    Map<Long, List<byte[]>> logCache = new HashMap<>();
    // 将日志文件的读取位置重置到开始
    lg.rewind();
    // 循环读取日志文件中的所有日志记录
    while (true) {
        // 读取下一条日志记录
        byte[] log = lg.next();
        // 如果读取到的日志记录为空，表示已经读取到日志文件的末尾，跳出循环
        if (log == null) break;
        // 判断日志记录的类型
        if (isInsertLog(log)) {
            // 如果是插入日志，解析日志记录，获取插入日志信息
            InsertLogInfo li = parseInsertLog(log);
            // 获取事务ID
            long xid = li.xid;
            // 如果当前事务仍然活跃，将日志记录添加到对应的日志列表中
            if (tm.isActive(xid)) {
                if (!logCache.containsKey(xid)) {
                    logCache.put(xid, new ArrayList<>());
                }
                logCache.get(xid).add(log);
            }
        } else {
            // 如果是更新日志，解析日志记录，获取更新日志信息
            UpdateLogInfo xi = parseUpdateLog(log);
            // 获取事务ID
            long xid = xi.xid;
            // 如果当前事务仍然活跃，将日志记录添加到对应的日志列表中
            if (tm.isActive(xid)) {
                if (!logCache.containsKey(xid)) {
                    logCache.put(xid, new ArrayList<>());
                }
                // 将事务id对应的log添加到集合中
                logCache.get(xid).add(log);
            }
        }
    }

    // 对所有活跃的事务的日志进行倒序撤销
    for (Entry<Long, List<byte[]>> entry : logCache.entrySet()) {
        List<byte[]> logs = entry.getValue();
        for (int i = logs.size() - 1; i >= 0; i--) {
            byte[] log = logs.get(i);
            // 判断日志记录的类型
            if (isInsertLog(log)) {
                // 如果是插入日志，进行撤销插入操作
                doInsertLog(pc, log, UNDO);
            } else {
                // 如果是更新日志，进行撤销更新操作
                doUpdateLog(pc, log, UNDO);
            }
        }
        // 中止当前事务
        tm.abort(entry.getKey());
    }
}

private static void doInsertLog(PageCache pc, byte[] log, int flag) {
    // 解析日志记录，获取插入日志信息
    InsertLogInfo li = parseInsertLog(log);
    Page pg = null;
    try {
        // 根据页码从页面缓存中获取页面，即AbstractCache.get()方法
        pg = pc.getPage(li.pgno);
    } catch (Exception e) {
        // 如果发生异常，调用Panic.panic方法处理
        Panic.panic(e);
    }
    try {
        // 如果标志位为UNDO，将数据项设置为无效
        if (flag == UNDO) {
            DataItem.setDataItemRawInvalid(li.raw);
        }
        // 在指定的页面和偏移量处插入数据
        PageX.recoverInsert(pg, li.raw, li.offset);
    } finally {
        // 无论是否发生异常，都要释放页面,即AbstractCache.release() 方法
        pg.release();
    }
}

private static void doUpdateLog(PageCache pc, byte[] log, int flag) {
    int pgno; // 用于存储页面编号
    short offset; // 用于存储偏移量
    byte[] raw; // 用于存储原始数据

    // 根据标志位判断是进行重做操作还是撤销操作
    if (flag == REDO) {
        // 如果是重做操作，解析日志记录，获取更新日志信息，主要获取新数据
        UpdateLogInfo xi = parseUpdateLog(log);
        pgno = xi.pgno;
        offset = xi.offset;
        raw = xi.newRaw;
    } else {
        // 如果是撤销操作，解析日志记录，获取更新日志信息，主要获取旧数据
        UpdateLogInfo xi = parseUpdateLog(log);
        pgno = xi.pgno;
        offset = xi.offset;
        raw = xi.oldRaw;
    }

    Page pg = null; // 用于存储获取到的页面
    try {
        // 尝试从页面缓存中获取指定页码的页面
        pg = pc.getPage(pgno);
    } catch (Exception e) {
        // 如果获取页面过程中发生异常，调用Panic.panic方法进行处理
        Panic.panic(e);
    }

    try {
        // 在指定的页面和偏移量处插入解析出的数据, 数据页缓存讲解了该方法
        PageX.recoverUpdate(pg, raw, offset);
    } finally {
        // 无论是否发生异常，都要释放页面
        pg.release();
    }
}

public class PageIndex {
    // 将一页划成40个区间，每个区间的大小为204字节
    private final List<PageInfo>[] lists;
    // 每个区间的大小为204字节
    private static final int INTERVALS_NO = 40;
    private static final int THRESHOLD = PageCache.PAGE_SIZE / INTERVALS_NO; //204

    private Lock lock;
}

/**
 * 根据给定的空间大小选择一个 PageInfo 对象。
 *
 * @param spaceSize 需要的空间大小
 * @return 一个 PageInfo 对象，其空闲空间大于或等于给定的空间大小。如果没有找到合适的 PageInfo，返回 null。
 */
public PageInfo select(int spaceSize) {
    lock.lock(); // 获取锁，确保线程安全
    try {
        int number = spaceSize / THRESHOLD; // 计算需要的空间大小对应的区间编号
        // 此处+1主要为了向上取整
        /*
            1、假需要存储的字节大小为5168，此时计算出来的区间号是25，但是25*204=5100显然是不满足条件的
            2、此时向上取整找到 26，而26*204=5304，是满足插入条件的
         */
        if (number < INTERVALS_NO) number++; // 如果计算出的区间编号小于总的区间数，编号加一
        while (number <= INTERVALS_NO) { // 从计算出的区间编号开始，向上寻找合适的 PageInfo
            if (lists[number].size() == 0) { // 如果当前区间没有 PageInfo，继续查找下一个区间
                number++;
                continue;
            }
            return lists[number].remove(0); // 如果当前区间有 PageInfo，返回第一个 PageInfo，并从列表中移除
        }
        return null; // 如果没有找到合适的 PageInfo，返回 null
    } finally {
        lock.unlock(); // 释放锁
    }
}

/**
 * 根据给定的页面编号和空闲空间大小添加一个 PageInfo 对象。
 * @param pgno      页面编号
 * @param freeSpace 页面的空闲空间大小
 */
public void add(int pgno, int freeSpace) {
    lock.lock(); // 获取锁，确保线程安全
    try {
        int number = freeSpace / THRESHOLD; // 计算空闲空间大小对应的区间编号
        lists[number].add(new PageInfo(pgno, freeSpace)); // 在对应的区间列表中添加一个新的 PageInfo 对象
    } finally {
        lock.unlock(); // 释放锁
    }
}

/**
 * 填充 PageIndex。
 * 遍历从第二页开始的每一页，将每一页的页面编号和空闲空间大小添加到 PageIndex 中。
 */
void fillPageIndex() {
    int pageNumber = pc.getPageNumber(); // 获取当前的页面数量
    for (int i = 2; i <= pageNumber; i++) { // 从第二页开始，对每一页进行处理
        Page pg = null;
        try {
            pg = pc.getPage(i); // 尝试获取页面
        } catch (Exception e) {
            Panic.panic(e); // 如果出现异常，处理异常
        }
        pIndex.add(pg.getPageNumber(), PageX.getFreeSpace(pg)); // 将页面编号和页面的空闲空间大小添加到 PageIndex 中
        pg.release(); // 释放页面
    }
}

public class DataItemImpl implements DataItem {
    private SubArray raw; //原始数据
    private byte[] oldRaw; //旧的原始数据
    private DataManagerImpl dm; //数据管理器
    private long uid; //唯一标识符
    private Page pg; //页面对象
}

@Override
public SubArray data() {
    // 返回 [data] 部分
    return new SubArray(raw.raw, raw.start+OF_DATA, raw.end);
}

@Override
public void before() {
    wLock.lock();
    pg.setDirty(true);
    //保存原始数据的副本，以便在需要时进行回滚
    System.arraycopy(raw.raw, raw.start, oldRaw, 0, oldRaw.length); 
}

@Override
public void unBefore() {
    System.arraycopy(oldRaw, 0, raw.raw, raw.start, oldRaw.length);
    wLock.unlock();
}

@Override
public void after(long xid) {
    dm.logDataItem(xid, this);
    wLock.unlock();
}

/**
 * 创建一个更新日志。
 *
 * @param xid 事务ID
 * @param di  DataItem对象
 * @return 更新日志，包含日志类型、事务ID、DataItem的唯一标识符、旧原始数据和新原始数据
 */
public static byte[] updateLog(long xid, DataItem di) {
    byte[] logType = {LOG_TYPE_UPDATE}; // 创建一个表示日志类型的字节数组，并设置其值为LOG_TYPE_UPDATE
    byte[] xidRaw = Parser.long2Byte(xid); // 将事务ID转换为字节数组
    byte[] uidRaw = Parser.long2Byte(di.getUid()); // 将DataItem对象的唯一标识符转换为字节数组
    byte[] oldRaw = di.getOldRaw(); // 获取DataItem对象的旧原始数据
    SubArray raw = di.getRaw(); // 获取DataItem对象的新原始数据
    byte[] newRaw = Arrays.copyOfRange(raw.raw, raw.start, raw.end); // 将新原始数据转换为字节数组
    return Bytes.concat(logType, xidRaw, uidRaw, oldRaw, newRaw); // 将所有字节数组连接在一起，形成一个完整的更新日志，并返回这个日志
}

@Override
public void release() {
    dm.releaseDataItem(this);
}

public class Types {
    public static long addressToUid(int pgno, short offset) {
        long u0 = (long)pgno;
        long u1 = (long)offset;
        return u0 << 32 | u1; //或运算是全0则0，见1则1
    }
}

// 从 uid 中提取出偏移量（offset），这是通过位操作实现的，偏移量是 uid 的低16位
// & 运算：有0则0，全1才1
short offset = (short) (uid & ((1L << 16) - 1)); 

// 将 uid 右移32位，以便接下来提取出页面编号（pgno）
uid >>>= 32;
// 从 uid 中提取出页面编号（pgno），页面编号是 uid 的高32位
// & 运算：有0则0，全1才1
int pgno = (int) (uid & ((1L << 32) - 1));

@Override
protected DataItem getForCache(long uid) throws Exception {
    // 从 uid 中提取出偏移量（offset），这是通过位操作实现的，偏移量是 uid 的低16位
    short offset = (short) (uid & ((1L << 16) - 1));
    // 将 uid 右移32位，以便接下来提取出页面编号（pgno）
    uid >>>= 32;
    // 从 uid 中提取出页面编号（pgno），页面编号是 uid 的高32位
    int pgno = (int) (uid & ((1L << 32) - 1));
    // 使用页面缓存（pc）的 getPage(int pgno) 方法根据页面编号获取一个 Page 对象
    Page pg = pc.getPage(pgno);
    // 使用 DataItem 接口的静态方法 parseDataItem(Page pg, short offset, DataManagerImpl dm)
    // 根据获取到的 Page 对象、偏移量和当前的 DataManagerImpl 对象（this）解析出一个 DataItem 对象，并返回这个对象
    return DataItem.parseDataItem(pg, offset, this);
}

@Override
protected void releaseForCache(DataItem di) {
    di.page().release();
}

// 静态方法，用于创建DataManager实例
public static DataManager create(String path, long mem, TransactionManager tm) {
    // 创建一个PageCache实例，path是文件路径，mem是内存大小
    PageCache pc = PageCache.create(path, mem);
    // 创建一个Logger实例，path是文件路径
    Logger lg = Logger.create(path);

    // 创建一个DataManagerImpl实例，pc是PageCache实例，lg是Logger实例，tm是TransactionManager实例
    DataManagerImpl dm = new DataManagerImpl(pc, lg, tm);
    // 初始化PageOne
    dm.initPageOne();
    // 返回创建的DataManagerImpl实例
    return dm;
}

// 静态方法，用于打开已存在的DataManager实例
public static DataManager open(String path, long mem, TransactionManager tm) {
    // 打开一个PageCache实例，path是文件路径，mem是内存大小
    PageCache pc = PageCache.open(path, mem);
    // 打开一个Logger实例，path是文件路径
    Logger lg = Logger.open(path);
    // 创建一个DataManagerImpl实例，pc是PageCache实例，lg是Logger实例，tm是TransactionManager实例
    DataManagerImpl dm = new DataManagerImpl(pc, lg, tm);
    // 加载并检查PageOne，如果检查失败，则进行恢复操作
    if (!dm.loadCheckPageOne()) {
        Recover.recover(tm, lg, pc);
    }
    // 填充PageIndex，遍历从第二页开始的每一页，将每一页的页面编号和空闲空间大小添加到 PageIndex 中
    dm.fillPageIndex();
    // 设置PageOne为打开状态
    PageOne.setVcOpen(dm.pageOne);
    // 将PageOne立即写入到磁盘中，确保PageOne的数据被持久化
    dm.pc.flushPage(dm.pageOne);

    // 返回创建的DataManagerImpl实例
    return dm;
}

@Override
public DataItem read(long uid) throws Exception {
    //从缓存页面中读取到DataItemImpl
    DataItemImpl di = (DataItemImpl) super.get(uid);
    //校验di是否有效
    if (!di.isValid()) {
        // 无效释放缓存
        di.release();
        return null;
    }
    return di;
}

@Override
protected DataItem getForCache(long uid) throws Exception {
    // 从 uid 中提取出偏移量（offset），这是通过位操作实现的，偏移量是 uid 的低16位
    short offset = (short) (uid & ((1L << 16) - 1));
    // 将 uid 右移32位，以便接下来提取出页面编号（pgno）
    uid >>>= 32;
    // 从 uid 中提取出页面编号（pgno），页面编号是 uid 的高32位
    int pgno = (int) (uid & ((1L << 32) - 1));
    // 使用页面缓存（pc）的 getPage(int pgno) 方法根据页面编号获取一个 Page 对象
    Page pg = pc.getPage(pgno);
    // 使用 DataItem 接口的静态方法 parseDataItem(Page pg, short offset, DataManagerImpl dm)
    // 根据获取到的 Page 对象、偏移量和当前的 DataManagerImpl 对象（this）解析出一个 DataItem 对象，并返回这个对象
    return DataItem.parseDataItem(pg, offset, this);
}

@Override
public long insert(long xid, byte[] data) throws Exception {
    // 将输入的数据包装成DataItem的原始格式
    byte[] raw = DataItem.wrapDataItemRaw(data);
    // 如果数据项的大小超过了页面的最大空闲空间，抛出异常
    if (raw.length > PageX.MAX_FREE_SPACE) {
        throw Error.DataTooLargeException;
    }

    // 初始化一个页面信息对象
    PageInfo pi = null;
    // 尝试5次找到一个可以容纳新数据项的页面
    for (int i = 0; i < 5; i++) {
        // 从页面索引中选择一个可以容纳新数据项的页面
        pi = pIndex.select(raw.length);
        // 如果找到了合适的页面，跳出循环
        if (pi != null) {
            break;
        } else {
            // 如果没有找到合适的页面，创建一个新的页面，并将其添加到页面索引中
            int newPgno = pc.newPage(PageX.initRaw());
            pIndex.add(newPgno, PageX.MAX_FREE_SPACE);
        }
    }
    // 如果还是没有找到合适的页面，抛出异常
    if (pi == null) {
        throw Error.DatabaseBusyException;
    }

    // 初始化一个页面对象
    Page pg = null;
    // 初始化空闲空间大小为0
    int freeSpace = 0;
    try {
        // 获取页面信息对象中的页面
        pg = pc.getPage(pi.pgno);
        // 生成插入日志
        byte[] log = Recover.insertLog(xid, pg, raw);
        // 将日志写入日志文件
        logger.log(log);

        // 在页面中插入新的数据项，并获取其在页面中的偏移量
        short offset = PageX.insert(pg, raw);

        // 释放页面
        pg.release();
        // 返回新插入的数据项的唯一标识符
        return Types.addressToUid(pi.pgno, offset);

    } finally {
        // 将页面重新添加到页面索引中
        if (pg != null) {
            pIndex.add(pi.pgno, PageX.getFreeSpace(pg));
        } else {
            pIndex.add(pi.pgno, freeSpace);
        }
    }
}

/**
 *  返回一个完整的 DataItem 结构数据
 *  dataItem 结构如下：
 *  [ValidFlag] [DataSize] [Data]
 *  ValidFlag 1字节，0为合法，1为非法
 *  DataSize  2字节，标识Data的长度
 * @param raw
 * @return
 */
public static byte[] wrapDataItemRaw(byte[] raw) {
    byte[] valid = new byte[1]; //证明此时为非法数据
    byte[] size = Parser.short2Byte((short)raw.length); //计算数据字节大小
    return Bytes.concat(valid, size, raw); //拼接DataItem 结构数据
}

/**
 * 根据给定的空间大小选择一个 PageInfo 对象。
 *
 * @param spaceSize 需要的空间大小
 * @return 一个 PageInfo 对象，其空闲空间大于或等于给定的空间大小。如果没有找到合适的 PageInfo，返回 null。
 */
public PageInfo select(int spaceSize) {
    lock.lock(); // 获取锁，确保线程安全
    try {
        int number = spaceSize / THRESHOLD; // 计算需要的空间大小对应的区间编号
        // 此处+1主要为了向上取整
        /*
            1、假需要存储的字节大小为5168，此时计算出来的区间号是25，但是25*204=5100显然是不满足条件的
            2、此时向上取整找到 26，而26*204=5304，是满足插入条件的
         */
        if (number < INTERVALS_NO) number++; // 如果计算出的区间编号小于总的区间数，编号加一
        while (number <= INTERVALS_NO) { // 从计算出的区间编号开始，向上寻找合适的 PageInfo
            if (lists[number].size() == 0) { // 如果当前区间没有 PageInfo，继续查找下一个区间
                number++;
                continue;
            }
            return lists[number].remove(0); // 如果当前区间有 PageInfo，返回第一个 PageInfo，并从列表中移除
        }
        return null; // 如果没有找到合适的 PageInfo，返回 null
    } finally {
        lock.unlock(); // 释放锁
    }
}

// 定义一个静态方法，用于创建插入日志
public static byte[] insertLog(long xid, Page pg, byte[] raw) {
    // 创建一个表示日志类型的字节数组，并设置其值为LOG_TYPE_INSERT
    byte[] logTypeRaw = {LOG_TYPE_INSERT};
    // 将事务ID转换为字节数组
    byte[] xidRaw = Parser.long2Byte(xid);
    // 将页面编号转换为字节数组
    byte[] pgnoRaw = Parser.int2Byte(pg.getPageNumber());
    // 获取页面的第一个空闲空间的偏移量，并将其转换为字节数组
    byte[] offsetRaw = Parser.short2Byte(PageX.getFSO(pg));
    // 将所有字节数组连接在一起，形成一个完整的插入日志，并返回这个日志
    return Bytes.concat(logTypeRaw, xidRaw, pgnoRaw, offsetRaw, raw);
}

// 将raw插入pg中，返回插入位置
public static short insert(Page pg, byte[] raw) {
    pg.setDirty(true); // 将pg的dirty标志设置为true，表示pg的数据已经被修改
    short offset = getFSO(pg.getData()); // 获取pg的空闲空间偏移量
    System.arraycopy(raw, 0, pg.getData(), offset, raw.length); // 将raw的数据复制到pg的数据中的offset位置
    setFSO(pg.getData(), (short) (offset + raw.length)); // 更新pg的空闲空间偏移量
    return offset; // 返回插入位置
}

@Override
public void close() {
    super.close();
    logger.close();

    PageOne.setVcClose(pageOne);
    pageOne.release();
    pc.close();
}

public class Entry {
    // 定义了XMIN的偏移量为0
    private static final int OF_XMIN = 0;
    // 定义了XMAX的偏移量为XMIN偏移量后的8个字节
    private static final int OF_XMAX = OF_XMIN+8;
    // 定义了DATA的偏移量为XMAX偏移量后的8个字节
    private static final int OF_DATA = OF_XMAX+8;

    // uid字段，可能是用来唯一标识一个Entry的
    private long uid;
    // DataItem对象，用来存储数据的
    private DataItem dataItem;
    // VersionManager对象，用来管理版本的
    private VersionManager vm;

    // 静态方法，用来加载一个Entry。它首先从VersionManager中读取数据，然后创建一个新的Entry
    public static Entry loadEntry(VersionManager vm, long uid) throws Exception {
        DataItem di = ((VersionManagerImpl)vm).dm.read(uid);
        return newEntry(vm, di, uid);
    }

    // 方法，用来移除一个Entry。它通过调用dataItem的release方法来实现
    public void remove() {
        dataItem.release();
    }
}

/**
 * 生成日志格式数据
 */
public static byte[] wrapEntryRaw(long xid, byte[] data) {
    // 将事务id转为8字节数组
    byte[] xmin = Parser.long2Byte(xid);
    // 创建一个空的8字节数组，等待版本修改或删除是才修改
    byte[] xmax = new byte[8];
    // 拼接成日志格式
    return Bytes.concat(xmin, xmax, data);
}

// 以拷贝的形式返回内容
public byte[] data() {
    // 加锁，确保数据安全
    dataItem.rLock();
    try {
        // 获取日志数据
        SubArray sa = dataItem.data();
        // 创建一个去除前16字节的数组，因为前16字节表示 xmin and xmax
        byte[] data = new byte[sa.end - sa.start - OF_DATA];
        // 拷贝数据到data数组上
        System.arraycopy(sa.raw, sa.start+OF_DATA, data, 0, data.length);
        return data;
    } finally {
        //释放锁
        dataItem.rUnLock();
    }
}