求助：Coro多线程采集网页的时候假死问题

laolun

现在在做一个项目，需要采集大量的网页，从其中提取信息。
我打算用Coro来实现，但是用的过程中，会出现假死的情况。不是因为内存或者CPU的原因，这个程序不占CPU或者内存，跟网速有关系。网络情况好的时候，几乎不会出现，反之就经常出现。已经写了一个线程来监控其他线程超时的问题，但是问题依然存在。可能我代码写的有问题，大家帮忙看看。谢谢啦！

1. use strict;
   
2. use warnings;
   
3. use Coro;
   
4. use Coro::Timer;
   
5. use Coro::LWP;
   
6. use LWP::UserAgent;
   
7. use LWP::ConnCache;
   
8. use HTTP::Cookies;
   
9. use Time::HiRes;
   
10. use JSON;
    
11. use Data::Dumper;
    
12. 
    
13. my @urls=();
    
14. open(my $in, "todo.txt") or die $!;
    
15. while(<$in>){
    
16.         chomp $_;
    
17.     push(@urls, $_);
    
18. }
    
19. close($in);
    
20. print scalar(@urls)." pages\n";
    
21. 
    
22. my $MAX_THREADS = 10;
    
23. my @data_queue;
    
24. my @result_queue;
    
25. my $processing_count = 0;
    
26. 
    
27. foreach (1 .. 10)
    
28. {
    
29.         async(\&thread_io);
    
30. }
    
31. 
    
32. # 监视是否线程超时
    
33. async{
    
34.     while(1){
    
35.         Coro::Timer::sleep 10;
    
36.         my @coros=Coro::State::list;
    
37.         #print Dumper(\@coros);
    
38.         my @lwp_coro = grep { $_->desc eq "LWP" } @coros;
    
39. 
    
40.         warn sprintf "%s lwp coro found.", scalar @lwp_coro;
    
41.         for my $coro (@lwp_coro) {
    
42.             my $now=time();
    
43.             if ($now > $coro->{timeout_at}) {
    
44.                 $coro->cancel("timeout");
    
45.                 if(@data_queue){
    
46.                                         async(\&thread_io);
    
47.                 }
    
48.             }
    
49.         }
    
50.         if(@lwp_coro==0){
    
51.             exit;
    
52.         }
    
53.     }
    
54. };
    
55. 
    
56. 
    
57. for (@urls)
    
58. {
    
59.         if ($#data_queue > $MAX_THREADS * 2)
    
60.         {
    
61.                 Coro::Timer::sleep(0.02);
    
62.                 redo;
    
63.         }
    
64.         push(@data_queue, $_);
    
65. 
    
66.         if ($#result_queue > -1)
    
67.         {
    
68.                 # warn $#result_queue;
    
69.                 while ($#result_queue > -1)
    
70.                 {
    
71.                         my $result = shift(@result_queue);
    
72.                         # print "$result\n";
    
73.                         append("content.txt", $result."\n");
    
74.                 }
    
75.         }
    
76. }
    
77. 
    
78. while ($processing_count > 0 or $#data_queue > -1 or $#result_queue > -1)
    
79. {
    
80.         while ($#result_queue > -1)
    
81.         {
    
82.                 my $result = shift(@result_queue);
    
83.                 # print "$result\n";
    
84.                 append("content.txt", $result."\n");
    
85.         }
    
86.         
    
87.         Coro::Timer::sleep(0.02);
    
88. }
    
89. 
    
90. print "done";
    
91. <STDIN>;
    
92. 
    
93. ##########################################################################
    
94. 
    
95. sub thread_io{
    
96.         my $lwp = LWP::UserAgent->new();
    
97.         $lwp->agent("Mozilla/5.0 (Windows NT 6.1; rv:36.0) Gecko/20100101 Firefox/36.0");#设置浏览器属性
    
98.         my $conncache = new LWP::ConnCache;
    
99.         $lwp->conn_cache($conncache);
    
100.         my $cookie_jar = HTTP::Cookies->new(
     
101.             file => "cookies",
     
102.             autosave => 1,
     
103.         );
     
104.         $lwp->cookie_jar($cookie_jar);
     
105.         $Coro::current->desc("LWP");
     
106.         $lwp->timeout(60);
     
107.         coro_timeout(60);
     
108.         while (1)
     
109.         {
     
110.                 if ($#data_queue == -1)
     
111.                 {
     
112.                         Coro::Timer::sleep(1);
     
113.                         next;
     
114.                 }
     
115. 
     
116.                 my $data = shift(@data_queue);
     
117.                 print $data."\t";
     
118.                 ++$processing_count;
     
119.                 my $r = $lwp->get($data);
     
120.                 my $status = $r->status_line();
     
121.                 my $c=filter($r->content());
     
122.                 # push(@result_queue, "$Coro::current $data $status");
     
123.                 if($r->is_success){
     
124.                     print "success\n";
     
125.                     push(@result_queue, "$data\t".dosth($c));
     
126.                 }else{
     
127.                     print "failed\n";
     
128.                     append("failed.txt", $data."\n");
     
129.                 }
     
130.                 --$processing_count;
     
131.         }
     
132. }
     
133. 
     
134. sub coro_timeout {
     
135.     my $timeout = shift;
     
136.     my $coro = $Coro::current;
     
137.     $coro->{timeout_at} = Time::HiRes::time() + $timeout;
     
138.     $coro->on_destroy(sub {
     
139.         my $message = shift;
     
140.         warn sprintf "coro:%s cancel because %s", $coro->desc, $message;
     
141.     });
     
142. }
     
143. 
     
144. sub dosth{
     
145.         my ($c)=@_;
     
146.         # 很简单的字符串处理，省略掉了。
     
147.         return 1;
     
148. }
     
149. 
     
150. 
     
151. sub append{
     
152.         my($fn,$c)=@_;
     
153.         open(my $out, ">>$fn") or die $!;
     
154.         print $out $c;
     
155.         close($out);
     
156. }
     
157. sub filter{
     
158.         my ($str)=@_;
     
159.         $str=~s/^\s+//;
     
160.         $str=~s/\s+$//;
     
161.         $str=~s/\s+/ /igs;
     
162.         return $str;
     
163. }

复制代码

laolun

感觉是访问url的线程发生了阻塞，导致没法调度到监控线程上，所以监控线程也就无效了。
然而，俺并不知道问题出在哪。好悲催，技术好low……

tc1989tc

sorry, kancuolege

tc1989tc

是不是线程cancel的问题啊
有个cancel_safe安全函数 用用试一试

tc1989tc

在获取url的时候$coro->ready 一下看啊

感觉即使发生阻塞 coro调度线程都会把自己加入的ready队列啊
还有就是确认你的监控线程是不是自己exit啦  exit之前打个日志看看嘛。。

104359176

Perl 内置并不支持并发。第三方模块实现的并发，是真的并发吗？有真正支持并发（内置)的语言，要是我的话，就用支持并发的语言实现，并发涉及到底层线程和进程的调度，如果出了问题，不是语言本身的能力，很难跟踪调试。

最稳定的工具是语言，也是最可信赖的东西，要使用第三方模块，必须能跟踪其中的问题。

laolun

谢谢兄的建议。查了下文档，稍微纠正下，应该是 safe_cancel 。从兄的建议中得到了启发，就是线程退出的方法问题。文档中关于线程退出的方法有好几种，等会我都试下。
谢谢！
回复 4# tc1989tc


   

laolun

你是说在：

1. my $data = shift(@data_queue);

复制代码

之后加上一句

1. $Coro::current->ready

复制代码

?

tc1989tc 发表于 2015-08-30 18:05
在获取url的时候$coro->ready 一下看啊

感觉即使发生阻塞 coro调度线程都会把自己加入的ready队列啊

laolun

我估计你说的是这一句：

1. my $r = $lwp->get($data);

复制代码

如果是这一句堵了，不大好弄啊。
回复 5# tc1989tc


   

laolun

我也有这种感觉。
现在网络很顺畅，毫无问题。不会出现卡死的情况，监控线程偶尔会杀掉几个超时的线程。
所以问题可能出在第三方模块（Coro）的get方法上

1. $lwp->get($url);

复制代码

，而这可能跟它模块本身的写法有关，而在外部不好监控和调试。
回复 6# 104359176