/[mldonkey]/mldonkey/donkey/donkeyClient.ml
ViewVC logotype

Contents of /mldonkey/donkey/donkeyClient.ml

Parent Directory Parent Directory | Revision Log Revision Log


Revision 1.19 - (show annotations) (download)
Wed Oct 23 09:54:57 2002 UTC (21 years, 7 months ago) by mldonkey
Branch: MAIN
CVS Tags: release-2-00+1
Changes since 1.18: +1 -0 lines
-

1 (* Copyright 2001, 2002 b8_bavard, b8_fee_carabine, INRIA *)
2 (*
3 This file is part of mldonkey.
4
5 mldonkey is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published by
7 the Free Software Foundation; either version 2 of the License, or
8 (at your option) any later version.
9
10 mldonkey is distributed in the hope that it will be useful,
11 but WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 GNU General Public License for more details.
14
15 You should have received a copy of the GNU General Public License
16 along with mldonkey; if not, write to the Free Software
17 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18 *)
19 (* The function handling the cooperation between two clients. Most used
20 functions are defined in downloadOneFile.ml *)
21
22 (* Some things about Emule:
23 - If a client asks for a file more often than 590 seconds, ie 10 minutes,
24 it gets a BAD REQUEST
25 - If a client has 2 BAD REQUESTS, it get banned !
26 *)
27
28 open CommonRoom
29 open CommonShared
30 open CommonGlobals
31 open CommonFile
32 open CommonClient
33 open CommonComplexOptions
34 open GuiProto
35 open CommonResult
36 open CommonTypes
37 open Options
38 open BasicSocket
39 open DonkeyMftp
40 open DonkeyProtoCom
41 open TcpBufferedSocket
42 open DonkeyTypes
43 open DonkeyOneFile
44 open DonkeyOptions
45 open CommonOptions
46 open DonkeyComplexOptions
47 open DonkeyGlobals
48
49 let client_send_if_possible sock msg =
50 if can_write_len sock (!!client_buffer_size/2) then
51 client_send sock msg
52
53 let verbose_group = ref false
54
55 let tag_udp_client = 203
56
57 let new_udp_client c group =
58 match c.client_kind with
59 Indirect_location _ -> ()
60 | Known_location (ip, port) ->
61 let uc =
62 try
63 let uc = Hashtbl.find udp_clients c.client_kind in
64 uc.udp_client_last_conn <- last_time ();
65 uc
66 with _ ->
67 let uc = {
68 udp_client_last_conn = last_time ();
69 udp_client_ip = ip;
70 udp_client_port = port;
71 udp_client_is_mldonkey = c.client_is_mldonkey >= 2;
72 }
73 in
74 Heap.set_tag uc tag_udp_client;
75 Hashtbl.add udp_clients c.client_kind uc;
76 uc
77 in
78 group.group <- UdpClientMap.add c.client_kind uc group.group
79
80
81 let udp_sock () =
82 match !udp_sock with
83 None -> failwith "No UDP socket"
84 | Some sock -> sock
85
86 let udp_client_send uc t =
87 DonkeyProtoCom.udp_send (udp_sock ())
88 (Unix.ADDR_INET (Ip.to_inet_addr uc.udp_client_ip,uc.udp_client_port+4))
89 t
90
91 let client_udp_send ip port t =
92 DonkeyProtoCom.udp_send (udp_sock ())
93 (Unix.ADDR_INET (Ip.to_inet_addr ip,port+4))
94 t
95
96 let find_sources_in_groups c md4 =
97 if !!propagate_sources then
98 try
99 let group = Hashtbl.find file_groups md4 in
100 try
101 let uc = UdpClientMap.find c.client_kind group.group in
102 uc.udp_client_last_conn <- last_time ()
103 (* the client is already known *)
104 with _ ->
105 (* a new client for this group *)
106 if c.client_is_mldonkey >= 2 then begin
107 match c.client_sock with
108 None -> ()
109 | Some sock ->
110 (* send the list of members of the group to the client *)
111 let list = ref [] in
112 UdpClientMap.iter (fun _ uc ->
113 list := (uc.udp_client_ip, uc.udp_client_port, uc.udp_client_ip) :: !list
114 ) group.group;
115 if !list <> [] then begin
116 if !verbose_group then begin
117 Printf.printf "Send %d sources from file groups to mldonkey peer" (List.length !list); print_newline ();
118 end;
119 let msg =
120 let module Q = DonkeyProtoClient.Sources in
121 DonkeyProtoClient.SourcesReq {
122 Q.md4 = md4;
123 Q.sources = !list;
124 }
125 in
126 client_send_if_possible sock msg
127 end
128 end else
129 begin
130 match c.client_kind with
131 Indirect_location _ -> ()
132 | Known_location (ip, port) ->
133 UdpClientMap.iter (fun _ uc ->
134 if !verbose_group then
135 (Printf.printf "Send source from file groups to UDP peer";
136 print_newline ());
137 client_udp_send ip port (
138 let module Q = DonkeyProtoServer.QueryLocationReply in
139 DonkeyProtoServer.QueryLocationReplyUdpReq {
140 Q.md4 = md4;
141 Q.locs = [{
142 Q.ip = uc.udp_client_ip;
143 Q.port = uc.udp_client_port;
144 }];
145 })
146 ) group.group;
147 end
148 ;
149 match c.client_kind with
150 Indirect_location _ -> ()
151 | Known_location (ip, port) ->
152 (* send this client as a source for the file to
153 all mldonkey clients in the group. add client to group *)
154
155 UdpClientMap.iter (fun _ uc ->
156 if !verbose_group then
157 (Printf.printf "Send new source to file groups UDP peers";
158 print_newline ());
159 udp_client_send uc (
160 let module M = DonkeyProtoServer in
161 M.QueryLocationReplyUdpReq (
162 let module Q = M.QueryLocationReply in
163 {
164 Q.md4 = md4;
165 Q.locs = [{ Q.ip = ip; Q.port = port }];
166 }))
167 ) group.group;
168 new_udp_client c group
169 with _ ->
170 let group = { group = UdpClientMap.empty } in
171 Hashtbl.add file_groups md4 group;
172 new_udp_client c group
173
174 let clean_groups () =
175 let one_day_before = last_time () -. one_day in
176 Hashtbl.iter (fun file group ->
177 let map = group.group in
178 group.group <- UdpClientMap.empty;
179 UdpClientMap.iter (fun v uc ->
180 if uc.udp_client_last_conn > one_day_before then
181 group.group <- UdpClientMap.add v uc group.group
182 ) map
183 ) file_groups
184
185 let client_wants_file c md4 =
186 if md4 <> Md4.null && md4 <> Md4.one then begin
187 find_sources_in_groups c md4;
188 end
189
190
191 let new_chunk up begin_pos end_pos =
192 if begin_pos <> end_pos then
193 let pair = (begin_pos, end_pos) in
194 match up.up_chunks with
195 [] ->
196 up.up_pos <- begin_pos;
197 up.up_end_chunk <- end_pos;
198 up.up_chunks <- [pair]
199 | chunks ->
200 if not (List.mem pair chunks) then
201 up.up_chunks <- chunks @ [pair]
202
203 let rec really_write fd s pos len =
204 if len = 0 then begin
205 Printf.printf "really_write 0 BYTES !!!!!!!!!"; print_newline ();
206 raise End_of_file
207 end else
208 let nwrite = Unix.write fd s pos len in
209 if nwrite = 0 then raise End_of_file else
210 if nwrite < len then
211 really_write fd s (pos + nwrite) (len - nwrite)
212
213
214 let identify_as_mldonkey c sock =
215 direct_client_send sock (
216 let module M = DonkeyProtoClient in
217 let module C = M.QueryFile in
218 M.QueryFileReq Md4.null);
219 direct_client_send sock (
220 let module M = DonkeyProtoClient in
221 let module C = M.QueryFile in
222 M.QueryFileReq Md4.one)
223
224
225 let query_files c sock =
226 List.iter (fun file ->
227 if file_state file = FileDownloading then
228 direct_client_send sock (
229 let module M = DonkeyProtoClient in
230 let module C = M.QueryFile in
231 M.QueryFileReq file.file_md4);
232 ) !current_files;
233 ()
234
235
236 let client_has_chunks c file chunks =
237
238 if not (Intmap.mem (client_num c) file.file_sources) then begin
239 new_source file c;
240 file.file_new_locations <- true
241 end;
242
243 if file.file_chunks_age = [||] then
244 file.file_chunks_age <- Array.create file.file_nchunks 0.0;
245 let change_last_seen = ref false in
246 for i = 0 to file.file_nchunks - 1 do
247 match file.file_chunks.(i) with
248 PresentVerified | PresentTemp ->
249 file.file_chunks_age.(i) <- last_time ()
250 | _ ->
251 if chunks.(i) then begin
252 change_last_seen := true;
253 file.file_chunks_age.(i) <- last_time ()
254 end;
255 done;
256
257 if !change_last_seen then begin
258 try
259 let last_seen = Array2.min file.file_chunks_age in
260 if last_seen > file.file_file.impl_file_last_seen then
261 begin
262 file.file_file.impl_file_last_seen <- last_seen;
263 file_must_update_downloaded (as_file file.file_file)
264 end
265 with _ -> ()
266 end;
267
268 add_client_chunks file chunks;
269
270 match c.client_file_queue with
271 [] ->
272 c.client_file_queue <- [file, chunks];
273 start_download c
274
275 | _ ->
276 c.client_file_queue <- c.client_file_queue @ [file, chunks]
277
278 let add_new_location sock file c =
279
280 let is_new = ref false in
281 let module M = DonkeyProtoClient in
282 if not (Intmap.mem (client_num c) file.file_sources) then begin
283 new_source file c;
284 is_new := true;
285 file.file_new_locations <- true;
286
287 end;
288
289 if last_time () -.
290 connection_last_conn c.client_connection_control < 300.
291 && !!propagate_sources then
292 let send_locations = ref 10 in
293
294 (* Send the known sources to the client. Use UDP for normal clients, TCP
295 for mldonkey clients .*)
296 begin
297 match c.client_kind with
298 Known_location (ip, port) ->
299 (try
300 (* send this location to at most 10 other locations,
301 connected in the last quarter *)
302 let counter = ref 10 in
303
304 (* send this location to all connected locations *)
305 let msg =
306 let module Q = M.Sources in
307 M.SourcesReq {
308 Q.md4 = file.file_md4;
309 Q.sources = [(ip, port, ip)];
310 }
311 in
312 let time = last_time () -. 900. in
313 Intmap.iter (fun _ cc ->
314 if cc.client_checked &&
315 connection_last_conn c.client_connection_control
316 > time then begin
317 match cc.client_kind with
318 Known_location (src_ip, src_port) ->
319 decr counter;
320 if !counter = 0 then raise Exit;
321
322 client_udp_send src_ip src_port (
323 let module Q = DonkeyProtoServer.QueryLocationReply in
324 DonkeyProtoServer.QueryLocationReplyUdpReq {
325 Q.md4 = file.file_md4;
326 Q.locs = [{
327 Q.ip = ip;
328 Q.port = port;
329 }];
330 });
331 | _ ->
332 if cc.client_is_mldonkey > 1 then
333 match cc.client_sock with
334 None -> ()
335 | Some sock -> client_send_if_possible sock msg
336
337 end
338 ) file.file_sources;
339 with _ -> ());
340
341 | _ -> ()
342 end;
343 begin
344 match c.client_kind with
345 Indirect_location _ when c.client_is_mldonkey < 2 -> ()
346 | _ ->
347
348 (* send at most 10 sources connected in the last quarter to this new location *)
349 let sources = ref [] in
350 let time = last_time () -. 900. in
351 (try (* simply send the last found location *)
352 Intmap.iter (fun _ c ->
353 if c.client_checked &&
354 connection_last_conn c.client_connection_control > time
355 then
356 match c.client_kind with
357 Known_location (ip, port) ->
358 sources := (ip, port, ip) :: !sources;
359 decr send_locations;
360 if !send_locations = 0 then
361 raise Not_found
362 | _ -> ()
363 ) file.file_sources;
364 with _ -> ());
365
366 if !sources <> [] then
367 if c.client_is_mldonkey = 2 then begin
368 client_send_if_possible sock ( (
369 let module Q = M.Sources in
370 M.SourcesReq {
371 Q.md4 = file.file_md4;
372 Q.sources = !sources;
373 }))
374 end else begin
375 match c.client_kind with
376 Indirect_location _ -> ()
377 | Known_location (ip, port) ->
378 List.iter (fun (src_ip, src_port, _ ) ->
379 client_udp_send ip port (
380 let module Q = DonkeyProtoServer.QueryLocationReply in
381 DonkeyProtoServer.QueryLocationReplyUdpReq {
382 Q.md4 = file.file_md4;
383 Q.locs = [{
384 Q.ip = src_ip;
385 Q.port = src_port;
386 }];
387 });
388 ) !sources;
389 end;
390 end
391
392
393
394
395 let client_to_client for_files c t sock =
396 let module M = DonkeyProtoClient in
397 match t with
398 M.ConnectReplyReq t ->
399 printf_string "******* [CCONN OK] ********";
400
401 c.client_checked <- true;
402
403 let module CR = M.ConnectReply in
404
405 if t.CR.md4 = !!client_md4 then
406 TcpBufferedSocket.close sock "connected to myself";
407
408 c.client_tags <- t.CR.tags;
409 List.iter (fun tag ->
410 match tag with
411 { tag_name = "name"; tag_value = String s } ->
412 set_client_name c s t.CR.md4
413 | _ -> ()
414 ) c.client_tags;
415
416 connection_ok c.client_connection_control;
417
418 if Ip.valid t.CR.ip_server && !!update_server_list then
419 ignore (add_server t.CR.ip_server t.CR.port_server);
420
421 set_client_state c Connected_idle;
422
423 identify_as_mldonkey c sock;
424 query_files c sock;
425 client_must_update c;
426 if client_type c <> NormalClient then begin
427 if last_time () > c.client_next_view_files then begin
428 (*
429 Printf.printf "****************************************";
430 print_newline ();
431 Printf.printf " ASK VIEW FILES ";
432 print_newline ();
433 *)
434 direct_client_send sock (
435 let module M = DonkeyProtoClient in
436 let module C = M.ViewFiles in
437 M.ViewFilesReq C.t);
438 end
439 end
440
441 | M.ViewFilesReplyReq t ->
442 (*
443 Printf.printf "****************************************";
444 print_newline ();
445 Printf.printf " VIEW FILES REPLY ";
446 print_newline ();
447 *)
448 let module Q = M.ViewFilesReply in
449 begin
450 try
451 let list = ref [] in
452 List.iter (fun f ->
453 match result_of_file f.f_md4 f.f_tags with
454 None -> ()
455 | Some r ->
456 let r = DonkeyIndexer.index_result_no_filter r in
457 client_new_file c r;
458 list := r :: !list
459 ) t;
460 c.client_all_files <- Some !list;
461 client_must_update c
462
463 with e ->
464 Printf.printf "Exception in ViewFilesReply %s"
465 (Printexc.to_string e); print_newline ();
466 end;
467
468 | M.AvailableSlotReq _ ->
469 printf_string "[QUEUED]";
470 set_rtimeout sock !!queued_timeout;
471 (* how long should we wait for a block ? *)
472 begin
473 match c.client_block with
474 None -> find_client_block c
475 | Some b ->
476 printf_string "[QUEUED WITH BLOCK]";
477 end
478 (*
479 | M.QueueReq t ->
480
481 printf_char 'Q';
482 List.iter (fun ip ->
483 let c = new_client (Known_location (ip, 4662)) in
484 if not (List.memq c !interesting_clients) then
485 interesting_clients := c :: !interesting_clients
486 ) t;
487
488 set_client_state c Connected_queued;
489 client_send sock (
490 let module M = DonkeyProtoClient in
491 M.QueueReq t);
492 *)
493
494 | M.JoinQueueReq _ ->
495 if Fifo.length upload_clients < !!max_upload_slots then begin
496 set_rtimeout sock !!upload_timeout;
497
498 direct_client_send sock (
499 let module M = DonkeyProtoClient in
500 let module Q = M.AvailableSlot in
501 M.AvailableSlotReq Q.t);
502 end
503
504 | M.CloseSlotReq _ ->
505 printf_string "[DOWN]"
506
507 | M.ReleaseSlotReq _ ->
508 direct_client_send sock (
509 let module M = DonkeyProtoClient in
510 let module Q = M.CloseSlot in
511 M.CloseSlotReq Q.t);
512
513 | M.QueryFileReplyReq t ->
514 let module Q = M.QueryFileReply in
515
516 begin
517 try
518 let file = find_file t.Q.md4 in
519 if file_state file = FileDownloading then begin
520 printf_string "[FOUND FILE]";
521 c.client_rating <- c.client_rating + 1;
522 if not (List.mem t.Q.name file.file_filenames) then begin
523 file.file_filenames <- file.file_filenames @ [t.Q.name] ;
524 update_best_name file
525 end;
526 if file_size file > block_size then
527 direct_client_send sock (
528 let module M = DonkeyProtoClient in
529 let module C = M.QueryChunks in
530 M.QueryChunksReq file.file_md4)
531 else
532 client_has_chunks c file [| true |]
533 end
534 with _ -> ()
535 end
536
537 | M.QueryChunksReplyReq t ->
538
539 let module Q = M.QueryChunksReply in
540 let file = find_file t.Q.md4 in
541
542 let chunks =
543 if t.Q.chunks = [||] then
544 Array.create file.file_nchunks true
545 else
546 if Array.length t.Q.chunks <> file.file_nchunks then begin
547 Printf.printf "BAD BAD BAD: number of chunks is different %d/%d for %s:%ld on peer" (Array.length t.Q.chunks) file.file_nchunks (Md4.to_string file.file_md4) (file_size file);
548 print_newline ();
549 Array.create file.file_nchunks false
550 (* What should we do ?
551
552 1) Try to recover the correct size of the file: we can use
553 ViewFilesReq on all clients having the file to test what is
554 the most widely used size for this file. Maybe create
555 different instances of the file for each proposed size ?
556
557 *)
558
559 end else
560 t.Q.chunks in
561
562 client_has_chunks c file chunks
563
564 | M.QueryChunkMd4ReplyReq t ->
565 begin
566 let module Q = M.QueryChunkMd4Reply in
567 let file = find_file t.Q.md4 in
568
569 let module Q = M.QueryChunkMd4Reply in
570 if !!verbose then begin
571 Printf.printf "MD4 FOR CHUNKS RECEIVED";
572 print_newline ();
573 end;
574
575 if t.Q.chunks = [||] then
576 file.file_md4s <- [file.file_md4]
577 else
578 if Array.length t.Q.chunks <> file.file_nchunks then begin
579 Printf.printf "BAD BAD BAD (2): number of chunks is different %d/%d for %s:%ld on peer" (Array.length t.Q.chunks) file.file_nchunks (Md4.to_string file.file_md4) (file_size file);
580 print_newline ();
581 (* What should we do ?
582
583 1) Try to recover the correct size of the file: we can use
584 ViewFilesReq on all clients having the file to test what is
585 the most widely used size for this file. Maybe create
586 different instances of the file for each proposed size ?
587
588 Maybe we should allow a degraded mode of download, where each client
589 is checked for the file.
590
591 *)
592
593 end else
594
595 file.file_md4s <- Array.to_list (t.Q.chunks);
596
597 (* if file.file_exists then verify_chunks file *)
598 end
599
600 | M.BlocReq t ->
601 let module Q = M.Bloc in
602 let file = client_file c in
603
604 if file.file_md4 <> t.Q.md4 then begin
605 Printf.printf "BLOC FOR BAD FILE %s/%s !!"
606 (Md4.to_string t.Q.md4) (Md4.to_string file.file_md4);
607 print_newline ();
608 raise Not_found
609 end;
610
611 c.client_rating <- c.client_rating + 10;
612 if file_state file = FilePaused then
613 (next_file c; raise Not_found);
614
615 let begin_pos = t.Q.start_pos in
616 let end_pos = t.Q.end_pos in
617
618 set_client_state c Connected_busy;
619 let len = Int32.sub end_pos begin_pos in
620 download_counter := Int64.add !download_counter (Int64.of_int32 len);
621 begin
622 match c.client_block with
623 None ->
624 printf_string "NO BLOCK EXPECTED FROM CLIENT";
625 raise Not_found
626 | Some b ->
627 let str_begin = Int32.of_int t.Q.bloc_begin in
628
629 if begin_pos < b.block_begin
630 || begin_pos >= b.block_end || end_pos > b.block_end
631 then
632 let chunk_num = Int32.to_int (Int32.div begin_pos block_size)
633 in
634 Printf.printf "%d: Exceeding block boundaries" (client_num c);
635 print_newline ();
636
637 Printf.printf "%ld-%ld (%ld-%ld)"
638 (begin_pos) (end_pos)
639 (b.block_begin) (b.block_end)
640 ;
641 print_newline ();
642
643 List.iter (fun z ->
644 Printf.printf "zone: %ld-%ld"
645 (z.zone_begin) (z.zone_end)
646 ) c.client_zones;
647
648 (* try to recover the corresponding block ... *)
649
650 (match file.file_chunks.(chunk_num) with
651 PresentTemp | PresentVerified ->
652 Printf.printf "ALREADY PRESENT"; print_newline ();
653 | AbsentTemp | AbsentVerified ->
654 Printf.printf "ABSENT (not implemented)";
655 print_newline ();
656 | PartialTemp b | PartialVerified b ->
657 Printf.printf "PARTIAL"; print_newline ();
658
659 (* try to find the corresponding zone *)
660 List.iter (fun z ->
661 if z.zone_begin >= begin_pos &&
662 end_pos > z.zone_begin then begin
663 Printf.printf "BEGIN ZONE MATCHES";
664 print_newline ();
665 end else
666 if z.zone_begin < begin_pos &&
667 begin_pos < z.zone_end &&
668 z.zone_end < end_pos then begin
669 Printf.printf "END ZONE MATCHES";
670 print_newline ();
671 end
672
673 ) b.block_zones
674 );
675 raise Not_found
676 else
677
678 let final_pos = Unix32.seek32 (file_fd file)
679 begin_pos Unix.SEEK_SET in
680 if final_pos <> begin_pos then begin
681 Printf.printf "BAD LSEEK %ld/%ld"
682 (final_pos)
683 (begin_pos); print_newline ();
684 raise Not_found
685 end;
686 if c.client_connected then
687 printf_string "#[OUT]"
688 else
689 printf_string "#[IN]";
690
691 (* if !!verbose then begin
692 Printf.printf "{%d-%d = %ld-%ld}" (t.Q.bloc_begin)
693 (t.Q.bloc_len) (begin_pos)
694 (end_pos);
695 print_newline ();
696 end; *)
697 try
698 let fd = try
699 Unix32.force_fd (file_fd file)
700 with e ->
701 Printf.printf "In Unix32.force_fd"; print_newline ();
702 raise e
703 in
704 really_write fd t.Q.bloc_str t.Q.bloc_begin t.Q.bloc_len;
705 List.iter (update_zone file begin_pos end_pos) c.client_zones;
706 find_client_zone c;
707 with
708 End_of_file ->
709 Printf.printf "WITH CLIENT %s" c.client_name;
710 print_newline ();
711 | e ->
712 Printf.printf "Error %s while writing block. Pausing download" (Printexc.to_string e);
713 print_newline ();
714 file_pause (as_file file.file_file);
715 info_change_file file
716
717 end;
718
719
720 (* Upload requests *)
721 | M.ViewFilesReq t when !has_upload = 0 ->
722 let files = DonkeyShare.all_shared () in
723 (*
724 Printf.printf "ASK VIEW FILES"; print_newline ();
725 *)
726 direct_client_send_files sock files
727
728 | M.QueryFileReq t when !has_upload = 0 ->
729
730 (try client_wants_file c t with _ -> ());
731 if t = Md4.null && c.client_is_mldonkey = 0 then
732 c.client_is_mldonkey <- 1;
733 if t = Md4.one && c.client_is_mldonkey = 1 then begin
734 printf_string "[MLDONKEY]";
735 c.client_is_mldonkey <- 2;
736 end;
737
738 begin try
739 let file = find_file t in
740 (match file.file_shared with
741 None -> ()
742 | Some impl ->
743 shared_must_update_downloaded (as_shared impl);
744 impl.impl_shared_requests <- impl.impl_shared_requests + 1);
745 direct_client_send sock (
746 let module Q = M.QueryFileReply in
747 M.QueryFileReplyReq {
748 Q.md4 = file.file_md4;
749 Q.name = file_best_name file
750 });
751
752 add_new_location sock file c
753
754 with _ -> () end
755
756 | M.SourcesReq t ->
757
758 let module Q = M.Sources in
759 begin
760 try
761 let file = find_file t.Q.md4 in
762 List.iter (fun (ip1, port, ip2) ->
763 if Ip.valid ip1 then
764 let c = new_client (Known_location (ip1, port)) in
765 if not (Intmap.mem (client_num c) file.file_sources) then
766 new_source file c;
767 (* we should probably only add this client if it is not already there ...
768 clients_list := (c, [file]) :: !clients_list;
769 incr clients_list_len;
770 *)
771 ) t.Q.sources
772 with _ -> ()
773 end
774
775 | M.SayReq s ->
776
777 let ad_opt =
778 match c.client_kind with
779 Known_location (ip, port) ->
780 (
781 match c.client_chat_port with
782 0 -> None
783 | p ->Some (Ip.to_string ip, p)
784 )
785 | Indirect_location _ -> None
786 in
787 (* A VOIR : historique à gérer *)
788 (* !say_hook c s *)
789 private_message_from (as_client c.client_client) s
790
791 | M.QueryChunkMd4Req t when !has_upload = 0 ->
792
793 let file = find_file t in
794 begin
795 match file.file_md4s with
796 [] -> () (* should not happen *)
797 | md4s ->
798 direct_client_send sock (
799 let module Q = M.QueryChunkMd4Reply in
800 M.QueryChunkMd4ReplyReq {
801 Q.md4 = file.file_md4;
802 Q.chunks = Array.of_list md4s
803 })
804
805 end
806
807 | M.QueryChunksReq t when !has_upload = 0 ->
808
809 let file = find_file t in
810 direct_client_send sock (
811 let module Q = M.QueryChunksReply in
812 M.QueryChunksReplyReq {
813 Q.md4 = file.file_md4;
814 Q.chunks = Array.map (fun state ->
815 match state with
816 PresentVerified -> true
817 | _ -> false
818 ) file.file_chunks;
819 })
820
821 | M.QueryBlocReq t when !has_upload = 0 ->
822
823 set_rtimeout sock !!upload_timeout;
824 let module Q = M.QueryBloc in
825 let file = find_file t.Q.md4 in
826
827 let up, waiting = match c.client_upload with
828 Some ({ up_file = f } as up) when f == file -> up, up.up_waiting
829 | Some old_up ->
830 {
831 up_file = file;
832 up_pos = Int32.zero;
833 up_end_chunk = Int32.zero;
834 up_chunks = [];
835 up_waiting = old_up.up_waiting;
836 }, old_up.up_waiting
837 | _ ->
838 {
839 up_file = file;
840 up_pos = Int32.zero;
841 up_end_chunk = Int32.zero;
842 up_chunks = [];
843 up_waiting = false;
844 }, false
845 in
846 new_chunk up t.Q.start_pos1 t.Q.end_pos1;
847 new_chunk up t.Q.start_pos2 t.Q.end_pos2;
848 new_chunk up t.Q.start_pos3 t.Q.end_pos3;
849 c.client_upload <- Some up;
850 if not waiting && !has_upload = 0 then begin
851 Fifo.put upload_clients c;
852 up.up_waiting <- true
853 end
854
855 | _ -> ()
856
857 let client_handler c sock event =
858 match event with
859 BASIC_EVENT (CLOSED s) ->
860 disconnect_client c
861 | _ -> ()
862
863 let client_handler2 c sock event =
864 match !c with
865 Some c -> client_handler c sock event
866 | None ->
867 match event with
868 BASIC_EVENT (CLOSED s) ->
869 printf_string "-c";
870 | _ -> ()
871
872 let init_connection sock =
873 ignore (setsock_iptos_throughput (fd (TcpBufferedSocket.sock sock)));
874 TcpBufferedSocket.set_read_controler sock download_control;
875 TcpBufferedSocket.set_write_controler sock upload_control;
876 set_rtimeout sock !!client_timeout;
877 set_handler sock (BASIC_EVENT RTIMEOUT) (fun s ->
878 printf_string "[TO?]";
879 close s "timeout"
880 )
881
882 let init_client sock c =
883 set_handler sock WRITE_DONE (fun s ->
884 match c.client_upload with
885 None -> ()
886 | Some up ->
887 if not up.up_waiting && !has_upload = 0 then begin
888 up.up_waiting <- true;
889 Fifo.put upload_clients c
890 end
891 );
892 set_handler sock (BASIC_EVENT RTIMEOUT) (fun s ->
893 connection_delay c.client_connection_control;
894 printf_string "-!C";
895 close s "timeout"
896 );
897 c.client_block <- None;
898 c.client_zones <- [];
899 c.client_file_queue <- []
900
901
902
903 let read_first_message t sock =
904 let module M = DonkeyProtoClient in
905 match t with
906
907 | M.ConnectReq t ->
908 printf_string "******* [PCONN OK] ********";
909
910 let module CR = M.Connect in
911
912 if t.CR.md4 = !!client_md4 then begin
913 TcpBufferedSocket.close sock "connected to myself";
914 raise End_of_file
915 end;
916
917 let name = ref "" in
918 List.iter (fun tag ->
919 match tag with
920 { tag_name = "name"; tag_value = String s } -> name := s
921 | _ -> ()
922 ) t.CR.tags;
923
924 let kind = Indirect_location (!name,t.CR.md4) in
925 let c = new_client kind in
926
927 begin
928 match c.client_sock with
929 None ->
930 c.client_sock <- Some sock;
931 c.client_connected <- false;
932 init_client sock c
933 | Some _ ->
934 close sock "already connected"; raise Not_found
935 end;
936
937 set_write_power sock c.client_power;
938 set_read_power sock c.client_power;
939
940 set_client_name c !name t.CR.md4;
941 connection_ok c.client_connection_control;
942 c.client_tags <- t.CR.tags;
943
944 if Ip.valid t.CR.ip_server && !!update_server_list then
945 ignore (add_server t.CR.ip_server t.CR.port_server);
946
947 direct_client_send sock (
948 let module M = DonkeyProtoClient in
949 let module C = M.ConnectReply in
950 M.ConnectReplyReq {
951 C.md4 = !!client_md4;
952 C.ip = client_ip (Some sock);
953 C.port = !client_port;
954 C.tags = !client_tags;
955 C.ip_server = t.CR.ip_server;
956 C.port_server = t.CR.port_server;
957 }
958 );
959
960 set_client_state c Connected_idle;
961 identify_as_mldonkey c sock;
962 query_files c sock;
963 if client_type c <> NormalClient then
964 if last_time () > c.client_next_view_files then begin
965 (*
966 Printf.printf "****************************************";
967 print_newline ();
968 Printf.printf " ASK VIEW FILES ";
969 print_newline ();
970 *)
971
972 direct_client_send sock (
973 let module M = DonkeyProtoClient in
974 let module C = M.ViewFiles in
975 M.ViewFilesReq C.t);
976 end;
977 client_must_update c;
978 Some c
979
980 | M.NewUserIDReq _ ->
981 M.print t; print_newline ();
982 None
983
984 | _ ->
985 Printf.printf "BAD MESSAGE FROM CONNECTING CLIENT"; print_newline ();
986 M.print t; print_newline ();
987 close sock "bad connecting message";
988 raise Not_found
989
990
991 let reconnect_client c =
992 if can_open_connection () then
993 match c.client_kind with
994 Indirect_location _ -> ()
995 | Known_location (ip, port) ->
996 try
997 set_client_state c Connecting;
998 connection_try c.client_connection_control;
999
1000 printf_string "?C";
1001 let sock = TcpBufferedSocket.connect "donkey to client" (
1002 Ip.to_inet_addr ip)
1003 port
1004 (client_handler c) (*client_msg_to_string*) in
1005 TcpBufferedSocket.set_write_power sock c.client_power;
1006 TcpBufferedSocket.set_read_power sock c.client_power;
1007 init_connection sock;
1008 init_client sock c;
1009
1010 set_reader sock (DonkeyProtoCom.cut_messages DonkeyProtoClient.parse
1011 (client_to_client files c));
1012
1013 c.client_sock <- Some sock;
1014 c.client_connected <- true;
1015 let server_ip, server_port =
1016 try
1017 let s = DonkeyGlobals.last_connected_server () in
1018 s.server_ip, s.server_port
1019 with _ -> Ip.localhost, 4665
1020 in
1021 direct_client_send sock (
1022 let module M = DonkeyProtoClient in
1023 let module C = M.Connect in
1024 M.ConnectReq {
1025 C.md4 = !!client_md4; (* we want a different id each conn *)
1026 C.ip = client_ip None;
1027 C.port = !client_port;
1028 C.tags = !client_tags;
1029 C.version = 16;
1030 C.ip_server = server_ip;
1031 C.port_server = server_port;
1032 }
1033 )
1034
1035 with e ->
1036 Printf.printf "Exception %s in client connection"
1037 (Printexc.to_string e);
1038 print_newline ();
1039 connection_failed c.client_connection_control;
1040 set_client_state c NotConnected
1041
1042 let schedule_client c =
1043 if not c.client_on_list then
1044 match c.client_sock with
1045 Some _ -> ()
1046 | None ->
1047 match c.client_kind with
1048 Known_location _ ->
1049 let next_try = connection_next_try c.client_connection_control in
1050 let delay = next_try -. last_time () in
1051 c.client_on_list <- delay < 240.;
1052 (*
1053 Printf.printf "NEXT TRY IN %f(last_try %s current %s)" delay
1054 (Date.to_string c.client_connection_control.control_last_try)
1055 (Date.to_string (last_time ()))
1056
1057 ; print_newline ();
1058 *)
1059 if delay < 0. then
1060 clients_lists.(0) <- c :: clients_lists.(0)
1061 else
1062 if delay < 60. then
1063 clients_lists.(1) <- c :: clients_lists.(1)
1064 else
1065 if delay < 120. then
1066 clients_lists.(2) <- c :: clients_lists.(2)
1067 else
1068 if delay < 180. then
1069 clients_lists.(3) <- c :: clients_lists.(3)
1070 else
1071 if delay < 240. then
1072 clients_lists.(4) <- c :: clients_lists.(4)
1073 else begin
1074 (*
1075 Printf.printf "NEXT TRY TOO LONG: %2.2f (%s)"
1076 delay (Date.to_string next_try);
1077 print_newline ();
1078 *)
1079 ()
1080 end
1081 | _ -> ()
1082
1083 let unschedule_client c =
1084 if c.client_on_list then begin
1085 clients_lists.(0) <- List2.removeq c clients_lists.(0);
1086 clients_lists.(1) <- List2.removeq c clients_lists.(1);
1087 clients_lists.(2) <- List2.removeq c clients_lists.(2);
1088 clients_lists.(3) <- List2.removeq c clients_lists.(3);
1089 clients_lists.(4) <- List2.removeq c clients_lists.(4);
1090 c.client_on_list <- false;
1091 end
1092
1093 let force_fast_connect_client c =
1094 unschedule_client c;
1095
1096 connection_must_try c.client_connection_control;
1097 if can_open_connection () && c.client_sock = None then
1098 reconnect_client c
1099
1100 else begin
1101 clients_lists.(0) <- c :: clients_lists.(0);
1102 c.client_on_list <- true
1103 end
1104
1105 let connect_as_soon_as_possible c =
1106 unschedule_client c;
1107 let control = c.client_connection_control in
1108 control.control_last_try <- control.control_last_ok;
1109 control.control_state <- 1.;
1110 schedule_client c
1111
1112 let query_id_reply s t =
1113 let module M = DonkeyProtoServer in
1114 let module Q = M.QueryIDReply in
1115 if Ip.valid t.Q.ip then
1116 let c = new_client (Known_location (t.Q.ip, t.Q.port)) in
1117 connect_as_soon_as_possible c
1118
1119 let query_id s sock ip =
1120 printf_string "[QUERY ID]";
1121 direct_server_send sock (
1122 let module M = DonkeyProtoServer in
1123 let module C = M.QueryID in
1124 M.QueryIDReq ip
1125 )
1126
1127 let udp_server_send s t =
1128 DonkeyProtoCom.udp_send (udp_sock ())
1129 (Unix.ADDR_INET (Ip.to_inet_addr s.server_ip,s.server_port+4))
1130 t
1131
1132 let query_locations_reply s t =
1133 let module M = DonkeyProtoServer in
1134 let module Q = M.QueryLocationReply in
1135
1136 let nlocs = List.length t.Q.locs in
1137 s.server_score <- s.server_score + 3;
1138
1139 List.iter (fun l ->
1140 let ip = l.Q.ip in
1141 let port = l.Q.port in
1142
1143 if Ip.valid ip then
1144 let c = new_client (Known_location (ip, port)) in
1145 connect_as_soon_as_possible c
1146 else
1147 match s.server_sock with
1148 None ->
1149 let module Q = M.QueryCallUdp in
1150 udp_server_send s
1151 (M.QueryCallUdpReq {
1152 Q.ip = client_ip None;
1153 Q.port = !client_port;
1154 Q.id = ip;
1155 })
1156
1157 | Some sock ->
1158 printf_string "QUERY ID";
1159 query_id s sock ip
1160 ) t.Q.locs
1161
1162 let client_connection_handler t event =
1163 printf_string "[REMOTE CONN]";
1164 match event with
1165 TcpServerSocket.CONNECTION (s, Unix.ADDR_INET (from_ip, from_port)) ->
1166
1167 if can_open_connection () then
1168 begin
1169 try
1170 let c = ref None in
1171 let sock =
1172 TcpBufferedSocket.create "donkey client connection" s
1173 (client_handler2 c)
1174 (*client_msg_to_string*)
1175 in
1176 init_connection sock;
1177
1178 (try
1179 set_reader sock
1180 (DonkeyProtoCom.client_handler2 c read_first_message
1181 (client_to_client []));
1182
1183 with e -> Printf.printf "Exception %s in init_connection"
1184 (Printexc.to_string e);
1185 print_newline ());
1186 with e ->
1187 Printf.printf "Exception %s in client_connection_handler"
1188 (Printexc.to_string e);
1189 print_newline ();
1190 Unix.close s
1191 end
1192 else begin
1193 Unix.close s
1194 end;
1195 | _ ->
1196 ()
1197
1198 let retry_connect_in c delay = ()
1199 (*
1200 let try_connection () =
1201 if CommonGlobals.can_open_connection () then
1202 match client_schedule.(0) with
1203 [] -> ()
1204 | c :: tail ->
1205 client_schedule.(0) <- tail;
1206 c.client_next_schedule <- -1;
1207 retry_connect_in c 20;
1208 match c.client_sock with
1209 None ->
1210 Printf.printf "Should try connection"; print_newline ();
1211 | Some sock -> ()
1212 *)
1213 let schedule_connections () =
1214 ()
1215 (*
1216 decr client_schedule_remaining_seconds;
1217 if !client_schedule_remaining_seconds = 0 then begin
1218 client_schedule_remaining_seconds := 60;
1219 incr client_schedule_minute;
1220 let remaining_connects = client_schedule.(0) in
1221 List.iter (fun c ->
1222 c.client_next_schedule <- !client_schedule_minute)
1223 remaining_connects;
1224 Array.blit client_schedule 1 client_schedule 0
1225 (client_schedule_size - 1);
1226 client_schedule.(0) <- remaining_connects @ client_schedule.(0);
1227 client_schedule.(client_schedule_size - 1) <- [];
1228 end;
1229 let nwaiting = List.length client_schedule.(0) in
1230 for i = 0 to min (nwaiting / !client_schedule_remaining_seconds + 1)
1231 !!max_clients_per_second do
1232 try_connection ()
1233 done
1234 *)

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26