[ViewVC] Annotation of: cvs/syncmail/syncmail

#!/usr/bin/perl

use Coro;
use Coro::Handle;
use Coro::Event;
use Coro::Semaphore;
use Coro::Channel;
use Coro::Signal;
use Coro::RWLock;

use Fcntl;
use MD5;

use constant PROTVERSION => 1;

$VERSION = 0.1;

require "config.pl";

use folder;
use vc;

$v = $VERBOSE;

# WARNING:
# Content-Length headers are deliberately being ignored. They
# are broken by design and will never be supported

# TODO: real message-id parsing

$|=1;

my $ecnt;

sub slog {
   if ($_[0] <= $v) {
      print STDERR "[$SLAVE] ", $_[1];
   }
}

$lockdisk = new Coro::Semaphore;

# give up a/the timeslice
sub give {
   Coro::Event::do_timer(after => 0);
}

my $quit = new Coro::RWLock;

sub rwlock::DESTROY {
   $quit->unlock;
}

sub quit_guard {
   $quit->rdlock;
   bless [], rwlock;
}

sub find_folders {
   my @folders;

   opendir my $dh, $PREFIX
      or die "$PREFIX: $!";

   while (defined (my $folder = readdir $dh)) {
      next if $folder =~ /^\./;
      next unless -f "$PREFIX/$folder";
      push @folders, $folder;
   }

   @folders;
}

my $sync_folder = new Coro::Semaphore 3; # max 3 folders in parallel

sub sync_folder {
   my $name = $_[0];

   my $quit_guard = quit_guard;
   async {
      my $guard = $sync_folder->guard;
      my $vc = new vc;

      my $folder = new folder name => $name;
      #my ($rfid, $rmtime) = split /\s+/, (request open => $name)->[1];
      ::give;

      $vc->snd("open", $name);
      $vc->snd("mtime");

      $folder->check;
      $folder->read_mdif;

      my $mtime = $folder->{host}{$OTHERNAME};
      my $rtime = $vc->rcv;

      $vc->snd("diff", $rtime);
      $folder->{host}{$OTHERNAME} = $folder->{mtime};

      #request "update $rfid $folder->{mtime}";
      #request "close $rfid";
      undef $quit_guard;
   }
}

sub main {
   my $vc = new vc;

   # time checking done symmetrically
   {
      my $time = time;
      $vc->snd("time");
      my $othertime = $vc->rcv;
      abs (($time + time)*0.5 - $othertime) <= $::MAXTIMEDIFF
         or die "ERROR: time difference between hosts larger than $::MAXTIMEDIFF";
   }
#Coro::Event::do_timer(after => 60);#d#

   $vc->snd("name");
   $OTHERNAME = $vc->rcv;

   if ($SLAVE) {
      #
   } else {
      $vc->snd("list");
      for (split /\0/, $vc->rcv) {
         if (!-e "$PREFIX/$_") {
            slog 2, "created new empty folder $_\n";
            sysopen my $fh, "$PREFIX/$_", O_RDWR|O_CREAT, 0666;
         }
      }

      for my $folder (find_folders) {
         sync_folder $folder;
      }

      $quit->wrlock;

      vc::snd_quit;
   }
}

sub serve_folder {
   my $vc = shift;
   my $name = $vc->rcv;
   my $folder = new folder name => $name;

   slog 8, "serving folder $name\n";

   $folder->check;
   $folder->read_mdif;

   while (my $msg = $vc->rcv) {
      if ($msg eq "mtime") {
         $vc->snd($folder->{mtime});
      } elsif ($msg =~ /^update (\d+) (\d+)$/) {
         #if ($folder[$1]->{host}{$OTHERNAME} != $2) {
         #   $folder[$1]->{host}{$OTHERNAME} = $2;
         #   $folder[$1]->dirty;
         #}
         #reply $id, 200, "ok";
      } else {
         die "protocol error, unknown folder command ($msg)\n";
      }
   }
}

sub serve {
   my $vc = shift;

   slog 8, "new connection $vc->{port}\n";

   while (my $msg = $vc->rcv) {
      if ($msg eq "name") {
         $vc->snd($::MYNAME);
      } elsif ($msg eq "pri") {
         $self->{pri} = $vc->rcv;
      } elsif ($msg eq "time") {
         $vc->snd(time);
      } elsif ($msg eq "list") {
         $vc->snd(join "\0", find_folders);
      } elsif ($msg eq "open") {
         serve_folder($vc);
      } else {
         die "protocol error, unknown command ($msg)\n";
      }
   }
}

if ($SLAVE) {
   $HOSTID = "slave";
   async \&main;
   vc::multiplex unblock \*STDIN, $SLAVE;
} else {
   $HOSTID = "master";
   {
      use Socket;
      socketpair S1, S2, AF_UNIX, SOCK_STREAM, PF_UNSPEC;
      if (fork == 0) {
         close S1;
         open STDIN, "<&S2" or die;
         open STDOUT, ">&S2" or die;
         exec $0, "--slave";
         exit 255;
      }
      close S2;
      async \&main;
      vc::multiplex unblock \*S1, $SLAVE;
   }
}

Revision:	1.3
Committed:	Sat Oct 27 23:53:49 2001 UTC (22 years, 6 months ago) by root
Branch:	MAIN
Changes since 1.2:	+78 -396 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#!/usr/bin/perl
2
3			use Coro;
4			use Coro::Handle;
5			use Coro::Event;
6			use Coro::Semaphore;
7			use Coro::Channel;
8			use Coro::Signal;
9	root	1.2	use Coro::RWLock;
10	root	1.1
11	root	1.2	use Fcntl;
12	root	1.1	use MD5;
13
14	root	1.2	use constant PROTVERSION => 1;
15
16			$VERSION = 0.1;
17
18	root	1.3	require "config.pl";
19
20			use folder;
21			use vc;
22
23			$v = $VERBOSE;
24	root	1.2
25			# WARNING:
26			# Content-Length headers are deliberately being ignored. They
27			# are broken by design and will never be supported
28	root	1.1
29			# TODO: real message-id parsing
30
31			$\|=1;
32
33			my $ecnt;
34
35			sub slog {
36			if ($_[0] <= $v) {
37	root	1.2	print STDERR "[$SLAVE] ", $_[1];
38	root	1.1	}
39			}
40
41	root	1.3	$lockdisk = new Coro::Semaphore;
42	root	1.2
43			# give up a/the timeslice
44			sub give {
45			Coro::Event::do_timer(after => 0);
46			}
47
48			my $quit = new Coro::RWLock;
49
50			sub rwlock::DESTROY {
51			$quit->unlock;
52			}
53
54			sub quit_guard {
55			$quit->rdlock;
56			bless [], rwlock;
57			}
58
59			sub find_folders {
60			my @folders;
61
62			opendir my $dh, $PREFIX
63			or die "$PREFIX: $!";
64
65			while (defined (my $folder = readdir $dh)) {
66			next if $folder =~ /^\./;
67			next unless -f "$PREFIX/$folder";
68			push @folders, $folder;
69			}
70
71			@folders;
72			}
73
74			my $sync_folder = new Coro::Semaphore 3; # max 3 folders in parallel
75
76			sub sync_folder {
77			my $name = $_[0];
78
79			my $quit_guard = quit_guard;
80			async {
81			my $guard = $sync_folder->guard;
82	root	1.3	my $vc = new vc;
83	root	1.2
84	root	1.3	my $folder = new folder name => $name;
85			#my ($rfid, $rmtime) = split /\s+/, (request open => $name)->[1];
86	root	1.2	::give;
87
88	root	1.3	$vc->snd("open", $name);
89			$vc->snd("mtime");
90	root	1.2
91			$folder->check;
92			$folder->read_mdif;
93	root	1.3
94	root	1.2	my $mtime = $folder->{host}{$OTHERNAME};
95	root	1.3	my $rtime = $vc->rcv;
96	root	1.1
97	root	1.3	$vc->snd("diff", $rtime);
98	root	1.2	$folder->{host}{$OTHERNAME} = $folder->{mtime};
99	root	1.1
100	root	1.3	#request "update $rfid $folder->{mtime}";
101			#request "close $rfid";
102	root	1.2	undef $quit_guard;
103	root	1.1	}
104			}
105
106	root	1.2	sub main {
107	root	1.3	my $vc = new vc;
108
109	root	1.2	# time checking done symmetrically
110			{
111			my $time = time;
112	root	1.3	$vc->snd("time");
113			my $othertime = $vc->rcv;
114	root	1.2	abs (($time + time)*0.5 - $othertime) <= $::MAXTIMEDIFF
115			or die "ERROR: time difference between hosts larger than $::MAXTIMEDIFF";
116			}
117	root	1.3	#Coro::Event::do_timer(after => 60);#d#
118
119			$vc->snd("name");
120			$OTHERNAME = $vc->rcv;
121	root	1.1
122	root	1.2	if ($SLAVE) {
123			#
124			} else {
125	root	1.3	$vc->snd("list");
126			for (split /\0/, $vc->rcv) {
127	root	1.2	if (!-e "$PREFIX/$_") {
128			slog 2, "created new empty folder $_\n";
129			sysopen my $fh, "$PREFIX/$_", O_RDWR\|O_CREAT, 0666;
130			}
131			}
132	root	1.1
133	root	1.2	for my $folder (find_folders) {
134			sync_folder $folder;
135	root	1.1	}
136	root	1.2
137			$quit->wrlock;
138	root	1.3
139			vc::snd_quit;
140			}
141			}
142
143			sub serve_folder {
144			my $vc = shift;
145			my $name = $vc->rcv;
146			my $folder = new folder name => $name;
147
148			slog 8, "serving folder $name\n";
149
150			$folder->check;
151			$folder->read_mdif;
152
153			while (my $msg = $vc->rcv) {
154			if ($msg eq "mtime") {
155			$vc->snd($folder->{mtime});
156			} elsif ($msg =~ /^update (\d+) (\d+)$/) {
157			#if ($folder[$1]->{host}{$OTHERNAME} != $2) {
158			# $folder[$1]->{host}{$OTHERNAME} = $2;
159			# $folder[$1]->dirty;
160			#}
161			#reply $id, 200, "ok";
162			} else {
163			die "protocol error, unknown folder command ($msg)\n";
164			}
165			}
166			}
167
168			sub serve {
169			my $vc = shift;
170
171			slog 8, "new connection $vc->{port}\n";
172
173			while (my $msg = $vc->rcv) {
174			if ($msg eq "name") {
175			$vc->snd($::MYNAME);
176			} elsif ($msg eq "pri") {
177			$self->{pri} = $vc->rcv;
178			} elsif ($msg eq "time") {
179			$vc->snd(time);
180			} elsif ($msg eq "list") {
181			$vc->snd(join "\0", find_folders);
182			} elsif ($msg eq "open") {
183			serve_folder($vc);
184			} else {
185			die "protocol error, unknown command ($msg)\n";
186			}
187	root	1.1	}
188			}
189
190			if ($SLAVE) {
191			$HOSTID = "slave";
192	root	1.2	async \&main;
193	root	1.3	vc::multiplex unblock \*STDIN, $SLAVE;
194	root	1.1	} else {
195			$HOSTID = "master";
196			{
197			use Socket;
198			socketpair S1, S2, AF_UNIX, SOCK_STREAM, PF_UNSPEC;
199			if (fork == 0) {
200	root	1.2	close S1;
201	root	1.1	open STDIN, "<&S2" or die;
202			open STDOUT, ">&S2" or die;
203			exec $0, "--slave";
204			exit 255;
205			}
206	root	1.2	close S2;
207			async \&main;
208	root	1.3	vc::multiplex unblock \*S1, $SLAVE;
209	root	1.1	}
210			}
211