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.396 by root, Wed Nov 14 08:09:46 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;
3320 Event::loop; 3333 EV::loop;
3321} 3334}
3322 3335
3323############################################################################# 3336#############################################################################
3324# initialisation and cleanup 3337# initialisation and cleanup
3325 3338
3326# install some emergency cleanup handlers 3339# install some emergency cleanup handlers
3327BEGIN { 3340BEGIN {
3341 our %SIGWATCHER = ();
3328 for my $signal (qw(INT HUP TERM)) { 3342 for my $signal (qw(INT HUP TERM)) {
3329 Event->signal ( 3343 $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"; 3344 cf::cleanup "SIG$signal";
3336 },
3337 ); 3345 };
3338 } 3346 }
3339} 3347}
3340 3348
3341sub write_runtime { 3349sub write_runtime {
3342 my $runtime = "$LOCALDIR/runtime"; 3350 my $runtime = "$LOCALDIR/runtime";
… …
3436 warn "syncing database to disk"; 3444 warn "syncing database to disk";
3437 BDB::db_env_txn_checkpoint $DB_ENV; 3445 BDB::db_env_txn_checkpoint $DB_ENV;
3438 3446
3439 # if anything goes wrong in here, we should simply crash as we already saved 3447 # if anything goes wrong in here, we should simply crash as we already saved
3440 3448
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"; 3449 warn "flushing outstanding aio requests";
3447 for (;;) { 3450 for (;;) {
3448 BDB::flush; 3451 BDB::flush;
3449 IO::AIO::flush; 3452 IO::AIO::flush;
3450 Coro::cede_notself; 3453 Coro::cede_notself;
… …
3544 3547
3545sub reload_perl() { 3548sub reload_perl() {
3546 # doing reload synchronously and two reloads happen back-to-back, 3549 # doing reload synchronously and two reloads happen back-to-back,
3547 # coro crashes during coro_state_free->destroy here. 3550 # coro crashes during coro_state_free->destroy here.
3548 3551
3549 $RELOAD_WATCHER ||= Event->timer ( 3552 $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; 3553 undef $RELOAD_WATCHER;
3556 }, 3554 do_reload_perl;
3557 ); 3555 };
3558} 3556}
3559 3557
3560register_command "reload" => sub { 3558register_command "reload" => sub {
3561 my ($who, $arg) = @_; 3559 my ($who, $arg) = @_;
3562 3560
… …
3592 my $signal = new Coro::Signal; 3590 my $signal = new Coro::Signal;
3593 push @WAIT_FOR_TICK_BEGIN, $signal; 3591 push @WAIT_FOR_TICK_BEGIN, $signal;
3594 $signal->wait; 3592 $signal->wait;
3595} 3593}
3596 3594
3597$TICK_WATCHER = Event->timer ( 3595$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) { 3596 if ($Coro::current != $Coro::main) {
3605 Carp::cluck "major BUG: server tick called outside of main coro, skipping it" 3597 Carp::cluck "major BUG: server tick called outside of main coro, skipping it"
3606 unless ++$bug_warning > 10; 3598 unless ++$bug_warning > 10;
3607 return; 3599 return;
3608 } 3600 }
3609 3601
3610 $NOW = $tick_start = Event::time; 3602 $NOW = $tick_start = EV::now;
3611 3603
3612 cf::server_tick; # one server iteration 3604 cf::server_tick; # one server iteration
3613 3605
3614 $RUNTIME += $TICK; 3606 $RUNTIME += $TICK;
3615 $NEXT_TICK += $TICK; 3607 $NEXT_TICK += $TICK;
3616 3608
3617 if ($NOW >= $NEXT_RUNTIME_WRITE) { 3609 if ($NOW >= $NEXT_RUNTIME_WRITE) {
3618 $NEXT_RUNTIME_WRITE = $NOW + 10; 3610 $NEXT_RUNTIME_WRITE = $NOW + 10;
3619 Coro::async_pool { 3611 Coro::async_pool {
3620 $Coro::current->{desc} = "runtime saver"; 3612 $Coro::current->{desc} = "runtime saver";
3621 write_runtime 3613 write_runtime
3622 or warn "ERROR: unable to write runtime file: $!"; 3614 or warn "ERROR: unable to write runtime file: $!";
3623 };
3624 } 3615 };
3616 }
3625 3617
3626 if (my $sig = shift @WAIT_FOR_TICK_BEGIN) { 3618 if (my $sig = shift @WAIT_FOR_TICK_BEGIN) {
3627 $sig->send; 3619 $sig->send;
3628 } 3620 }
3629 while (my $sig = shift @WAIT_FOR_TICK) { 3621 while (my $sig = shift @WAIT_FOR_TICK) {
3630 $sig->send; 3622 $sig->send;
3631 } 3623 }
3632 3624
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; 3625 $LOAD = ($NOW - $tick_start) / $TICK;
3642 $LOADAVG = $LOADAVG * 0.75 + $LOAD * 0.25; 3626 $LOADAVG = $LOADAVG * 0.75 + $LOAD * 0.25;
3643 3627
3644 _post_tick; 3628 _post_tick;
3645 }, 3629};
3646); 3630$TICK_WATCHER->priority (EV::MAXPRI);
3647 3631
3648{ 3632{
3649 BDB::min_parallel 8; 3633 BDB::min_parallel 8;
3650 BDB::max_poll_time $TICK * 0.1; 3634 BDB::max_poll_time $TICK * 0.1;
3651 $BDB_POLL_WATCHER = Event->io ( 3635 $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 3636
3660 BDB::set_sync_prepare { 3637 BDB::set_sync_prepare {
3661 my $status; 3638 my $status;
3662 my $current = $Coro::current; 3639 my $current = $Coro::current;
3663 ( 3640 (
… …
3693 3670
3694 cf::cleanup "db_env_open(db): $@" if $@; 3671 cf::cleanup "db_env_open(db): $@" if $@;
3695 }; 3672 };
3696 } 3673 }
3697 3674
3698 $BDB_DEADLOCK_WATCHER = Event->timer ( 3675 $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 { }; 3676 BDB::db_env_lock_detect $DB_ENV, 0, BDB::LOCK_DEFAULT, 0, sub { };
3706 },
3707 ); 3677 };
3708 $BDB_CHECKPOINT_WATCHER = Event->timer ( 3678 $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 { }; 3679 BDB::db_env_txn_checkpoint $DB_ENV, 0, 0, 0, sub { };
3716 },
3717 ); 3680 };
3718 $BDB_TRICKLE_WATCHER = Event->timer ( 3681 $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 { }; 3682 BDB::db_env_memp_trickle $DB_ENV, 20, 0, sub { };
3726 },
3727 ); 3683 };
3728} 3684}
3729 3685
3730{ 3686{
3731 IO::AIO::min_parallel 8; 3687 IO::AIO::min_parallel 8;
3732 3688
3733 undef $Coro::AIO::WATCHER; 3689 undef $Coro::AIO::WATCHER;
3734 IO::AIO::max_poll_time $TICK * 0.1; 3690 IO::AIO::max_poll_time $TICK * 0.1;
3735 $AIO_POLL_WATCHER = Event->io ( 3691 $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} 3692}
3744 3693
3745my $_log_backtrace; 3694my $_log_backtrace;
3746 3695
3747sub _log_backtrace { 3696sub _log_backtrace {

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines