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