Axel之 -axel_do剖析

axel_do主体部分，尝试从多个连接select方式去读取数据，如果读取失败或者连接超时就重新连接。

下面是代码分析.

1. //下载的主循环
   
2. void axel_do( axel_t *axel )
   
3. {
   
4. fd_set fds[1];
   
5. int hifd, i;
   
6. long long int remaining,size;
   
7. struct timeval timeval[1];
   
8. 
   
9. /* Create statefile if necessary                                */
   
10. //如果到了保存状态的时间，保存当前状态到状态文件
    
11. if( gettime() > axel->next_state )
    
12. {
    
13. //把连接的个数，当前下载量，每个连接当前的下载进度保存起来，
    
14.                 //如果状态文件已经存在，就清空重写，目的是只保存一份
    
15.                 //这样的话，如果程序异常退出，下次开启任务时，就能从状态文件中
    
16.                 //重新加载最近的下载状态(不是100%准确的状态，因为保存状态是有周期的),
    
17.                 //接着下载
    
18. save_state( axel );
    
19. //计算下一次状态保存时间
    
20. axel->next_state = gettime() + axel->conf->save_state_interval;
    
21. }
    
22. 
    
23.                 //采用多线程做连接，但是还是单线程select做传输
    
24.         /* Wait for data on (one of) the connections                        */
    
25. FD_ZERO( fds );
    
26. hifd = 0;
    
27. //经典的select方式，把有用的socket描述符加入到集合中
    
28.         //而且，限于文件下载的特殊任务方式，基本上每个建立好的连接
    
29.         //都不会太空，有恒定持续的数据传输，因此在这里select的效率并不弱于epoll
    
30. for( i = 0; i < axel->conf->num_connections; i ++ )
    
31. {
    
32. if( axel->conn[i].enabled )
    
33. FD_SET( axel->conn[i].fd, fds );
    
34. hifd = max( hifd, axel->conn[i].fd );
    
35. }
    
36. //没有任何连接，等待重试
    
37. if( hifd == 0 )
    
38. {
    
39.   /* No connections yet. Wait...                                */
    
40. usleep( 100000 );
    
41. goto conn_check;
    
42. }
    
43. else
    
44. {
    
45. //超时0.1秒
    
46. timeval->tv_sec = 0;
    
47. timeval->tv_usec = 100000;
    
48. /* A select() error probably means it was interrupted
    
49.                    by a signal, or that something else's very wrong...        */
    
50. //select等待数据到来
    
51. if( select( hifd + 1, fds, NULL, NULL, timeval ) == -1 )
    
52. {
    
53. //值位ready为-1，这样，下载的主循环将退出，下载失败
    
54. axel->ready = -1;
    
55. return;
    
56. }
    
57. }
    
58. 
    
59.   /* Handle connections which need attention                        */
    
60.         //循环读取每个socket的数据
    
61. for( i = 0; i < axel->conf->num_connections; i ++ )
    
62. if( axel->conn[i].enabled ) {
    
63.   //检测，如果该socket有数据到来，就读取
    
64. if( FD_ISSET( axel->conn[i].fd, fds ) )
    
65. {
    
66. //更新最后一次读取数据的时间
    
67. axel->conn[i].last_transfer = gettime();
    
68. //尝试读取数据
    
69. size = read( axel->conn[i].fd, buffer, axel->conf->buffer_size );
    
70.   //读取失败
    
71. if( size == -1 )
    
72. {
    
73. if( axel->conf->verbose )
    
74. {
    
75. axel_message( axel, _("Error on connection %i! "
    
76. "Connection closed"), i );
    
77. }
    
78.   //关闭当前连接，并不等于放弃，还可能重新连接。。
    
79. axel->conn[i].enabled = 0;
    
80. conn_disconnect( &axel->conn[i] );
    
81. continue;
    
82. }//当前连接的数据读取结束
    
83. else if( size == 0 )
    
84. {
    
85. if( axel->conf->verbose )
    
86. {
    
87. /* Only abnormal behaviour if:                */
    
88. if( axel->conn[i].currentbyte < axel->conn[i].lastbyte && axel->size != INT_MAX )
    
89. {
    
90. axel_message( axel, _("Connection %i unexpectedly closed"), i );
    
91. }
    
92. else
    
93. {
    
94. axel_message( axel, _("Connection %i finished"), i );
    
95. }
    
96. }
    
97. //如果是不支持并发分片下载(也就是说是单连接下载)，表明下载完成
    
98. if( !axel->conn[0].supported )
    
99. {
    
100. axel->ready = 1;
     
101. }
     
102. axel->conn[i].enabled = 0;
     
103. conn_disconnect( &axel->conn[i] );
     
104. continue;
     
105. }
     
106.   /* remaining == Bytes to go                                        */
     
107. remaining = axel->conn[i].lastbyte - axel->conn[i].currentbyte + 1;
     
108. //需要填充的小于读取的，下载完成
     
109. if( remaining < size )
     
110. {
     
111. if( axel->conf->verbose )
     
112. {
     
113. axel_message( axel, _("Connection %i finished"), i );
     
114. }
     
115. axel->conn[i].enabled = 0;
     
116. //关闭连接
     
117. conn_disconnect( &axel->conn[i] );
     
118. //修改需要的数据量大小，比如需要20字节，下载了30字节，那么就只要20字节
     
119. size = remaining;
     
120. /* Don't terminate, still stuff to write!        */
     
121. }
     
122.     /* This should always succeed..                                */
     
123.                 //调整偏移，写文件
     
124. lseek( axel->outfd, axel->conn[i].currentbyte, SEEK_SET );
     
125. if( write( axel->outfd, buffer, size ) != size )
     
126. {
     
127. //写失败，退出
     
128. axel_message( axel, _("Write error!") );
     
129. axel->ready = -1;
     
130. return;
     
131. }
     
132. //修改偏移
     
133. axel->conn[i].currentbyte += size;
     
134. axel->bytes_done += size;
     
135. }
     
136. else //当前socket描述符不在select中，检查超时
     
137. {
     
138. //传输超时，关闭连接
     
139. if( gettime() > axel->conn[i].last_transfer + axel->conf->connection_timeout )
     
140. {
     
141. if( axel->conf->verbose )
     
142. axel_message( axel, _("Connection %i timed out"), i );
     
143. conn_disconnect( &axel->conn[i] );
     
144. axel->conn[i].enabled = 0;
     
145. }
     
146. } }
     
147. //如果一切ok，就返回
     
148. if( axel->ready )
     
149. return;
     
150. 
     
151. conn_check:
     
152.   /* Look for aborted connections and attempt to restart them.        */
     
153.         //检查有问题的连接,如果未下载完，并且出错了，重新启动线程，开始传输
     
154. for( i = 0; i < axel->conf->num_connections; i ++ )
     
155. {
     
156. //连接无效并且未下载完
     
157. if( !axel->conn[i].enabled && axel->conn[i].currentbyte < axel->conn[i].lastbyte )
     
158. {
     
159. //状态为0,表明setup_thread是成功执行了的，并且已经执行问，因此，连接的初始化没问题，调用join回收
     
160. if( axel->conn[i].state == 0 )
     
161. {        
     
162. // Wait for termination of this thread
     
163. pthread_join(*(axel->conn[i].setup_thread), NULL);
     
164. 
     
165. conn_set( &axel->conn[i], axel->url->text );
     
166. axel->url = axel->url->next;
     
167. /* axel->conn[i].local_if = axel->conf->interfaces->text;
     
168.                                 axel->conf->interfaces = axel->conf->interfaces->next; */
     
169. if( axel->conf->verbose >= 2 )
     
170. axel_message( axel, _("Connection %i downloading from %s:%i using interface %s"),
     
171. i, axel->conn[i].host, axel->conn[i].port, axel->conn[i].local_if );
     
172. 
     
173. //状态设置为1，表示setup_thread开始执行，设置为0，表示setup_thread线程执行结束
     
174. axel->conn[i].state = 1;
     
175. if( pthread_create( axel->conn[i].setup_thread, NULL, setup_thread, &axel->conn[i] ) == 0 )
     
176. {
     
177. axel->conn[i].last_transfer = gettime();
     
178. }
     
179. else
     
180. {
     
181. axel_message( axel, _("pthread error!!!") );
     
182. axel->ready = -1;
     
183. }
     
184. }
     
185. else //setup_thread线程还未执行完，也就是说连接建立过程还未完成，需要检查连接超时
     
186. {
     
187. //超时了，就取消她...
     
188. if( gettime() > axel->conn[i].last_transfer + axel->conf->reconnect_delay )
     
189. {
     
190. pthread_cancel( *axel->conn[i].setup_thread );
     
191. axel->conn[i].state = 0;
     
192. }
     
193. }
     
194. }
     
195. }
     
196. 
     
197. /* Calculate current average speed and finish_time                */
     
198.         //计算平均速度
     
199. axel->bytes_per_second = (int) ( (double) ( axel->bytes_done - axel->start_byte ) / ( gettime() - axel->start_time ) );
     
200. //估算结束时间
     
201. axel->finish_time = (int) ( axel->start_time + (double) ( axel->size - axel->start_byte ) / axel->bytes_per_second );
     
202. 
     
203. /* Check speed. If too high, delay for some time to slow things
     
204.            down a bit. I think a 5% deviation should be acceptable.        */
     
205. //速度调整
     
206. if( axel->conf->max_speed > 0 )
     
207. {
     
208. //如果超速了
     
209. if( (float) axel->bytes_per_second / axel->conf->max_speed > 1.05 )
     
210. axel->delay_time += 10000;
     
211. //速度太低，少休息会儿
     
212. else if( ( (float) axel->bytes_per_second / axel->conf->max_speed < 0.95 ) && ( axel->delay_time >= 10000 ) )
     
213. axel->delay_time -= 10000;
     
214.   //速度太低，干脆不休息
     
215. else if( ( (float) axel->bytes_per_second / axel->conf->max_speed < 0.95 ) )
     
216. axel->delay_time = 0;
     
217. usleep( axel->delay_time );
     
218. }
     
219. 
     
220. /* Ready?                                                        */
     
221.         //下载完了？
     
222. if( axel->bytes_done == axel->size )
     
223. axel->ready = 1;
     
224. }