ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/deliantra/Deliantra-Client/DC/DB.pm
Revision: 1.41
Committed: Wed Dec 24 04:09:27 2008 UTC (15 years, 5 months ago) by root
Branch: MAIN
CVS Tags: rel-2_02
Changes since 1.40: +3 -0 lines
Log Message:
*** empty log message ***

File Contents

# User Rev Content
1 root 1.1 =head1 NAME
2    
3 root 1.31 DC::DB - async. database and filesystem access for deliantra
4 root 1.1
5     =head1 SYNOPSIS
6    
7 root 1.29 use DC::DB;
8 root 1.1
9     =head1 DESCRIPTION
10    
11     =over 4
12    
13     =cut
14    
15 root 1.29 package DC::DB;
16 root 1.1
17     use strict;
18     use utf8;
19    
20 root 1.35 use File::Path ();
21 root 1.1 use Carp ();
22 root 1.5 use Storable ();
23 root 1.9 use Config;
24 root 1.24 use BDB;
25 root 1.1
26 root 1.29 use DC;
27 root 1.3
28 root 1.38 our $ODBDIR = "cfplus-" . BDB::VERSION_MAJOR . "." . BDB::VERSION_MINOR . "-$Config{archname}";
29     our $DBDIR = "client-" . BDB::VERSION_MAJOR . "." . BDB::VERSION_MINOR . "-$Config{archname}";
30 root 1.27 our $DB_HOME = "$Deliantra::VARDIR/$DBDIR";
31    
32 root 1.40 sub FIRST_TILE_ID () { 64 }
33    
34 root 1.35 unless (-d $DB_HOME) {
35     if (-d "$Deliantra::VARDIR/$ODBDIR") {
36     rename "$Deliantra::VARDIR/$ODBDIR", $DB_HOME;
37     print STDERR "INFO: moved old database from $Deliantra::VARDIR/$ODBDIR to $DB_HOME\n";
38     } elsif (-d "$Deliantra::OLDDIR/$ODBDIR") {
39     rename "$Deliantra::OLDDIR/$DBDIR", $DB_HOME;
40     print STDERR "INFO: moved old database from $Deliantra::OLDDIR/$ODBDIR to $DB_HOME\n";
41     } else {
42     File::Path::mkpath [$DB_HOME]
43     or die "unable to create database directory $DB_HOME: $!";
44     }
45 root 1.27 }
46 root 1.9
47 root 1.30 BDB::max_poll_time 0.03;
48 root 1.32 BDB::max_parallel 1;
49 root 1.30
50 root 1.24 our $DB_ENV;
51     our $DB_STATE;
52     our %DB_TABLE;
53 root 1.35 our $TILE_SEQ;
54 root 1.24
55 root 1.33 sub try_open_db {
56 root 1.35 File::Path::mkpath [$DB_HOME];
57 root 1.24
58 root 1.36 my $env = db_env_create;
59 root 1.24
60 root 1.36 $env->set_errfile (\*STDERR);
61     $env->set_msgfile (\*STDERR);
62     $env->set_verbose (-1, 1);
63 root 1.24
64 root 1.39 $env->set_flags (BDB::AUTO_COMMIT | BDB::REGION_INIT);
65     $env->set_flags (&BDB::LOG_AUTOREMOVE ) if BDB::VERSION v0, v4.7;
66     $env->log_set_config (&BDB::LOG_AUTO_REMOVE) if BDB::VERSION v4.7;
67    
68     $env->set_timeout (3, BDB::SET_TXN_TIMEOUT);
69     $env->set_timeout (3, BDB::SET_LOCK_TIMEOUT);
70    
71 root 1.36 $env->set_cachesize (0, 2048 * 1024, 0);
72 root 1.24
73 root 1.36 db_env_open $env, $DB_HOME,
74 root 1.24 BDB::CREATE | BDB::REGISTER | BDB::RECOVER | BDB::INIT_MPOOL | BDB::INIT_LOCK | BDB::INIT_TXN,
75     0666;
76    
77     $! and die "cannot open database environment $DB_HOME: " . BDB::strerror;
78    
79 root 1.36 $DB_ENV = $env;
80    
81 root 1.24 1
82     }
83    
84     sub table($) {
85     $DB_TABLE{$_[0]} ||= do {
86     my ($table) = @_;
87    
88     $table =~ s/([^a-zA-Z0-9_\-])/sprintf "=%x=", ord $1/ge;
89    
90 root 1.41 $DB_ENV#d#
91     or return ::clienterror ("trying to create table $_[0] with empty db_env $DB_ENV" => 1);#d#
92    
93 root 1.24 my $db = db_create $DB_ENV;
94     $db->set_flags (BDB::CHKSUM);
95    
96     db_open $db, undef, $table, undef, BDB::BTREE,
97     BDB::AUTO_COMMIT | BDB::CREATE | BDB::READ_UNCOMMITTED, 0666;
98    
99     $! and "unable to open/create database table $_[0]: ". BDB::strerror;
100    
101     $db
102     }
103 root 1.9 }
104    
105 root 1.24 #############################################################################
106    
107 root 1.33 our $WATCHER;
108     our $SYNC;
109 root 1.34 our $facemap;
110 root 1.24
111 root 1.9 sub exists($$$) {
112 root 1.24 my ($db, $key, $cb) = @_;
113    
114     my $data;
115     db_get table $db, undef, $key, $data, 0, sub {
116     $cb->($! ? () : length $data);
117     };
118 root 1.9 }
119    
120 root 1.1 sub get($$$) {
121 root 1.24 my ($db, $key, $cb) = @_;
122    
123     my $data;
124     db_get table $db, undef, $key, $data, 0, sub {
125     $cb->($! ? () : $data);
126     };
127 root 1.1 }
128    
129     sub put($$$$) {
130 root 1.24 my ($db, $key, $data, $cb) = @_;
131    
132     db_put table $db, undef, $key, $data, 0, sub {
133     $cb->($!);
134     $SYNC->again unless $SYNC->is_active;
135     };
136 root 1.1 }
137    
138 root 1.24 sub do_table {
139     my ($db, $cb) = @_;
140    
141     $db = table $db;
142    
143     my $cursor = $db->cursor;
144     my %kv;
145    
146     for (;;) {
147     db_c_get $cursor, my $k, my $v, BDB::NEXT;
148     last if $!;
149     $kv{$k} = $v;
150     }
151    
152     $cb->(\%kv);
153 root 1.9 }
154    
155 root 1.24 sub do_get_tile_id {
156     my ($name, $cb) = @_;
157    
158     my $table = table "facemap";
159     my $id;
160    
161 root 1.35 db_get $table, undef, $name => $id, 0;
162 root 1.34 $! or return $cb->($id);
163 root 1.24
164 root 1.35 unless ($TILE_SEQ) {
165     $TILE_SEQ = $table->sequence;
166 root 1.40 $TILE_SEQ->initial_value (FIRST_TILE_ID);
167 root 1.35 $TILE_SEQ->set_cachesize (0);
168     db_sequence_open $TILE_SEQ, undef, "id", BDB::CREATE;
169     }
170 root 1.24
171 root 1.35 db_sequence_get $TILE_SEQ, undef, 1, my $id;
172 root 1.24
173 root 1.35 die "unable to allocate tile id: $!"
174     if $!;
175    
176     db_put $table, undef, $name => $id, 0;
177     $cb->($id);
178 root 1.24
179 root 1.15 }
180    
181 root 1.1 sub get_tile_id_sync($) {
182 root 1.20 my ($name) = @_;
183 root 1.1
184 root 1.34 $facemap->{$name} ||= do {
185 root 1.1 my $id;
186 root 1.24 do_get_tile_id $name, sub {
187     $id = $_[0];
188     };
189     BDB::flush;
190 root 1.1 $id
191     }
192     }
193    
194 root 1.24 #############################################################################
195    
196     sub path_of_res($) {
197     utf8::downgrade $_[0]; # bug in unpack "H*"
198     "$DB_HOME/res-data-" . unpack "H*", $_[0]
199     }
200    
201     sub sync {
202     # for debugging
203 root 1.29 #DC::DB::Server::req (sync => sub { });
204     DC::DB::Server::sync ();
205 root 1.24 }
206 root 1.1
207 root 1.24 sub unlink($$) {
208 root 1.29 DC::DB::Server::req (unlink => @_);
209 root 1.24 }
210 root 1.1
211 root 1.24 sub read_file($$) {
212 root 1.29 DC::DB::Server::req (read_file => @_);
213 root 1.24 }
214 root 1.1
215 root 1.24 sub write_file($$$) {
216 root 1.29 DC::DB::Server::req (write_file => @_);
217 root 1.24 }
218 root 1.1
219 root 1.24 sub prefetch_file($$$) {
220 root 1.29 DC::DB::Server::req (prefetch_file => @_);
221 root 1.24 }
222 root 1.1
223 root 1.24 sub logprint($$$) {
224 root 1.29 DC::DB::Server::req (logprint => @_);
225 root 1.1 }
226    
227 root 1.30 #############################################################################
228    
229 root 1.29 package DC::DB::Server;
230 root 1.1
231 root 1.24 use strict;
232 root 1.1
233 root 1.24 use EV ();
234     use Fcntl;
235 root 1.1
236     our %CB;
237     our $FH;
238     our $ID = "aaa0";
239     our ($fh_r_watcher, $fh_w_watcher);
240     our $sync_timer;
241     our $write_buf;
242     our $read_buf;
243    
244     sub fh_write {
245     my $len = syswrite $FH, $write_buf;
246    
247     substr $write_buf, 0, $len, "";
248    
249 root 1.14 $fh_w_watcher->stop
250 root 1.1 unless length $write_buf;
251     }
252    
253     sub fh_read {
254     my $status = sysread $FH, $read_buf, 16384, length $read_buf;
255    
256     die "FATAL: database process died\n"
257     if $status == 0 && defined $status;
258    
259     while () {
260     return if 4 > length $read_buf;
261     my $len = unpack "N", $read_buf;
262    
263     return if $len + 4 > length $read_buf;
264    
265     substr $read_buf, 0, 4, "";
266     my $res = Storable::thaw substr $read_buf, 0, $len, "";
267    
268     my ($id, @args) = @$res;
269     (delete $CB{$id})->(@args);
270     }
271     }
272    
273     sub sync {
274     # biggest mess evarr
275     my $fds; (vec $fds, fileno $FH, 1) = 1;
276    
277     while (1 < scalar keys %CB) {
278     my $r = $fds;
279     my $w = length $write_buf ? $fds : undef;
280     select $r, $w, undef, undef;
281    
282     fh_write if vec $w, fileno $FH, 1;
283     fh_read if vec $r, fileno $FH, 1;
284     }
285     }
286    
287     sub req {
288     my ($type, @args) = @_;
289     my $cb = pop @args;
290    
291     my $id = ++$ID;
292     $write_buf .= pack "N/a*", Storable::freeze [$id, $type, @args];
293     $CB{$id} = $cb;
294    
295 root 1.14 $fh_w_watcher->start;
296 root 1.1 }
297    
298 root 1.9 sub do_unlink {
299     unlink $_[0];
300     }
301    
302 root 1.16 sub do_read_file {
303     my ($path) = @_;
304    
305     utf8::downgrade $path;
306     open my $fh, "<:raw", $path
307     or return;
308     sysread $fh, my $buf, -s $fh;
309    
310     $buf
311     }
312    
313 root 1.9 sub do_write_file {
314 root 1.16 my ($path, $data) = @_;
315 root 1.9
316 root 1.16 utf8::downgrade $path;
317 root 1.9 utf8::downgrade $data;
318 root 1.16 open my $fh, ">:raw", $path
319 root 1.9 or return;
320 root 1.16 syswrite $fh, $data;
321 root 1.9 close $fh;
322    
323     1
324     }
325    
326 root 1.11 sub do_prefetch_file {
327 root 1.16 my ($path, $size) = @_;
328 root 1.11
329 root 1.16 utf8::downgrade $path;
330     open my $fh, "<:raw", $path
331 root 1.11 or return;
332     sysread $fh, my $buf, $size;
333    
334     1
335     }
336    
337 root 1.15 our %LOG_FH;
338    
339     sub do_logprint {
340     my ($path, $line) = @_;
341    
342     $LOG_FH{$path} ||= do {
343     open my $fh, ">>:utf8", $path
344     or warn "Couldn't open logfile $path: $!";
345    
346     $fh->autoflush (1);
347    
348     $fh
349     };
350    
351     my ($sec, $min, $hour, $mday, $mon, $year) = localtime time;
352    
353     my $ts = sprintf "%04d-%02d-%02d %02d:%02d:%02d",
354     $year + 1900, $mon + 1, $mday, $hour, $min, $sec;
355    
356     print { $LOG_FH{$path} } "$ts $line\n"
357     }
358    
359 root 1.1 sub run {
360 root 1.29 ($FH, my $fh) = DC::socketpipe;
361 root 1.1
362     my $oldfh = select $FH; $| = 1; select $oldfh;
363     my $oldfh = select $fh; $| = 1; select $oldfh;
364    
365     my $pid = fork;
366    
367     if (defined $pid && !$pid) {
368 root 1.30 local $SIG{QUIT} = "IGNORE";
369 root 1.1 local $SIG{__DIE__};
370 root 1.15 local $SIG{__WARN__};
371 root 1.1 eval {
372     close $FH;
373    
374     while () {
375     4 == read $fh, my $len, 4
376     or last;
377     $len = unpack "N", $len;
378     $len == read $fh, my $req, $len
379     or die "unexpected eof while reading request";
380    
381     $req = Storable::thaw $req;
382    
383     my ($id, $type, @args) = @$req;
384 root 1.29 my $cb = DC::DB::Server->can ("do_$type")
385 root 1.1 or die "$type: unknown database request type\n";
386     my $res = pack "N/a*", Storable::freeze [$id, $cb->(@args)];
387     (syswrite $fh, $res) == length $res
388 root 1.21 or die "DB::write: $!";
389 root 1.1 }
390     };
391    
392     my $error = $@;
393    
394     eval {
395     Storable::store_fd [die => $error], $fh;
396     };
397    
398 root 1.25 warn $error
399     if $error;
400 root 1.24
401 root 1.29 DC::_exit 0;
402 root 1.1 }
403    
404     close $fh;
405 root 1.29 DC::fh_nonblocking $FH, 1;
406 root 1.1
407 root 1.2 $CB{die} = sub { die shift };
408 root 1.1
409 root 1.21 $fh_r_watcher = EV::io $FH, EV::READ , \&fh_read;
410     $fh_w_watcher = EV::io $FH, EV::WRITE, \&fh_write;
411 root 1.1 }
412    
413 root 1.6 sub stop {
414     close $FH;
415     }
416    
417 root 1.33 package DC::DB;
418    
419 root 1.35 sub nuke_db {
420     File::Path::mkpath [$DB_HOME];
421     eval { File::Path::rmtree $DB_HOME };
422     }
423    
424 root 1.33 sub open_db {
425     unless (eval { try_open_db }) {
426     warn "$@";#d#
427 root 1.35 eval { nuke_db };
428 root 1.33 try_open_db;
429     }
430    
431     # fetch the full face table first
432 root 1.34 unless ($facemap) {
433 root 1.33 do_table facemap => sub {
434 root 1.34 $facemap = $_[0];
435     delete $facemap->{id};
436     my %maptile = reverse %$facemap;#d#
437     if ((scalar keys %$facemap) != (scalar keys %maptile)) {#d#
438     $facemap = { };#d#
439 root 1.33 DC::error "FATAL: facemap is not a 1:1 mapping, please report this and delete your $DB_HOME directory!\n";#d#
440     }#d#
441     };
442     }
443    
444     $WATCHER = EV::io BDB::poll_fileno, EV::READ, \&BDB::poll_cb;
445     $SYNC = EV::timer_ns 0, 60, sub {
446     $_[0]->stop;
447     db_env_txn_checkpoint $DB_ENV, 0, 0, 0, sub { };
448     };
449     }
450    
451     END {
452 root 1.37 db_env_txn_checkpoint $DB_ENV, 0, 0, 0
453     if $DB_ENV;
454    
455 root 1.35 undef $TILE_SEQ;
456 root 1.33 %DB_TABLE = ();
457     undef $DB_ENV;
458     }
459    
460 root 1.1 1;
461    
462     =back
463    
464     =head1 AUTHOR
465    
466     Marc Lehmann <schmorp@schmorp.de>
467     http://home.schmorp.de/
468    
469     =cut
470