[ViewVC] Annotation of: cvs/nbdcache/nbdcache

#!/opt/bin/perl

my ($cachemeta, $cachedata, $sizegb, $orig, $apply) = @ARGV
   or die "Usage: $0 /cache/meta /cache/data size_gb origdevice\n";

use common::sense;

use IO::AIO ();
use Coro::AIO ();

use Linux::NBD;
use Linux::NBD::Server;

use POSIX ();

use IO::Socket::UNIX;

sub BLKSIZE() { 512 }

my $blocks = qx<blockdev --getsz \Q$orig> * 512 / BLKSIZE;

sysopen my $orig_fh, $orig, IO::AIO::O_RDWR
   or die "$orig: $!";

sysopen my $meta_fh, "$cachemeta", IO::AIO::O_RDWR | IO::AIO::O_CREAT
   or die "$cachemeta: $!";

sysopen my $data_fh, "$cachedata", IO::AIO::O_RDWR | IO::AIO::O_CREAT
   or die "$cachedata: $!";

my $cacheblocks = $sizegb * 1024 * 1024 * 1024 / BLKSIZE;

my $meta_data;
my $data_data;

sub mapin {
   Coro::AIO::aio_truncate $meta_fh, $blocks * 4;
   Coro::AIO::aio_truncate $data_fh, $cacheblocks * BLKSIZE;

   IO::AIO::mmap $meta_data, $blocks * 4           , IO::AIO::PROT_READ | IO::AIO::PROT_WRITE, IO::AIO::MAP_SHARED, $meta_fh, 0;
   IO::AIO::mmap $data_data, $cacheblocks * BLKSIZE, IO::AIO::PROT_READ | IO::AIO::PROT_WRITE, IO::AIO::MAP_SHARED, $data_fh, 0;

#   IO::AIO::madvise $meta_data, 0, undef, IO::AIO::MADV_RANDOM;
#   IO::AIO::madvise $data_data, 0, undef, IO::AIO::MADV_RANDOM;
}

system "chattr", "+C", $cachemeta;
system "chattr", "+C", $cachedata;

mapin;

IO::AIO::mmap my $orig_data, $blocks * BLKSIZE, IO::AIO::PROT_READ | IO::AIO::PROT_WRITE, IO::AIO::MAP_SHARED, $orig_fh, 0;

package myserver;

use common::sense;

sub BLKSIZE() { 512 }

our @ISA = Linux::NBD::Server::;

my $alloc = 0;
#my $need_sync = 0;

sub apply {
   my ($reset) = @_;

   my ($blk, $idx);

   my ($blk_nxt, $blk_buf, $blk_total);

   my $blk_flush = sub {
      return unless length $blk_buf;

      $blk_nxt -= (length $blk_buf) / BLKSIZE;
      $blk_total += length $blk_buf;

      printf "write back blk %d len %d (total: %.3fGiB)\n", $blk_nxt, length $blk_buf, $blk_total / (1 << 30);
      Coro::AIO::aio_write $orig_fh, $blk_nxt * BLKSIZE, undef, $blk_buf, 0;
      #substr $orig_data, $blk_nxt * BLKSIZE, length $blk_buf, $blk_buf;

      $blk_buf = "";
   };

   my $nul = (pack "L", 0) x 65536;
   for (my $base = 0; $base * 4 < length $meta_data; $base += 65536) {
      if ($nul ne substr $meta_data, $base * 4, 65536 * 4) {
         for $blk ($base .. $base + 65535) {
            $idx = unpack "L", substr $meta_data, $blk * 4, 4;

            if ($idx) {
               if ($blk_nxt != $blk or (64 * 1024 * 1024) <= length $blk_buf) {
                  $blk_flush->();
               }

               #warn "apply cache blk $idx to orig blk $blk\n";
               $blk_nxt = $blk + 1;
               $blk_buf .= substr $data_data, $idx * BLKSIZE, BLKSIZE;

               #IO::AIO::aio_write $orig_fh, $blk * BLKSIZE, BLKSIZE, $data_data, $idx * BLKSIZE, sub { };
               #substr $orig_data, $blk * BLKSIZE, BLKSIZE, substr $data_data, $idx * BLKSIZE, BLKSIZE;
            }
         }
      }
   }

   print "done, syncing...\n";
   $blk_flush->();
   Coro::AIO::aio_fsync $orig_fh;

   if ($reset) {
      print "resetting\n";

      undef $meta_data;
      undef $data_data;

      Coro::AIO::aio_truncate $meta_fh, 0;
      Coro::AIO::aio_truncate $data_fh, 0;

      ::mapin;

      Coro::AIO::aio_fsync $meta_fh;
      Coro::AIO::aio_fsync $data_fh;
   }

   $alloc = 0;
}

apply $apply eq "r" if $apply;

{
   my $nul = "\x00" x 65536;
   for (my $ofs = 0; $ofs < length $meta_data; $ofs += 65536) {
      if ($nul ne substr $meta_data, $ofs, 65536) {
         for (unpack "L*", substr $meta_data, $ofs, 65536) {
            $alloc = $_ if $_ > $alloc;
         }
      }
   }
}

print "ALLOC is $alloc\n";

sub sync {
   Coro::AIO::aio_fsync $data_fh;
   Coro::AIO::aio_fsync $meta_fh;
}

sub req_read {
   my ($self, $handle, $ofs, $len) = @_;

#   warn "get read $ofs,$len\n";#d#

   my $buf;

   die if $ofs & (BLKSIZE - 1);
   die if $len & (BLKSIZE - 1);

   $ofs /= BLKSIZE;
   $len /= BLKSIZE;

   while ($len > 0) {
      my $idx = unpack "L", substr $meta_data, $ofs * 4, 4;

      if ($idx) {
         $buf .= substr $data_data, $idx * BLKSIZE, BLKSIZE;
      } else {
         $buf .= substr $orig_data, $ofs * BLKSIZE, BLKSIZE;
      }

      ++$ofs;
      --$len;
   }

   $self->reply ($handle, 0, $buf);
}

sub req_write {
   my ($self, $handle, $ofs, $buf) = @_;

#   print "write $ofs, ", length $buf, "\n";

   die if $ofs & (BLKSIZE - 1);
   die if (length $buf) & (BLKSIZE - 1);

   $ofs /= BLKSIZE;

   my $len = (length $buf) / BLKSIZE;

   while (length $buf) {
      my $idx = unpack "L", substr $meta_data, $ofs * 4, 4;

      if ($idx) {
         substr $data_data, $idx * BLKSIZE, BLKSIZE, substr $buf, 0, BLKSIZE;
      } else {
         if ($alloc + 1 >= $cacheblocks) {
            print "cache full, applying...\n";
            #$need_sync = 1;
            sync;
            apply 1;
            print "cache applied\n";
         }

         $idx = ++$alloc;

         substr $data_data, $idx * BLKSIZE, BLKSIZE, substr $buf, 0, BLKSIZE;
         substr $meta_data,$ofs * 4, 4, pack "L", $idx;
      }

      ++$ofs;
      substr $buf, 0, BLKSIZE, "";
   }

   $self->reply ($handle, 0);
}

sub req_flush {
   my ($self, $handle) = @_;

#   sync if $need_sync;

   $self->reply ($handle, 0);

   #print "flushed.\n";
}

package main;

#my $listen = new IO::Socket::INET LocalPort => 10809, Listen => 1;
unlink "/tmp/nbdsock";
my $listen = new IO::Socket::UNIX Local => "/tmp/nbdsock", Listen => 1;

while () {
   print "waiting for connections...\n";
   my $fh = $listen->accept;
   print "accepted.\n";

   syswrite $fh,
      ("NBDMAGIC\x00\x00\x42\x02\x81\x86\x12\x53")
      . (pack "Q> N x124", $blocks * BLKSIZE, 1 + 4); # has_flags + can_flush

   my $server = new myserver socket => $fh;

   $server->run;

   myserver::sync;
}

Revision:	1.2
Committed:	Tue Jun 6 06:38:52 2017 UTC (6 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.1:	+6 -1 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#!/opt/bin/perl
2
3			my ($cachemeta, $cachedata, $sizegb, $orig, $apply) = @ARGV
4			or die "Usage: $0 /cache/meta /cache/data size_gb origdevice\n";
5
6			use common::sense;
7
8			use IO::AIO ();
9			use Coro::AIO ();
10
11			use Linux::NBD;
12			use Linux::NBD::Server;
13
14			use POSIX ();
15
16			use IO::Socket::UNIX;
17
18			sub BLKSIZE() { 512 }
19
20			my $blocks = qx<blockdev --getsz \Q$orig> * 512 / BLKSIZE;
21
22			sysopen my $orig_fh, $orig, IO::AIO::O_RDWR
23			or die "$orig: $!";
24
25			sysopen my $meta_fh, "$cachemeta", IO::AIO::O_RDWR \| IO::AIO::O_CREAT
26			or die "$cachemeta: $!";
27
28			sysopen my $data_fh, "$cachedata", IO::AIO::O_RDWR \| IO::AIO::O_CREAT
29			or die "$cachedata: $!";
30
31			my $cacheblocks = $sizegb * 1024 * 1024 * 1024 / BLKSIZE;
32
33			my $meta_data;
34			my $data_data;
35
36			sub mapin {
37			Coro::AIO::aio_truncate $meta_fh, $blocks * 4;
38			Coro::AIO::aio_truncate $data_fh, $cacheblocks * BLKSIZE;
39
40			IO::AIO::mmap $meta_data, $blocks * 4 , IO::AIO::PROT_READ \| IO::AIO::PROT_WRITE, IO::AIO::MAP_SHARED, $meta_fh, 0;
41			IO::AIO::mmap $data_data, $cacheblocks * BLKSIZE, IO::AIO::PROT_READ \| IO::AIO::PROT_WRITE, IO::AIO::MAP_SHARED, $data_fh, 0;
42	root	1.2
43			# IO::AIO::madvise $meta_data, 0, undef, IO::AIO::MADV_RANDOM;
44			# IO::AIO::madvise $data_data, 0, undef, IO::AIO::MADV_RANDOM;
45	root	1.1	}
46
47			system "chattr", "+C", $cachemeta;
48			system "chattr", "+C", $cachedata;
49
50			mapin;
51
52			IO::AIO::mmap my $orig_data, $blocks * BLKSIZE, IO::AIO::PROT_READ \| IO::AIO::PROT_WRITE, IO::AIO::MAP_SHARED, $orig_fh, 0;
53
54			package myserver;
55
56			use common::sense;
57
58			sub BLKSIZE() { 512 }
59
60			our @ISA = Linux::NBD::Server::;
61
62			my $alloc = 0;
63			#my $need_sync = 0;
64
65			sub apply {
66			my ($reset) = @_;
67
68			my ($blk, $idx);
69
70			my ($blk_nxt, $blk_buf, $blk_total);
71
72			my $blk_flush = sub {
73			return unless length $blk_buf;
74
75			$blk_nxt -= (length $blk_buf) / BLKSIZE;
76			$blk_total += length $blk_buf;
77
78	root	1.2	printf "write back blk %d len %d (total: %.3fGiB)\n", $blk_nxt, length $blk_buf, $blk_total / (1 << 30);
79	root	1.1	Coro::AIO::aio_write $orig_fh, $blk_nxt * BLKSIZE, undef, $blk_buf, 0;
80			#substr $orig_data, $blk_nxt * BLKSIZE, length $blk_buf, $blk_buf;
81
82			$blk_buf = "";
83			};
84
85			my $nul = (pack "L", 0) x 65536;
86			for (my $base = 0; $base * 4 < length $meta_data; $base += 65536) {
87			if ($nul ne substr $meta_data, $base * 4, 65536 * 4) {
88			for $blk ($base .. $base + 65535) {
89			$idx = unpack "L", substr $meta_data, $blk * 4, 4;
90
91			if ($idx) {
92			if ($blk_nxt != $blk or (64 * 1024 * 1024) <= length $blk_buf) {
93			$blk_flush->();
94			}
95
96			#warn "apply cache blk $idx to orig blk $blk\n";
97			$blk_nxt = $blk + 1;
98			$blk_buf .= substr $data_data, $idx * BLKSIZE, BLKSIZE;
99
100			#IO::AIO::aio_write $orig_fh, $blk * BLKSIZE, BLKSIZE, $data_data, $idx * BLKSIZE, sub { };
101			#substr $orig_data, $blk * BLKSIZE, BLKSIZE, substr $data_data, $idx * BLKSIZE, BLKSIZE;
102			}
103			}
104			}
105			}
106
107			print "done, syncing...\n";
108			$blk_flush->();
109			Coro::AIO::aio_fsync $orig_fh;
110
111			if ($reset) {
112			print "resetting\n";
113
114			undef $meta_data;
115			undef $data_data;
116
117			Coro::AIO::aio_truncate $meta_fh, 0;
118			Coro::AIO::aio_truncate $data_fh, 0;
119
120			::mapin;
121
122			Coro::AIO::aio_fsync $meta_fh;
123			Coro::AIO::aio_fsync $data_fh;
124			}
125
126			$alloc = 0;
127			}
128
129			apply $apply eq "r" if $apply;
130
131			{
132			my $nul = "\x00" x 65536;
133			for (my $ofs = 0; $ofs < length $meta_data; $ofs += 65536) {
134			if ($nul ne substr $meta_data, $ofs, 65536) {
135			for (unpack "L*", substr $meta_data, $ofs, 65536) {
136			$alloc = $_ if $_ > $alloc;
137			}
138			}
139			}
140			}
141
142			print "ALLOC is $alloc\n";
143
144			sub sync {
145			Coro::AIO::aio_fsync $data_fh;
146			Coro::AIO::aio_fsync $meta_fh;
147			}
148
149			sub req_read {
150			my ($self, $handle, $ofs, $len) = @_;
151
152			# warn "get read $ofs,$len\n";#d#
153
154			my $buf;
155
156			die if $ofs & (BLKSIZE - 1);
157			die if $len & (BLKSIZE - 1);
158
159			$ofs /= BLKSIZE;
160			$len /= BLKSIZE;
161
162			while ($len > 0) {
163			my $idx = unpack "L", substr $meta_data, $ofs * 4, 4;
164
165			if ($idx) {
166			$buf .= substr $data_data, $idx * BLKSIZE, BLKSIZE;
167			} else {
168			$buf .= substr $orig_data, $ofs * BLKSIZE, BLKSIZE;
169			}
170
171			++$ofs;
172			--$len;
173			}
174
175			$self->reply ($handle, 0, $buf);
176			}
177
178			sub req_write {
179			my ($self, $handle, $ofs, $buf) = @_;
180
181			# print "write $ofs, ", length $buf, "\n";
182
183			die if $ofs & (BLKSIZE - 1);
184			die if (length $buf) & (BLKSIZE - 1);
185
186			$ofs /= BLKSIZE;
187
188			my $len = (length $buf) / BLKSIZE;
189
190			while (length $buf) {
191			my $idx = unpack "L", substr $meta_data, $ofs * 4, 4;
192
193			if ($idx) {
194			substr $data_data, $idx * BLKSIZE, BLKSIZE, substr $buf, 0, BLKSIZE;
195			} else {
196			if ($alloc + 1 >= $cacheblocks) {
197			print "cache full, applying...\n";
198			#$need_sync = 1;
199			sync;
200			apply 1;
201			print "cache applied\n";
202			}
203
204			$idx = ++$alloc;
205
206			substr $data_data, $idx * BLKSIZE, BLKSIZE, substr $buf, 0, BLKSIZE;
207			substr $meta_data,$ofs * 4, 4, pack "L", $idx;
208			}
209
210			++$ofs;
211			substr $buf, 0, BLKSIZE, "";
212			}
213
214			$self->reply ($handle, 0);
215			}
216
217			sub req_flush {
218			my ($self, $handle) = @_;
219
220			# sync if $need_sync;
221
222			$self->reply ($handle, 0);
223
224			#print "flushed.\n";
225			}
226
227			package main;
228
229			#my $listen = new IO::Socket::INET LocalPort => 10809, Listen => 1;
230			unlink "/tmp/nbdsock";
231			my $listen = new IO::Socket::UNIX Local => "/tmp/nbdsock", Listen => 1;
232
233			while () {
234			print "waiting for connections...\n";
235			my $fh = $listen->accept;
236			print "accepted.\n";
237
238			syswrite $fh,
239			("NBDMAGIC\x00\x00\x42\x02\x81\x86\x12\x53")
240			. (pack "Q> N x124", $blocks * BLKSIZE, 1 + 4); # has_flags + can_flush
241
242			my $server = new myserver socket => $fh;
243
244			$server->run;
245	root	1.2
246			myserver::sync;
247	root	1.1	}
248