ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork-RPC/RPC.pm
(Generate patch)

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.32 by root, Sat Aug 31 19:51:25 2013 UTC vs.
Revision 1.45 by root, Thu Sep 12 15:15:49 2019 UTC

175you really I<are> done. 175you really I<are> done.
176 176
177=head2 Example 2: Asynchronous Backend 177=head2 Example 2: Asynchronous Backend
178 178
179This example implements multiple count-downs in the child, using 179This example implements multiple count-downs in the child, using
180L<AnyEvent> timers. While this is a bit silly (one could use timers in te 180L<AnyEvent> timers. While this is a bit silly (one could use timers in the
181parent just as well), it illustrates the ability to use AnyEvent in the 181parent just as well), it illustrates the ability to use AnyEvent in the
182child and the fact that responses can arrive in a different order then the 182child and the fact that responses can arrive in a different order then the
183requests. 183requests.
184 184
185It also shows how to embed the actual child code into a C<__DATA__> 185It also shows how to embed the actual child code into a C<__DATA__>
371 371
372There is an important twist - the C<AnyEvent::Fork::RPC::event> function 372There is an important twist - the C<AnyEvent::Fork::RPC::event> function
373is only defined when the child is fully initialised. If you redirect the 373is only defined when the child is fully initialised. If you redirect the
374log messages in your C<init> function for example, then the C<event> 374log messages in your C<init> function for example, then the C<event>
375function might not yet be available. This is why the log callback checks 375function might not yet be available. This is why the log callback checks
376whether the fucntion is there using C<defined>, and only then uses it to 376whether the function is there using C<defined>, and only then uses it to
377log the message. 377log the message.
378 378
379=head1 PARENT PROCESS USAGE 379=head1 PARENT PROCESS USAGE
380 380
381This module exports nothing, and only implements a single function: 381This module exports nothing, and only implements a single function:
391use Errno (); 391use Errno ();
392use Guard (); 392use Guard ();
393 393
394use AnyEvent; 394use AnyEvent;
395 395
396our $VERSION = 1.2; 396our $VERSION = 1.25;
397 397
398=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...] 398=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
399 399
400The traditional way to call it. But it is way cooler to call it in the 400The traditional way to call it. But it is way cooler to call it in the
401following way: 401following way:
439been successfully handled. This is useful when you queue some requests and 439been successfully handled. This is useful when you queue some requests and
440want the child to go away after it has handled them. The problem is that 440want the child to go away after it has handled them. The problem is that
441the parent must not exit either until all requests have been handled, and 441the parent must not exit either until all requests have been handled, and
442this can be accomplished by waiting for this callback. 442this can be accomplished by waiting for this callback.
443 443
444=item init => $function (default none) 444=item init => $function (default: none)
445 445
446When specified (by name), this function is called in the child as the very 446When specified (by name), this function is called in the child as the very
447first thing when taking over the process, with all the arguments normally 447first thing when taking over the process, with all the arguments normally
448passed to the C<AnyEvent::Fork::run> function, except the communications 448passed to the C<AnyEvent::Fork::run> function, except the communications
449socket. 449socket.
454It is called very early - before the serialisers are created or the 454It is called very early - before the serialisers are created or the
455C<$function> name is resolved into a function reference, so it could be 455C<$function> name is resolved into a function reference, so it could be
456used to load any modules that provide the serialiser or function. It can 456used to load any modules that provide the serialiser or function. It can
457not, however, create events. 457not, however, create events.
458 458
459=item done => $function (default C<CORE::exit>) 459=item done => $function (default: C<CORE::exit>)
460 460
461The function to call when the asynchronous backend detects an end of file 461The function to call when the asynchronous backend detects an end of file
462condition when reading from the communications socket I<and> there are no 462condition when reading from the communications socket I<and> there are no
463outstanding requests. It's ignored by the synchronous backend. 463outstanding requests. It's ignored by the synchronous backend.
464 464
465By overriding this you can prolong the life of a RPC process after e.g. 465By overriding this you can prolong the life of a RPC process after e.g.
466the parent has exited by running the event loop in the provided function 466the parent has exited by running the event loop in the provided function
467(or simply calling it, for example, when your child process uses L<EV> you 467(or simply calling it, for example, when your child process uses L<EV> you
468could provide L<EV::loop> as C<done> function). 468could provide L<EV::run> as C<done> function).
469 469
470Of course, in that case you are responsible for exiting at the appropriate 470Of course, in that case you are responsible for exiting at the appropriate
471time and not returning from 471time and not returning from
472 472
473=item async => $boolean (default: 0) 473=item async => $boolean (default: C<0>)
474 474
475The default server used in the child does all I/O blockingly, and only 475The default server used in the child does all I/O blockingly, and only
476allows a single RPC call to execute concurrently. 476allows a single RPC call to execute concurrently.
477 477
478Setting C<async> to a true value switches to another implementation that 478Setting C<async> to a true value switches to another implementation that
488synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode. 488synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
489 489
490If you use a template process and want to fork both sync and async 490If you use a template process and want to fork both sync and async
491children, then it is permissible to load both modules. 491children, then it is permissible to load both modules.
492 492
493=item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER) 493=item serialiser => $string (default: C<$AnyEvent::Fork::RPC::STRING_SERIALISER>)
494 494
495All arguments, result data and event data have to be serialised to be 495All arguments, result data and event data have to be serialised to be
496transferred between the processes. For this, they have to be frozen and 496transferred between the processes. For this, they have to be frozen and
497thawed in both parent and child processes. 497thawed in both parent and child processes.
498 498
499By default, only octet strings can be passed between the processes, which 499By default, only octet strings can be passed between the processes,
500is reasonably fast and efficient and requires no extra modules. 500which is reasonably fast and efficient and requires no extra modules
501(the C<AnyEvent::Fork::RPC> distribution does not provide these extra
502serialiser modules).
501 503
502For more complicated use cases, you can provide your own freeze and thaw 504For more complicated use cases, you can provide your own freeze and thaw
503functions, by specifying a string with perl source code. It's supposed to 505functions, by specifying a string with perl source code. It's supposed to
504return two code references when evaluated: the first receives a list of 506return two code references when evaluated: the first receives a list of
505perl values and must return an octet string. The second receives the octet 507perl values and must return an octet string. The second receives the octet
507 509
508If you need an external module for serialisation, then you can either 510If you need an external module for serialisation, then you can either
509pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use> 511pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
510or C<require> statement into the serialiser string. Or both. 512or C<require> statement into the serialiser string. Or both.
511 513
512Here are some examples - some of them are also available as global 514Here are some examples - all of them are also available as global
513variables that make them easier to use. 515variables that make them easier to use.
514 516
515=over 4 517=over 4
516 518
517=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER> 519=item C<$AnyEvent::Fork::RPC::STRING_SERIALISER> - octet strings only
518 520
519This serialiser concatenates length-prefixes octet strings, and is the 521This serialiser (currently the default) concatenates length-prefixes octet
520default. That means you can only pass (and return) strings containing 522strings, and is the default. That means you can only pass (and return)
521character codes 0-255. 523strings containing character codes 0-255.
524
525The main advantages of this serialiser are the high speed and that it
526doesn't need another module. The main disadvantage is that you are very
527limited in what you can pass - only octet strings.
522 528
523Implementation: 529Implementation:
524 530
525 ( 531 (
526 sub { pack "(w/a*)*", @_ }, 532 sub { pack "(w/a*)*", @_ },
527 sub { unpack "(w/a*)*", shift } 533 sub { unpack "(w/a*)*", shift }
528 ) 534 )
529 535
530=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER> 536=item C<$AnyEvent::Fork::RPC::CBOR_XS_SERIALISER> - uses L<CBOR::XS>
537
538This serialiser creates CBOR::XS arrays - you have to make sure the
539L<CBOR::XS> module is installed for this serialiser to work. It can be
540beneficial for sharing when you preload the L<CBOR::XS> module in a template
541process.
542
543L<CBOR::XS> is about as fast as the octet string serialiser, but supports
544complex data structures (similar to JSON) and is faster than any of the
545other serialisers. If you have the L<CBOR::XS> module available, it's the
546best choice.
547
548The encoder enables C<allow_sharing> (so this serialisation method can
549encode cyclic and self-referencing data structures).
550
551Implementation:
552
553 use CBOR::XS ();
554 (
555 sub { CBOR::XS::encode_cbor_sharing \@_ },
556 sub { @{ CBOR::XS::decode_cbor shift } }
557 )
558
559=item C<$AnyEvent::Fork::RPC::JSON_SERIALISER> - uses L<JSON::XS> or L<JSON>
531 560
532This serialiser creates JSON arrays - you have to make sure the L<JSON> 561This serialiser creates JSON arrays - you have to make sure the L<JSON>
533module is installed for this serialiser to work. It can be beneficial for 562module is installed for this serialiser to work. It can be beneficial for
534sharing when you preload the L<JSON> module in a template process. 563sharing when you preload the L<JSON> module in a template process.
535 564
543 ( 572 (
544 sub { JSON::encode_json \@_ }, 573 sub { JSON::encode_json \@_ },
545 sub { @{ JSON::decode_json shift } } 574 sub { @{ JSON::decode_json shift } }
546 ) 575 )
547 576
548=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER> 577=item C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER> - L<Storable>
549 578
550This serialiser uses L<Storable>, which means it has high chance of 579This serialiser uses L<Storable>, which means it has high chance of
551serialising just about anything you throw at it, at the cost of having 580serialising just about anything you throw at it, at the cost of having
552very high overhead per operation. It also comes with perl. It should be 581very high overhead per operation. It also comes with perl. It should be
553used when you need to serialise complex data structures. 582used when you need to serialise complex data structures.
558 ( 587 (
559 sub { Storable::freeze \@_ }, 588 sub { Storable::freeze \@_ },
560 sub { @{ Storable::thaw shift } } 589 sub { @{ Storable::thaw shift } }
561 ) 590 )
562 591
563=item portable storable - C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER> 592=item C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER> - portable Storable
564 593
565This serialiser also uses L<Storable>, but uses it's "network" format 594This serialiser also uses L<Storable>, but uses it's "network" format
566to serialise data, which makes it possible to talk to different 595to serialise data, which makes it possible to talk to different
567perl binaries (for example, when talking to a process created with 596perl binaries (for example, when talking to a process created with
568L<AnyEvent::Fork::Remote>). 597L<AnyEvent::Fork::Remote>).
575 sub { @{ Storable::thaw shift } } 604 sub { @{ Storable::thaw shift } }
576 ) 605 )
577 606
578=back 607=back
579 608
609=item buflen => $bytes (default: C<512 - 16>)
610
611The starting size of the read buffer for request and response data.
612
613C<AnyEvent::Fork::RPC> ensures that the buffer for reeading request and
614response data is large enough for at leats aingle request or response, and
615will dynamically enlarge the buffer if needed.
616
617While this ensures that memory is not overly wasted, it typically leads
618to having to do one syscall per request, which can be inefficient in some
619cases. In such cases, it can be beneficient to increase the buffer size to
620hold more than one request.
621
622=item buflen_req => $bytes (default: same as C<buflen>)
623
624Overrides C<buflen> for request data (as read by the forked process).
625
626=item buflen_res => $bytes (default: same as C<buflen>)
627
628Overrides C<buflen> for response data (replies read by the parent process).
629
580=back 630=back
581 631
582See the examples section earlier in this document for some actual 632See the examples section earlier in this document for some actual
583examples. 633examples.
584 634
585=cut 635=cut
586 636
587our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 637our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
638our $CBOR_XS_SERIALISER = 'use CBOR::XS (); (sub { CBOR::XS::encode_cbor_sharing \@_ }, sub { @{ CBOR::XS::decode_cbor shift } })';
588our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })'; 639our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
589our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })'; 640our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
590our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })'; 641our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })';
591 642
592sub run { 643sub run {
593 my ($self, $function, %arg) = @_; 644 my ($self, $function, %arg) = @_;
606 $on_event ||= sub { $on_error->("event received, but no on_event handler") }; 657 $on_event ||= sub { $on_error->("event received, but no on_event handler") };
607 658
608 my ($f, $t) = eval $serialiser; die $@ if $@; 659 my ($f, $t) = eval $serialiser; die $@ if $@;
609 660
610 my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww); 661 my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
611 my ($rlen, $rbuf, $rw) = 512 - 16; 662 my ($rlen, $rbuf, $rw) = $arg{buflen_res} || $arg{buflen} || 512 - 16;
612 663
613 my $wcb = sub { 664 my $wcb = sub {
614 my $len = syswrite $fh, $wbuf; 665 my $len = syswrite $fh, $wbuf;
615 666
616 unless (defined $len) { 667 unless (defined $len) {
628 } 679 }
629 }; 680 };
630 681
631 my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync"); 682 my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
632 683
633 $self->require ($module) 684 $self->eval ("use $module 2 ()")
634 ->send_arg ($function, $arg{init}, $serialiser, $arg{done} || "CORE::exit") 685 ->send_arg (
686 function => $function,
687 init => $arg{init},
688 serialiser => $serialiser,
689 done => $arg{done} || "$module\::do_exit",
690 rlen => $arg{buflen_req} || $arg{buflen} || 512 - 16,
691 -10 # the above are 10 arguments
692 )
635 ->run ("$module\::run", sub { 693 ->run ("$module\::run", sub {
636 $fh = shift; 694 $fh = shift
695 or return $on_error->("connection failed");
637 696
638 my ($id, $len); 697 my ($id, $len);
639 $rw = AE::io $fh, 0, sub { 698 $rw = AE::io $fh, 0, sub {
640 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf; 699 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
641 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf; 700 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
753values. 812values.
754 813
755See the examples section earlier in this document for some actual 814See the examples section earlier in this document for some actual
756examples. 815examples.
757 816
817Note: the event data, like any data send to the parent, might not be sent
818immediatelly but queued for later sending, so there is no guarantee that
819the event has been sent to the parent when the call returns - when you
820e.g. exit directly after calling this function, the parent might never
821receive the event.
822
758=back 823=back
759 824
760=head2 PROCESS EXIT 825=head2 PROCESS EXIT
761 826
762If and when the child process exits depends on the backend and 827If and when the child process exits depends on the backend and
786 851
787For the asynchronous backend, things are more complicated: Whenever it 852For the asynchronous backend, things are more complicated: Whenever it
788listens for another request by the parent, it might detect that the socket 853listens for another request by the parent, it might detect that the socket
789was closed (e.g. because the parent exited). It will sotp listening for 854was closed (e.g. because the parent exited). It will sotp listening for
790new requests and instead try to write out any remaining data (if any) or 855new requests and instead try to write out any remaining data (if any) or
791simply check whether the socket cna be written to. After this, the RPC 856simply check whether the socket can be written to. After this, the RPC
792process is effectively done - no new requests are incoming, no outstanding 857process is effectively done - no new requests are incoming, no outstanding
793request data can be written back. 858request data can be written back.
794 859
795Since chances are high that there are event watchers that the RPC server 860Since chances are high that there are event watchers that the RPC server
796knows nothing about (why else would one use the async backend if not for 861knows nothing about (why else would one use the async backend if not for
867are queued and the jobs are slow, they will all run concurrently. The 932are queued and the jobs are slow, they will all run concurrently. The
868child must implement some queueing/limiting mechanism if this causes 933child must implement some queueing/limiting mechanism if this causes
869problems. Alternatively, the parent could limit the amount of rpc calls 934problems. Alternatively, the parent could limit the amount of rpc calls
870that are outstanding. 935that are outstanding.
871 936
872Blocking use of condvars is not supported. 937Blocking use of condvars is not supported (in the main thread, outside of
938e.g. L<Coro> threads).
873 939
874Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is 940Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
875easy. 941easy.
876 942
877=back 943=back
943gory details. 1009gory details.
944 1010
945=head1 EXCEPTIONS 1011=head1 EXCEPTIONS
946 1012
947There are no provisions whatsoever for catching exceptions at this time - 1013There are no provisions whatsoever for catching exceptions at this time -
948in the child, exeptions might kill the process, causing calls to be lost 1014in the child, exceptions might kill the process, causing calls to be lost
949and the parent encountering a fatal error. In the parent, exceptions in 1015and the parent encountering a fatal error. In the parent, exceptions in
950the result callback will not be caught and cause undefined behaviour. 1016the result callback will not be caught and cause undefined behaviour.
951 1017
952=head1 SEE ALSO 1018=head1 SEE ALSO
953 1019

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines