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.397 by root, Mon Nov 19 15:47:14 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;
… …
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
… …
336 }; 334 };
337 $TICK_WATCHER->stop; 335 $TICK_WATCHER->stop;
338 $guard 336 $guard
339} 337}
340 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
352}
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>:
344This call blocks and returns only when you have at least C<$time> seconds 357This call blocks and returns only when you have at least C<$time> seconds
345of cpu time till the next tick. The slot is only valid till the next cede. 358of cpu time till the next tick. The slot is only valid till the next cede.
… …
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) = @_;
… …
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