ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/syncmail/syncmail
(Generate patch)

Comparing syncmail/syncmail (file contents):
Revision 1.5 by root, Sun Oct 28 04:00:59 2001 UTC vs.
Revision 1.7 by root, Mon Oct 29 00:37:41 2001 UTC

… …
10 10
11use Set::Scalar; 11use Set::Scalar;
12 12
13use Fcntl; 13use Fcntl;
14 14
15use constant PROTVERSION => 1;
16
17$VERSION = 0.1; 15our $VERSION = 0.1;
16
17our $NOW = time;
18 18
19require "config.pl"; 19require "config.pl";
20 20
21use folder; 21use folder;
22use vc; 22use vc;
… …
63 opendir my $dh, $PREFIX 63 opendir my $dh, $PREFIX
64 or die "$PREFIX: $!"; 64 or die "$PREFIX: $!";
65 65
66 while (defined (my $folder = readdir $dh)) { 66 while (defined (my $folder = readdir $dh)) {
67 next if $folder =~ /^\./; 67 next if $folder =~ /^\./;
68 next if $folder =~ /~$/;
68 next unless -f "$PREFIX/$folder"; 69 next unless -f "$PREFIX/$folder";
69 push @folders, $folder; 70 push @folders, $folder;
70 } 71 }
71 72
72 @folders; 73 @folders;
73} 74}
74 75
75my $sync_folder = new Coro::Semaphore 3; # max 3 folders in parallel 76my $sync_folder = new Coro::Semaphore 3; # max 3 folders in parallel
77
78sub sync_offer {
79 my ($folder, $avc, $diff) = @_;
80
81 my $vc = create vc pri => 1;
82 $avc->snd("offer", $vc->{port});
83
84 async {
85 $vc->rcv; # need to synchronize, argl. should open on other side
86 $vc->snd(join "\0", @$diff);
87 my %dup; @dup{split /\0/, $vc->rcv} = ();
88
89 for (@$diff) {
90 $vc->snd($folder->fetch($_)) unless exists $dup{$_};
91 }
92
93 defined $vc->rcv and die "protocol error, expected close";
94 };
95}
76 96
77sub sync_folder { 97sub sync_folder {
78 my $name = $_[0]; 98 my $name = $_[0];
79 99
80 my $quit_guard = quit_guard; 100 my $quit_guard = quit_guard;
81 async { 101 async {
82 my $guard = $sync_folder->guard; 102 my $guard = $sync_folder->guard;
83 my $vc = new vc; 103 my $vc = create vc;
84 104
85 my $folder = new folder name => $name; 105 my $folder = new folder name => $name;
86 106
87 $vc->snd("open", $name); 107 $vc->snd("open", $name);
88 $vc->snd("mtime"); 108 $vc->snd("mtime");
89 109
90 $folder->check;
91 $folder->read_mdif; 110 $folder->read_mdif;
92 111
93 my $ctime = $folder->{host}{$OTHERNAME} || -1; 112 my $ctime = $folder->{host}{$OTHERNAME} || -1;
94 my $rtime = $vc->rcv; 113 my $rtime = $vc->rcv;
95 114
96 $vc->snd("diff", $ctime); 115 $vc->snd("diff", $ctime);
97 116
98 my (%ladd, %ldel, %radd, %rdel); 117 my %diff; #
118 # 00 - local del
119 # 01 - local add
120 # 10 - remote del
121 # 11 - remote add
99 122
100 my @diff = grep { $_->[0] > $ctime } @{$folder->{diff}}; 123 my @diff = grep { $_->[0] > $ctime } @{$folder->{diff}};
124
101 $diff = $vc->rcv; 125 $diff = $vc->rcv;
102
103 while () { 126 while () {
104 if ($diff >= 0 and (!@diff or $diff <= $diff[0][0])) { 127 if ($diff >= 0 and (!@diff or $diff < $diff[0][0])) {
105 my @add = split /\0/, $vc->rcv; 128 $diff{$_} = 0b01 for split /\0/, $vc->rcv; # add
106 my @del = split /\0/, $vc->rcv; 129 $diff{$_} = 0b00 for split /\0/, $vc->rcv; # del
107 slog 0, "applying remote diff $diff\n"; 130
108 for (@del) { undef $rdel{$_}; delete $radd{$_}; delete $ladd{$_}; delete $ldel{$_}; }
109 for (@add) { undef $radd{$_}; delete $rdel{$_}; delete $ladd{$_}; delete $ldel{$_}; }
110 $diff = $vc->rcv; 131 $diff = $vc->rcv;
111 } elsif (@diff) { 132 } elsif (@diff) {
112 slog 0, "applying local diff $diff[0][0]\n"; 133 $diff{$_} = 0b11 for @{$diff[0][1]};
113 for (@{$diff[0][2]}) { undef $rdel{$_}; delete $radd{$_}; delete $ladd{$_}; delete $ldel{$_}; } 134 $diff{$_} = 0b10 for @{$diff[0][2]};
114 for (@{$diff[0][1]}) { undef $radd{$_}; delete $rdel{$_}; delete $ladd{$_}; delete $ldel{$_}; } 135
115 shift @diff; 136 shift @diff;
116 } else { 137 } else {
117 slog 0, "no more diffing\n";
118 last; 138 last;
119 } 139 }
120 } 140 }
121 141
122 slog 0, "LADD ".join(" ", keys %ladd)."\n"; 142 # append or update, depending on wether there are messages to be deleted
123 slog 0, "LDEL ".join(" ", keys %ldel)."\n"; 143 $vc->snd("begin");
124 slog 0, "RADD ".join(" ", keys %radd)."\n"; 144 $folder->begin_update;
125 slog 0, "RDEL ".join(" ", keys %rdel)."\n"; 145
146 while (my ($k,$v) = each %diff) {
147 push @{$diff[$v]}, $k;
148 slog 0, "DIFF $k : $v\n";#d#
149 }
150
151 $vc->snd("delete", join "\0", @{$diff[2]});
152 $folder->delete(@{$diff[0]});
153
154 # offer ours
155 my $offer_coro = sync_offer($folder, $vc, $diff[3]);
156
157 # request theirs
158 {
159 my @send = grep { !$folder->exists($_) } @{$diff[1]};
160 $vc->snd("send", join "\0", @send);
161 $folder->append($_, $vc->rcv) for @send;
162 $vc->snd("-"); # sync
163 }
164
165 slog 0, "waiting...\n";#d#
166 $offer_coro->join;
167
168 # sanity check
169 $vc->snd("inventory");
170 if ($folder->inventory ne $vc->rcv) {
171 $folder->write_mdif;
172 slog 0, "FATAL: folder inventory mismatch after update\n";
173 }
174
175 $vc->snd("end");
176 $folder->end_update;
126 177
178 $vc->snd("setctime", $folder->{ctime});
179
180 $vc->snd("mtime");
127 #$folder->{host}{$OTHERNAME} = time; 181 $folder->{host}{$OTHERNAME} = $vc->rcv;
128 #$vc->snd("setctime", time); 182
183 $vc->snd("close");
184 $folder->close;
129 185
130 undef $quit_guard; 186 undef $quit_guard;
131 } 187 }
132} 188}
133 189
134sub main { 190sub main {
135 my $vc = new vc; 191 my $vc = create vc;
136 192
137 # time checking done symmetrically 193 # time checking done symmetrically
138 { 194 {
139 my $time = time; 195 my $time = time;
140 $vc->snd("time"); 196 $vc->snd("time");
141 my $othertime = $vc->rcv; 197 my $othertime = $vc->rcv;
142 abs (($time + time)*0.5 - $othertime) <= $::MAXTIMEDIFF 198 abs (($time + time)*0.5 - $othertime) <= $::MAXTIMEDIFF
143 or die "ERROR: time difference between hosts larger than $::MAXTIMEDIFF"; 199 or die "ERROR: time difference between hosts larger than $::MAXTIMEDIFF";
144 } 200 }
145#Coro::Event::do_timer(after => 60);#d#
146 201
147 $vc->snd("name"); 202 $vc->snd("setname", $MYNAME); $OTHERNAME = $vc->rcv;
148 $OTHERNAME = $vc->rcv;
149 203
150 if ($SLAVE) { 204 if ($SLAVE) {
151 # 205 #
152 } else { 206 } else {
153 $vc->snd("list"); 207 $vc->snd("list");
… …
173 my $name = $vc->rcv; 227 my $name = $vc->rcv;
174 my $folder = new folder name => $name; 228 my $folder = new folder name => $name;
175 229
176 slog 8, "serving folder $name\n"; 230 slog 8, "serving folder $name\n";
177 231
178 $folder->check;
179 $folder->read_mdif; 232 $folder->read_mdif;
180 233
181 while (my $msg = $vc->rcv) { 234 while (my $msg = $vc->rcv) {
182 if ($msg eq "mtime") { 235 if ($msg eq "mtime") {
183 $vc->snd($folder->{mtime}); 236 $vc->snd($folder->{mtime});
237 } elsif ($msg eq "inventory") {
238 $vc->snd($folder->inventory);
184 } elsif ($msg eq "diff") { 239 } elsif ($msg eq "diff") {
185 my $time = $vc->rcv; 240 my $time = $vc->rcv;
186 for (@{$folder->{diff}}) { 241 for (@{$folder->{diff}}) {
187 next if $_->[0] <= $time; 242 next if $_->[0] <= $time;
188 $vc->snd($_->[0], 243 $vc->snd($_->[0],
189 (join "\0", @{$_->[1]}), 244 (join "\0", @{$_->[1]}),
190 (join "\0", @{$_->[2]}), 245 (join "\0", @{$_->[2]}),
191 ); 246 );
192 } 247 }
193 $vc->snd(-1); 248 $vc->snd(-1);
249 } elsif ($msg eq "begin") {
250 $folder->begin_update;
251 } elsif ($msg eq "delete") {
252 $folder->delete(split /\0/, $vc->rcv);
253 } elsif ($msg eq "offer") {
254 my $ovc = catch vc port => $vc->rcv;
255 async {
256 my @offer;
257 {
258 my @dup;
259
260 $ovc->snd("-"); # synchronize
261 for (split /\0/, $ovc->rcv) {
262 if ($folder->exists($_)) {
263 push @dup, $_;
264 } else {
265 push @offer, $_;
266 }
267 }
268
269 $ovc->snd(join "\0", @dup);
270 }
271
272 # now we'll get everything in @offer, in order
273 $folder->append($_, $ovc->rcv) for @offer;
274 $ovc->close;
275 };
276 } elsif ($msg eq "send") {
277 $vc->pri(1);
278 $vc->snd($folder->fetch($_)) for split /\0/, $vc->rcv;
279 $vc->rcv; # sync
280 $vc->pri(0);
281 } elsif ($msg eq "end") {
282 $folder->end_update;
283 } elsif ($msg eq "mtime") {
284 $vc->snd($folder->{mtime});
194 } elsif ($msg eq "setctime") { 285 } elsif ($msg eq "setctime") {
195 $folder->{host}{$OTHERNAME} = $vc->rcv; 286 $folder->{host}{$OTHERNAME} = $vc->rcv;
287 } elsif ($msg eq "close") {
288 $folder->close;
196 } else { 289 } else {
197 die "protocol error, unknown folder command ($msg)\n"; 290 die "protocol error, unknown folder command ($msg)\n";
198 } 291 }
199 } 292 }
200} 293}
… …
203 my $vc = shift; 296 my $vc = shift;
204 297
205 slog 8, "new connection $vc->{port}\n"; 298 slog 8, "new connection $vc->{port}\n";
206 299
207 while (my $msg = $vc->rcv) { 300 while (my $msg = $vc->rcv) {
208 if ($msg eq "name") { 301 if ($msg eq "setname") {
209 $vc->snd($::MYNAME); 302 $vc->snd($MYNAME);
303 $OTHERNAME = $vc->rcv;
210 } elsif ($msg eq "pri") { 304 } elsif ($msg eq "pri") {
211 $self->{pri} = $vc->rcv; 305 $vc->{pri} = $vc->rcv;
212 } elsif ($msg eq "time") { 306 } elsif ($msg eq "time") {
213 $vc->snd(time); 307 $vc->snd(time);
214 } elsif ($msg eq "list") { 308 } elsif ($msg eq "list") {
215 $vc->snd(join "\0", find_folders); 309 $vc->snd(join "\0", find_folders);
216 } elsif ($msg eq "open") { 310 } elsif ($msg eq "open") {

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines