数据库内核月报 － 2016 / 04

struct ftnode {
    MSN max_msn_applied_to_node_on_disk;  // applied最大的msn
    unsigned int flags; // 创建标志，等于ft->h->flags
    BLOCKNUM blocknum; // 节点对应的块号
    int    height; // 高度：0表示leaf节点，>0表示中间节点
    int    dirty; // 脏页标记，用于cachetable脏页回刷
    uint32_t fullhash; // 在cachetable的哪个bucket上
    int n_children; // partition个数
    ftnode_pivot_keys pivotkeys; // 每个partition的key值区间
    TXNID oldest_referenced_xid_known; // 最近一次push_into_root操作时系统可能的最老事务id
struct ftnode_partition *bp; // internal节点：子节点的msg_buffer; leaf节点：有序数据结构
    struct ctpair *ct_pair; // 对应cachetable的pair（cache entry）
};

struct ftnode_partition {
BLOCKNUM  blocknum; // internal节点：子节点块号;  leaf节点：unused
uint64_t     workdone; // internal节点：applied message字节数; leaf节点：unused
struct ftnode_child_pointer ptr; // internal节点：子节点msg_buffer信息; leaf节点：basement节点
enum pt_state state; // internal节点：子节点状态; leaf节点：basement节点状态
    uint8_t clock_count; // evictor线程根据此变量判断是否可以partial evict
};
enum pt_state {
    PT_INVALID = 0, // 无效
    PT_ON_DISK = 1, // 在磁盘上
    PT_COMPRESSED = 2, // 已读入内存，压缩格式
    PT_AVAIL = 3 // 已读入内存，解压缩格式
};
typedef struct ftnode_child_pointer {
    union {
        struct sub_block *subblock;  //压缩后的buffer
        struct ftnode_nonleaf_childinfo *nonleaf;  // internal节点：msg_buffer
        struct ftnode_leaf_basement_node *leaf; // leaf节点：basement节点
    } u;
 } FTNODE_CHILD_POINTER;

struct ftnode_nonleaf_childinfo {
    message_buffer msg_buffer; // 缓存的message
    off_omt_t broadcast_list; // 用于支持on-line add index, on-line add column
    marked_off_omt_t fresh_message_tree; // 未apply的message在msg_buffer 里的offset，按（key，msn）顺序排序
    off_omt_t stale_message_tree; // 已经applied过的message在msg_buffer里的offset，按（key，msn）顺序排序
    uint64_t flow[2]; // 流控
};

struct ftnode_leaf_basement_node {
    bn_data data_buffer; // 有序序列，弱平衡树形结构（或者数组）
    unsigned int seqinsert; // 判断顺序 insert的hint
    MSN max_msn_applied; // applied最大的msn
};

struct ft {
    FT_HEADER h; // header
    FT_HEADER checkpoint_header; // checkpoint开始时刻header的克隆
    CACHEFILE cf; // 描述了FT对应的磁盘文件和数据节点信息。
    toku::comparator cmp; // DBT compare函数
    block_table blocktable; // FT逻辑块号到文件offset的映射表
    struct toku_list live_ft_handles; // 打开此FT的handle列表
    uint32_t num_txns;  // 访问此FT的事务个数
    bool pinned_by_checkpoint; // 表示checkpoint正在进行
    BLOCKNUM rightmost_blocknum; // 最右leaf节点块号，用于顺序insert优化
} ;
typedef struct ft *FT;
struct ft_header {
    enum ft_type type; // header类型
    int dirty; // 脏标记
    uint64_t checkpoint_count; // checkpoint计数
    LSN checkpoint_lsn; // checkpoint开始时刻的lsn
    const uint64_t time_of_creation; // FT创建时间
    TXNID root_xid_that_created; //创建FT的事务ID
    uint64_t time_of_last_modification; // 最近一次更新时间
    BLOCKNUM root_blocknum; // root节点块号
    const unsigned int flags; // 创建标志
    unsigned int nodesize; // 节点大小，缺省值4M字节
    unsigned int basementnodesize; // basement节点大小，缺省值128K字节
    enum toku_compression_method compression_method; // 压缩算法
    unsigned int fanout; // internal节点的fanout
    MSN max_msn_in_ft; // FT最大的msn
};
enum ft_type {
    FT_CURRENT = 1, // FT header
    FT_CHECKPOINT_INPROGRESS // checkpoint开始时刻FT header的克隆
};

void toku_ft_root_put_msg(FT ft, const ft_msg &msg, txn_gc_info *gc_info) {
    ftnode_fetch_extra bfe;
    bfe.create_for_full_read(ft); // fetch整个node
    pair_lock_type lock_type = PL_READ; // 初始状态：申请读锁
change_lock_type:
    toku_pin_ftnode(ft, root_key, fullhash, &bfe, lock_type, &node, true);
    enum reactivity re = toku_ftnode_get_reactivity(ft, node);
    switch (re) {
    case RE_STABLE:
case RE_FUSIBLE:
        // root节点不支持merge操作
        if (lock_type != PL_READ) {
            // 其他thread抢先做了split
            toku_unpin_ftnode_read_only(ft, node);
            lock_type = PL_READ;
            goto change_lock_type;
        }
        break;
    case RE_FISSIBLE:
        if (lock_type == PL_READ) {
            // 需要split，升级为写锁
            toku_unpin_ftnode_read_only(ft, node);
            lock_type = PL_WRITE_CHEAP;
            goto change_lock_type;
        } else {
            // root节点split
            ft_init_new_root(ft, node, &node);
            // split完成，降级为读锁
            toku_unpin_ftnode(ft, node);
            lock_type = PL_READ;
            goto change_lock_type;
        }
        break;
    }
    if (root ->height == 0 || ft_msg_type_is_broadcast(msg.type())) {
        // root是leaf节点或者messsage是广播消息，把message放到root节点上
        // 这里释放锁的原因是：inject_message_at_this_blocknum里面会重新拿锁
        toku_unpin_ftnode_read_only(ft, root);
        inject_message_at_this_blocknum(ft, root_key, fullhash, msg, flow_deltas, gc_info);
    } else if (root->height > 1) {
        // root高度大于1时，把message加到root节点的某个子树上
        push_something_in_subtree(ft, node, -1, msg, flow_deltas, gc_info, 0, LEFT_EXTREME |  RIGHT_EXTREME, false);
    } else {
        // root高度等于1时，确定message应该放到哪个leaf节点上
        int childnum = toku_ftnode_which_child(node, msg.kdbt(), ft->cmp);
        // 目标leaf节点是root的最左或最右子节点时，把message放到相应的leaf节点上
        if (childnum == 0 || childnum == node->n_children - 1) {
            push_something_in_subtree(ft, node, childnum, msg, flow_deltas, gc_info, 0, LEFT_EXTREME | RIGHT_EXTREME, false);
        } else {
            // 把message放到root节点上
            // 这里释放锁的原因是：inject_message_at_this_blocknum里面会重新拿锁
            toku_unpin_ftnode_read_only(ft, node);
            inject_message_at_this_blocknum(ft, root_key, fullhash, msg, flow_deltas, gc_info);
        }
    }
}
toku_ftnode_get_reactivity返回的reactivity定义如下：
enum reactivity {
    RE_STABLE,    // 正常状态
    RE_FUSIBLE,   // 需要merge
    RE_FISSIBLE    // 需要split
};

static void inject_message_in_locked_node(FT ft, FTNODE node, int childnum, const ft_msg &msg, size_t flow_deltas[],txn_gc_info *gc_info) {
    // 生成全局（FT内）唯一的msn
    MSN msg_msn = { .msn = toku_sync_add_and_fetch(&ft->h->max_msn_in_ft.msn, 1) };
ft_msg msg_with_msn(msg.kdbt(), msg.vdbt(), msg.type(), msg_msn, msg.xids());
    toku_ftnode_put_msg(ft->cmp, ft->update_fun, node, childnum, msg_with_msn, true, gc_info, flow_deltas, &stats_delta);
if (node->height > 0 && toku_ftnode_nonleaf_is_gorged(node, ft->h->nodesize)) {
    // 如果node节点是internal节点，msg_buffer非空并且node节点所需的磁盘空间 + 各个子节点的workdone（query过程中同步flush的message大小的总和）大于nodesize（缺省4M），trigger client线程池的工作线程异步flush这个节点。
    // msg_buffer非空并且node节点所需的磁盘空间 + 各个子节点的workdone: 表示node节点逻辑上的size
   toku_ft_flush_node_on_background_thread(ft, node);
    } else {
        toku_unpin_ftnode(ft, node);
}
}

toku_le_apply_msg(const ft_msg &msg,
                 LEAFENTRY old_leafentry,    // 老的LEAFENTRY
                 bn_data* data_buffer,       // basement节点的bn，如果是basement节点上第一个message，这个值NULL
                 uint32_t idx,               // old_leafentry在bn上的index
                 uint32_t old_keylen, txn_gc_info *gc_info,
                 LEAFENTRY *new_leafentry_p,  //新的LEAFENTRY
                 int64_t * numbytes_delta_p) {
    if (old_leafentry == NULL) {
        msg_init_empty_ule(&ule);
    } else {
        le_unpack(&ule, old_leafentry); // 把LEAFENTRY转成ULE
    }
    msg_modify_ule(&ule, msg);          // 把message加到ULE的MVCC
    // 把ULE转成LEAFENTRY
    // 若data_buffer == NULL，le_pack分配basement节点的bn
    int r = le_pack(&ule, data_buffer, idx, msg.kdbt()->data, keylen, old_keylen, oldmemsize, new_leafentry_p, &maybe_free);
}

msg_modify_ule(ULE ule, const ft_msg &msg) {
    switch (type) {
    case FT_INSERT_NO_OVERWRITE: {
        UXR old_innermost_uxr = ule_get_innermost_uxr(ule);
        // key已存在，保留老值
        if (uxr_is_insert(old_innermost_uxr)) break;
    }
    case FT_INSERT: {
        uint32_t vallen = msg.vdbt()->size;
        void * valp = msg.vdbt()->data;
        ule_apply_insert(ule, xids, vallen, valp);
        break;
}

msg_modify_ule(ULE ule, const ft_msg &msg) {
    switch (type) {
    case FT_DELETE_ANY:
        ule_apply_delete(ule, xids);
        break;
    }
}

static void
msg_modify_ule(ULE ule, const ft_msg &msg) {
    XIDS xids = msg.xids();
    enum ft_msg_type type = msg.type();
    if (type != FT_OPTIMIZE && type != FT_OPTIMIZE_FOR_UPGRADE) {
        ule_do_implicit_promotions(ule, xids);
    }
}
static void
ule_do_implicit_promotions(ULE ule, XIDS xids) {
    if (ule->num_puxrs > 0) {
        int num_xids = toku_xids_get_num_xids(xids);
        uint32_t max_index = ule->num_cuxrs + min_i32(ule->num_puxrs, num_xids) - 1;
        uint32_t ica_index = max_index;
        uint32_t index;
        for (index = ule->num_cuxrs; index <= max_index; index++) {
            TXNID current_msg_xid = toku_xids_get_xid(xids, index - ule->num_cuxrs);
            TXNID current_ule_xid = ule_get_xid(ule, index);
            if (current_msg_xid != current_ule_xid) {
                //ICA表示innermost common ancestor
                ica_index = index - 1;
                break;
            }
        }
        if (ica_index < ule->num_cuxrs) {
            // 隐式提交上一次对这个LEAFENTRY的修改
            invariant(ica_index == ule->num_cuxrs - 1);
            ule_promote_provisional_innermost_to_committed(ule);
        }
        else if (ica_index < ule->num_cuxrs + ule->num_puxrs - 1) {
            ule_promote_provisional_innermost_to_index(ule, ica_index);
        }
    }
}

enum split_mode {
    SPLIT_EVENLY,
    SPLIT_LEFT_HEAVY,
    SPLIT_RIGHT_HEAVY
};

int toku_ft_search(FT_HANDLE ft_handle, ft_search *search, FT_GET_CALLBACK_FUNCTION getf, void *getf_v, FT_CURSOR ftcursor, bool can_bulk_fetch)
{
try_again:
    trycount++;
    ftnode_fetch_extra bfe;
    // 只读取包含search key的那个子节点
    bfe.create_for_subset_read(ft, search, &ftcursor->range_lock_left_key,
                           &ftcursor->range_lock_right_key, ftcursor->left_is_neg_infty,
                           Ftcursor->right_is_pos_infty,ftcursor->disable_prefetching, true);
    FTNODE node = NULL;
    {
        //读取root节点，并设置bfe.child_to_read为包含search key的子节点
        toku_pin_ftnode(ft, root_key, fullhash, &bfe, PL_READ, &node, true);
    }
    struct unlock_ftnode_extra unlock_extra   = {ft_handle,node,false};
    struct unlockers  unlockers      = {true, unlock_ftnode_fun, (void*)&unlock_extra,          (UNLOCKERS)NULL};
    {
        bool doprefetch = false; bfe.child_to_read
        r = ft_search_node(ft_handle, node, search, bfe.child_to_read, getf, getf_v, &doprefetch, ftcursor, &unlockers, (ANCESTORS)NULL, pivot_bounds::infinite_bounds(), can_bulk_fetch);
        if (r==TOKUDB_TRY_AGAIN) {
            // 获取锁时需要等待或者需要partial fetch
            if (unlockers.locked) {
                toku_unpin_ftnode_read_only(ft_handle->ft, node);
            }
            goto try_again;
        } else {
            assert(unlockers.locked);
        }
    }
    assert(unlockers.locked);
    toku_unpin_ftnode_read_only(ft_handle->ft, node);
}