--- deliantra/server/lib/cf.pm 2007/09/15 13:25:33 1.372 +++ deliantra/server/lib/cf.pm 2007/12/17 07:03:32 1.402 @@ -6,30 +6,31 @@ use Symbol; use List::Util; use Socket; -use Storable; -use Event; +use EV; use Opcode; use Safe; use Safe::Hole; +use Storable (); -use Coro 3.64 (); +use Coro 4.32 (); use Coro::State; use Coro::Handle; -use Coro::Event; +use Coro::EV; use Coro::Timer; use Coro::Signal; use Coro::Semaphore; use Coro::AIO; +use Coro::BDB; use Coro::Storable; use Coro::Util (); -use JSON::XS (); +use JSON::XS 2.01 (); use BDB (); use Data::Dumper; use Digest::MD5; use Fcntl; use YAML::Syck (); -use IO::AIO 2.32 (); +use IO::AIO 2.51 (); use Time::HiRes; use Compress::LZF; use Digest::MD5 (); @@ -40,8 +41,6 @@ Coro::State::cctx_stacksize 256000; # 1-2MB stack, for deep recursions in maze generator Compress::LZF::sfreeze_cr { }; # prime Compress::LZF so it does not use require later -$Event::Eval = 1; # no idea why this is required, but it is - # work around bug in YAML::Syck - bad news for perl6, will it be as broken wrt. unicode? $YAML::Syck::ImplicitUnicode = 1; @@ -78,7 +77,6 @@ our $AIO_POLL_WATCHER; our $NEXT_RUNTIME_WRITE; # when should the runtime file be written our $NEXT_TICK; -our $NOW; our $USE_FSYNC = 1; # use fsync to write maps - default off our $BDB_POLL_WATCHER; @@ -91,6 +89,7 @@ our $UPTIME; $UPTIME ||= time; our $RUNTIME; +our $NOW; our (%PLAYER, %PLAYER_LOADING); # all users our (%MAP, %MAP_LOADING ); # all maps @@ -164,7 +163,7 @@ =item %cf::CFG -Configuration for the server, loaded from C, or +Configuration for the server, loaded from C, or from wherever your confdir points to. =item cf::wait_for_tick, cf::wait_for_tick_begin @@ -217,7 +216,7 @@ @{"safe::$pkg\::wrap::ISA"} = @{"$pkg\::wrap::ISA"} = $pkg; } -$Event::DIED = sub { +$EV::DIED = sub { warn "error in event callback: @_"; }; @@ -250,11 +249,11 @@ } || "[unable to dump $_[0]: '$@']"; } -=item $ref = cf::from_json $json +=item $ref = cf::decode_json $json Converts a JSON string into the corresponding perl data structure. -=item $json = cf::to_json $ref +=item $json = cf::encode_json $ref Converts a perl data structure into its JSON representation. @@ -262,8 +261,8 @@ our $json_coder = JSON::XS->new->utf8->max_size (1e6); # accept ~1mb max -sub to_json ($) { $json_coder->encode ($_[0]) } -sub from_json ($) { $json_coder->decode ($_[0]) } +sub encode_json($) { $json_coder->encode ($_[0]) } +sub decode_json($) { $json_coder->decode ($_[0]) } =item cf::lock_wait $string @@ -338,6 +337,21 @@ $guard } +=item cf::periodic $interval, $cb + +Like EV::periodic, but randomly selects a starting point so that the actions +get spread over timer. + +=cut + +sub periodic($$) { + my ($interval, $cb) = @_; + + my $start = rand List::Util::min 180, $interval; + + EV::periodic $start, $interval, 0, $cb +} + =item cf::get_slot $time[, $priority[, $name]] Allocate $time seconds of blocking CPU time at priority C<$priority>: @@ -356,6 +370,8 @@ $SLOT_QUEUE->cancel if $SLOT_QUEUE; $SLOT_QUEUE = Coro::async { + $Coro::current->desc ("timeslot manager"); + my $signal = new Coro::Signal; while () { @@ -373,7 +389,7 @@ } if (@SLOT_QUEUE) { - # we do not use wait_For_tick() as it returns immediately when tick is inactive + # we do not use wait_for_tick() as it returns immediately when tick is inactive push @cf::WAIT_FOR_TICK, $signal; $signal->wait; } else { @@ -406,8 +422,8 @@ =item cf::sync_job { BLOCK } -The design of Crossfire TRT requires that the main coroutine ($Coro::main) -is always able to handle events or runnable, as Crossfire TRT is only +The design of Deliantra requires that the main coroutine ($Coro::main) +is always able to handle events or runnable, as Deliantra is only partly reentrant. Thus "blocking" it by e.g. waiting for I/O is not acceptable. @@ -422,11 +438,13 @@ my ($job) = @_; if ($Coro::current == $Coro::main) { - my $time = Event::time; + my $time = EV::time; # this is the main coro, too bad, we have to block # till the operation succeeds, freezing the server :/ + LOG llevError, Carp::longmess "sync job";#d# + # TODO: use suspend/resume instead # (but this is cancel-safe) my $freeze_guard = freeze_mainloop; @@ -435,16 +453,21 @@ my @res; (async { + $Coro::current->desc ("sync job coro"); @res = eval { $job->() }; warn $@ if $@; undef $busy; })->prio (Coro::PRIO_MAX); while ($busy) { - Coro::cede or Event::one_event; + if (Coro::nready) { + Coro::cede_notself; + } else { + EV::loop EV::LOOP_ONESHOT; + } } - $time = Event::time - $time; + $time = EV::time - $time; LOG llevError | logBacktrace, Carp::longmess "long sync job" if $time > $TICK * 0.5 && $TICK_WATCHER->is_active; @@ -484,7 +507,7 @@ Executes the given code block with the given arguments in a seperate process, returning the results. Everything must be serialisable with Coro::Storable. May, of course, block. Note that the executed sub may -never block itself or use any form of Event handling. +never block itself or use any form of event handling. =cut @@ -655,7 +678,7 @@ In the following description, CLASS can be any of C, C C, C or C (i.e. the attachable objects in -Crossfire TRT). +Deliantra). =over 4 @@ -963,16 +986,47 @@ ############################################################################# # object support -# +sub _object_equal($$); +sub _object_equal($$) { + my ($a, $b) = @_; + + return 0 unless (ref $a) eq (ref $b); + + if ("HASH" eq ref $a) { + my @ka = keys %$a; + my @kb = keys %$b; + + return 0 if @ka != @kb; + + for (0 .. $#ka) { + return 0 unless $ka[$_] eq $kb[$_]; + return 0 unless _object_equal $a->{$ka[$_]}, $b->{$kb[$_]}; + } + + } elsif ("ARRAY" eq ref $a) { + + return 0 if @$a != @$b; + + for (0 .. $#$a) { + return 0 unless _object_equal $a->[$_], $b->[$_]; + } + + } elsif ($a ne $b) { + return 0; + } + + 1 +} + +our $SLOW_MERGES;#d# sub _can_merge { my ($ob1, $ob2) = @_; - local $Storable::canonical = 1; - my $fob1 = Storable::freeze $ob1; - my $fob2 = Storable::freeze $ob2; + ++$SLOW_MERGES;#d# - $fob1 eq $fob2 + # we do the slow way here + return _object_equal $ob1, $ob2 } sub reattach { @@ -1004,7 +1058,7 @@ on_instantiate => sub { my ($obj, $data) = @_; - $data = from_json $data; + $data = decode_json $data; for (@$data) { my ($name, $args) = @$_; @@ -1028,8 +1082,6 @@ sub object_freezer_save { my ($filename, $rdata, $objs) = @_; - my $guard = cf::lock_acquire "io"; - sync_job { if (length $$rdata) { utf8::decode (my $decname = $filename); @@ -1045,7 +1097,7 @@ if (@$objs) { if (my $fh = aio_open "$filename.pst~", O_WRONLY | O_CREAT, 0600) { chmod SAVE_MODE, $fh; - my $data = Storable::nfreeze { version => 1, objs => $objs }; + my $data = Coro::Storable::nfreeze { version => 1, objs => $objs }; aio_write $fh, 0, (length $data), $data, 0; aio_fsync $fh if $cf::USE_FSYNC; close $fh; @@ -1064,8 +1116,6 @@ aio_unlink "$filename.pst"; } }; - - undef $guard; } sub object_freezer_as_string { @@ -1081,8 +1131,6 @@ my ($data, $av); - my $guard = cf::lock_acquire "io"; - (aio_load $filename, $data) >= 0 or return; @@ -1090,8 +1138,8 @@ (aio_load "$filename.pst", $av) >= 0 or return; - undef $guard; - $av = eval { (Storable::thaw $av)->{objs} }; + my $st = eval { Coro::Storable::thaw $av }; + $av = $st->{objs}; } utf8::decode (my $decname = $filename); @@ -1516,46 +1564,88 @@ =cut +use re 'eval'; + +my $group; +my $interior; $interior = qr{ + # match a pod interior sequence sans C<< >> + (?: + \ (.*?)\ (?{ $group = $^N }) + | < (??{$interior}) > + ) +}x; + sub expand_cfpod { - ((my $self), (local $_)) = @_; + my ($self, $pod) = @_; + + my $xml; + + while () { + if ($pod =~ /\G( (?: [^BCGHITU]+ | .(?!<) )+ )/xgcs) { + $group = $1; + + $group =~ s/&/&/g; + $group =~ s/]*) (?{ $group = $^N }) + | < $interior > + ) + > + %gcsx + ) { + my ($code, $data) = ($1, $group); + + if ($code eq "B") { + $xml .= "" . expand_cfpod ($self, $data) . ""; + } elsif ($code eq "I") { + $xml .= "" . expand_cfpod ($self, $data) . ""; + } elsif ($code eq "U") { + $xml .= "" . expand_cfpod ($self, $data) . ""; + } elsif ($code eq "C") { + $xml .= "" . expand_cfpod ($self, $data) . ""; + } elsif ($code eq "T") { + $xml .= "" . expand_cfpod ($self, $data) . ""; + } elsif ($code eq "G") { + my ($male, $female) = split /\|/, $data; + $data = $self->gender ? $female : $male; + $xml .= expand_cfpod ($self, $data); + } elsif ($code eq "H") { + $xml .= ("[" . expand_cfpod ($self, $data) . " (Use hintmode to suppress hints)]", + "[Hint suppressed, see hintmode]", + "") + [$self->{hintmode}]; + } else { + $xml .= "error processing '$code($data)' directive"; + } + } else { + if ($pod =~ /\G(.+)/) { + warn "parse error while expanding $pod (at $1)"; + } + last; + } + } + + for ($xml) { + # create single paragraphs (very hackish) + s/(?<=\S)\n(?=\w)/ /g; - # escape & and < - s/&/&/g; - s/(?, I<>, U<> etc. - s/B<([^\>]*)>/$1<\/b>/ - || s/I<([^\>]*)>/$1<\/i>/ - || s/U<([^\>]*)>/$1<\/u>/ - || s/T<([^\>]*)>/$1<\/b><\/big>/ - # replace G tags - || s{G<([^>|]*)\|([^>]*)>}{ - $self->gender ? $2 : $1 - }ge - # replace H - || s{H<([^\>]*)>} - { - ("[$1 (Use hintmode to suppress hints)]", - "[Hint suppressed, see hintmode]", - "") - [$self->{hintmode}] - }ge; - - # create single paragraphs (very hackish) - s/(?<=\S)\n(?=\w)/ /g; - - # compress some whitespace - s/\s+\n/\n/g; # ws line-ends - s/\n\n+/\n/g; # double lines - s/^\n+//; # beginning lines - s/\n+$//; # ending lines + # compress some whitespace + s/\s+\n/\n/g; # ws line-ends + s/\n\n+/\n/g; # double lines + s/^\n+//; # beginning lines + s/\n+$//; # ending lines + } - $_ + $xml } +no re 'eval'; + sub hintmode { $_[0]{hintmode} = $_[1] if @_ > 1; $_[0]{hintmode} @@ -2045,6 +2135,8 @@ $MAP_PREFETCH{$path} |= $load; $MAP_PREFETCHER ||= cf::async { + $Coro::current->{desc} = "map prefetcher"; + while (%MAP_PREFETCH) { while (my ($k, $v) = each %MAP_PREFETCH) { if (my $map = find $k) { @@ -2080,6 +2172,7 @@ local $self->{last_access} = $self->last_access;#d# cf::async { + $Coro::current->{desc} = "map player save"; $_->contr->save for $self->players; }; @@ -2276,10 +2369,7 @@ sub deref { my ($ref) = @_; - # temporary compatibility#TODO#remove - $ref =~ s{^<}{player/<}; - - if ($ref =~ m{^player\/(<1\.\d+>)/(.*)$}) { + if ($ref =~ m{^player\/(<1\.[0-9a-f]+>)/(.*)$}) { my ($uuid, $name) = ($1, $2); my $pl = $cf::PLAYER_LOADING{$name} || cf::player::find $name or return; @@ -2460,6 +2550,8 @@ $self->enter_link; (async { + $Coro::current->{desc} = "player::goto $path $x $y"; + # *tag paths override both path and x|y if ($path =~ /^\*(.*)$/) { if (my @obs = grep $_->map, ext::map_tags::find $1) { @@ -2549,7 +2641,7 @@ $rmp->{random_seed} ||= $exit->random_seed; - my $data = cf::to_json $rmp; + my $data = cf::encode_json $rmp; my $md5 = Digest::MD5::md5_hex $data; my $meta = "$RANDOMDIR/$md5.meta"; @@ -2586,6 +2678,8 @@ if $exit->flag (FLAG_DAMNED); (async { + $Coro::current->{desc} = "enter_exit $slaying $hp $sp"; + $self->deactivate_recursive; # just to be sure unless (eval { $self->goto ($slaying, $hp, $sp); @@ -2630,19 +2724,56 @@ =cut +# non-persistent channels (usually the info channel) our %CHANNEL = ( "c/identify" => { - id => "identify", + id => "infobox", title => "Identify", reply => undef, tooltip => "Items recently identified", }, "c/examine" => { - id => "examine", + id => "infobox", title => "Examine", reply => undef, tooltip => "Signs and other items you examined", }, + "c/book" => { + id => "infobox", + title => "Book", + reply => undef, + tooltip => "The contents of a note or book", + }, + "c/lookat" => { + id => "infobox", + title => "Look", + reply => undef, + tooltip => "What you saw there", + }, + "c/who" => { + id => "infobox", + title => "Players", + reply => undef, + tooltip => "Shows players who are currently online", + }, + "c/body" => { + id => "infobox", + title => "Body Parts", + reply => undef, + tooltip => "Shows which body parts you posess and are available", + }, + "c/uptime" => { + id => "infobox", + title => "Uptime", + reply => undef, + tooltip => "How long the server has been running since last restart", + }, + "c/mapinfo" => { + id => "infobox", + title => "Map Info", + reply => undef, + tooltip => "Information related to the maps", + }, ); sub cf::client::send_msg { @@ -2653,9 +2784,15 @@ $color &= cf::NDI_CLIENT_MASK; # just in case... # check predefined channels, for the benefit of C - $channel = $CHANNEL{$channel} if $CHANNEL{$channel}; + if ($CHANNEL{$channel}) { + $channel = $CHANNEL{$channel}; + + $self->ext_msg (channel_info => $channel) + if $self->can_msg; + + $channel = $channel->{id}; - if (ref $channel) { + } elsif (ref $channel) { # send meta info to client, if not yet sent unless (exists $self->{channel}{$channel->{id}}) { $self->{channel}{$channel->{id}} = $channel; @@ -2882,7 +3019,7 @@ The following functions and methods are available within a safe environment: cf::object - contr pay_amount pay_player map x y force_find force_add + contr pay_amount pay_player map x y force_find force_add destroy insert remove name archname title slaying race decrease_ob_nr cf::object::player @@ -2899,7 +3036,7 @@ for ( ["cf::object" => qw(contr pay_amount pay_player map force_find force_add x y insert remove inv name archname title slaying race - decrease_ob_nr)], + decrease_ob_nr destroy)], ["cf::object::player" => qw(player)], ["cf::player" => qw(peaceful)], ["cf::map" => qw(trigger)], @@ -3184,7 +3321,8 @@ local $Coro::idle = sub { Carp::cluck "FATAL: Coro::idle was called, major BUG, use cf::sync_job!\n";#d# (async { - Event::one_event; + $Coro::current->{desc} = "IDLE BUG HANDLER"; + EV::loop EV::LOOP_ONESHOT; })->prio (Coro::PRIO_MAX); }; @@ -3193,7 +3331,8 @@ load_extensions; $TICK_WATCHER->start; - Event::loop; + $Coro::current->prio (Coro::PRIO_MAX); # give the main loop max. priority + EV::loop; } ############################################################################# @@ -3201,16 +3340,11 @@ # install some emergency cleanup handlers BEGIN { + our %SIGWATCHER = (); for my $signal (qw(INT HUP TERM)) { - Event->signal ( - reentrant => 0, - data => WF_AUTOCANCEL, - signal => $signal, - prio => 0, - cb => sub { - cf::cleanup "SIG$signal"; - }, - ); + $SIGWATCHER{$signal} = EV::signal $signal, sub { + cf::cleanup "SIG$signal"; + }; } } @@ -3265,6 +3399,7 @@ for my $login (keys %cf::PLAYER) { my $pl = $cf::PLAYER{$login} or next; $pl->valid or next; + delete $pl->{unclean_save}; # not strictly necessary, but cannot hurt $pl->save; } warn "end emergency player save\n"; @@ -3313,16 +3448,11 @@ # if anything goes wrong in here, we should simply crash as we already saved - warn "cancelling all WF_AUTOCANCEL watchers"; - for (Event::all_watchers) { - $_->cancel if $_->data & WF_AUTOCANCEL; - } - warn "flushing outstanding aio requests"; for (;;) { BDB::flush; IO::AIO::flush; - Coro::cede; + Coro::cede_notself; last unless IO::AIO::nreqs || BDB::nreqs; warn "iterate..."; } @@ -3421,15 +3551,10 @@ # doing reload synchronously and two reloads happen back-to-back, # coro crashes during coro_state_free->destroy here. - $RELOAD_WATCHER ||= Event->timer ( - reentrant => 0, - after => 0, - data => WF_AUTOCANCEL, - cb => sub { - do_reload_perl; - undef $RELOAD_WATCHER; - }, - ); + $RELOAD_WATCHER ||= EV::timer 0, 0, sub { + undef $RELOAD_WATCHER; + do_reload_perl; + }; } register_command "reload" => sub { @@ -3437,7 +3562,10 @@ if ($who->flag (FLAG_WIZ)) { $who->message ("reloading server."); - async { reload_perl }; + async { + $Coro::current->{desc} = "perl_reload"; + reload_perl; + }; } }; @@ -3466,82 +3594,48 @@ $signal->wait; } -$TICK_WATCHER = Event->timer ( - reentrant => 0, - parked => 1, - prio => 0, - at => $NEXT_TICK || $TICK, - data => WF_AUTOCANCEL, - cb => sub { - if ($Coro::current != $Coro::main) { - Carp::cluck "major BUG: server tick called outside of main coro, skipping it" - unless ++$bug_warning > 10; - return; - } - - $NOW = $tick_start = Event::time; - - cf::server_tick; # one server iteration - - $RUNTIME += $TICK; - $NEXT_TICK += $TICK; - - if ($NOW >= $NEXT_RUNTIME_WRITE) { - $NEXT_RUNTIME_WRITE = $NOW + 10; - Coro::async_pool { - write_runtime - or warn "ERROR: unable to write runtime file: $!"; - }; - } +$TICK_WATCHER = EV::periodic_ns 0, $TICK, 0, sub { + if ($Coro::current != $Coro::main) { + Carp::cluck "major BUG: server tick called outside of main coro, skipping it" + unless ++$bug_warning > 10; + return; + } - if (my $sig = shift @WAIT_FOR_TICK_BEGIN) { - $sig->send; - } - while (my $sig = shift @WAIT_FOR_TICK) { - $sig->send; - } + $NOW = $tick_start = EV::now; - $NOW = Event::time; + cf::server_tick; # one server iteration - # if we are delayed by four ticks or more, skip them all - $NEXT_TICK = $NOW if $NOW >= $NEXT_TICK + $TICK * 4; + $RUNTIME += $TICK; + $NEXT_TICK += $TICK; - $TICK_WATCHER->at ($NEXT_TICK); - $TICK_WATCHER->start; + if ($NOW >= $NEXT_RUNTIME_WRITE) { + $NEXT_RUNTIME_WRITE = List::Util::max $NEXT_RUNTIME_WRITE + 10, $NOW + 5.; + Coro::async_pool { + $Coro::current->{desc} = "runtime saver"; + write_runtime + or warn "ERROR: unable to write runtime file: $!"; + }; + } - $LOAD = ($NOW - $tick_start) / $TICK; - $LOADAVG = $LOADAVG * 0.75 + $LOAD * 0.25; + if (my $sig = shift @WAIT_FOR_TICK_BEGIN) { + $sig->send; + } + while (my $sig = shift @WAIT_FOR_TICK) { + $sig->send; + } - _post_tick; - }, -); + $LOAD = ($NOW - $tick_start) / $TICK; + $LOADAVG = $LOADAVG * 0.75 + $LOAD * 0.25; + + _post_tick; +}; +$TICK_WATCHER->priority (EV::MAXPRI); { - BDB::min_parallel 8; - BDB::max_poll_time $TICK * 0.1; - $BDB_POLL_WATCHER = Event->io ( - reentrant => 0, - fd => BDB::poll_fileno, - poll => 'r', - prio => 0, - data => WF_AUTOCANCEL, - cb => \&BDB::poll_cb, - ); + # configure BDB - BDB::set_sync_prepare { - my $status; - my $current = $Coro::current; - ( - sub { - $status = $!; - $current->ready; undef $current; - }, - sub { - Coro::schedule while defined $current; - $! = $status; - }, - ) - }; + BDB::min_parallel 8; + BDB::max_poll_reqs $TICK * 0.1; unless ($DB_ENV) { $DB_ENV = BDB::db_env_create; @@ -3566,51 +3660,22 @@ }; } - $BDB_DEADLOCK_WATCHER = Event->timer ( - after => 3, - interval => 1, - hard => 1, - prio => 0, - data => WF_AUTOCANCEL, - cb => sub { - BDB::db_env_lock_detect $DB_ENV, 0, BDB::LOCK_DEFAULT, 0, sub { }; - }, - ); - $BDB_CHECKPOINT_WATCHER = Event->timer ( - after => 11, - interval => 60, - hard => 1, - prio => 0, - data => WF_AUTOCANCEL, - cb => sub { - BDB::db_env_txn_checkpoint $DB_ENV, 0, 0, 0, sub { }; - }, - ); - $BDB_TRICKLE_WATCHER = Event->timer ( - after => 5, - interval => 10, - hard => 1, - prio => 0, - data => WF_AUTOCANCEL, - cb => sub { - BDB::db_env_memp_trickle $DB_ENV, 20, 0, sub { }; - }, - ); + $BDB_DEADLOCK_WATCHER = EV::periodic 0, 3, 0, sub { + BDB::db_env_lock_detect $DB_ENV, 0, BDB::LOCK_DEFAULT, 0, sub { }; + }; + $BDB_CHECKPOINT_WATCHER = EV::periodic 0, 60, 0, sub { + BDB::db_env_txn_checkpoint $DB_ENV, 0, 0, 0, sub { }; + }; + $BDB_TRICKLE_WATCHER = EV::periodic 0, 10, 0, sub { + BDB::db_env_memp_trickle $DB_ENV, 20, 0, sub { }; + }; } { - IO::AIO::min_parallel 8; + # configure IO::AIO - undef $Coro::AIO::WATCHER; + IO::AIO::min_parallel 8; IO::AIO::max_poll_time $TICK * 0.1; - $AIO_POLL_WATCHER = Event->io ( - reentrant => 0, - data => WF_AUTOCANCEL, - fd => IO::AIO::poll_fileno, - poll => 'r', - prio => 6, - cb => \&IO::AIO::poll_cb, - ); } my $_log_backtrace; @@ -3624,6 +3689,8 @@ if ($_log_backtrace < 2) { ++$_log_backtrace; async { + $Coro::current->{desc} = "abt $msg"; + my @bt = fork_call { @addr = map { sprintf "%x", $_ } @addr; my $self = (-f "/proc/$$/exe") ? "/proc/$$/exe" : $^X;