FastDFS分布式文件系统点滴记录5 -- upload上传机制剖析2

首先分析下 tracker ， tracker_service.c 3137行：int tracker_deal_task(struct fast_task_info *pTask)

对TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE命令的处理是

case TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE:
            result = tracker_deal_service_query_storage( \
                    pTask, pHeader->cmd);
            break;

我们进入 tracker_deal_service_query_storage， tracker_service.c 1955行：

static int tracker_deal_service_query_storage( \
        struct fast_task_info *pTask, char cmd)

这个函数比较长，由于里面包含了负载均衡相关的逻辑判断，我们暂时关注点只分析cmd为TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE，涉及到负载均衡的部分，之后其他文章会单独分析。

tracker_service.c 2184行： 

    if (cmd == TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITH_GROUP_ONE
      || cmd == TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE)
    {
        pStorageServer = tracker_get_writable_storage(pStoreGroup);
        if (pStorageServer == NULL)
        {
            pTask->length = sizeof(TrackerHeader);
            return ENOENT;
        }
    }
    else  //query store server list, use the first to check
    {
        pStorageServer = *(pStoreGroup->active_servers);
    }

代码分析：

1.通过负载均衡, 获得可利用的pStoreGroup 结构；

2. 调用tracker_get_writable_storage ，找到合适的pStorageServer

tracker_service.c 2199行：

write_path_index = pStorageServer->current_write_path;
    if (write_path_index >= pStoreGroup->store_path_count)
    {
        write_path_index = 0;
    }

    avg_reserved_mb = g_storage_reserved_mb / \
              pStoreGroup->store_path_count;
    if (pStorageServer->path_free_mbs[write_path_index] <= avg_reserved_mb)
    {
        int i;
        for (i=0; i<pStoreGroup->store_path_count; i++)
        {
            if (pStorageServer->path_free_mbs[i] > avg_reserved_mb)
            {
                pStorageServer->current_write_path = i;
                write_path_index = i;
                break;
            }
        }

        if (i == pStoreGroup->store_path_count)
        {
            if (!g_if_use_trunk_file)
            {
                pTask->length = sizeof(TrackerHeader);
                return ENOSPC;
            }

            for (i=write_path_index; i<pStoreGroup-> \
                store_path_count; i++)
            {
                if (pStorageServer->path_free_mbs[i] + \
                    pStoreGroup->trunk_free_mb > avg_reserved_mb)
                {
                    pStorageServer->current_write_path = i;
                    write_path_index = i;
                    break;
                }
            }
            if ( i == pStoreGroup->store_path_count)
            {
                for (i=0; i<write_path_index; i++)
                {
                if (pStorageServer->path_free_mbs[i] + \
                    pStoreGroup->trunk_free_mb > avg_reserved_mb)
                {
                    pStorageServer->current_write_path = i;
                    write_path_index = i;
                    break;
                }
                }

                if (i == write_path_index)
                {
                    pTask->length = sizeof(TrackerHeader);
                    return ENOSPC;
                }
            }
        }
    }

    if (g_groups.store_path == FDFS_STORE_PATH_ROUND_ROBIN)
    {
        pStorageServer->current_write_path++;
        if (pStorageServer->current_write_path >= \
                pStoreGroup->store_path_count)
        {
            pStorageServer->current_write_path = 0;
        }
    }

    /*
    //printf("pStoreGroup->current_write_server: %d, " \
    "pStoreGroup->active_count=%d\n", \
    pStoreGroup->current_write_server, \
    pStoreGroup->active_count);
    */

    p = pTask->data + sizeof(TrackerHeader);
    memcpy(p, pStoreGroup->group_name, FDFS_GROUP_NAME_MAX_LEN);
    p += FDFS_GROUP_NAME_MAX_LEN;

    if (cmd == TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ALL
     || cmd == TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITH_GROUP_ALL)
    {
        int active_count;
        FDFSStorageDetail **ppServer;
        FDFSStorageDetail **ppEnd;

        active_count = pStoreGroup->active_count;
        if (active_count == 0)
        {
            pTask->length = sizeof(TrackerHeader);
            return ENOENT;
        }

        ppEnd = pStoreGroup->active_servers + active_count;
        for (ppServer=pStoreGroup->active_servers; ppServer<ppEnd; \
            ppServer++)
        {
            memcpy(p, (*ppServer)->ip_addr, IP_ADDRESS_SIZE - 1);
            p += IP_ADDRESS_SIZE - 1;

            long2buff(pStoreGroup->storage_port, p);
            p += FDFS_PROTO_PKG_LEN_SIZE;
        }
    }
    else
    {
        memcpy(p, pStorageServer->ip_addr, IP_ADDRESS_SIZE - 1);
        p += IP_ADDRESS_SIZE - 1;

        long2buff(pStoreGroup->storage_port, p);
        p += FDFS_PROTO_PKG_LEN_SIZE;
    }

    *p++ = (char)write_path_index;

    pTask->length = p - pTask->data;

    return 0;

代码分析：

1. 根据配置的条件，查找合适的write_path_index，

2. 将找到的storage server 信息，group_name、ip_addr、storage_port 拼接到响应包里面。

最后， tracker_service.c 3287行 ：

    pHeader = (TrackerHeader *)pTask->data;
    pHeader->status = result;
    pHeader->cmd = TRACKER_PROTO_CMD_RESP;
    long2buff(pTask->length - sizeof(TrackerHeader), pHeader->pkg_len);

    send_add_event(pTask);

会将应答包通过socket 发送至客户端。

至此，client 与 tracker 的请求storage server 的过程结束。这里只是对TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE 命令进行分析。其余获取store 的命令和它类似，读者可以自行分析。

欢迎感兴趣的朋友一起交流研究，提出意见。

FastDFS技术交流群：164684842