ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/deliantra/Deliantra-Client/DC/DB.pm
Revision: 1.4
Committed: Fri Apr 6 23:00:52 2007 UTC (17 years, 1 month ago) by root
Branch: MAIN
Changes since 1.3: +1 -1 lines
Log Message:
*** empty log message ***

File Contents

# Content
1 =head1 NAME
2
3 CFPlus::DB - async. database access for cfplus
4
5 =head1 SYNOPSIS
6
7 use CFPlus::DB;
8
9 =head1 DESCRIPTION
10
11 =over 4
12
13 =cut
14
15 package CFPlus::DB;
16
17 use strict;
18 use utf8;
19
20 use Carp ();
21 use AnyEvent ();
22 use Storable (); # finally
23
24 use CFPlus;
25
26 sub sync {
27 # for debugging
28 #CFPlus::DB::Server::req (sync => sub { });
29 CFPlus::DB::Server::sync ();
30 }
31
32 sub get($$$) {
33 CFPlus::DB::Server::req (get => @_);
34 }
35
36 sub put($$$$) {
37 CFPlus::DB::Server::req (put => @_);
38 }
39
40 our $tilemap;
41
42 sub get_tile_id_sync($) {
43 my ($hash) = @_;
44
45 # fetch the full face table first
46 unless ($tilemap) {
47 CFPlus::DB::Server::req (table => facemap => sub { $tilemap = $_[0] });
48 sync;
49 }
50
51 $tilemap->{$hash} ||= do {
52 my $id;
53 CFPlus::DB::Server::req (get_tile_id => $hash, sub { $id = $_[0] });
54 sync;
55 $id
56 }
57 }
58
59 package CFPlus::DB::Server;
60
61 use strict;
62
63 use Fcntl;
64 use BerkeleyDB;
65
66 our $DB_HOME = "$Crossfire::VARDIR/cfplus";
67 our $DB_ENV;
68 our $DB_STATE;
69 our %DB_TABLE;
70
71 sub open_db {
72 mkdir $DB_HOME, 0777;
73 my $recover = $BerkeleyDB::db_version >= 4.4
74 ? eval "DB_REGISTER | DB_RECOVER"
75 : 0;
76
77 $DB_ENV = new BerkeleyDB::Env
78 -Home => $DB_HOME,
79 -Cachesize => 8_000_000,
80 -ErrFile => "$DB_HOME/errorlog.txt",
81 # -ErrPrefix => "DATABASE",
82 -Verbose => 1,
83 -Flags => DB_CREATE | DB_RECOVER | DB_INIT_MPOOL | DB_INIT_LOCK | DB_INIT_TXN | $recover,
84 -SetFlags => DB_AUTO_COMMIT | DB_LOG_AUTOREMOVE,
85 or die "unable to create/open database home $DB_HOME: $BerkeleyDB::Error";
86
87 1
88 }
89
90 sub table($) {
91 $DB_TABLE{$_[0]} ||= do {
92 my ($table) = @_;
93
94 $table =~ s/([^a-zA-Z0-9_\-])/sprintf "=%x=", ord $1/ge;
95
96 new BerkeleyDB::Btree
97 -Env => $DB_ENV,
98 -Filename => $table,
99 # -Filename => "database",
100 # -Subname => $table,
101 -Property => DB_CHKSUM,
102 -Flags => DB_CREATE | DB_UPGRADE,
103 or die "unable to create/open database table $_[0]: $BerkeleyDB::Error"
104 }
105 }
106
107 our $SYNC_INTERVAL = 60;
108
109 our %CB;
110 our $FH;
111 our $ID = "aaa0";
112 our ($fh_r_watcher, $fh_w_watcher);
113 our $sync_timer;
114 our $write_buf;
115 our $read_buf;
116
117 sub fh_write {
118 my $len = syswrite $FH, $write_buf;
119
120 substr $write_buf, 0, $len, "";
121
122 undef $fh_w_watcher
123 unless length $write_buf;
124 }
125
126 sub fh_read {
127 my $status = sysread $FH, $read_buf, 16384, length $read_buf;
128
129 die "FATAL: database process died\n"
130 if $status == 0 && defined $status;
131
132 while () {
133 return if 4 > length $read_buf;
134 my $len = unpack "N", $read_buf;
135
136 return if $len + 4 > length $read_buf;
137
138 substr $read_buf, 0, 4, "";
139 my $res = Storable::thaw substr $read_buf, 0, $len, "";
140
141 my ($id, @args) = @$res;
142 (delete $CB{$id})->(@args);
143 }
144 }
145
146 sub sync {
147 # biggest mess evarr
148 my $fds; (vec $fds, fileno $FH, 1) = 1;
149
150 while (1 < scalar keys %CB) {
151 my $r = $fds;
152 my $w = length $write_buf ? $fds : undef;
153 select $r, $w, undef, undef;
154
155 fh_write if vec $w, fileno $FH, 1;
156 fh_read if vec $r, fileno $FH, 1;
157 }
158 }
159
160 sub req {
161 my ($type, @args) = @_;
162 my $cb = pop @args;
163
164 my $id = ++$ID;
165 $write_buf .= pack "N/a*", Storable::freeze [$id, $type, @args];
166 $CB{$id} = $cb;
167
168 $fh_w_watcher = AnyEvent->io (fh => $FH, poll => 'w', cb => \&fh_write);
169 }
170
171 sub sync_tick {
172 req "sync", sub { };
173 $sync_timer = AnyEvent->timer (after => $SYNC_INTERVAL, cb => \&sync_tick);
174 }
175
176 sub do_sync {
177 $DB_ENV->txn_checkpoint (0, 0, 0);
178 ()
179 }
180
181 sub do_get {
182 my ($db, $key) = @_;
183
184 my $data;
185 (table $db)->db_get ($key, $data) == 0
186 ? $data
187 : ()
188 }
189
190 sub do_put {
191 my ($db, $key, $data) = @_;
192
193 (table $db)->db_put ($key => $data)
194 }
195
196 sub do_table {
197 my ($db) = @_;
198
199 $db = table $db;
200
201 my $cursor = $db->db_cursor;
202 my %kv;
203 my ($k, $v);
204 $kv{$k} = $v while $cursor->c_get ($k, $v, BerkeleyDB::DB_NEXT) == 0;
205
206 \%kv
207 }
208
209 sub do_get_tile_id {
210 my ($hash) = @_;
211
212 my $id;
213 my $table = table "facemap";
214
215 return $id
216 if $table->db_get ($hash, $id) == 0;
217
218 for (1..100) {
219 my $txn = $DB_ENV->txn_begin;
220 my $status = $table->db_get (id => $id);
221 if ($status == 0 || $status == BerkeleyDB::DB_NOTFOUND) {
222 $id = ($id || 64) + 1;
223 if ($table->db_put (id => $id) == 0
224 && $table->db_put ($hash => $id) == 0) {
225 $txn->txn_commit;
226
227 return $id;
228 }
229 }
230 $txn->txn_abort;
231 }
232
233 die "maximum number of transaction retries reached - database problems?";
234 }
235
236 sub run {
237 ($FH, my $fh) = CFPlus::socketpipe;
238
239 my $oldfh = select $FH; $| = 1; select $oldfh;
240 my $oldfh = select $fh; $| = 1; select $oldfh;
241
242 my $pid = fork;
243
244 if (defined $pid && !$pid) {
245 local $SIG{__DIE__};
246 eval {
247 close $FH;
248
249 unless (eval { open_db }) {
250 File::Path::rmtree $DB_HOME;
251 open_db;
252 }
253
254 while () {
255 4 == read $fh, my $len, 4
256 or last;
257 $len = unpack "N", $len;
258 $len == read $fh, my $req, $len
259 or die "unexpected eof while reading request";
260
261 $req = Storable::thaw $req;
262
263 my ($id, $type, @args) = @$req;
264 my $cb = CFPlus::DB::Server->can ("do_$type")
265 or die "$type: unknown database request type\n";
266 my $res = pack "N/a*", Storable::freeze [$id, $cb->(@args)];
267 (syswrite $fh, $res) == length $res
268 or die;
269 }
270 };
271
272 my $error = $@;
273
274 eval {
275 undef %DB_TABLE;
276 undef $DB_ENV;
277
278 Storable::store_fd [die => $error], $fh;
279 };
280
281 CFPlus::_exit 0;
282 }
283
284 close $fh;
285 CFPlus::fh_nonblocking $FH, 1;
286
287 $CB{die} = sub { die shift };
288
289 $fh_r_watcher = AnyEvent->io (fh => $FH, poll => 'r', cb => \&fh_read);
290
291 sync_tick;
292 }
293
294 1;
295
296 =back
297
298 =head1 AUTHOR
299
300 Marc Lehmann <schmorp@schmorp.de>
301 http://home.schmorp.de/
302
303 =cut
304