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