lib/MogileFS/Server.pm

   1 package MogileFS::Server;
   2 use strict;
   3 use warnings;
   4 use vars qw($VERSION);
   5 $VERSION = "2.32";
   6
   7 =head1 NAME
   8
   9 MogileFS::Server - MogileFS (distributed filesystem) server
  10
  11 =head1 SYNOPSIS
  12
  13  $s = MogileFS::Server->server;
  14  $s->run;
  15
  16 =cut
  17
  18 # based on where we found this file (a pure-perl module),
  19 # add the mogdeps/ subdirectory of that base to our @INC search
  20 # path, where all the misc Mogile dependencies are installed.
  21 BEGIN {
  22     my $libpath;
  23     if (! $ENV{MOGILE_NO_BUILTIN_DEPS} &&
  24         ($libpath = $INC{"MogileFS/Server.pm"}) &&
  25         $libpath =~ s!MogileFS/Server.pm$!!)
  26     {
  27         my $dep_dir = "${libpath}mogdeps";
  28         push @INC, $dep_dir;
  29         unless (($ENV{PERL5LIB} || "") =~ /$dep_dir/) {
  30             $ENV{PERL5LIB} = join(":",
  31                                   split(/:/, $ENV{PERL5LIB} || ""),
  32                                   $dep_dir);
  33         }
  34     }
  35 }
  36
  37 use IO::Socket;
  38 use Symbol;
  39 use POSIX;
  40 use File::Copy ();
  41 use Carp;
  42 use File::Basename ();
  43 use File::Path ();
  44 use Sys::Syslog ();
  45 use Time::HiRes ();
  46 use Net::Netmask;
  47 use LWP::UserAgent;
  48 use List::Util;
  49 use Socket ();
  50
  51 use MogileFS::Util qw(daemonize);
  52 use MogileFS::Sys;
  53 use MogileFS::Config;
  54
  55 use MogileFS::ProcManager;
  56 use MogileFS::Connection::Client;
  57 use MogileFS::Connection::Worker;
  58
  59 use MogileFS::Worker::Query;
  60 use MogileFS::Worker::Delete;
  61 use MogileFS::Worker::Replicate;
  62 use MogileFS::Worker::Reaper;
  63 use MogileFS::Worker::Monitor;
  64 use MogileFS::Worker::Fsck;
  65 use MogileFS::Worker::JobMaster;
  66
  67 use MogileFS::HTTPFile;
  68 use MogileFS::Class;
  69 use MogileFS::Device;
  70 use MogileFS::Host;
  71 use MogileFS::FID;
  72 use MogileFS::Domain;
  73 use MogileFS::DevFID;
  74
  75 use MogileFS::Store;
  76 use MogileFS::Store::MySQL;  # FIXME: don't load this until after reading their config, but before fork.
  77
  78 use MogileFS::ReplicationPolicy::MultipleHosts;
  79
  80 my $server; # server singleton
  81 sub server {
  82     my ($pkg) = @_;
  83     return $server ||= bless {}, $pkg;
  84 }
  85
  86 # --------------------------------------------------------------------------
  87 # instance methods:
  88 # --------------------------------------------------------------------------
  89
  90 sub run {
  91     my $self = shift;
  92
  93     MogileFS::Config->load_config;
  94
  95     # don't run as root
  96     die "mogilefsd cannot be run as root\n"
  97         if $< == 0 && MogileFS->config('user') ne "root";
  98
  99     MogileFS::Config->check_database;
 100     daemonize() if MogileFS->config("daemonize");
 101
 102     MogileFS::ProcManager->set_min_workers('queryworker' => MogileFS->config('query_jobs'));
 103     MogileFS::ProcManager->set_min_workers('delete'      => MogileFS->config('delete_jobs'));
 104     MogileFS::ProcManager->set_min_workers('replicate'   => MogileFS->config('replicate_jobs'));
 105     MogileFS::ProcManager->set_min_workers('reaper'      => MogileFS->config('reaper_jobs'));
 106     MogileFS::ProcManager->set_min_workers('monitor'     => MogileFS->config('monitor_jobs'));
 107     MogileFS::ProcManager->set_min_workers('fsck'        => MogileFS->config('fsck_jobs'));
 108     MogileFS::ProcManager->set_min_workers('job_master'  => 1);
 109
 110     # open up our log
 111     Sys::Syslog::openlog('mogilefsd', 'pid', 'daemon');
 112     Mgd::log('info', 'beginning run');
 113
 114     unless (MogileFS::ProcManager->write_pidfile) {
 115         Mgd::log('info', "Couldn't write pidfile, ending run");
 116         Sys::Syslog::closelog();
 117         exit 1;
 118     }
 119
 120     # Install signal handlers.
 121     $SIG{TERM}  = sub {
 122         my @children = MogileFS::ProcManager->child_pids;
 123         print STDERR scalar @children, " children to kill.\n" if $DEBUG;
 124         my $count = kill( 'TERM' => @children );
 125         print STDERR "Sent SIGTERM to $count children.\n" if $DEBUG;
 126         MogileFS::ProcManager->remove_pidfile;
 127         Mgd::log('info', 'ending run due to SIGTERM');
 128         Sys::Syslog::closelog();
 129
 130         exit 0;
 131     };
 132
 133     $SIG{INT}  = sub {
 134         my @children = MogileFS::ProcManager->child_pids;
 135         print STDERR scalar @children, " children to kill.\n" if $DEBUG;
 136         my $count = kill( 'INT' => @children );
 137         print STDERR "Sent SIGINT to $count children.\n" if $DEBUG;
 138         MogileFS::ProcManager->remove_pidfile;
 139         Mgd::log('info', 'ending run due to SIGINT');
 140         exit 0;
 141     };
 142     $SIG{PIPE} = 'IGNORE';  # catch them by hand
 143
 144     # setup server sockets to listen for client connections
 145     my @servers;
 146     foreach my $listen (@{ MogileFS->config('listen') }) {
 147         my $server = IO::Socket::INET->new(LocalAddr => $listen,
 148                                            Type      => SOCK_STREAM,
 149                                            Proto     => 'tcp',
 150                                            Blocking  => 0,
 151                                            Reuse     => 1,
 152                                            Listen    => 10 )
 153             or die "Error creating socket: $@\n";
 154
 155         # save sub to accept a client
 156         push @servers, $server;
 157         Danga::Socket->AddOtherFds( fileno($server) => sub {
 158                 my $csock = $server->accept
 159                     or return;
 160                 MogileFS::Connection::Client->new($csock);
 161             } );
 162     }
 163
 164     MogileFS::ProcManager->push_pre_fork_cleanup(sub {
 165         # so children don't hold server connection open
 166         close($_) foreach @servers;
 167     });
 168
 169     # setup the post event loop callback to spawn jobs, and the timeout
 170     Danga::Socket->DebugLevel(3);
 171     Danga::Socket->SetLoopTimeout( 250 ); # 250 milliseconds
 172     Danga::Socket->SetPostLoopCallback(MogileFS::ProcManager->PostEventLoopChecker);
 173
 174     # and now, actually start listening for events
 175     eval {
 176         print( "Starting event loop for frontend job on pid $$.\n" ) if $DEBUG;
 177         Danga::Socket->EventLoop();
 178     };
 179
 180     if ($@) {
 181         Mgd::log('err', "crash log: $@");
 182         exit 1;
 183     }
 184     Mgd::log('info', 'ending run');
 185     Sys::Syslog::closelog();
 186     exit(0);
 187 }
 188
 189 # --------------------------------------------------------------------------
 190
 191 package MogileFS;
 192 # just so MogileFS->config($key) will work:
 193 use MogileFS::Config qw(config);
 194
 195 my %hooks;
 196
 197 sub register_worker_command {
 198     # just pass this through to the Worker class
 199     return MogileFS::Worker::Query::register_command(@_);
 200 }
 201
 202 sub register_global_hook {
 203     $hooks{$_[0]} = $_[1];
 204     return 1;
 205 }
 206
 207 sub unregister_global_hook {
 208     delete $hooks{$_[0]};
 209     return 1;
 210 }
 211
 212 sub run_global_hook {
 213     my $hookname = shift;
 214     my $ref = $hooks{$hookname};
 215     return $ref->(@_) if defined $ref;
 216     return undef;
 217 }
 218
 219 # --------------------------------------------------------------------------
 220
 221 package Mgd;  # conveniently short name
 222 use strict;
 223 use warnings;
 224 use MogileFS::Config;
 225 use MogileFS::Util qw(error fatal debug); # for others calling Mgd::foo()
 226
 227 sub server {
 228     return MogileFS::Server->server;
 229 }
 230
 231 # database checking/connecting
 232 sub validate_dbh { Mgd::get_store()->recheck_dbh }
 233 sub get_dbh      { return Mgd::get_store()->dbh  }
 234
 235 # the eventual replacement for callers asking for a dbh directly:
 236 # they'll ask for the current store, which is a database abstraction
 237 # layer.
 238 my ($store, $store_pid);
 239 sub get_store {
 240     return $store if $store && $store_pid == $$;
 241     $store_pid = $$;
 242     return $store = MogileFS::Store->new;
 243 }
 244
 245 # only for t/ scripts to explicitly set a store, without loading in a config
 246 sub set_store {
 247     my ($s) = @_;
 248     $store = $s;
 249     $store_pid = $$;
 250 }
 251
 252 # log stuff to syslog or the screen
 253 sub log {
 254     # simple logging functionality
 255     if (! $MogileFS::Config::daemonize) {
 256         # syslog acts like printf so we have to use printf and append a \n
 257         shift; # ignore the first parameter (info, warn, critical, etc)
 258         my $mask = shift; # format string
 259         $mask .= "\n" unless $mask =~ /\n$/;
 260         my $message = @_ ? sprintf($mask, @_) : $mask;
 261         print $message;
 262     } else {
 263         # just pass the parameters to syslog
 264         Sys::Syslog::syslog(@_);
 265     }
 266 }
 267
 268 1;
 269 __END__
 270 #Just for MakeMaker's kinda lame regexp for ABSTRACT_FROM
 271 =dummypod
 272 mogilefs::server - MogileFS (distributed filesystem) server.
 273