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