ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/deliantra/server/lib/cf.pm
(Generate patch)

Comparing deliantra/server/lib/cf.pm (file contents):
Revision 1.395 by root, Sat Nov 10 22:41:59 2007 UTC vs.
Revision 1.398 by root, Wed Dec 5 11:08:34 2007 UTC

4use strict; 4use strict;
5 5
6use Symbol; 6use Symbol;
7use List::Util; 7use List::Util;
8use Socket; 8use Socket;
9use Event; 9use EV;
10use Opcode; 10use Opcode;
11use Safe; 11use Safe;
12use Safe::Hole; 12use Safe::Hole;
13use Storable (); 13use Storable ();
14 14
15use Coro 4.1 (); 15use Coro 4.1 ();
16use Coro::State; 16use Coro::State;
17use Coro::Handle; 17use Coro::Handle;
18use Coro::Event; 18use Coro::EV;
19use Coro::Timer; 19use Coro::Timer;
20use Coro::Signal; 20use Coro::Signal;
21use Coro::Semaphore; 21use Coro::Semaphore;
22use Coro::AIO; 22use Coro::AIO;
23use Coro::Storable; 23use Coro::Storable;
24use Coro::Util (); 24use Coro::Util ();
25 25
26use JSON::XS (); 26use JSON::XS 2.01 ();
27use BDB (); 27use BDB ();
28use Data::Dumper; 28use Data::Dumper;
29use Digest::MD5; 29use Digest::MD5;
30use Fcntl; 30use Fcntl;
31use YAML::Syck (); 31use YAML::Syck ();
37# configure various modules to our taste 37# configure various modules to our taste
38# 38#
39$Storable::canonical = 1; # reduce rsync transfers 39$Storable::canonical = 1; # reduce rsync transfers
40Coro::State::cctx_stacksize 256000; # 1-2MB stack, for deep recursions in maze generator 40Coro::State::cctx_stacksize 256000; # 1-2MB stack, for deep recursions in maze generator
41Compress::LZF::sfreeze_cr { }; # prime Compress::LZF so it does not use require later 41Compress::LZF::sfreeze_cr { }; # prime Compress::LZF so it does not use require later
42
43$Event::Eval = 1; # no idea why this is required, but it is
44 42
45# work around bug in YAML::Syck - bad news for perl6, will it be as broken wrt. unicode? 43# work around bug in YAML::Syck - bad news for perl6, will it be as broken wrt. unicode?
46$YAML::Syck::ImplicitUnicode = 1; 44$YAML::Syck::ImplicitUnicode = 1;
47 45
48$Coro::main->prio (Coro::PRIO_MAX); # run main coroutine ("the server") with very high priority 46$Coro::main->prio (Coro::PRIO_MAX); # run main coroutine ("the server") with very high priority
215)) { 213)) {
216 no strict 'refs'; 214 no strict 'refs';
217 @{"safe::$pkg\::wrap::ISA"} = @{"$pkg\::wrap::ISA"} = $pkg; 215 @{"safe::$pkg\::wrap::ISA"} = @{"$pkg\::wrap::ISA"} = $pkg;
218} 216}
219 217
220$Event::DIED = sub { 218$EV::DIED = sub {
221 warn "error in event callback: @_"; 219 warn "error in event callback: @_";
222}; 220};
223 221
224############################################################################# 222#############################################################################
225 223
248 $d =~ s/([\x00-\x07\x09\x0b\x0c\x0e-\x1f])/sprintf "\\x%02x", ord($1)/ge; 246 $d =~ s/([\x00-\x07\x09\x0b\x0c\x0e-\x1f])/sprintf "\\x%02x", ord($1)/ge;
249 $d 247 $d
250 } || "[unable to dump $_[0]: '$@']"; 248 } || "[unable to dump $_[0]: '$@']";
251} 249}
252 250
253=item $ref = cf::from_json $json 251=item $ref = cf::decode_json $json
254 252
255Converts a JSON string into the corresponding perl data structure. 253Converts a JSON string into the corresponding perl data structure.
256 254
257=item $json = cf::to_json $ref 255=item $json = cf::encode_json $ref
258 256
259Converts a perl data structure into its JSON representation. 257Converts a perl data structure into its JSON representation.
260 258
261=cut 259=cut
262 260
263our $json_coder = JSON::XS->new->utf8->max_size (1e6); # accept ~1mb max 261our $json_coder = JSON::XS->new->utf8->max_size (1e6); # accept ~1mb max
264 262
265sub to_json ($) { $json_coder->encode ($_[0]) } 263sub encode_json($) { $json_coder->encode ($_[0]) }
266sub from_json ($) { $json_coder->decode ($_[0]) } 264sub decode_json($) { $json_coder->decode ($_[0]) }
267 265
268=item cf::lock_wait $string 266=item cf::lock_wait $string
269 267
270Wait until the given lock is available. See cf::lock_acquire. 268Wait until the given lock is available. See cf::lock_acquire.
271 269
334 my $guard = Coro::guard { 332 my $guard = Coro::guard {
335 $TICK_WATCHER->start; 333 $TICK_WATCHER->start;
336 }; 334 };
337 $TICK_WATCHER->stop; 335 $TICK_WATCHER->stop;
338 $guard 336 $guard
337}
338
339=item cf::periodic $interval, $cb
340
341Like EV::periodic, but randomly selects a starting point so that the actions
342get spread over timer.
343
344=cut
345
346sub periodic($$) {
347 my ($interval, $cb) = @_;
348
349 my $start = rand List::Util::min 180, $interval;
350
351 EV::periodic $start, $interval, 0, $cb
339} 352}
340 353
341=item cf::get_slot $time[, $priority[, $name]] 354=item cf::get_slot $time[, $priority[, $name]]
342 355
343Allocate $time seconds of blocking CPU time at priority C<$priority>: 356Allocate $time seconds of blocking CPU time at priority C<$priority>:
422 435
423sub sync_job(&) { 436sub sync_job(&) {
424 my ($job) = @_; 437 my ($job) = @_;
425 438
426 if ($Coro::current == $Coro::main) { 439 if ($Coro::current == $Coro::main) {
427 my $time = Event::time; 440 my $time = EV::time;
428 441
429 # this is the main coro, too bad, we have to block 442 # this is the main coro, too bad, we have to block
430 # till the operation succeeds, freezing the server :/ 443 # till the operation succeeds, freezing the server :/
431 444
432 LOG llevError, Carp::longmess "sync job";#d# 445 LOG llevError, Carp::longmess "sync job";#d#
447 460
448 while ($busy) { 461 while ($busy) {
449 if (Coro::nready) { 462 if (Coro::nready) {
450 Coro::cede_notself; 463 Coro::cede_notself;
451 } else { 464 } else {
452 Event::one_event; 465 EV::loop EV::LOOP_ONESHOT;
453 } 466 }
454 } 467 }
455 468
456 $time = Event::time - $time; 469 $time = EV::time - $time;
457 470
458 LOG llevError | logBacktrace, Carp::longmess "long sync job" 471 LOG llevError | logBacktrace, Carp::longmess "long sync job"
459 if $time > $TICK * 0.5 && $TICK_WATCHER->is_active; 472 if $time > $TICK * 0.5 && $TICK_WATCHER->is_active;
460 473
461 $tick_start += $time; # do not account sync jobs to server load 474 $tick_start += $time; # do not account sync jobs to server load
491=item fork_call { }, $args 504=item fork_call { }, $args
492 505
493Executes the given code block with the given arguments in a seperate 506Executes the given code block with the given arguments in a seperate
494process, returning the results. Everything must be serialisable with 507process, returning the results. Everything must be serialisable with
495Coro::Storable. May, of course, block. Note that the executed sub may 508Coro::Storable. May, of course, block. Note that the executed sub may
496never block itself or use any form of Event handling. 509never block itself or use any form of event handling.
497 510
498=cut 511=cut
499 512
500sub fork_call(&@) { 513sub fork_call(&@) {
501 my ($cb, @args) = @_; 514 my ($cb, @args) = @_;
1042cf::attachable->attach ( 1055cf::attachable->attach (
1043 prio => -1000000, 1056 prio => -1000000,
1044 on_instantiate => sub { 1057 on_instantiate => sub {
1045 my ($obj, $data) = @_; 1058 my ($obj, $data) = @_;
1046 1059
1047 $data = from_json $data; 1060 $data = decode_json $data;
1048 1061
1049 for (@$data) { 1062 for (@$data) {
1050 my ($name, $args) = @$_; 1063 my ($name, $args) = @$_;
1051 1064
1052 $obj->attach ($name, %{$args || {} }); 1065 $obj->attach ($name, %{$args || {} });
2625 $rmp->{origin_y} = $exit->y; 2638 $rmp->{origin_y} = $exit->y;
2626 } 2639 }
2627 2640
2628 $rmp->{random_seed} ||= $exit->random_seed; 2641 $rmp->{random_seed} ||= $exit->random_seed;
2629 2642
2630 my $data = cf::to_json $rmp; 2643 my $data = cf::encode_json $rmp;
2631 my $md5 = Digest::MD5::md5_hex $data; 2644 my $md5 = Digest::MD5::md5_hex $data;
2632 my $meta = "$RANDOMDIR/$md5.meta"; 2645 my $meta = "$RANDOMDIR/$md5.meta";
2633 2646
2634 if (my $fh = aio_open "$meta~", O_WRONLY | O_CREAT, 0666) { 2647 if (my $fh = aio_open "$meta~", O_WRONLY | O_CREAT, 0666) {
2635 aio_write $fh, 0, (length $data), $data, 0; 2648 aio_write $fh, 0, (length $data), $data, 0;
3306 # we must not ever block the main coroutine 3319 # we must not ever block the main coroutine
3307 local $Coro::idle = sub { 3320 local $Coro::idle = sub {
3308 Carp::cluck "FATAL: Coro::idle was called, major BUG, use cf::sync_job!\n";#d# 3321 Carp::cluck "FATAL: Coro::idle was called, major BUG, use cf::sync_job!\n";#d#
3309 (async { 3322 (async {
3310 $Coro::current->{desc} = "IDLE BUG HANDLER"; 3323 $Coro::current->{desc} = "IDLE BUG HANDLER";
3311 Event::one_event; 3324 EV::loop EV::LOOP_ONESHOT;
3312 })->prio (Coro::PRIO_MAX); 3325 })->prio (Coro::PRIO_MAX);
3313 }; 3326 };
3314 3327
3315 reload_config; 3328 reload_config;
3316 db_init; 3329 db_init;
3317 load_extensions; 3330 load_extensions;
3318 3331
3319 $TICK_WATCHER->start; 3332 $TICK_WATCHER->start;
3333 $Coro::current->prio (Coro::PRIO_MAX); # give the main loop max. priority
3320 Event::loop; 3334 EV::loop;
3321} 3335}
3322 3336
3323############################################################################# 3337#############################################################################
3324# initialisation and cleanup 3338# initialisation and cleanup
3325 3339
3326# install some emergency cleanup handlers 3340# install some emergency cleanup handlers
3327BEGIN { 3341BEGIN {
3342 our %SIGWATCHER = ();
3328 for my $signal (qw(INT HUP TERM)) { 3343 for my $signal (qw(INT HUP TERM)) {
3329 Event->signal ( 3344 $SIGWATCHER{$signal} = EV::signal $signal, sub {
3330 reentrant => 0,
3331 data => WF_AUTOCANCEL,
3332 signal => $signal,
3333 prio => 0,
3334 cb => sub {
3335 cf::cleanup "SIG$signal"; 3345 cf::cleanup "SIG$signal";
3336 },
3337 ); 3346 };
3338 } 3347 }
3339} 3348}
3340 3349
3341sub write_runtime { 3350sub write_runtime {
3342 my $runtime = "$LOCALDIR/runtime"; 3351 my $runtime = "$LOCALDIR/runtime";
3436 warn "syncing database to disk"; 3445 warn "syncing database to disk";
3437 BDB::db_env_txn_checkpoint $DB_ENV; 3446 BDB::db_env_txn_checkpoint $DB_ENV;
3438 3447
3439 # if anything goes wrong in here, we should simply crash as we already saved 3448 # if anything goes wrong in here, we should simply crash as we already saved
3440 3449
3441 warn "cancelling all WF_AUTOCANCEL watchers";
3442 for (Event::all_watchers) {
3443 $_->cancel if $_->data & WF_AUTOCANCEL;
3444 }
3445
3446 warn "flushing outstanding aio requests"; 3450 warn "flushing outstanding aio requests";
3447 for (;;) { 3451 for (;;) {
3448 BDB::flush; 3452 BDB::flush;
3449 IO::AIO::flush; 3453 IO::AIO::flush;
3450 Coro::cede_notself; 3454 Coro::cede_notself;
3544 3548
3545sub reload_perl() { 3549sub reload_perl() {
3546 # doing reload synchronously and two reloads happen back-to-back, 3550 # doing reload synchronously and two reloads happen back-to-back,
3547 # coro crashes during coro_state_free->destroy here. 3551 # coro crashes during coro_state_free->destroy here.
3548 3552
3549 $RELOAD_WATCHER ||= Event->timer ( 3553 $RELOAD_WATCHER ||= EV::timer 0, 0, sub {
3550 reentrant => 0,
3551 after => 0,
3552 data => WF_AUTOCANCEL,
3553 cb => sub {
3554 do_reload_perl;
3555 undef $RELOAD_WATCHER; 3554 undef $RELOAD_WATCHER;
3556 }, 3555 do_reload_perl;
3557 ); 3556 };
3558} 3557}
3559 3558
3560register_command "reload" => sub { 3559register_command "reload" => sub {
3561 my ($who, $arg) = @_; 3560 my ($who, $arg) = @_;
3562 3561
3592 my $signal = new Coro::Signal; 3591 my $signal = new Coro::Signal;
3593 push @WAIT_FOR_TICK_BEGIN, $signal; 3592 push @WAIT_FOR_TICK_BEGIN, $signal;
3594 $signal->wait; 3593 $signal->wait;
3595} 3594}
3596 3595
3597$TICK_WATCHER = Event->timer ( 3596$TICK_WATCHER = EV::periodic_ns 0, $TICK, 0, sub {
3598 reentrant => 0,
3599 parked => 1,
3600 prio => 0,
3601 at => $NEXT_TICK || $TICK,
3602 data => WF_AUTOCANCEL,
3603 cb => sub {
3604 if ($Coro::current != $Coro::main) { 3597 if ($Coro::current != $Coro::main) {
3605 Carp::cluck "major BUG: server tick called outside of main coro, skipping it" 3598 Carp::cluck "major BUG: server tick called outside of main coro, skipping it"
3606 unless ++$bug_warning > 10; 3599 unless ++$bug_warning > 10;
3607 return; 3600 return;
3608 } 3601 }
3609 3602
3610 $NOW = $tick_start = Event::time; 3603 $NOW = $tick_start = EV::now;
3611 3604
3612 cf::server_tick; # one server iteration 3605 cf::server_tick; # one server iteration
3613 3606
3614 $RUNTIME += $TICK; 3607 $RUNTIME += $TICK;
3615 $NEXT_TICK += $TICK; 3608 $NEXT_TICK += $TICK;
3616 3609
3617 if ($NOW >= $NEXT_RUNTIME_WRITE) { 3610 if ($NOW >= $NEXT_RUNTIME_WRITE) {
3618 $NEXT_RUNTIME_WRITE = $NOW + 10; 3611 $NEXT_RUNTIME_WRITE = $NOW + 10;
3619 Coro::async_pool { 3612 Coro::async_pool {
3620 $Coro::current->{desc} = "runtime saver"; 3613 $Coro::current->{desc} = "runtime saver";
3621 write_runtime 3614 write_runtime
3622 or warn "ERROR: unable to write runtime file: $!"; 3615 or warn "ERROR: unable to write runtime file: $!";
3623 };
3624 } 3616 };
3617 }
3625 3618
3626 if (my $sig = shift @WAIT_FOR_TICK_BEGIN) { 3619 if (my $sig = shift @WAIT_FOR_TICK_BEGIN) {
3627 $sig->send; 3620 $sig->send;
3628 } 3621 }
3629 while (my $sig = shift @WAIT_FOR_TICK) { 3622 while (my $sig = shift @WAIT_FOR_TICK) {
3630 $sig->send; 3623 $sig->send;
3631 } 3624 }
3632 3625
3633 $NOW = Event::time;
3634
3635 # if we are delayed by four ticks or more, skip them all
3636 $NEXT_TICK = $NOW if $NOW >= $NEXT_TICK + $TICK * 4;
3637
3638 $TICK_WATCHER->at ($NEXT_TICK);
3639 $TICK_WATCHER->start;
3640
3641 $LOAD = ($NOW - $tick_start) / $TICK; 3626 $LOAD = ($NOW - $tick_start) / $TICK;
3642 $LOADAVG = $LOADAVG * 0.75 + $LOAD * 0.25; 3627 $LOADAVG = $LOADAVG * 0.75 + $LOAD * 0.25;
3643 3628
3644 _post_tick; 3629 _post_tick;
3645 }, 3630};
3646); 3631$TICK_WATCHER->priority (EV::MAXPRI);
3647 3632
3648{ 3633{
3649 BDB::min_parallel 8; 3634 BDB::min_parallel 8;
3650 BDB::max_poll_time $TICK * 0.1; 3635 BDB::max_poll_time $TICK * 0.1;
3651 $BDB_POLL_WATCHER = Event->io ( 3636 $BDB_POLL_WATCHER = EV::io BDB::poll_fileno, EV::READ, \&BDB::poll_cb;
3652 reentrant => 0,
3653 fd => BDB::poll_fileno,
3654 poll => 'r',
3655 prio => 0,
3656 data => WF_AUTOCANCEL,
3657 cb => \&BDB::poll_cb,
3658 );
3659 3637
3660 BDB::set_sync_prepare { 3638 BDB::set_sync_prepare {
3661 my $status; 3639 my $status;
3662 my $current = $Coro::current; 3640 my $current = $Coro::current;
3663 ( 3641 (
3693 3671
3694 cf::cleanup "db_env_open(db): $@" if $@; 3672 cf::cleanup "db_env_open(db): $@" if $@;
3695 }; 3673 };
3696 } 3674 }
3697 3675
3698 $BDB_DEADLOCK_WATCHER = Event->timer ( 3676 $BDB_DEADLOCK_WATCHER = EV::periodic 0, 3, 0, sub {
3699 after => 3,
3700 interval => 1,
3701 hard => 1,
3702 prio => 0,
3703 data => WF_AUTOCANCEL,
3704 cb => sub {
3705 BDB::db_env_lock_detect $DB_ENV, 0, BDB::LOCK_DEFAULT, 0, sub { }; 3677 BDB::db_env_lock_detect $DB_ENV, 0, BDB::LOCK_DEFAULT, 0, sub { };
3706 },
3707 ); 3678 };
3708 $BDB_CHECKPOINT_WATCHER = Event->timer ( 3679 $BDB_CHECKPOINT_WATCHER = EV::periodic 0, 60, 0, sub {
3709 after => 11,
3710 interval => 60,
3711 hard => 1,
3712 prio => 0,
3713 data => WF_AUTOCANCEL,
3714 cb => sub {
3715 BDB::db_env_txn_checkpoint $DB_ENV, 0, 0, 0, sub { }; 3680 BDB::db_env_txn_checkpoint $DB_ENV, 0, 0, 0, sub { };
3716 },
3717 ); 3681 };
3718 $BDB_TRICKLE_WATCHER = Event->timer ( 3682 $BDB_TRICKLE_WATCHER = EV::periodic 0, 10, 0, sub {
3719 after => 5,
3720 interval => 10,
3721 hard => 1,
3722 prio => 0,
3723 data => WF_AUTOCANCEL,
3724 cb => sub {
3725 BDB::db_env_memp_trickle $DB_ENV, 20, 0, sub { }; 3683 BDB::db_env_memp_trickle $DB_ENV, 20, 0, sub { };
3726 },
3727 ); 3684 };
3728} 3685}
3729 3686
3730{ 3687{
3731 IO::AIO::min_parallel 8; 3688 IO::AIO::min_parallel 8;
3732 3689
3733 undef $Coro::AIO::WATCHER; 3690 undef $Coro::AIO::WATCHER;
3734 IO::AIO::max_poll_time $TICK * 0.1; 3691 IO::AIO::max_poll_time $TICK * 0.1;
3735 $AIO_POLL_WATCHER = Event->io ( 3692 $AIO_POLL_WATCHER = EV::io IO::AIO::poll_fileno, EV::READ, \&IO::AIO::poll_cb;
3736 reentrant => 0,
3737 data => WF_AUTOCANCEL,
3738 fd => IO::AIO::poll_fileno,
3739 poll => 'r',
3740 prio => 0,
3741 cb => \&IO::AIO::poll_cb,
3742 );
3743} 3693}
3744 3694
3745my $_log_backtrace; 3695my $_log_backtrace;
3746 3696
3747sub _log_backtrace { 3697sub _log_backtrace {

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines