tpb/common.pl

#!/opt/bin/perl

use common::sense;
use Compress::LZF;
use EV;
use AnyEvent::HTTP;
use BDB;
use Coro;
use Coro::BDB;
use AnyEvent::DNS;
use Set::IntSpan;
use CBOR::XS;
use Compress::Zlib ();
use Compress::Raw::Zlib ();
use AnyEvent::TLS;

#d# cloudflare requires SNI
my $get_session = \&AnyEvent::TLS::_get_session;
*AnyEvent::TLS::_get_session = sub ($$;$$) {
   my ($self, $mode, $ref, $cn) = @_;
   my $session = $get_session->($self, $mode, $ref, $cn);
   Net::SSLeay::set_tlsext_host_name $session, $cn;
   $session
};

BDB::min_parallel 8;

our @TPB = qw(thepiratebay.la);
our $TPB = $TPB[0];

our $TODAY = int time / 86400;

#our $TPB_SWITCHER = AE::timer 1, 0.02, sub {
#   state $i;
#   $TPB = $TPB[++$i % @TPB];
#};

#############################################################################

$AnyEvent::HTTP::MAX_PER_HOST =
$AnyEvent::HTTP::MAX_PERSISTENT_PER_HOST = 400;

our %stat;

our $statprint = AE::timer 1, 1, sub {
   syswrite STDOUT, "\r" . (join " ", map "$_=$stat{$_}", sort keys %stat) . " "
      if $ENV{VERBOSE};
};

our @DICTIONARY = (do {
   open my $fh, "<:perlio", "dictionary0"
      or die "dictionary0: $!";
   local $/;
   <$fh>
});

#############################################################################

our $db_env = db_env_create;

$db_env->set_flags (BDB::AUTO_COMMIT | BDB::TXN_WRITE_NOSYNC, 1);
$db_env->log_set_config (BDB::LOG_AUTO_REMOVE);
$db_env->set_cachesize (0, 8 * 1024 * 1024);
$db_env->set_lk_max_lockers (20480);
$db_env->set_lk_max_locks (20480);
$db_env->set_lk_max_objects (20480);

mkdir "db", 0777;
db_env_open
  $db_env,
  "db",
  BDB::INIT_LOCK | BDB::INIT_LOG | BDB::INIT_MPOOL
  | BDB::INIT_TXN | BDB::RECOVER | BDB::REGISTER | BDB::USE_ENVIRON | BDB::CREATE,
  0666;

our %DB;

sub table($) {
   my $db = $DB{$_[0]} = db_create $db_env;
   db_open $db, undef, $_[0], undef, BDB::BTREE, BDB::AUTO_COMMIT | BDB::CREATE, 0666;
   die if $!;
   $db
}

our $db_state = table "state"; # global state
our $db_info  = table "info"; # per torrent state

our $TXN;

sub get($$) {
   db_get $_[0], $TXN, $_[1], my $data, 0;
   $data # will be undef on error
}

sub put($$$) {
   db_put $_[0], $TXN, $_[1], $_[2], 0;
}

sub del($$) {
   db_del $_[0], $TXN, $_[1], 0, sub { };
}

sub sget($) {
   get $db_state, $_[0]
}

sub sput($$) {
   put $db_state, $_[0], $_[1];
}

sub iget($) {
   my $data = get $db_info, $_[0];
   length $data ? decode_cbor $data : []
}

sub iput($$) {
   put $db_info, $_[0], encode_cbor $_[1];
}

our $syncer = AE::timer 5, 5, sub {
   db_env_txn_checkpoint $db_env, 0, 0, 0, sub { };
};

sub dbsync() {
   db_env_txn_checkpoint $db_env;
}

sub dbflush {
   return if $::dbflush_called++;
   db_env_txn_checkpoint $db_env, sub { };
   BDB::set_sync_prepare undef;
   db_close $_ for values %DB;
   db_env_txn_checkpoint $db_env;
   db_env_close $db_env;
}

END {
   dbflush;
}

sub quit {
   EV::unloop;
   #exit 1;
};

our $sigint  = AE::signal INT  => \&quit;
our $sigterm = AE::signal TERM => \&quit;

sub dic_compress($$) {
   my $d = new Compress::Raw::Zlib::Deflate
      -Level        => 9,
      -WindowBits   => -15,
      -Dictionary   => $DICTIONARY[$_[0]],
      -AppendOutput => 1,
   ;

   my $o;

   $d->deflate ($_[1], $o);
   $d->flush ($o);

   $o
}

sub dic_decompress($$) {
   my $i = new Compress::Raw::Zlib::Inflate
      -WindowBits   => -15,
      -Dictionary   => $DICTIONARY[$_[0]],
      -AppendOutput => 1,
      -ConsumeInput => 0,
   ;

   $i->inflate ($_[1], my $o, 1);

   $o
}

#############################################################################

#############################################################################

sub GET {
   my ($data, $hdr);

   for (1..10) {
      http_get $_[0],
         proxy => undef,
         recurse => 0,
         headers => {
#            "user-agent" => "Mozilla/5.0 (X11; Linux x86_64; rv:20.0) Gecko/20100101 Firefox/20.0 Iceweasel/20.0",
            "referer" => undef,
         },
         Coro::rouse_cb;
      ($data, $hdr) = Coro::rouse_wait;

      $hdr->{Status} = 580
         if $data =~ /^(?:Upgrading software|Could not connect to caching server)/;

      $hdr->{Status} = 581
         unless length $data;

      $data = Compress::Zlib::memGunzip $data
         if $hdr->{"content-encoding"} eq "gzip";

      last
         if $hdr->{Status} < 500;

      Coro::AnyEvent::sleep 0.5;
      ++$stat{"retry-$hdr->{Status}"};
   }

   ++$stat{$hdr->{Status}};
   ($data, $hdr)
}

1

Revision:	1.1
Committed:	Sun Sep 27 07:55:20 2015 UTC (8 years, 8 months ago) by root
Content type:	text/plain
Branch:	MAIN
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#!/opt/bin/perl
2
3			use common::sense;
4			use Compress::LZF;
5			use EV;
6			use AnyEvent::HTTP;
7			use BDB;
8			use Coro;
9			use Coro::BDB;
10			use AnyEvent::DNS;
11			use Set::IntSpan;
12			use CBOR::XS;
13			use Compress::Zlib ();
14			use Compress::Raw::Zlib ();
15			use AnyEvent::TLS;
16
17			#d# cloudflare requires SNI
18			my $get_session = \&AnyEvent::TLS::_get_session;
19			*AnyEvent::TLS::_get_session = sub ($$;$$) {
20			my ($self, $mode, $ref, $cn) = @_;
21			my $session = $get_session->($self, $mode, $ref, $cn);
22			Net::SSLeay::set_tlsext_host_name $session, $cn;
23			$session
24			};
25
26			BDB::min_parallel 8;
27
28			our @TPB = qw(thepiratebay.la);
29			our $TPB = $TPB[0];
30
31			our $TODAY = int time / 86400;
32
33			#our $TPB_SWITCHER = AE::timer 1, 0.02, sub {
34			# state $i;
35			# $TPB = $TPB[++$i % @TPB];
36			#};
37
38			#############################################################################
39
40			$AnyEvent::HTTP::MAX_PER_HOST =
41			$AnyEvent::HTTP::MAX_PERSISTENT_PER_HOST = 400;
42
43			our %stat;
44
45			our $statprint = AE::timer 1, 1, sub {
46			syswrite STDOUT, "\r" . (join " ", map "$_=$stat{$_}", sort keys %stat) . " "
47			if $ENV{VERBOSE};
48			};
49
50			our @DICTIONARY = (do {
51			open my $fh, "<:perlio", "dictionary0"
52			or die "dictionary0: $!";
53			local $/;
54			<$fh>
55			});
56
57			#############################################################################
58
59			our $db_env = db_env_create;
60
61			$db_env->set_flags (BDB::AUTO_COMMIT \| BDB::TXN_WRITE_NOSYNC, 1);
62			$db_env->log_set_config (BDB::LOG_AUTO_REMOVE);
63			$db_env->set_cachesize (0, 8 * 1024 * 1024);
64			$db_env->set_lk_max_lockers (20480);
65			$db_env->set_lk_max_locks (20480);
66			$db_env->set_lk_max_objects (20480);
67
68			mkdir "db", 0777;
69			db_env_open
70			$db_env,
71			"db",
72			BDB::INIT_LOCK \| BDB::INIT_LOG \| BDB::INIT_MPOOL
73			\| BDB::INIT_TXN \| BDB::RECOVER \| BDB::REGISTER \| BDB::USE_ENVIRON \| BDB::CREATE,
74			0666;
75
76			our %DB;
77
78			sub table($) {
79			my $db = $DB{$_[0]} = db_create $db_env;
80			db_open $db, undef, $_[0], undef, BDB::BTREE, BDB::AUTO_COMMIT \| BDB::CREATE, 0666;
81			die if $!;
82			$db
83			}
84
85			our $db_state = table "state"; # global state
86			our $db_info = table "info"; # per torrent state
87
88			our $TXN;
89
90			sub get($$) {
91			db_get $_[0], $TXN, $_[1], my $data, 0;
92			$data # will be undef on error
93			}
94
95			sub put($$$) {
96			db_put $_[0], $TXN, $_[1], $_[2], 0;
97			}
98
99			sub del($$) {
100			db_del $_[0], $TXN, $_[1], 0, sub { };
101			}
102
103			sub sget($) {
104			get $db_state, $_[0]
105			}
106
107			sub sput($$) {
108			put $db_state, $_[0], $_[1];
109			}
110
111			sub iget($) {
112			my $data = get $db_info, $_[0];
113			length $data ? decode_cbor $data : []
114			}
115
116			sub iput($$) {
117			put $db_info, $_[0], encode_cbor $_[1];
118			}
119
120			our $syncer = AE::timer 5, 5, sub {
121			db_env_txn_checkpoint $db_env, 0, 0, 0, sub { };
122			};
123
124			sub dbsync() {
125			db_env_txn_checkpoint $db_env;
126			}
127
128			sub dbflush {
129			return if $::dbflush_called++;
130			db_env_txn_checkpoint $db_env, sub { };
131			BDB::set_sync_prepare undef;
132			db_close $_ for values %DB;
133			db_env_txn_checkpoint $db_env;
134			db_env_close $db_env;
135			}
136
137			END {
138			dbflush;
139			}
140
141			sub quit {
142			EV::unloop;
143			#exit 1;
144			};
145
146			our $sigint = AE::signal INT => \&quit;
147			our $sigterm = AE::signal TERM => \&quit;
148
149			sub dic_compress($$) {
150			my $d = new Compress::Raw::Zlib::Deflate
151			-Level => 9,
152			-WindowBits => -15,
153			-Dictionary => $DICTIONARY[$_[0]],
154			-AppendOutput => 1,
155			;
156
157			my $o;
158
159			$d->deflate ($_[1], $o);
160			$d->flush ($o);
161
162			$o
163			}
164
165			sub dic_decompress($$) {
166			my $i = new Compress::Raw::Zlib::Inflate
167			-WindowBits => -15,
168			-Dictionary => $DICTIONARY[$_[0]],
169			-AppendOutput => 1,
170			-ConsumeInput => 0,
171			;
172
173			$i->inflate ($_[1], my $o, 1);
174
175			$o
176			}
177
178			#############################################################################
179
180			#############################################################################
181
182			sub GET {
183			my ($data, $hdr);
184
185			for (1..10) {
186			http_get $_[0],
187			proxy => undef,
188			recurse => 0,
189			headers => {
190			# "user-agent" => "Mozilla/5.0 (X11; Linux x86_64; rv:20.0) Gecko/20100101 Firefox/20.0 Iceweasel/20.0",
191			"referer" => undef,
192			},
193			Coro::rouse_cb;
194			($data, $hdr) = Coro::rouse_wait;
195
196			$hdr->{Status} = 580
197			if $data =~ /^(?:Upgrading software\|Could not connect to caching server)/;
198
199			$hdr->{Status} = 581
200			unless length $data;
201
202			$data = Compress::Zlib::memGunzip $data
203			if $hdr->{"content-encoding"} eq "gzip";
204
205			last
206			if $hdr->{Status} < 500;
207
208			Coro::AnyEvent::sleep 0.5;
209			++$stat{"retry-$hdr->{Status}"};
210			}
211
212			++$stat{$hdr->{Status}};
213			($data, $hdr)
214			}
215
216			1
217