Coro如何动态的创建新协程？

yvmu

我刚接触Coro和AnyEvent::HTTP。原理方面还不是很懂，故请教：

网上看到很多例子都是直接for(1..1000)这样直接创建运行全部协程，最多用Coro::Semaphore让几个先跑。

现在，我想在for(1..10000)循环里每次创建完一个协程的时候就$coro->ready，这样让出当前的协程给其他跑（比如用AnyEvent::HTTP的WEB请求）。然后等这些IO非阻塞的协程处理完，再回主程序创建剩余协程。

请问，该方案可行吗？底层的细节又是怎样的？

laputa73

协程可以自动切换cede,也可以手动切换cede to.,可以自动调度shedule,也可以同步等待join
async=new+ready
coro还是要靠自己多写程序测试.
原理可以看py发的中文入门文档
http://bbs.chinaunix.net/forum.php?mod=viewthread&tid=3759030

yvmu

1. #!/usr/bin/perl
   
2. use strict;
   
3. use Coro;
   
4. use AnyEvent::HTTP;
   
5. 
   
6. my @coro;
   
7. for (1..10000){
   
8.    push @coro, async {
   
9.       http_get 'http://www.baidu.com', cb => Coro::rouse_cb;  
   
10.       my @res = Coro::rouse_wait;  
    
11.    }
    
12. }
    

复制代码

上面的代码一次创建了10000个异步执行块，然后根据Coro的调度机制执行。

我在想有没有不是一次性创建10000个执行块的方法，而是根据当前IO吞吐流量的自动创建足够适量的执行块。

我目前想到的方式是：

1. #!/usr/bin/perl
   
2. use strict;
   
3. use Coro;
   
4. use AnyEvent::HTTP;
   
5. 
   
6. my @coro;
   
7. for (1..10000){
   
8.    push @coro, async {
   
9.       http_get 'http://www.baidu.com', cb => Coro::rouse_cb;  
   
10.       my @res = Coro::rouse_wait;  
    
11.       cede；#没创建一个异步执行块就将让出当前的CPU资源，等待调度执行完其他程序块，然后继续创建新异步程序块。
    
12.    }
    
13. }
    

复制代码

可能这样放上代码能更加清楚的说明我的问题。

laputa73

不明白的你的想法.
你是觉得一次10000个并发太多吗?
那就限制一下并发不就好了?

1. use Coro;
   
2. use AnyEvent::Ping;
   
3. 
   
4. my $cocurrent = 100;    # 并发数
   
5. my @todoList = map { "192.168.6." . $_ } (1 .. 254); #
   
6. 
   
7. my @coro;
   
8. my $ping = AnyEvent::Ping->new;
   
9. 
   
10. doit() foreach 1..$cocurrent;
    
11. sub doit{
    
12.     my $ip = shift @todoList;
    
13.     return if not defined $ip;
    
14.          push @coro, async {
    
15.             $ping->ping($ip, 1, Coro::rouse_cb );
    
16.              my @result = Coro::rouse_wait;
    
17.              print "Result($ip): ", $result[0][0][0]," in ", $result[0][0][1], " seconds\n";
    
18.               doit();
    
19.          }     
    
20. }
    
21. 
    
22. 
    

复制代码

yvmu

1. doit() foreach 1..$cocurrent;

复制代码

是这里做了限制吗？那$cocurrent(100)以后的IP怎么去Ping呢？
还有doit()里面调用doit()自己是什么意思？
我感觉用Coro::Semaphore做限制正规点。

我的意思是如何在网络阻塞的时候少创建点并发，在网络状况比较流畅的时候多创点并发。

不知道是否能用Coro的调度机制去实现？

laputa73

你可以转换一下思路,预先创建好协程,需要时才调度
变为需要时才创建协程

以一定的并发去抓取网页.
网络好就做得快点,网络差就做得慢点
尽量不要自己去干涉协程调度.

协程要做得就是从任务队列里面取任务
my $ip = shift @todoList;
然后再递归创建新的协程

yvmu

laputa73 发表于 2013-08-04 20:28
以一定的并发去抓取网页.
网络好就做得快点,网络差就做得慢点
尽量不要自己去干涉协程调度.

那就是手动控制咯？我就是再自动的方法。

laputa73 发表于 2013-08-04 20:28
协程要做得就是从任务队列里面取任务
my $ip = shift @todoList;
然后再迭代新的协程

那你这样写的话，最后是怎样控制住并发的数量的呢？

laputa73

例子都给你了.
你自己把ping的改成http的试试看呗
把你的10000个url放到todolist里面去就ok.

yvmu

1. #!/usr/bin/perl
   
2. use Coro;
   
3. use AnyEvent::HTTP;
   
4. 
   
5. my $cocurrent = 100;
   
6. my @todoList = ('http://www.baidu.com') x 1000;
   
7. 
   
8. my @coro;
   
9. my $i=1;
   
10. 
    
11. doit() foreach 1..$cocurrent;
    
12. sub doit{
    
13.     my $url = shift @todoList;
    
14.     return if not defined $url;
    
15.     push @coro, async {
    
16.         http_get $url, cb => Coro::rouse_cb;
    
17.         print "开始并发$i...\n";
    
18.         my @res = Coro::rouse_wait;
    
19.         print "$i返回结果$res[1]->{Status}\n";
    
20.         $i++;
    
21.         doit();
    
22.     }
    
23. }
    
24. $_->join foreach @coro;
    

复制代码

你确认这样写能控制并发？我觉得不能。

无论是用Coro::Semaphore还是for (1..100){while(){async{}}}都比这个好吧。

yvmu

1. #!/usr/bin/perl
   
2. use Coro;
   
3. use AnyEvent::HTTP;
   
4. 
   
5. my @coro;
   
6. 
   
7. foreach my $i (1..1000){
   
8.     push @coro, async{
   
9.         http_get 'http://www.baidu.com', cb => Coro::rouse_cb;
   
10.         print "开始并发$i...\n";
    
11.         my @res = Coro::rouse_wait;
    
12.         print "$i返回结果$res[1]->{Status}\n";
    
13.     };
    
14.     cede;#请问这里跳转了吗？
    
15. }
    
16. $_->join foreach @coro;
    

复制代码

请问cede这里跳转了吗？为什么程序还是顺序返回了？