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