VPF::Реализация асинхронного клиента к memcached

slavenk

Дата 18.4.2014, 20:10 (ссылка)

(нет голосов)

Загрузка ...

TheAthlete

Профиль
Группа: Участник
Сообщений: 69
Регистрация: 22.10.2009

Репутация: нет
Всего: нет

Здравствуйте!
Реализовываю клиент к memcached по текстовому протоколу. Реализовал синхронный вариант
А как реализовать асинхронный вариант не пойму. Планирую использовать AnyEvent::Socket
Но не пойму, как реализовать алгоритм отправки запросов и парсинга ответов. Подскажите пожайлуста, в какую строну копать?

По идее нужно подключаться с помощью tcp_connect, а потом что делать с файлхендлером (который неблокирующий)?

arto

Дата 18.4.2014, 21:36 (ссылка)

(нет голосов)

Загрузка ...

Эксперт

Профиль
Группа: Завсегдатай
Сообщений: 1495
Регистрация: 31.10.2004

Репутация: 38
Всего: 40

писать, читать и парсить ответы.

можно посмотреть как сделано в Cache::Memcached::AnyEvent

slavenk

Дата 19.4.2014, 15:25 (ссылка)

(нет голосов)

Загрузка ...

TheAthlete

Профиль
Группа: Участник
Сообщений: 69
Регистрация: 22.10.2009

Репутация: нет
Всего: нет

Реализовал вот такой вот код:

Код


#!/usr/bin/perl -w
use strict;
use warnings;
use feature 'say';

use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Socket;
use EV;

my $cv = AnyEvent->condvar;
my $cnt = 0;

tcp_connect localhost => 11211, sub {
  my $fh = shift or die "unable to connect: $!";

  my $handle;
  $handle = new AnyEvent::Handle
    fh => $fh,
    on_error => sub {
      my ($handler, $fatal, $msg) = @_;
      AE::log error => $msg;
      $handler->destroy;
      $cv->send;
    },
    on_eof => sub {
      $handle->destroy; # destroy handle
      AE::log info => "Done.";
      $cv->send;
    };

    #my @commands_get = (
      #"get data\r\n",
      #"get someKey\r\n",
    #);

    #$handle->push_write($_) for @commands_get;

    $handle->push_write("get data\r\n");
    $handle->push_write("get someKey\r\n");

    #$handle->push_write("set dataKey 0 0 5\r\nhello\r\n");

    $handle->push_read(line => sub {
        my ($handle, $line) = @_;

        say "line: $line";

        $handle->on_read(sub {
            say $_[0]->rbuf;
            $_[0]->rbuf = "";
            $cv->send if ++$cnt == 2;
          });
      });

};

$cv->recv;

Подскажите, правильно ли я реализовал. Суть кода вот в чем: сначала заполняем очередь запросов (в данном случае get) и как только она заполнится, выполняем чтение. Но почему-то вариант с массивом и циклом (закомментирован) не работает - программа что-то ждет. Подскажите пожайлуста, в чем может быть проблема и правильно ли написана программа?

noize

Дата 19.4.2014, 19:10 (ссылка)

(нет голосов)

Загрузка ...

5.18.2

Профиль
Группа: Участник
Сообщений: 89
Регистрация: 10.7.2010
Где: СПб

Репутация: 3
Всего: 3

AnyEvent::Memcached не ?

slavenk

Дата 22.4.2014, 17:25 (ссылка)

(нет голосов)

Загрузка ...

TheAthlete

Профиль
Группа: Участник
Сообщений: 69
Регистрация: 22.10.2009

Репутация: нет
Всего: нет

Оно, просто хочу разобраться с AnyEvent, в частности с AnyEvent::Handler и AnyEvent::Socket. Реализовал вот такой код:

Код


#!/usr/bin/perl
use strict;
use warnings;
use feature 'say';
use Carp qw/croak carp/;

use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Socket;
use EV;

sub get {
  my ($host, $port, $cv, $keys, %args) = @_; # $args -> cb => sub {$value, $err}

  $cv->begin if $cv;

  tcp_connect $host => $port, sub {
    my $fh = shift or die "unable to connect: $!";

    my $handle;
    $handle = new AnyEvent::Handle
      fh => $fh,
      on_error => sub {
        my ($handler, $fatal, $msg) = @_;
        AE::log error => $msg;
        $handler->destroy;
        $cv->send;
      },
      on_eof => sub {
        $handle->destroy; # destroy handle
        AE::log info => "Done.";
        $cv->send;
      };

      my $array = 1 if ref $keys and ref $keys eq 'ARRAY';
      my $key = $array
        ? 'get ' . join(' ', @$keys) . "\r\n"
        : "get $keys\r\n";
        
      $cv->begin;
      $handle->push_write($key);

      my $result;

      my $reader; $reader = sub {
          my ($handle, $line) = @_;

          if ($line eq 'END') {
            undef $reader;
            $args{cb}($result);
          }
          elsif (substr($line, 0, 5) eq 'ERROR') {
            undef $reader;
            $args{cb}(undef, $line);
          }
          elsif (!length($line)) {
            warn "Skip empty line";
            $handle->unshift_read(line => $reader);
          }
          elsif($line =~ /^VALUE (\S+) (\d+) (\d+)(?:| (.+))$/ ) {
            my ($key, $flags, $len, $cas) = ($1, $2, $3, $4);
            $handle->push_read(chunk => $len+2, sub { # data length + \r\n
                my ($handle, $data) = @_;
                substr($data, $len) = ''; # trim \r\n
                if ($array) {
                  $result->{$key} = $data;
                } else {
                  $result = $data;
                }
                $handle->unshift_read(line => $reader);
                $cv->send;
              });
          }
        };

      $handle->push_read(line => $reader);
      $cv->end;
  };
}

my $cv = AnyEvent->condvar;

my $host = '127.0.0.1';
my $port = 11211;

get($host, $port, $cv, 'data', cb => sub {
    my ($value, $err) = @_;
    $err and return warn "Get failed: @_";
    warn "Value for key is '$value'";
  });

get($host, $port, $cv, [qw/data someKey/], cb => sub {
    my ($values, $err)  = @_;
    $err and return warn "Get failed: @_";
    warn "Value for key1 is '$values->{data}' and value for key2 is '$values->{someKey}'";
  });

$cv->recv;

Код почему-то работает нестабильно - отрабатывают то один get, то два.

Цитата

$ perl test_anyevent.pl
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at test_anyevent.pl line 95.
Value for key is 'hello memcached world' at test_anyevent.pl line 89.
$ perl test_anyevent.pl
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at test_anyevent.pl line 95.
$ perl test_anyevent.pl
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at test_anyevent.pl line 95.
Value for key is 'hello memcached world' at test_anyevent.pl line 89.
$ perl test_anyevent.pl
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at test_anyevent.pl line 95.
Value for key is 'hello memcached world' at test_anyevent.pl line 89.

Не могу понять, в чем проблема. Если использовать модуль AnyEvent::Memcached то все впорядке:

Код


#!/usr/bin/perl -w
use strict;
use warnings;
use feature 'say';

use AnyEvent;
use EV;
use AnyEvent::Memcached;

my $cv = AnyEvent->condvar;

my $memd = AnyEvent::Memcached->new(
  servers => ['127.0.0.1:11211'],
  cv => $cv,
);

$memd->get('data', cb => sub {
    my ($value, $err) = shift;
    $err and return warn "Get failed: @_";
    warn "Value for key is $value";
  });

$memd->get([qw/data someKey/], cb => sub {
    my ($values, $err)  = shift;
    $err and return warn "Get failed: @_";
    warn "Value for key1 is '$values->{data}' and value for key2 is '$values->{someKey}'";
  });

$cv->recv;

Вывод:

Цитата

$ perl anyevent_memcached_mons.pl
Value for key is hello memcached world at anyevent_memcached_mons.pl line 20.
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at anyevent_memcached_mons.pl line 26.
$ perl anyevent_memcached_mons.pl
Value for key is hello memcached world at anyevent_memcached_mons.pl line 20.
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at anyevent_memcached_mons.pl line 26.
$ perl anyevent_memcached_mons.pl
Value for key is hello memcached world at anyevent_memcached_mons.pl line 20.
Value for key1 is 'hello memcached world' and value for key2 is 'hello world' at anyevent_memcached_mons.pl line 26.

0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| Perl: Общие вопросы \| Следующая тема »