Perl&Stuff

Stuff&Perl

[ukautz-blog]

Concurrency in DynamoDB

I need concurrency. So had to figure out howto. The general idea is to use the UpdateItem method. Either on a separate (lock) table or on the entry you want to update. So what you need to to is encapsulate your update code within a block like so:

# wait for lock if entry.lockkey == 0 entry.lockkey = 1 update entry # have lock, do stuff entry.counter1 ++ some_other_stuff() # release lock, write data entry.lockkey = 0 update entry

So here is an example in Perl:

#!/usr/bin/perl use strict; use warnings; use Data::Dumper; use FindBin qw/ $Bin /; use Net::Amazon::DynamoDB; $| = 1; my ( $forks, $writes ) = @ARGV; die "Usage $0 " unless $forks && $writes; my $ddb = Net::Amazon::DynamoDB->new( secret_key => $ENV{ AWS_SECRET_ACCESS_KEY }, access_key => $ENV{ AWS_ACCESS_KEY_ID }, host => sprintf( 'dynamodb.%s.amazonaws.com', $ENV{ EC2_REGION } || 'us' ), read_consistent => 1, raise_error => 1, max_retries => 1000, retry_timeout => 0.1, use_keepalives => 0, tables => { mytable =>{ hash_key => 'hkey', range_key => 'rkey', attributes => { hkey => 'S', rkey => 'S', ckey => 'N', # << keyfor concurrency counter1 => 'N', # << one counter counter2 => 'N', # << another counter } }, } ); # init my %pk = ( hkey => 'mycounterhash', rkey => 'mycounterrange', ); $ddb->put_item( mytable => { %pk, counter1 => 0, counter2 => 1, ckey => 0 } ); # fork my @pids; my $orig_pid = my $pid = $$; $pid && do { push @pids, $pid = fork() } for 1..$forks; # each fork writes multiple entries foreach my $write( 1..$writes ) { my $uref = undef; while( ! eval { $uref = $ddb->update_item( mytable => { ckey => 1 }, { %pk, ckey => 0 }, { return_mode => 'ALL_OLD' } ) } ) { print "WAIT $write / $$\n"; sleep 1; } print "WRITING IN $write / $$ ". Dumper( $uref ); print "** UPDATE AND RELEASE $write / $$ **\n\n"; $ddb->update_item( mytable => { ckey => 0, counter1 => $uref->{ counter1 } + 1, counter2 => $uref->{ counter2 } * 2 }, \%pk ); } if ( $orig_pid == $$ ) { waitpid( $_, 0 ) for @pids; print "\n\n************\nALL DONE\n\n"; my $ref = $ddb->get_item( mytable => \%pk ); print Dumper( $ref ); }

Thats all

[ukautz-blog]

Perl vs Event::Lib vs UDP

Just playing around with Event::Lib (based on libevent) and high performance UDP server. Need something for real time analysis behind HAProxy.

UDP Server 

#!/usr/bin/perl use strict; use Event::Lib; use Event::Lib::UDPPump; use IO::Socket::INET; $|=1; my $counter = 0; my $last_counter = 0; my $num_forks = 5; # create server my $server = IO::Socket::INET->new( Proto => "udp", LocalPort => 23000 ); # do fork my $pid = $$; $pid && do { $pid = fork() } foreach 1..$num_forks; # create udp pump event my $udp_event_server = udppump_new( $server, \&udp_connection_cb ); # create a timer event my $timer = timer_new( \&timer_timeout_cb ); # add events $udp_event_server->add(); $timer->add(1); # loop.. event_mainloop(); # methods sub udp_connection_cb { $counter++; } sub timer_timeout_cb { print "HANDLED $counter CONNS IN PID $$\n" if $counter != $last_counter; $last_counter = $counter; shift->add(1) }

UDP Client

#!/usr/bin/perl use strict; use IO::Socket::INET; foreach ( 1..10000 ) { # open new connection my $sock = IO::Socket::INET->new( PeerAddr => "127.0.0.1:23000", Proto => "udp" ) or die "Failed to open: $!\n"; # send data $sock->send( "Bla\n" ); # done $sock->close() }

Good article about async IO in Perl: http://stason.org/articles/perl/Developing-High-Performance-Asynchronous-IO-Applications.html