源码搜藏网

1. /* struct for network transfering */  
2. typedef struct  
3. {  
4.     char sz_total_upload_count[8];  
5.     char sz_success_upload_count[8];  
6.     char sz_total_append_count[8];  
7.     char sz_success_append_count[8];  
8.     char sz_total_modify_count[8];  
9.     char sz_success_modify_count[8];  
10.     char sz_total_truncate_count[8];  
11.     char sz_success_truncate_count[8];  
12.     char sz_total_set_meta_count[8];  
13.     char sz_success_set_meta_count[8];  
14.     char sz_total_delete_count[8];  
15.     char sz_success_delete_count[8];  
16.     char sz_total_download_count[8];  
17.     char sz_success_download_count[8];  
18.     char sz_total_get_meta_count[8];  
19.     char sz_success_get_meta_count[8];  
20.     char sz_total_create_link_count[8];  
21.     char sz_success_create_link_count[8];  
22.     char sz_total_delete_link_count[8];  
23.     char sz_success_delete_link_count[8];  
24.     char sz_total_upload_bytes[8];  
25.     char sz_success_upload_bytes[8];  
26.     char sz_total_append_bytes[8];  
27.     char sz_success_append_bytes[8];  
28.     char sz_total_modify_bytes[8];  
29.     char sz_success_modify_bytes[8];  
30.     char sz_total_download_bytes[8];  
31.     char sz_success_download_bytes[8];  
32.     char sz_total_sync_in_bytes[8];  
33.     char sz_success_sync_in_bytes[8];  
34.     char sz_total_sync_out_bytes[8];  
35.     char sz_success_sync_out_bytes[8];  
36.     char sz_total_file_open_count[8];  
37.     char sz_success_file_open_count[8];  
38.     char sz_total_file_read_count[8];  
39.     char sz_success_file_read_count[8];  
40.     char sz_total_file_write_count[8];  
41.     char sz_success_file_write_count[8];  
42.     char sz_last_source_update[8];  
43.     char sz_last_sync_update[8];  
44.     char sz_last_synced_timestamp[8];  
45.     char sz_last_heart_beat_time[8];  
46. } FDFSStorageStatBuff;  

tracker主要是做了什么呢？

对其进行解包，然后对这个保存在本地的storage的信息进行保存到文件中，调用

[cpp] view plain copy  

1. status = tracker_save_storages();  

调用

[cpp] view plain copy  

1. tracker_mem_active_store_server(pClientInfo->pGroup, \  
2.             pClientInfo->pStorage);  

将这个存储服务器如果没有，就插入到group中。

最后调用

[cpp] view plain copy  

1. static int tracker_check_and_sync(struct fast_task_info *pTask, \  
2.             const int status)  

检查相应的改变状态，并将其同步等。（需要再详细看看）

2、报告相应同步时间

[cpp] view plain copy  

1. #define TRACKER_PROTO_CMD_STORAGE_SYNC_REPORT       89  //report src last synced time as dest server  

同样在storage的report线程执行

[cpp] view plain copy  

1. if (sync_time_chg_count != g_sync_change_count && \  
2.                 current_time - last_sync_report_time >= \  
3.                     g_heart_beat_interval)  
4.             {  
5.                 if (tracker_report_sync_timestamp( \  
6.                     pTrackerServer, &bServerPortChanged)!=0)  
7.                 {  
8.                     break;  
9.                 }  
10.   
11.                 sync_time_chg_count = g_sync_change_count;  
12.                 last_sync_report_time = current_time;  
13.             }  

具体的数据包为

[cpp] view plain copy  

1. pEnd = g_storage_servers + g_storage_count;  
2.     for (pServer=g_storage_servers; pServer<pEnd; pServer++)  
3.     {  
4.         memcpy(p, pServer->server.id, FDFS_STORAGE_ID_MAX_SIZE);  
5.         p += FDFS_STORAGE_ID_MAX_SIZE;  
6.         int2buff(pServer->last_sync_src_timestamp, p);  
7.         p += 4;  
8.     }  

也就是遍历当前进程的本组所有storage服务器，和上次同步的时间戳，给tracker服务器。

然后tracker的服务器存储结构为

[cpp] view plain copy  

1. pClientInfo->pGroup->last_sync_timestamps \  
2.                 [src_index][dest_index] = sync_timestamp;  

dest_index 值为当前连接所在组的索引值

[cpp] view plain copy  

1. dest_index = tracker_mem_get_storage_index(pClientInfo->pGroup,  
2.             pClientInfo->pStorage);  
3.     if (dest_index < 0 || dest_index >= pClientInfo->pGroup->count)  
4.     {  
5.         status = 0;  
6.         break;  
7.     }  

因为 本链接的storage是固定不变的，而src_index就是为本组的其他storage的id索引，

首相通过id，（ip地址）找到具体的storage，然后在通过指针找到索引位置，

最后，调用

[cpp] view plain copy  

1. if (++g_storage_sync_time_chg_count % \  
2.         TRACKER_SYNC_TO_FILE_FREQ == 0)  
3. {  
4.     status = tracker_save_sync_timestamps();  
5. }  
6. else  
7. {  
8.     status = 0;  
9. }  
10. } while (0);  
11.   
12. return tracker_check_and_sync(pTask, status);  

定时保存文件和检查等

3、上报磁盘情况

#define TRACKER_PROTO_CMD_STORAGE_REPORT_DISK_USAGE 84  //report disk usage

同样线程定时调用，

[cpp] view plain copy  

1. if (current_time - last_df_report_time >= \  
2.                 g_stat_report_interval)  
3.         {  
4.             if (tracker_report_df_stat(pTrackerServer, \  
5.                     &bServerPortChanged) != 0)  
6.             {  
7.                 break;  
8.             }  
9.   
10.             last_df_report_time = current_time;  
11.         }  

同样上报这些数据

[cpp] view plain copy  

1. for (i=0; i<g_fdfs_store_paths.count; i++)  
2.     {  
3.         if (statvfs(g_fdfs_store_paths.paths[i], &sbuf) != 0)  
4.         {  
5.             logError("file: "__FILE__", line: %d, " \  
6.                 "call statfs fail, errno: %d, error info: %s.",\  
7.                 __LINE__, errno, STRERROR(errno));  
8.   
9.             if (pBuff != out_buff)  
10.             {  
11.                 free(pBuff);  
12.             }  
13.             return errno != 0  errno : EACCES;  
14.         }  
15.   
16.         g_path_space_list[i].total_mb = ((int64_t)(sbuf.f_blocks) * \  
17.                     sbuf.f_frsize) / FDFS_ONE_MB;  
18.         g_path_space_list[i].free_mb = ((int64_t)(sbuf.f_bavail) * \  
19.                     sbuf.f_frsize) / FDFS_ONE_MB;  
20.         long2buff(g_path_space_list[i].total_mb, pStatBuff->sz_total_mb);  
21.         long2buff(g_path_space_list[i].free_mb, pStatBuff->sz_free_mb);  
22.   
23.         pStatBuff++;  
24.     }  

tracker这边存储在

int64_t *path_total_mbs; //total disk storage in MB
int64_t *path_free_mbs;  //free disk storage in MB

这里

[cpp] view plain copy  

1. path_total_mbs[i] = buff2long(pStatBuff->sz_total_mb);  
2.         path_free_mbs[i] = buff2long(pStatBuff->sz_free_mb);  
3.   
4.         pClientInfo->pStorage->total_mb += path_total_mbs[i];  
5.         pClientInfo->pStorage->free_mb += path_free_mbs[i];  

4、storage服加入到tracker

[cpp] view plain copy  

1. #define TRACKER_PROTO_CMD_STORAGE_JOIN              81  

storage线程同样在该处调用

[cpp] view plain copy  

1. if (tracker_report_join(pTrackerServer, tracker_index, \  
2.                     sync_old_done) != 0)  
3.         {  
4.             sleep(g_heart_beat_interval);  
5.             continue;  
6.         }  

发送的包体数据包为：

[cpp] view plain copy  

1. typedef struct  
2. {  
3.     char group_name[FDFS_GROUP_NAME_MAX_LEN+1];  
4.     char storage_port[FDFS_PROTO_PKG_LEN_SIZE];  
5.     char storage_http_port[FDFS_PROTO_PKG_LEN_SIZE];  
6.     char store_path_count[FDFS_PROTO_PKG_LEN_SIZE];  
7.     char subdir_count_per_path[FDFS_PROTO_PKG_LEN_SIZE];  
8.     char upload_priority[FDFS_PROTO_PKG_LEN_SIZE];  
9.     char join_time[FDFS_PROTO_PKG_LEN_SIZE]; //storage join timestamp  
10.     char up_time[FDFS_PROTO_PKG_LEN_SIZE];   //storage service started timestamp  
11.     char version[FDFS_VERSION_SIZE];   //storage version  
12.     char domain_name[FDFS_DOMAIN_NAME_MAX_SIZE];  
13.     char init_flag;  
14.     signed char status;  
15.     char tracker_count[FDFS_PROTO_PKG_LEN_SIZE];  //all tracker server count  
16. } TrackerStorageJoinBody;  

当赋值完成后，在气候变加入

[cpp] view plain copy  

1. p = out_buff + sizeof(TrackerHeader) + sizeof(TrackerStorageJoinBody);  
2.     pServerEnd = g_tracker_group.servers + g_tracker_group.server_count;  
3.     for (pServer=g_tracker_group.servers; pServer<pServerEnd; pServer++)  
4.     {  
5.         /* 
6.         if (strcmp(pServer->ip_addr, pTrackerServer->ip_addr) == 0 && \ 
7.             pServer->port == pTrackerServer->port) 
8.         { 
9.             continue; 
10.         } 
11.         tracker_count++; 
12.         */  
13.   
14.         sprintf(p, "%s:%d", pServer->ip_addr, pServer->port);  
15.         p += FDFS_PROTO_IP_PORT_SIZE;  
16.     }  

加入所有tracker的服务器信息格式为ip:port

tracker 服务器接收

[cpp] view plain copy  

1. case TRACKER_PROTO_CMD_STORAGE_JOIN:  
2.         result = tracker_deal_storage_join(pTask);  
3.         break;  

获取到的相关信息存储到

[cpp] view plain copy  

1. typedef struct  
2. {  
3.     int storage_port;  
4.     int storage_http_port;  
5.     int store_path_count;  
6.     int subdir_count_per_path;  
7.     int upload_priority;  
8.     int join_time; //storage join timestamp (create timestamp)  
9.     int up_time;   //storage service started timestamp  
10.         char version[FDFS_VERSION_SIZE];   //storage version  
11.     char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];  
12.         char domain_name[FDFS_DOMAIN_NAME_MAX_SIZE];  
13.         char init_flag;  
14.     signed char status;  
15.     int tracker_count;  
16.     ConnectionInfo tracker_servers[FDFS_MAX_TRACKERS];  
17. } FDFSStorageJoinBody;  

这些结构体内

同时插入本地内存

result = tracker_mem_add_group_and_storage(pClientInfo, \
pTask->client_ip, &joinBody, true);

同时把发消息报的id传过来

[cpp] view plain copy  

1. pJoinBodyResp = (TrackerStorageJoinBodyResp *)(pTask->data + \  
2.             sizeof(TrackerHeader));  
3. memset(pJoinBodyResp, 0, sizeof(TrackerStorageJoinBodyResp));  
4.   
5. if (pClientInfo->pStorage->psync_src_server != NULL)  
6. {  
7.     strcpy(pJoinBodyResp->src_id, \  
8.         pClientInfo->pStorage->psync_src_server->id);  
9. }  

5、报告存储状态

[cpp] view plain copy  

1. #define TRACKER_PROTO_CMD_STORAGE_REPORT_STATUS     76  //report specified storage server status  

storage服务器调用

[cpp] view plain copy  

1. int tracker_report_storage_status(ConnectionInfo *pTrackerServer, \  
2.         FDFSStorageBrief *briefServer)  

内容主要是组名字

[cpp] view plain copy  

1. strcpy(out_buff + sizeof(TrackerHeader), g_group_name);  

和简要信息

[cpp] view plain copy  

1. memcpy(out_buff + sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN, \  
2.         briefServer, sizeof(FDFSStorageBrief));  

其结构体如下

[cpp] view plain copy  

1. typedef struct  
2. {  
3.     char status;  
4.     char port[4];  
5.     char id[FDFS_STORAGE_ID_MAX_SIZE];  
6.     char ip_addr[IP_ADDRESS_SIZE];  
7. } FDFSStorageBrief;  

6、从tracker获取storage状态。

[cpp] view plain copy  

1. #define TRACKER_PROTO_CMD_STORAGE_GET_STATUS        71  //get storage status from tracker  

该协议是由client发起

调用流程如下：

[cpp] view plain copy  

1. int tracker_get_storage_status(ConnectionInfo *pTrackerServer, \  
2.         const char *group_name, const char *ip_addr, \  
3.         FDFSStorageBrief *pDestBuff)  
4. int tracker_get_storage_max_status(TrackerServerGroup *pTrackerGroup, \  
5.         const char *group_name, const char *ip_addr, \  
6.         char *storage_id, int *status)    
7. int tracker_get_storage_status(ConnectionInfo *pTrackerServer, \  
8.         const char *group_name, const char *ip_addr, \  
9.         FDFSStorageBrief *pDestBuff)  

获取自己的状态，

包体格式   组名 ip的字符串

tracker通过获取了相应的数据，查找到storage的信息

结构体为：

[cpp] view plain copy  

1. typedef struct  
2. {  
3.     char status;  
4.     char port[4];  
5.     char id[FDFS_STORAGE_ID_MAX_SIZE];  
6.     char ip_addr[IP_ADDRESS_SIZE];  
7. } FDFSStorageBrief;  

赋值后，返回

7、通过tracker获取storageid

#define TRACKER_PROTO_CMD_STORAGE_GET_SERVER_ID     70  //get storage server id from tracker

和上以协议请求一样 groupname+ip 组成。

tracker处理方法

[cpp] view plain copy  

1. static int tracker_deal_get_storage_id(struct fast_task_info *pTask)  

tracker最后通过

[cpp] view plain copy  

1. FDFSStorageIdInfo *fdfs_get_storage_id_by_ip(const char *group_name, \  
2.         const char *pIpAddr)  
3. {  
4.     FDFSStorageIdInfo target;  
5.     memset(&target, 0, sizeof(FDFSStorageIdInfo));  
6.     snprintf(target.group_name, sizeof(target.group_name), "%s", group_name);  
7.     snprintf(target.ip_addr, sizeof(target.ip_addr), "%s", pIpAddr);  
8.     return (FDFSStorageIdInfo *)bsearch(&target, g_storage_ids_by_ip, \  
9.         g_storage_id_count, sizeof(FDFSStorageIdInfo), \  
10.         fdfs_cmp_group_name_and_ip);  
11. }  

该方法获取了了

[cpp] view plain copy  

1. FDFSStorageIdInfo  

信息，然后赋值，返回。

8、通过tracker获取所有storage服务器

[cpp] view plain copy  

1. #define TRACKER_PROTO_CMD_STORAGE_FETCH_STORAGE_IDS 69  //get all storage ids from tracker  

[cpp] view plain copy  

1. for (i=0; i<5; i++)  
2. {  
3.     for (pGServer=pServerStart; pGServer<pServerEnd; pGServer++)  
4.     {  
5.         memcpy(pTServer, pGServer, sizeof(ConnectionInfo));  
6.         pTServer->sock = -1;  
7.         result = fdfs_get_storage_ids_from_tracker_server(pTServer);  
8.         if (result == 0)  
9.         {  
10.             return result;  
11.         }  
12.     }  
13.   
14.     if (pServerStart != pTrackerGroup->servers)  
15.     {  
16.         pServerStart = pTrackerGroup->servers;  
17.     }  
18.     sleep(1);  
19. }  

调用顺序

[cpp] view plain copy  

1. int storage_func_init(const char *filename, char *bind_addr, const int addr_size)  
2. int fdfs_get_storage_ids_from_tracker_group(TrackerServerGroup *pTrackerGroup)  
3. int fdfs_get_storage_ids_from_tracker_server(ConnectionInfo *pTrackerServer)  

tracker函数，每秒钟中调用，遍历所有的trackers服务器

tracker服务器获取

[cpp] view plain copy  

1. case TRACKER_PROTO_CMD_STORAGE_FETCH_STORAGE_IDS:  
2.             result = tracker_deal_fetch_storage_ids(pTask);  
3.             break;  

然后通过这种协议格式

[cpp] view plain copy  

1. </pre></p><p>返回的数据</p><p></p><pre name="code" class="cpp">pIdsStart = g_storage_ids_by_ip + start_index;  
2.     pIdsEnd = g_storage_ids_by_ip + g_storage_id_count;  
3.     for (pIdInfo = pIdsStart; pIdInfo < pIdsEnd; pIdInfo++)  
4.     {  
5.         if ((int)(p - pTask->data) > pTask->size - 64)  
6.         {  
7.             break;  
8.         }  
9.   
10.         p += sprintf(p, "%s %s %s\n", pIdInfo->id, \  
11.             pIdInfo->group_name, pIdInfo->ip_addr);  
12.     }  

 返回给请求者。

9、回复给新的storage

[cpp] view plain copy  

1. #define TRACKER_PROTO_CMD_STORAGE_REPLICA_CHG       85  //repl new storage servers  

storage服务器调用流程：

剩下的协议

[cpp] view plain copy  

1. static int tracker_merge_servers(ConnectionInfo *pTrackerServer, \  
2.         FDFSStorageBrief *briefServers, const int server_count)  

[cpp] view plain copy  

1. case TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH_ONE:  
2.             result = tracker_deal_service_query_fetch_update( \  
3.                     pTask, pHeader->cmd);  
4.             break;  
5.         case TRACKER_PROTO_CMD_SERVICE_QUERY_UPDATE:  
6.             result = tracker_deal_service_query_fetch_update( \  
7.                     pTask, pHeader->cmd);  
8.             break;  
9.         case TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH_ALL:  
10.             result = tracker_deal_service_query_fetch_update( \  
11.                     pTask, pHeader->cmd);  
12.             break;  
13.         case TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ONE:  
14.             result = tracker_deal_service_query_storage( \  
15.                     pTask, pHeader->cmd);  
16.             break;  
17.         case TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITH_GROUP_ONE:  
18.             result = tracker_deal_service_query_storage( \  
19.                     pTask, pHeader->cmd);  
20.             break;  
21.         case TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITHOUT_GROUP_ALL:  
22.             result = tracker_deal_service_query_storage( \  
23.                     pTask, pHeader->cmd);  
24.             break;  
25.         case TRACKER_PROTO_CMD_SERVICE_QUERY_STORE_WITH_GROUP_ALL:  
26.             result = tracker_deal_service_query_storage( \  
27.                     pTask, pHeader->cmd);  
28.             break;  
29.         case TRACKER_PROTO_CMD_SERVER_LIST_ONE_GROUP:  
30.             result = tracker_deal_server_list_one_group(pTask);  
31.             break;  
32.         case TRACKER_PROTO_CMD_SERVER_LIST_ALL_GROUPS:  
33.             result = tracker_deal_server_list_all_groups(pTask);  
34.             break;  
35.         case TRACKER_PROTO_CMD_SERVER_LIST_STORAGE:  
36.             result = tracker_deal_server_list_group_storages(pTask);  
37.             break;  
38.         case TRACKER_PROTO_CMD_STORAGE_SYNC_SRC_REQ:  
39.             result = tracker_deal_storage_sync_src_req(pTask);  
40.             break;  
41.         case TRACKER_PROTO_CMD_STORAGE_SYNC_DEST_REQ:  
42.             TRACKER_CHECK_LOGINED(pTask)  
43.             result = tracker_deal_storage_sync_dest_req(pTask);  
44.             break;  
45.         case TRACKER_PROTO_CMD_STORAGE_SYNC_NOTIFY:  
46.             result = tracker_deal_storage_sync_notify(pTask);  
47.             break;  
48.         case TRACKER_PROTO_CMD_STORAGE_SYNC_DEST_QUERY:  
49.             result = tracker_deal_storage_sync_dest_query(pTask);  
50.             break;  
51.         case TRACKER_PROTO_CMD_SERVER_DELETE_STORAGE:  
52.             result = tracker_deal_server_delete_storage(pTask);  
53.             break;  
54.         case TRACKER_PROTO_CMD_SERVER_SET_TRUNK_SERVER:  
55.             result = tracker_deal_server_set_trunk_server(pTask);  
56.             break;  
57.         case TRACKER_PROTO_CMD_STORAGE_REPORT_IP_CHANGED:  
58.             result = tracker_deal_storage_report_ip_changed(pTask);  
59.             break;  
60.         case TRACKER_PROTO_CMD_STORAGE_CHANGELOG_REQ:  
61.             result = tracker_deal_changelog_req(pTask);  
62.             break;  
63.         case TRACKER_PROTO_CMD_STORAGE_PARAMETER_REQ:  
64.             result = tracker_deal_parameter_req(pTask);  
65.             break;  
66.         case FDFS_PROTO_CMD_QUIT:  
67.             close(pTask->ev_read.ev_fd);  
68.             task_finish_clean_up(pTask);  
69.             return 0;  
70.         case FDFS_PROTO_CMD_ACTIVE_TEST:  
71.             result = tracker_deal_active_test(pTask);  
72.             break;  
73.         case TRACKER_PROTO_CMD_TRACKER_GET_STATUS:  
74.             result = tracker_deal_get_tracker_status(pTask);  
75.             break;  
76.         case TRACKER_PROTO_CMD_TRACKER_GET_SYS_FILES_START:  
77.             result = tracker_deal_get_sys_files_start(pTask);  
78.             break;  
79.         case TRACKER_PROTO_CMD_TRACKER_GET_ONE_SYS_FILE:  
80.             result = tracker_deal_get_one_sys_file(pTask);  
81.             break;  
82.         case TRACKER_PROTO_CMD_TRACKER_GET_SYS_FILES_END:  
83.             result = tracker_deal_get_sys_files_end(pTask);  
84.             break;  
85.         case TRACKER_PROTO_CMD_STORAGE_REPORT_TRUNK_FID:  
86.             TRACKER_CHECK_LOGINED(pTask)  
87.             result = tracker_deal_report_trunk_fid(pTask);  
88.             break;  
89.         case TRACKER_PROTO_CMD_STORAGE_FETCH_TRUNK_FID:  
90.             TRACKER_CHECK_LOGINED(pTask)  
91.             result = tracker_deal_get_trunk_fid(pTask);  
92.             break;  
93.         case TRACKER_PROTO_CMD_STORAGE_REPORT_TRUNK_FREE:  
94.             TRACKER_CHECK_LOGINED(pTask)  
95.             result = tracker_deal_report_trunk_free_space(pTask);  
96.             break;  
97.         case TRACKER_PROTO_CMD_TRACKER_PING_LEADER:  
98.             result = tracker_deal_ping_leader(pTask);  
99.             break;  
100.         case TRACKER_PROTO_CMD_TRACKER_NOTIFY_NEXT_LEADER:  
101.             result = tracker_deal_notify_next_leader(pTask);  
102.             break;  
103.         case TRACKER_PROTO_CMD_TRACKER_COMMIT_NEXT_LEADER:  
104.             result = tracker_deal_commit_next_leader(pTask);  
105.             break;  

顶

0

踩

0