Coro/myhttpd/queue.pl

package transferqueue;

use Scalar::Util;

my @reserve = (
      [  1_200_000, 1],
      [  8_000_000, 1],
      [ 75_000_000, 1],
);

sub new {
   my $class = shift;
   my $self = bless {
      slots   => $_[0],
      lastspb => 0,
      avgspb  => 0,
   }, $class;
   $self->{reschedule} = EV::timer 10, 3, sub { $self->wake_next };
   $self;
}

sub start_transfer {
   my $self = shift;
   my $size = $_[0];

   my $transfer = bless {
      queue   => $self,
      time    => $::NOW,
      size    => $size,
      coro    => $Coro::current,
      started => 0,
   }, transfer::;

   push @{$self->{wait}}, $transfer;

   $self->wake_next;

   $transfer;
}

sub sort {
   my @queue = grep $_, @{$_[0]{wait}};

   $_->{spb} = ($::NOW-$_->{time}) / ($_->{size} || 1) for @queue;
   
   $_[0]{wait} = [sort { $b->{spb} <=> $a->{spb} } @queue];

   Scalar::Util::weaken $_ for @{$_[0]{wait}};
}

sub wake_next {
   my $self = shift;

   $self->sort;

   while (@{$self->{wait}}) {
      my $size = $self->{wait}[0]{size};
      my $min = 0;
      for (@reserve) {
         last if $size <= $_->[0];
         $min += $_->[1];
      }
      last unless $self->{slots} > $min;
      my $transfer = shift @{$self->{wait}};
      $self->{lastspb} = $transfer->{spb};
      $self->{avgspb} = $self->{avgspb} * 0.999 + $transfer->{spb} * 0.001;
      $self->{started}++;
      $transfer->wake;
      last;
   }
}

sub force_wake_next {
   my $self = shift;

   $self->{slots} += $::MAX_TRANSFERS;
   $self->wake_next;
   $self->{slots} -= $::MAX_TRANSFERS;
}

sub waiters {
   $_[0]->sort;
   @{$_[0]{wait}};
}

package transfer;

use Coro::Timer ();

sub wake {
   my $self = shift;

   $self->{alloc} = 1;
   $self->{queue}{slots}--;
   $self->{wake} and $self->{wake}->ready;
}

sub try {
   my $self = shift;

   $self->{alloc} || do {
      my $timeout = Coro::Timer::timeout $_[0];
      local $self->{wake} = $self->{coro};

      Coro::schedule;

      $self->{alloc};
   }
}

sub DESTROY {
   my $self = shift;

   if ($self->{alloc}) {
      $self->{queue}{slots}++;
      $self->{queue}->wake_next;
   }
}

1;

Revision:	1.5
Committed:	Sat Dec 8 21:01:17 2007 UTC (18 years, 10 months ago) by root
Content type:	text/plain
Branch:	MAIN
CVS Tags:	rel-4_91, rel-5_151, rel-5_1, rel-5_0, rel-4_748, rel-4_8, rel-4_9, rel-4_741, rel-4_743, rel-4_742, rel-4_744, rel-4_747, rel-6_13, rel-5_161, rel-4_74, rel-4_71, rel-4_72, rel-4_73, rel-5_162, rel-5_2, rel-4_802, rel-4_803, rel-4_801, rel-4_804, rel-4_479, rel-4_50, rel-4_51, rel-4_4, rel-4_45, rel-4_745, rel-4_901, rel-4_49, rel-4_48, rel-4_746, rel-5_11, rel-5_12, rel-5_15, rel-5_14, rel-5_17, rel-5_16, rel-4_47, rel-4_46, rel-4_7, rel-5_132, rel-5_131, rel-4_911, rel-4_912, rel-4_32, rel-4_33, rel-4_34, rel-4_35, rel-4_36, rel-4_37
Changes since 1.4:	+1 -11 lines
Log Message:	convert from Event to EV
#	User	Rev	Content
1	root	1.1	package transferqueue;
2
3			use Scalar::Util;
4
5			my @reserve = (
6			[ 1_200_000, 1],
7			[ 8_000_000, 1],
8			[ 75_000_000, 1],
9			);
10
11			sub new {
12			my $class = shift;
13			my $self = bless {
14			slots => $_[0],
15			lastspb => 0,
16			avgspb => 0,
17			}, $class;
18	root	1.5	$self->{reschedule} = EV::timer 10, 3, sub { $self->wake_next };
19	root	1.1	$self;
20			}
21
22			sub start_transfer {
23			my $self = shift;
24			my $size = $_[0];
25
26			my $transfer = bless {
27			queue => $self,
28			time => $::NOW,
29			size => $size,
30			coro => $Coro::current,
31			started => 0,
32			}, transfer::;
33
34			push @{$self->{wait}}, $transfer;
35
36			$self->wake_next;
37
38			$transfer;
39			}
40
41			sub sort {
42			my @queue = grep $_, @{$_[0]{wait}};
43
44			$_->{spb} = ($::NOW-$_->{time}) / ($_->{size} \|\| 1) for @queue;
45
46			$_[0]{wait} = [sort { $b->{spb} <=> $a->{spb} } @queue];
47
48			Scalar::Util::weaken $_ for @{$_[0]{wait}};
49			}
50
51			sub wake_next {
52			my $self = shift;
53
54			$self->sort;
55
56			while (@{$self->{wait}}) {
57			my $size = $self->{wait}[0]{size};
58			my $min = 0;
59			for (@reserve) {
60			last if $size <= $_->[0];
61			$min += $_->[1];
62			}
63			last unless $self->{slots} > $min;
64			my $transfer = shift @{$self->{wait}};
65			$self->{lastspb} = $transfer->{spb};
66	root	1.4	$self->{avgspb} = $self->{avgspb} * 0.999 + $transfer->{spb} * 0.001;
67	root	1.1	$self->{started}++;
68			$transfer->wake;
69			last;
70			}
71			}
72
73	root	1.2	sub force_wake_next {
74			my $self = shift;
75
76	root	1.3	$self->{slots} += $::MAX_TRANSFERS;
77	root	1.2	$self->wake_next;
78	root	1.3	$self->{slots} -= $::MAX_TRANSFERS;
79	root	1.2	}
80
81	root	1.1	sub waiters {
82			$_[0]->sort;
83			@{$_[0]{wait}};
84			}
85
86			package transfer;
87
88			use Coro::Timer ();
89
90			sub wake {
91			my $self = shift;
92
93			$self->{alloc} = 1;
94			$self->{queue}{slots}--;
95			$self->{wake} and $self->{wake}->ready;
96			}
97
98			sub try {
99			my $self = shift;
100
101			$self->{alloc} \|\| do {
102			my $timeout = Coro::Timer::timeout $_[0];
103			local $self->{wake} = $self->{coro};
104
105			Coro::schedule;
106
107			$self->{alloc};
108			}
109			}
110
111			sub DESTROY {
112			my $self = shift;
113
114			if ($self->{alloc}) {
115			$self->{queue}{slots}++;
116			$self->{queue}->wake_next;
117			}
118			}
119
120			1;
121