view release on metacpan or  search on metacpan

lib/Any/Daemon/FCGI/ClientConn.pm  view on Meta::CPAN

# Copyrights 2013-2020 by [Mark Overmeer].
#  For other contributors see ChangeLog.
# See the manual pages for details on the licensing terms.
# Pod stripped from pm file by OODoc 2.02.
# This code is part of distribution Any-Daemon-HTTP. Meta-POD processed
# with OODoc into POD and HTML manual-pages.  See README.md
# Copyright Mark Overmeer.  Licensed under the same terms as Perl itself.

package Any::Daemon::FCGI::ClientConn;
use vars '$VERSION';
$VERSION = '0.30';


use warnings;
use strict;

use Log::Report      'any-daemon-http';

use HTTP::Request ();
use Time::HiRes   qw(usleep);
use Errno         qw(EAGAIN EINTR EWOULDBLOCK);
use IO::Select    ();
use Socket        qw/inet_aton PF_INET AF_INET SHUT_RD SHUT_WR/;

use Any::Daemon::FCGI::Request ();

use constant
  { FCGI_VERSION    => 1
  , FCGI_KEEP_CONN  => 1    # flag bit
  , MAX_FRAME_SEND  => 32 * 1024   # may have 65535 bytes content
  , MAX_READ_CHUNKS => 16 * 1024
  , CRLF            => "\x0D\x0A"
  , RESERVED        => 0
  };

# Implementation heavily based on Net::Async::FastCGI::Request and
# Mojo::Server::FastCGI

my %server_role_name2id =
  ( RESPONDER          => 1
  , AUTHORIZER         => 2
  , FILTER             => 3
  );

my %frame_name2id =
  ( BEGIN_REQUEST      => 1
  , ABORT_REQUEST      => 2
  , END_REQUEST        => 3
  , PARAMS             => 4
  , STDIN              => 5
  , STDOUT             => 6
  , STDERR             => 7
  , DATA               => 8
  , GET_VALUES         => 9
  , GET_VALUES_RESULT  => 10
  , UNKNOWN_TYPE       => 11
  );

my %end_status2id =
  ( REQUEST_COMPLETE   => 0
  , CANT_MPX_CONN      => 1
  , OVERLOADED         => 2
  , UNKNOWN_ROLE       => 3
  );

my %server_role_id2name = reverse %server_role_name2id;
my %frame_id2name       = reverse %frame_name2id;


sub new($%) { (bless {}, $_[0])->init($_[1]) }

sub init($)
{   my ($self, $args) = @_;
    $self->{ADFC_requests}  = {};
    $self->{ADFC_max_conns} = $args->{max_childs} or panic;
    $self->{ADFC_max_reqs}  = $args->{max_childs};

    $self->{ADFC_select}    = my $select = IO::Select->new;
    $self->{ADFC_socket}    = my $socket = $args->{socket} or panic;
    $self->{ADFC_stdin}     = \my $stdin;
    $self->{ADFC_keep_conn} = 0;
    $select->add($socket);

    $self;
}

#----------------

sub socket() { shift->{ADFC_socket} }

#----------------

sub _next_record()
{   my $self = shift;
    my $leader = $self->_read_chunk(8);
    length $leader==8 or return;

    my ($version, $type_id, $req_id, $clen, $plen) = unpack 'CCnnC', $leader;
    my $body = $self->_read_chunk($clen + $plen);

    substr $body, -$plen, $plen, '' if $plen;   # remove padding bytes
    length $body==$clen or return;

    ($frame_id2name{$type_id} || 'UNKNOWN_TYPE', $req_id, \$body);
}

sub _reply_record($$$)
{   my ($self, $type, $req_id, $body) = @_;
    my $type_id = $frame_name2id{$type} or panic $type;
    my $empty   = ! length $body;  # write one empty frame

    while(length $body || $empty)
    {   my $chunk  = substr $body, 0, MAX_FRAME_SEND, '';
        my $size   = length $chunk;
        my $pad    = (-$size) % 8;    # advise to pad on 8 bytes
        my $frame  = pack "CCnnCxa${size}x${pad}"
          , FCGI_VERSION, $type_id, $req_id, $size, $pad, $chunk;

        while(length $frame)
        {   my $wrote = syswrite $self->socket, $frame;
            if(defined $wrote)
            {   substr $frame, 0, $wrote, '';
                next;
            }

            return unless $! == EAGAIN || $! == EINTR || $! == EWOULDBLOCK;
            usleep 1000;  # 1 ms
        }

        last if $empty;
    }
}


sub get_request()
{   my $self = shift;
    my $requests = $self->{ADFC_requests};
    my $reqdata;

    ### At the moment, we will only support processing of whole requests
    #   and full replies: no chunking inside the server.

    while(1)
    {   my ($type, $req_id, $body) = $self->_next_record
            or return;

        if($req_id==0)
        {   $self->_management_record($body);
            next;
        }

        if($type eq 'BEGIN_REQUEST')
        {   my ($role_id, $flags) = unpack 'nC', $$body;
            my $role = $server_role_id2name{$role_id}
                or $self->_fcgi_end_request(UNKNOWN_ROLE => $req_id);

            $requests->{$req_id} =
              { request_id      => $req_id
              , data_complete   => $role ne 'FILTER'
              , stdin_complete  => $role eq 'AUTHORIZER'
              , params_complete => 0
              , role            => $role
              , params          => undef,
              , stdin           => undef,
              , data            => undef,
              };

            unless($flags & FCGI_KEEP_CONN)
            {   # Actually, this flag is incorrectly: more threads may still be
                # active.  So, let's close when they all have ceased to exist.
                info __x"fcgi {id} is last request", id => $req_id;
                $self->{ADFC_keep_conn} = 0;
            }

            next;
        }

        defined $req_id or panic;
        $reqdata = $requests->{$req_id};
        unless($reqdata)
        {   notice __x"fcgi received {type} for {id} which does not exist now"
              , type => $type, id => $req_id;