Changeset 16071 for lang/perl/Data-Valve

Timestamp:

07/22/08 09:38:42 (5 years ago)

Author:

daisuke

Message:

implement strict interval mode

Location:

lang/perl/Data-Valve/trunk

Files:

• Valve.xs (modified) (2 diffs)
• dv_bucket.c (modified) (7 diffs)
• dv_bucket.h (modified) (4 diffs)
• lib/Data/Valve.pm (modified) (4 diffs)
• lib/Data/Valve/Bucket.pm (modified) (1 diff)
• lib/Data/Valve/BucketStore.pm (modified) (1 diff)
• lib/Data/Valve/BucketStore/Memory.pm (modified) (3 diffs)
• lib/Data/Valve/BucketStore/Object.pm (modified) (1 diff)
• t/01_basic.t (modified) (2 diffs)

lang/perl/Data-Valve/trunk/Valve.xs

@@ -8 +8 @@
 
 dv_bucket *
-dv_bucket_create(double interval, unsigned long max)
+dv_bucket_create(double interval, unsigned long max, int strict_interval = 0)
 
 void
@@ -44 +44 @@
 
 dv_bucket *
-dv_bucket__deserialize(SV *buf, double interval, long max)
+dv_bucket__deserialize(SV *buf, double interval, long max, int strict_interval = 0)
     PREINIT:
         STRLEN len;
         char *c_buf = (char *)SvPV(ST(0), len);
     CODE:
-        RETVAL = dv_bucket_deserialize(c_buf, len, interval, max);
+        RETVAL = dv_bucket_deserialize(c_buf, len, interval, max, strict_interval);
     OUTPUT:
         RETVAL

lang/perl/Data-Valve/trunk/dv_bucket.c

@@ -33 +33 @@
 
 dv_bucket*
-dv_bucket_create(double interval, unsigned long max)
+dv_bucket_create(double interval, unsigned long max, int strict_interval)
 {
     dv_bucket *bucket;
@@ -41 +41 @@
     bucket->interval = interval * DV_1E6;
     bucket->count = 0;
+    bucket->strict_interval = strict_interval;
     bucket->head = NULL;
     bucket->tail = NULL;
@@ -106 +107 @@
      */
     size_t expired = 0;
+    double dtime   = dv_bucket_timeval2double(tp);
 
     while ( 
         bucket->head != NULL &&
-        bucket->interval < dv_bucket_timeval2double(tp) - bucket->head->time
+        bucket->interval < dtime - bucket->head->time
     ) {
         dv_bucket_item *tmp = bucket->head;
@@ -125 +127 @@
 
 int
-dv_bucket_is_full(dv_bucket *bucket)
-{
+dv_bucket_is_full(dv_bucket *bucket, double dtime)
+{
+    if (bucket->count == 0 || bucket->head == NULL) {
+        /* safety net */
+        return 0;
+    }
+
+    /* if we're in strict_interval mode, then we check for the last entry
+     * in the list, and make sure that current time is more than 
+     * last entry + interval
+     */
+    if (bucket->strict_interval) {
+        return bucket->head->time + bucket->interval > dtime;
+    }
+
+    /* Otherwise, we care about how many items are in the list */
     return bucket->max <= bucket->count;
 }
@@ -149 +165 @@
 {
     struct timeval t;
+    double dtime;
 
     gettimeofday(&t, &tzp_not_used);
@@ -154 +171 @@
     dv_bucket_expire( bucket, &t );
 
+    dtime = dv_bucket_timeval2double(&t);
+
     if ( dv_bucket_count( bucket ) == 0 ) {
-        dv_bucket_push( bucket, dv_bucket_timeval2double(&t) );
+        dv_bucket_push( bucket, dtime );
         return 1;
     }
 
-    if ( dv_bucket_is_full(bucket) ) {
+    if ( dv_bucket_is_full(bucket, dtime) ) {
         return 0;
     }
 
-    dv_bucket_push( bucket, dv_bucket_timeval2double(&t) );
+    dv_bucket_push( bucket, dtime );
     return 1;
 }
@@ -183 +202 @@
 
 dv_bucket *
-dv_bucket_deserialize(char *buf, size_t len, double interval, unsigned long max)
-{
-    dv_bucket *bucket = dv_bucket_create(interval, max);
+dv_bucket_deserialize(char *buf, size_t len, double interval, unsigned long max, int strict_interval)
+{
+    dv_bucket *bucket = dv_bucket_create(interval, max, strict_interval);
     char *end = buf + len;
 

lang/perl/Data-Valve/trunk/dv_bucket.h

@@ -18 +18 @@
     unsigned long max;
     double interval;
+    int strict_interval;
     unsigned long count;
     dv_bucket_item *head;
@@ -29 +30 @@
 /* Creates a new bucket */
 dv_bucket *
-    dv_bucket_create(double interval, unsigned long max);
+    dv_bucket_create(double interval, unsigned long max, int strict_interval);
 
 /* Frees a bucket */
@@ -57 +58 @@
 /* Returns true if count >= max */
 int
-    dv_bucket_is_full(dv_bucket *bucket);
+    dv_bucket_is_full(dv_bucket *bucket, double dtime);
 
 /* Pushes a new item on to the bucket */
@@ -73 +74 @@
 /* Deserialize a bucket from a string */
 dv_bucket *
-    dv_bucket_deserialize(char *buf, size_t len, double interval, unsigned long max);
+    dv_bucket_deserialize(char *buf, size_t len, double interval, unsigned long max, int strict_interval);
 
 /* returns the first bucket item */

lang/perl/Data-Valve/trunk/lib/Data/Valve.pm

@@ -23 +23 @@
 );
 
+has 'strict_interval' => (
+    is => 'rw',
+    isa => 'Bool',
+    required => 1,
+    default => 0
+);
+
 has '__bucket_store' => (
     accessor => 'bucket_store',
     is => 'rw',
     does => 'Data::Valve::BucketStore',
+    required => 1,
 );
 
@@ -33 +41 @@
 no Moose;
 
-sub BUILD {
-    my ($self, $args) = @_;
+sub BUILDARGS {
+    my ($self, %args) = @_;
 
-    my $store = delete $args->{bucket_store} || { module => 'Memory' };
+    my $store = delete $args{bucket_store} || { module => 'Memory' };
     if (! blessed $store) {
         my $module = $store->{module};
@@ -44 +52 @@
         Class::MOP::load_class($module);
 
-        $store = $module->new( %{ $store->{args} }, context => $self );
+        $store = $module->new( %{ $store->{args} } );
     }
-    $self->bucket_store($store);
+
+    if ($args{strict_interval}) {
+        # in strict_interval mode, max_items doesn't mean anything
+        $args{max_items} = 0;
+    }
+
+    return { %args, __bucket_store => $store };
+}
+
+sub BUILD {
+    my $self = shift;
+    $self->bucket_store->context($self);
 }
 
@@ -90 +109 @@
 mechanism is much simpler than Data::Throttler, and so is faster.
 
-It also comes with Memcached support for a distributed throttling.
+It also comes with Memcached support for a distributed throttling via
+memcached + keyedmutexd. This means that you can have multiple hosts throttling
+on the same "key". For example, multiple crawler instances can throttle 
+requests against a single host safely. To enable distributed throttling,
+you simply need to specify a Data::Valve::BucketStore instance that supports
+distribution (i.e. Data::Valve::BucketStore::Memcached) or create your own
+instance, and pass it to the constructor:
+
+  Data::Valve->new(
+    ...,
+    bucket_store => {
+      module => "Memcached", # to use Data::Valve::BucketStore::Memcached
+      args   => {
+        servers => [ '127.0.0.1:11211' ]
+      }
+    }
+  );
+
+Please note that for distributed throttling to work, you must specify the
+correct values in max_items, interval, and so forth, for each Data::Valve
+instance. Data::Valve will not try to automatically adjust this for you.
+You must coordinate it in the client side (i.e., whatever that's using
+Data::Valve)
+
+Since version 0.00006, Data::Valve supports "strict_interval" mode, where
+instead of counting the number of items over a range of time, it simply
+calculates the amount of time elapsed since the last logged request.
+
+To enable, specify it in the constrctor:
+
+  # This specifies that at least 5 seconds should have passed before
+  # the next item can go
+  Data::Valve->new(
+    interval        => 5,
+    strict_interval => 1, 
+  );
 
 =head1 METHODS
+
+=head2 new(%args)
+
+=over 4
+
+=item max_items
+
+In strict interval mode, does not mean anything. If NOT in strict interval
+mode, specifies the max number of items that can go through this throttler
+in the given interval.
+
+=item interval
+
+In strict interval mode, this specifies the number of seconds to wait between
+each request. If NOT in strict interval mode, specifies the number of seconds
+to span the requests, up to the value specified in max_items
+
+C<interval> may be a fractional number, denoting fractional seconds.
+
+=item strict_interval
+
+Boolean. Enable/Disable strict interval mode. Default is off.
+
+=cut
 
 =head2 try_push([key => $key_name])

lang/perl/Data-Valve/trunk/lib/Data/Valve/Bucket.pm

@@ -6 +6 @@
 sub new {
     my ($class, %args) = @_;
-    my $self  = bless create($args{interval}, $args{max_items}), $class;
+    $args{strict_interval} ||= 0;
+    my $self  = bless create($args{interval}, $args{max_items}, $args{strict_interval}), $class;
     
     return $self;

lang/perl/Data-Valve/trunk/lib/Data/Valve/BucketStore.pm

@@ -9 +9 @@
     is       => 'rw',
     isa      => 'Data::Valve',
-    handles  => [ qw(max_items interval) ],
+    handles  => [ qw(max_items interval strict_interval) ],
 );
 

lang/perl/Data-Valve/trunk/lib/Data/Valve/BucketStore/Memory.pm

@@ -9 +9 @@
     is => 'rw',
     isa => 'HashRef',
+    required => 1,
+    default => sub { +{} }
 );
 
@@ -15 +17 @@
 no Moose;
 
-sub BUILD { 
-    my $self = shift;
-    $self->store( {
-        __default => $self->create_bucket()
-    } );
-}
-
 sub create_bucket
 {
@@ -27 +22 @@
     return Data::Valve::Bucket->new(
         max_items => $self->max_items,
-        interval  => $self->interval
+        interval  => $self->interval,
+        strict_interval => $self->strict_interval,
     );
 }

lang/perl/Data-Valve/trunk/lib/Data/Valve/BucketStore/Object.pm

@@ -40 +40 @@
         my $bucket;
         if ($bucket_source) {
-            $bucket = Data::Valve::Bucket->deserialize($bucket_source, $self->interval, $self->max_items);
+            $bucket = Data::Valve::Bucket->deserialize($bucket_source, $self->interval, $self->max_items, $self->strict_interval);
         } else {
             $bucket = Data::Valve::Bucket->new(
-                interval  => $self->interval,
-                max_items => $self->max_items,
+                interval        => $self->interval,
+                max_items       => $self->max_items,
+                strict_interval => $self->strict_interval
             );
         }

lang/perl/Data-Valve/trunk/t/01_basic.t

@@ -1 +1 @@
 use strict;
-use Test::More (tests => 16);
+use Test::More (tests => 20);
 
 BEGIN
@@ -45 +45 @@
     ok( $valve->try_push(key => "foo"), "try after 3 seconds should work");
 }
+
+{
+    my $valve = Data::Valve->new(
+        max_items => 5,
+        interval  => 3,
+        strict_interval => 1,
+    );
+
+    ok(  $valve->try_push(key => "foo"), "try 1 should succeed" );
+    ok(! $valve->try_push(key => "foo"), "try 2 should fail" );
+    diag("sleeping for 3 seconds...");
+    sleep 3;
+    ok(  $valve->try_push(key => "foo"), "try 3 should succeed" );
+    ok(! $valve->try_push(key => "foo"), "try 4 should fail" );
+}