2020 Perl Advent Calendar - Day 15

<< First | < Prev | Next >

Yesterday we took a deeper dive into the insides of one asynchronous IO system, Future::IO, to get an impression of the sorts of things that go into creating it. Before we conclude our 2020 update on the concept of futures, I'd like to do one more deep dive into some real-world application code, to show some more details of the insides of an implementation. Today we'll take a look inside Net::Async::HTTP.

This module has existed on CPAN for a long time. It certainly predates Future::AsyncAwait and the mechanisms to provide parser plugin modules that allow async/await syntax. It even predates the Future module. In fact, the initial version of futures, CPS::Future, was built partly as an experiment to provide Net::Async::HTTP with a way to cancel pending requests. As a result of this history, a desire to keep the module as lightweight as possible in dependencies, and to support as many older Perl versions as possible, it does not use async/await syntax in its implementation. Therefore parts of the code are somewhat more convoluted and harder to read or follow, as would be the case for a cleaner version built on more modern techniques.

Keep that in mind on today's exploration, as much of this will be a tour of what tricks are otherwise required, when async/await is not available. As with yesterday's overview of Future::IO, the point today is not necessarily to follow and understand what is going on, but rather to simply see the amount of work is required and the steps necessary to take when writing large asynchronous systems when async/await is not available.

The inner mechanism of Net::Async::HTTP which makes the whole thing work is actually in an internal class which represents a connection to one HTTP server, named Net::Async::HTTP::Connection. The actual method that starts the request/response cycle is called request. The full version is 150 lines long at current count, but for our purposes the basic outline is as follows.

sub request
{
    ...
    
    my $f = $self->loop->new_future;
    
    push @{$self->{request_queue}}, RequestContext(
        on_read => $self->_mk_on_read_header(...),
        f       => $f,
        ...
    );
    
    ...
    $self->write(join($CRLF, @headers) . $CLRF . $CRLF);
    $self->write($req->content);
    
    return $f;
}

This creates a new future instance, builds a context structure containing that (among other things), pushes this structure to an internal array, arranges for the request itself to be sent, and then returns the future instance to the caller, which propagates up to become the return value of the actual GET call. The context structure also captures many variables and state about the request process, because those will be needed later on when response data starts to arrive. In particular is the on_read field, which stores a code reference for actually handling the response.

The response half of the process is a little more distributed around a few places. Since the connection class is a subclass of IO::Async::Stream it has an on_read event which invokes a method of the same name, which is the start of the control path here. Its job is relatively tiny, in that it mostly just has to invoke the current on_read handler code for the topmost item in the request queue.

sub on_read
{
    my $self = shift;
    my ($buffref, $closed) = @_;

    while(my $head = $self->{request_queue}[0]) {
        my $ret = $head->on_read->($self, $buffref, $closed, $head);
        ...
    }
}

Initially, the on_read handler code is set up from the start of the request for handling a response header. This will be the anonymous sub returned by the _mk_on_read_header method. The code there begins:

sub _mk_on_read_header
{
    my $self = shift;
    ...
    
    return sub {
        my ($self, $buffref, $closed, $ctx) = @_;
        ...
        unless($$buffref =~ s/(.*?$CRLF$CRLF)//s) {
            return 0;
        }
        my $header = HTTP::Response->parse($1);
        my $content_length = $header->content_length;
        ...
        
        if(defined $content_length) {
            $ctx->on_read =
                $self->_mk_on_read_length($content_length, ...);
            return 1;
        }
        
        ...
    };
}

This will continue to read bytes of response until it encounters the double-CRLF sequence that marks the end of the header. At that point it will need to decide what kind of handler is best for the response content - whether it should just read until the filehandle is closed, or a fixed amount of content, or use chunked encoding. This choice is made by inspecting various headers, then calling one of three more _mk_on_read_* methods to generate a new on_read handler to put back into the context structure.

This part needs emphasising again, as it is the important central part of this whole "request context" technique. The anonymous function returned by _mk_on_read_header is only for reading the header part of the response. Once it has determined the appropriate way to read the body content of the response, it creates a new anonymous function for that instead, and stores it in the on_read field of the request context. This piece of mutable data, in effect, stores the running state of this particular request/response cycle. Having stored it there, it just returns back to the main on_read method, which will repeat its operation again, now with the new handler in place to read the response body content.

For example, the handler for reading a fixed length of content (from the Content-Length header) will maintain a count of the number of bytes remaining, counting down to zero:

sub _mk_on_read_length
{
    my $self = shift;
    my ($content_length, ...) = @_;
    
    return sub {
        my ($self, $buffref, $closed, $ctx) = @_;
        
        my $content = $$buffref;
        $$buffref = "";
        
        $content_length -= length $content;
        $response->add_content($content);
        
        if($content_length == 0) {
            $ctx->f->done($response);
        }
        
        return 0;
    };
}

Once the required amount of content has been received, this loop will finally finish. The response object, having finally been constructed and filled with content, is passed to the done method of the future in the request context object. This is the one that the original request method first created at the start of the cycle.

We've now taken a very brief, and very summarized tour of the code responsible for a single request/response cycle in the HTTP client. These code examples have been simplified and elided many details, such as content encoding or handling of streaming responses that don't have to store the full content in memory. Already we have seen a number of layers of code, and encountered places where the running state is stored in explicit data structures, rather than being implied by the position of execution within the code itself.

In summary - there's a lot of complex code here, all loosely connected together by passing around a context object. That context object stores a bunch of state - some of it mutable and changing over time - relating to this particular request/response cycle. The context object takes the place of simply storing the state of the program in normal lexical variables, as would be the case for synchronous code, or asynchronous code when using the async/await syntax. Lacking that syntax, we have had to reconstruct many of its abilities using other, less familiar techniques. If we had been able to use async/await then a lot of this code could be much improved, by moving a lot of the "machinery" parts out from custom explicit implementation and into regular Perl code.

<< First | < Prev | Next >

2020 Perl Advent Calendar - Day 9

<< First | < Prev | Next >

Yesterday we saw some ways to write concurrent asynchronous code which waits on a few different tasks to complete. Sometimes we want to do the same thing multiple times concurrently but with different data each time. Often it's the case that each item can be processed independently of the others, so it makes sense to try to do several at once.

One approach here is to simply await a future to process each individual item inside a regular foreach loop. This will only work on one item at once, so we won't make use of concurrency.

## A poor idea for iterating a list ##
use Future::AsyncAwait;

foreach my $item (@ITEMS) {
    await PROCESS($item);
}

Another idea is to use map to apply an asynchronous function to every item in the list, and thus start all the items at once, then wait on all those futures using needs_all. This may end up being too concurrent - if the list contained many thousand items we might do too many at once and overload whatever external service or system we are talking to.

## Another poor idea ##
use Future::AsyncAwait;

await Future->needs_all(
    map { PROCCESS($_) } @ITEMS
);

A better middle-ground between these two extremes was introduced in the original blog series on day 18, in the form of the Future::Utils::fmap collection of helper functions. The basic idea of fmap is that it is a future-aware equivalent of Perl's map operator. The fmap function is given a block of code which is expected to return a future, and a list of items. It invokes the code block once for each item in the list, collecting up and waiting on the returned future values, until all the items are done. fmap returns a future to represent the entire operation, which will complete with the results of each individual item.

The fmap family actually contains three individual functions, which all operate in the same basic manner. The difference between them is in how they handle return values from each individual item block - fmap_concat can handle an entire list from each item and concatenates all the results together for its overall result, fmap_scalar expects exactly one result per item, and fmap_void does not collect up any results at all; running the code block simply for its side-effects.

Since fmap expects a future-returning function and itself returns a future it is also idea for use with async/await syntax. It can be invoked in an await expression and passed an async sub to operate on.

This somewhat-paraphrased example uses the GET method of a Net::Async::HTTP user agent object to concurrently fetch and JSON-decode a collection of data from multiple API endpoints of some remote service.

use feature 'signatures';
use Future::AsyncAwait;
use Future::Utils qw( fmap_scalar );
use JSON::MaybeUTF8 qw( decode_json_utf8 );

use Net::Async::HTTP;
my $ua = Net::Async::HTTP->new;
...

my @urls = ...

my @data = await fmap_scalar(async sub ($url) {
    my $response = await $ua->GET($url);
    return decode_json_utf8($response->content);
}, foreach => \@urls, concurrent => 16);

A few things should be noted about this example. First is that the async sub syntax is used explicitly to create an asynchronous function to pass as the first argument to the fmap_scalar function. Second is the use of the concurrent parameter, telling the function how many items to keep running concurrently. Finally, the list of items has to be passed in an array reference rather than a plain flat list.

These facts all come about because the fmap functions are just plain Perl functions and not special syntax, as opposed to the real syntax provided by the async and await keywords. Whereas these two keywords were inspired by a whole collection of other languages which have all adopted it as a standard pattern, there is not much existing design on the problem of bounded concurrency map-style syntax. This particular area remains a matter of ongoing design and discussion. Thoughts welcome ;)

<< First | < Prev | Next >

Net::Async::HTTP::Server

Recently, I've been working on a way to serve HTTP with IO::Async. For a while it's been possible to use HTTP as a client with Net::Async::HTTP, so it seemed only natural to provide a server named Net::Async::HTTP::Server.

I've tried to design the interface fairly similar to some of the other server modules, most notably Net::Async::FastCGI. This allows for similar code to be written one way or the other, for responding via HTTP or FastCGI.

use Net::Async::HTTP::Server;
use IO::Async::Loop;
 
use HTTP::Response;
 
my $loop = IO::Async::Loop->new();
 
my $httpserver = Net::Async::HTTP::Server->new(
   on_request => sub {
      my $self = shift;
      my ( $req ) = @_;
 
      my $response = HTTP::Response->new( 200 );
      $response->add_content( "Hello, world!\n" );
      $response->content_type( "text/plain" );
 
      $req->respond( $response );
   },
);
 
$loop->add( $httpserver );
 
$httpserver->listen(
   addr => { family => "inet6", socktype => "stream", port => 8080 },
   on_listen_error => sub { die "Cannot listen - $_[-1]\n" },
);
 
$loop->run;

As with Net::Async::FastCGI the distribution also includes modules to support hosting a PSGI application via Plack. The Plack::Handler module in particular allows the use of Net::Async::HTTP::Server directly from the plackup commandline:

plackup -s Net::Async::HTTP::Server --listen ":8080" application.psgi

It's still in somewhat of an early state - while it supports sending streaming responses (as required by PSGI) it doesn't yet support streaming receiving of requests. That shouldn't be too hard to add, but so far I haven't found a need to add it yet.

General Updates

I have no big specific updates today. Instead, a list of lots of little things I've been working on:

• IO::Socket::IP now has preliminary non-blocking connect support (version 0.05_003). This isn't quite a perfect solution because of blocking name resolvers, but see also Net::LibAsyncNS below.


• Created a new CPAN dist, wrapping libasyncns, called Net::LibAsyncNS. This allows a simple way to asynchronise name resolver lookups.


• Have fixed a few bugs in IO::KQueue, relating to dodgy handling of Perl scalars in the udata field. Some memory leak bugs still exist, but I believe these to be the kernel's fault. See below.


• Spent some time on Freebsd-hackers@ arguing about kqueue and managing user data pointers. Long story short I believe kqueue API itself is missing a feature, making generic wrapping of it impossible from any high-level language, or properly by C libraries also.


• Both talks I submitted for LPW2010 were accepted; on the subjects of CPS and IO::Async.


• Net::Async::HTTP now has SSL support and can stream response body content as it arrives, rather than waiting for the whole response (version 0.08).

That's all for a quick update, but I may write about any or all of these topics in more detail later...