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